機械学習と情報技術

AIエージェントの設計パターン — ReAct・Plan-and-Execute

NLP

LLMの能力を最大限に活用するために、AIエージェントという概念が注目されています。エージェントは、目標を達成するために自律的に計画し、ツールを使い、環境とインタラクションするシステムです。

本記事では、AIエージェントの主要なアーキテクチャパターンを解説し、Pythonでの実装方法を説明します。

本記事の内容

  • AIエージェントの基本概念
  • ReActパターン
  • Plan-and-Executeパターン
  • Multi-Agentアーキテクチャ
  • Pythonでの実装例

前提知識

この記事を読む前に、以下の記事を読んでおくと理解が深まります。

AIエージェントとは

定義

AIエージェントは、以下の要素を持つシステムです。

  1. 知覚（Perception）: 環境からの入力を受け取る
  2. 推論（Reasoning）: 状況を分析し、次のアクションを決定
  3. 行動（Action）: ツールを使って環境に作用
  4. 記憶（Memory）: 過去のインタラクションを保持

エージェントループ

エージェントの基本的な動作ループ:

$$ \text{State}_{t+1} = f(\text{State}_t, \text{Observation}_t, \text{Action}_t) $$

while not done:
    observation = perceive(environment)
    thought = reason(observation, memory)
    action = decide(thought)
    result = execute(action)
    memory.update(observation, action, result)

ReActパターン

概要

ReAct（Reasoning + Acting）は、推論と行動を交互に行うパターンです。「考える」→「行動する」→「観察する」を繰り返します。

処理フロー

Thought: 何をすべきか考える
Action: ツールを呼び出す
Observation: 結果を観察する
... (繰り返し)
Thought: 十分な情報が得られた
Answer: 最終回答

数学的定式化

状態 $s_t$ から行動 $a_t$ を選択する方策:

$$ \pi(a_t | s_t, h_t) = \text{LLM}(\text{Prompt}(s_t, h_t, \mathcal{T})) $$

ここで $h_t$ は履歴、$\mathcal{T}$ はツール集合です。

Pythonでの実装

from typing import List, Dict, Any, Optional
from dataclasses import dataclass
import json
import re

@dataclass
class AgentAction:
    """エージェントのアクション"""
    tool: str
    tool_input: Dict[str, Any]
    thought: str

@dataclass
class AgentFinish:
    """エージェントの終了"""
    output: str
    thought: str

class ReActAgent:
    """ReActパターンのエージェント"""

    def __init__(self, llm_client, tools: Dict[str, callable], max_iterations: int = 10):
        """
        Args:
            llm_client: LLM APIクライアント
            tools: ツール名 → 関数のマッピング
            max_iterations: 最大反復回数
        """
        self.llm_client = llm_client
        self.tools = tools
        self.max_iterations = max_iterations

    def _build_prompt(self, question: str, history: List[str]) -> str:
        """プロンプトを構築"""
        tools_desc = "\n".join([
            f"- {name}: {func.__doc__ or '説明なし'}"
            for name, func in self.tools.items()
        ])

        history_text = "\n".join(history) if history else "（なし）"

        return f"""あなたは問題を解決するエージェントです。以下のツールを使用できます。

ツール一覧:
{tools_desc}

以下の形式で思考と行動を出力してください:

Thought: [考えていること]
Action: [ツール名]
Action Input: {{"param": "value"}}

または、答えがわかったら:

Thought: [考えていること]
Final Answer: [最終回答]

質問: {question}

これまでの履歴:
{history_text}

続けてください:"""

    def _parse_response(self, response: str) -> Any:
        """レスポンスを解析"""
        # Final Answerを探す
        if "Final Answer:" in response:
            thought = response.split("Final Answer:")[0]
            answer = response.split("Final Answer:")[1].strip()
            thought = thought.replace("Thought:", "").strip()
            return AgentFinish(output=answer, thought=thought)

        # Actionを探す
        thought_match = re.search(r"Thought:\s*(.+?)(?=Action:|$)", response, re.DOTALL)
        action_match = re.search(r"Action:\s*(\w+)", response)
        input_match = re.search(r"Action Input:\s*(\{.+?\})", response, re.DOTALL)

        if action_match:
            thought = thought_match.group(1).strip() if thought_match else ""
            tool = action_match.group(1).strip()
            tool_input = json.loads(input_match.group(1)) if input_match else {}
            return AgentAction(tool=tool, tool_input=tool_input, thought=thought)

        return None

    def run(self, question: str) -> str:
        """エージェントを実行"""
        history = []

        for i in range(self.max_iterations):
            # プロンプトを構築
            prompt = self._build_prompt(question, history)

            # LLMを呼び出し
            response = self.llm_client.generate(prompt)

            # レスポンスを解析
            parsed = self._parse_response(response)

            if parsed is None:
                history.append(f"Error: 解析できませんでした\n{response}")
                continue

            if isinstance(parsed, AgentFinish):
                return parsed.output

            if isinstance(parsed, AgentAction):
                # ツールを実行
                history.append(f"Thought: {parsed.thought}")
                history.append(f"Action: {parsed.tool}")
                history.append(f"Action Input: {json.dumps(parsed.tool_input, ensure_ascii=False)}")

                try:
                    if parsed.tool in self.tools:
                        result = self.tools[parsed.tool](**parsed.tool_input)
                        observation = json.dumps(result, ensure_ascii=False)
                    else:
                        observation = f"Error: Unknown tool '{parsed.tool}'"
                except Exception as e:
                    observation = f"Error: {str(e)}"

                history.append(f"Observation: {observation}")

        return "最大反復回数に達しました。"

# 使用例
def search(query: str) -> Dict:
    """Web検索を実行"""
    return {"results": [f"検索結果: {query}についての情報"]}

def calculator(expression: str) -> Dict:
    """数式を計算"""
    return {"result": eval(expression)}

# agent = ReActAgent(llm_client, {"search": search, "calculator": calculator})
# result = agent.run("富士山の標高は何メートルですか？それをフィートに変換すると？")

Plan-and-Executeパターン

概要

Plan-and-Executeは、最初に計画を立ててから実行するパターンです。複雑なタスクを小さなステップに分解します。

処理フロー

1. Plan: タスクをサブタスクに分解
2. Execute: 各サブタスクを順番に実行
3. Replan: 必要に応じて計画を修正
4. Complete: 全サブタスクが完了したら終了

Pythonでの実装

from dataclasses import dataclass, field
from typing import List, Optional
from enum import Enum

class TaskStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class Task:
    """タスク"""
    id: int
    description: str
    status: TaskStatus = TaskStatus.PENDING
    result: Optional[str] = None
    dependencies: List[int] = field(default_factory=list)

class PlanAndExecuteAgent:
    """Plan-and-Executeパターンのエージェント"""

    def __init__(self, llm_client, tools: Dict[str, callable]):
        self.llm_client = llm_client
        self.tools = tools
        self.tasks: List[Task] = []

    def plan(self, goal: str) -> List[Task]:
        """目標をタスクに分解"""
        prompt = f"""以下の目標を達成するためのステップを列挙してください。
各ステップは具体的で実行可能である必要があります。

目標: {goal}

ステップ（番号付きリストで）:"""

        response = self.llm_client.generate(prompt)

        # ステップを解析
        tasks = []
        lines = response.strip().split("\n")

        for i, line in enumerate(lines):
            # 番号を除去
            description = re.sub(r"^\d+[\.\)]\s*", "", line).strip()
            if description:
                tasks.append(Task(id=i, description=description))

        self.tasks = tasks
        return tasks

    def execute_task(self, task: Task) -> str:
        """単一タスクを実行"""
        task.status = TaskStatus.IN_PROGRESS

        # タスクに最適なツールを選択
        tools_desc = "\n".join([
            f"- {name}: {func.__doc__ or '説明なし'}"
            for name, func in self.tools.items()
        ])

        prompt = f"""以下のタスクを実行してください。

タスク: {task.description}

利用可能なツール:
{tools_desc}

ツールを使用する場合は以下の形式で出力:
Tool: [ツール名]
Input: {{"param": "value"}}

ツールが不要な場合は直接結果を出力してください。"""

        response = self.llm_client.generate(prompt)

        # ツール呼び出しを解析
        tool_match = re.search(r"Tool:\s*(\w+)", response)
        input_match = re.search(r"Input:\s*(\{.+?\})", response, re.DOTALL)

        if tool_match and tool_match.group(1) in self.tools:
            tool_name = tool_match.group(1)
            tool_input = json.loads(input_match.group(1)) if input_match else {}

            try:
                result = self.tools[tool_name](**tool_input)
                task.result = json.dumps(result, ensure_ascii=False)
                task.status = TaskStatus.COMPLETED
            except Exception as e:
                task.result = f"Error: {str(e)}"
                task.status = TaskStatus.FAILED
        else:
            task.result = response
            task.status = TaskStatus.COMPLETED

        return task.result

    def should_replan(self, task: Task) -> bool:
        """再計画が必要か判断"""
        if task.status == TaskStatus.FAILED:
            return True

        # 結果を評価
        prompt = f"""タスク: {task.description}
結果: {task.result}

この結果は満足できるものですか？「はい」または「いいえ」で答えてください:"""

        response = self.llm_client.generate(prompt)
        return "いいえ" in response

    def replan(self, failed_task: Task, remaining_tasks: List[Task]) -> List[Task]:
        """計画を修正"""
        prompt = f"""以下のタスクが失敗しました:
タスク: {failed_task.description}
結果: {failed_task.result}

残りのタスク:
{chr(10).join([t.description for t in remaining_tasks])}

計画を修正してください。新しいステップを列挙:"""

        response = self.llm_client.generate(prompt)

        # 新しいタスクを解析
        new_tasks = []
        lines = response.strip().split("\n")
        start_id = max(t.id for t in self.tasks) + 1

        for i, line in enumerate(lines):
            description = re.sub(r"^\d+[\.\)]\s*", "", line).strip()
            if description:
                new_tasks.append(Task(id=start_id + i, description=description))

        return new_tasks

    def synthesize_result(self, goal: str) -> str:
        """最終結果を統合"""
        completed_tasks = [t for t in self.tasks if t.status == TaskStatus.COMPLETED]

        task_results = "\n".join([
            f"- {t.description}: {t.result}"
            for t in completed_tasks
        ])

        prompt = f"""目標: {goal}

完了したタスクと結果:
{task_results}

これらの結果を統合して、目標に対する最終的な回答を生成してください:"""

        return self.llm_client.generate(prompt)

    def run(self, goal: str) -> str:
        """エージェントを実行"""
        # 計画
        self.plan(goal)
        print(f"計画: {len(self.tasks)} タスク")

        # 実行
        for task in self.tasks:
            print(f"実行中: {task.description}")
            self.execute_task(task)

            if self.should_replan(task):
                remaining = [t for t in self.tasks if t.status == TaskStatus.PENDING]
                new_tasks = self.replan(task, remaining)
                self.tasks.extend(new_tasks)

        # 結果を統合
        return self.synthesize_result(goal)

# 使用例
# agent = PlanAndExecuteAgent(llm_client, tools)
# result = agent.run("Pythonでシンプルな電卓アプリを作成する")

Multi-Agentアーキテクチャ

概要

複数の専門化されたエージェントが協力して問題を解決するパターンです。

パターンの種類

  1. Supervisor: 監督エージェントが他のエージェントを調整
  2. Hierarchical: 階層的な指揮系統
  3. Peer-to-Peer: エージェント間で対等に協力
  4. Debate: 複数エージェントが議論して結論を出す

Supervisorパターンの実装

from dataclasses import dataclass
from typing import Dict, List, Callable

@dataclass
class AgentSpec:
    """エージェントの仕様"""
    name: str
    description: str
    system_prompt: str
    tools: List[str]

class SupervisorMultiAgent:
    """Supervisorパターンのマルチエージェント"""

    def __init__(self, llm_client, tools: Dict[str, Callable]):
        self.llm_client = llm_client
        self.tools = tools
        self.agents: Dict[str, AgentSpec] = {}
        self.conversation_history: List[Dict] = []

    def register_agent(self, spec: AgentSpec):
        """エージェントを登録"""
        self.agents[spec.name] = spec

    def _supervisor_decide(self, task: str, history: List[Dict]) -> str:
        """Supervisorが次に動くエージェントを決定"""
        agents_desc = "\n".join([
            f"- {name}: {spec.description}"
            for name, spec in self.agents.items()
        ])

        history_text = "\n".join([
            f"{h['agent']}: {h['message']}"
            for h in history[-5:]  # 直近5件
        ]) if history else "（なし）"

        prompt = f"""あなたはマルチエージェントシステムのSupervisorです。

利用可能なエージェント:
{agents_desc}

現在のタスク: {task}

これまでの会話:
{history_text}

タスクが完了している場合は「FINISH」と出力してください。
そうでない場合は、次に動くべきエージェントの名前を出力してください:"""

        response = self.llm_client.generate(prompt)
        return response.strip()

    def _run_agent(self, agent_name: str, task: str, context: str) -> str:
        """指定されたエージェントを実行"""
        spec = self.agents[agent_name]

        prompt = f"""{spec.system_prompt}

タスク: {task}

コンテキスト:
{context}

回答:"""

        return self.llm_client.generate(prompt)

    def run(self, task: str, max_rounds: int = 10) -> str:
        """マルチエージェントシステムを実行"""
        history = []

        for round_num in range(max_rounds):
            # Supervisorが次のエージェントを決定
            next_agent = self._supervisor_decide(task, history)

            if "FINISH" in next_agent.upper():
                break

            # エージェント名を抽出
            agent_name = None
            for name in self.agents.keys():
                if name.lower() in next_agent.lower():
                    agent_name = name
                    break

            if agent_name is None:
                continue

            # エージェントを実行
            context = "\n".join([
                f"{h['agent']}: {h['message']}"
                for h in history
            ])

            response = self._run_agent(agent_name, task, context)

            history.append({
                "agent": agent_name,
                "message": response
            })

            print(f"[{agent_name}]: {response[:100]}...")

        # 最終結果を統合
        return self._synthesize_final_answer(task, history)

    def _synthesize_final_answer(self, task: str, history: List[Dict]) -> str:
        """最終回答を生成"""
        history_text = "\n".join([
            f"{h['agent']}: {h['message']}"
            for h in history
        ])

        prompt = f"""タスク: {task}

エージェントの議論:
{history_text}

上記の議論を踏まえて、タスクに対する最終的な回答を生成してください:"""

        return self.llm_client.generate(prompt)

# 使用例
# multi_agent = SupervisorMultiAgent(llm_client, tools)
#
# multi_agent.register_agent(AgentSpec(
#     name="researcher",
#     description="情報を検索・収集する専門家",
#     system_prompt="あなたは情報収集の専門家です。",
#     tools=["search"]
# ))
#
# multi_agent.register_agent(AgentSpec(
#     name="analyst",
#     description="データを分析する専門家",
#     system_prompt="あなたはデータ分析の専門家です。",
#     tools=["calculator"]
# ))
#
# result = multi_agent.run("日本のGDPの推移を調査し、傾向を分析してください")

メモリとコンテキスト管理

メモリの種類

  1. 短期メモリ: 現在の会話コンテキスト
  2. 長期メモリ: 永続化された情報（ベクトルDB等）
  3. エピソード記憶: 過去のインタラクションの記録

メモリ実装

from datetime import datetime
import numpy as np

class AgentMemory:
    """エージェントのメモリシステム"""

    def __init__(self, embedding_model, max_short_term: int = 10):
        self.embedding_model = embedding_model
        self.max_short_term = max_short_term

        self.short_term: List[Dict] = []
        self.long_term: List[Dict] = []  # (text, embedding, metadata)

    def add_to_short_term(self, content: str, role: str = "system"):
        """短期メモリに追加"""
        self.short_term.append({
            "content": content,
            "role": role,
            "timestamp": datetime.now().isoformat()
        })

        # 最大数を超えたら古いものを長期に移動
        if len(self.short_term) > self.max_short_term:
            old = self.short_term.pop(0)
            self.add_to_long_term(old["content"], {"role": old["role"]})

    def add_to_long_term(self, content: str, metadata: Dict = None):
        """長期メモリに追加"""
        embedding = self.embedding_model.encode(content)
        self.long_term.append({
            "content": content,
            "embedding": embedding,
            "metadata": metadata or {},
            "timestamp": datetime.now().isoformat()
        })

    def retrieve_relevant(self, query: str, k: int = 5) -> List[str]:
        """関連する長期メモリを検索"""
        if not self.long_term:
            return []

        query_embedding = self.embedding_model.encode(query)

        # コサイン類似度で検索
        similarities = []
        for item in self.long_term:
            sim = np.dot(query_embedding, item["embedding"]) / (
                np.linalg.norm(query_embedding) * np.linalg.norm(item["embedding"])
            )
            similarities.append((sim, item["content"]))

        # 上位k件を返す
        similarities.sort(reverse=True)
        return [content for _, content in similarities[:k]]

    def get_context(self, query: str = None) -> str:
        """現在のコンテキストを取得"""
        parts = []

        # 短期メモリ
        if self.short_term:
            short_term_text = "\n".join([
                f"[{m['role']}] {m['content']}"
                for m in self.short_term
            ])
            parts.append(f"最近の会話:\n{short_term_text}")

        # 関連する長期メモリ
        if query and self.long_term:
            relevant = self.retrieve_relevant(query, k=3)
            if relevant:
                parts.append(f"関連する過去の情報:\n" + "\n".join(relevant))

        return "\n\n".join(parts)

エージェント評価

評価指標

指標 説明
Task Success Rate タスク成功率
Tool Selection Accuracy 適切なツール選択率
Efficiency 解決までのステップ数
Robustness エラー回復能力

評価実装

class AgentEvaluator:
    """エージェントの評価"""

    def __init__(self):
        self.results = []

    def evaluate_task(self, agent, task: str, expected_output: str,
                      max_steps: int = 20) -> Dict:
        """タスクを評価"""
        start_time = datetime.now()

        # エージェントを実行
        actual_output = agent.run(task)

        end_time = datetime.now()
        duration = (end_time - start_time).total_seconds()

        # 成功判定（簡易）
        success = expected_output.lower() in actual_output.lower()

        result = {
            "task": task,
            "expected": expected_output,
            "actual": actual_output,
            "success": success,
            "duration": duration,
            "steps": getattr(agent, 'step_count', None)
        }

        self.results.append(result)
        return result

    def summary(self) -> Dict:
        """評価結果のサマリー"""
        total = len(self.results)
        successes = sum(1 for r in self.results if r["success"])

        return {
            "total_tasks": total,
            "success_rate": successes / total if total > 0 else 0,
            "avg_duration": np.mean([r["duration"] for r in self.results]),
            "avg_steps": np.mean([r["steps"] for r in self.results if r["steps"]])
        }

まとめ

本記事では、AIエージェントの主要なアーキテクチャパターンを解説しました。

  • ReAct: 推論と行動を交互に行う基本パターン
  • Plan-and-Execute: 計画を立ててから実行
  • Multi-Agent: 複数エージェントが協力
  • メモリ: 短期・長期メモリで文脈を維持

次のステップとして、以下の記事も参考にしてください。

関連記事

前の記事

根軌跡法の理論と描き方をわかりやすく解説

次の記事

対数微分法の仕組みと使い方を解説

関連タグ: LLM アーキテクチャ エージェント 機械学習 自然言語処理