Agentive
AIエージェント活用

AIエージェント設計パターン入門:ReAct・Plan-and-Execute・Multi-Agent

約12分で読めます

AIエージェントを「作る」側に回ると、すぐに壁にぶつかる。「どう動けばいい?」という設計の問いだ。LLMに道具を渡してループさせるだけでは、複雑なタスクは失敗する。実用的なエージェントを作るには、目的に合った設計パターンを選ぶ必要がある。

本記事では実装者向けに、現在主流の3つのパターンを解説する。どれを選ぶかは、タスクの複雑さと予測可能性によって変わる。

ReAct:逐次推論と行動の繰り返し

概念

ReAct(Reasoning + Acting)は、LLMに「考える→行動する→観察する」を繰り返させる最もシンプルなパターンだ。Yao et al. (2022) が提案し、現在のエージェントフレームワークの土台になっている。

各ステップで LLM は以下を出力する。

  • Thought — 今何を考えているか
  • Action — どのツールを使うか、何を引数に渡すか
  • Observation — ツールの実行結果

この3つを繰り返し、最終的に Final Answer を出して終了する。

実装例(Python)

from anthropic import Anthropic

client = Anthropic()

TOOLS = [
    {
        "name": "search_web",
        "description": "Web検索を行い、結果を返す",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "検索クエリ"}
            },
            "required": ["query"]
        }
    },
    {
        "name": "calculate",
        "description": "数式を計算して結果を返す",
        "input_schema": {
            "type": "object",
            "properties": {
                "expression": {"type": "string", "description": "計算式"}
            },
            "required": ["expression"]
        }
    }
]

def run_tool(name: str, inputs: dict) -> str:
    # 実際の実装ではここで外部APIやコードを呼び出す
    if name == "search_web":
        return f"[検索結果] {inputs['query']} に関する情報..."
    elif name == "calculate":
        # 注意: eval()は任意コード実行のリスクがある。本番ではsimpleeval等の安全なパーサーを使うこと
        import ast
        try:
            tree = ast.parse(inputs["expression"], mode="eval")
            return str(eval(compile(tree, "<string>", "eval"), {"__builtins__": {}}, {}))
        except Exception as e:
            return f"計算エラー: {e}"
    return "ツールが見つかりません"

def react_agent(task: str, max_steps: int = 10) -> str:
    messages = [{"role": "user", "content": task}]

    for step in range(max_steps):
        response = client.messages.create(
            model="claude-opus-4-5",
            max_tokens=1024,
            tools=TOOLS,
            messages=messages
        )

        # ツール呼び出しがない場合は最終回答
        if response.stop_reason == "end_turn":
            for block in response.content:
                if hasattr(block, "text"):
                    return block.text
            break

        # ツール呼び出しを処理
        messages.append({"role": "assistant", "content": response.content})
        tool_results = []

        for block in response.content:
            if block.type == "tool_use":
                result = run_tool(block.name, block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result
                })

        messages.append({"role": "user", "content": tool_results})

    return "最大ステップ数に達しました"

# 実行例
answer = react_agent("東京の現在の気温を調べて、摂氏から華氏に変換してください")
print(answer)

利点と欠点

利点

  • 実装がシンプルで、フレームワークなしでも動く
  • 各ステップの推論が LLM の Thought に残るため、デバッグしやすい
  • 短いタスク(3〜5ステップ程度)では安定して動作する

欠点

  • 長いタスクになると文脈が膨張し、推論の精度が落ちる
  • 全体計画を持たないため、途中で方向が変わりやすい
  • 一度始めると途中で「立ち止まって考え直す」ことが苦手

Plan-and-Execute:先に計画、後で実行

概念

Plan-and-Execute は、実行前に全体計画を立てるパターンだ。LLM を「Planner」と「Executor」の2つの役割に分離する。

  1. Planner:タスクを受け取り、実行可能なステップリストに分解する
  2. Executor:各ステップを順番に実行し、結果を記録する
  3. Replanner(省略可):途中の結果を見て計画を修正する

この分離によって、複雑なタスクを扱いやすい粒度に落とし込める。

実装例(Python)

from anthropic import Anthropic
import json
from dataclasses import dataclass, field
from typing import List

client = Anthropic()

@dataclass
class Plan:
    steps: List[str] = field(default_factory=list)
    completed: List[str] = field(default_factory=list)
    current_step: int = 0

def create_plan(task: str) -> Plan:
    """Planner: タスクをステップに分解する"""
    response = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": f"""以下のタスクを実行可能なステップに分解してください。
JSONフォーマットで返してください: {{"steps": ["ステップ1", "ステップ2", ...]}}

タスク: {task}"""
        }]
    )

    text = response.content[0].text
    # JSON部分を抽出
    start = text.find("{")
    end = text.rfind("}") + 1
    data = json.loads(text[start:end])
    return Plan(steps=data["steps"])

def execute_step(step: str, context: str) -> str:
    """Executor: 個別ステップを実行する"""
    response = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": f"""これまでの作業結果:
{context}

次のステップを実行してください:
{step}

実行結果のみを返してください。"""
        }]
    )
    return response.content[0].text

def plan_and_execute(task: str) -> str:
    # ステップ1: 計画を立てる
    plan = create_plan(task)
    print(f"計画: {plan.steps}")

    context = ""

    # ステップ2: 各ステップを実行する
    for i, step in enumerate(plan.steps):
        print(f"\n実行中 [{i+1}/{len(plan.steps)}]: {step}")
        result = execute_step(step, context)
        plan.completed.append(result)
        context += f"\nステップ{i+1}{step})の結果:\n{result}\n"
        print(f"結果: {result[:100]}...")

    # ステップ3: 最終回答を生成する
    response = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": f"""以下の実行結果をもとに、元のタスクへの最終回答を作成してください。

元のタスク: {task}

実行結果:
{context}"""
        }]
    )
    return response.content[0].text

# 実行例
result = plan_and_execute("競合3社のSaaSプロダクトの特徴を比較するレポートを作成してください")
print(result)

利点と欠点

利点

  • 複雑なタスクを構造化できる
  • 各ステップが明確なので、失敗箇所を特定しやすい
  • 計画を事前にユーザーに見せて承認を取れる

欠点

  • 最初の計画が間違っていると、後続ステップも崩れる
  • 動的な情報(検索結果など)に応じた計画変更が苦手(Replanner を追加すれば改善できる)
  • ReAct より API 呼び出し回数が増える

Multi-Agent:役割分担による協調動作

概念

Multi-Agent は複数の LLM インスタンスを並列・直列に動かし、役割を分担させるパターンだ。大規模・長時間のタスクや、専門性が必要なタスクに向いている。

代表的な構成:

  • Orchestrator + Worker:上位が指示を出し、下位が実行する階層構造
  • Pipeline:エージェントを直列につなぎ、出力を次の入力にする
  • Debate / Review:複数エージェントが同じ問いに回答し、相互評価する

実装例(Python)

from anthropic import Anthropic
from concurrent.futures import ThreadPoolExecutor
from typing import Dict

client = Anthropic()

def run_agent(role: str, system_prompt: str, task: str) -> str:
    """単一エージェントを実行する汎用関数"""
    response = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=1024,
        system=system_prompt,
        messages=[{"role": "user", "content": task}]
    )
    return response.content[0].text

def orchestrator_agent(task: str) -> Dict[str, str]:
    """オーケストレーター: タスクを分割して各Workerに割り当てる"""
    breakdown = run_agent(
        role="orchestrator",
        system_prompt="あなたはタスクを専門家チームに振り分けるマネージャーです。",
        task=f"""以下のタスクを、リサーチ担当・ライター担当・レビュー担当の3人に振り分けてください。
各担当の具体的な指示をJSON形式で返してください:
{{"researcher": "指示", "writer": "指示", "reviewer": "指示"}}

タスク: {task}"""
    )

    import json
    start = breakdown.find("{")
    end = breakdown.rfind("}") + 1
    return json.loads(breakdown[start:end])

WORKER_PROMPTS = {
    "researcher": "あなたはリサーチの専門家です。正確な情報収集と整理を行います。",
    "writer": "あなたはテクニカルライターです。明確で読みやすい文章を書きます。",
    "reviewer": "あなたはQAエンジニアです。内容の正確性と完成度を評価します。"
}

def multi_agent_pipeline(task: str) -> str:
    # フェーズ1: オーケストレーターが指示を生成
    print("オーケストレーターが計画中...")
    instructions = orchestrator_agent(task)

    # フェーズ2: Workerを並列実行
    print("Workerエージェントを並列実行中...")
    results = {}

    with ThreadPoolExecutor(max_workers=3) as executor:
        futures = {
            role: executor.submit(
                run_agent,
                role,
                WORKER_PROMPTS[role],
                instructions[role]
            )
            for role in ["researcher", "writer", "reviewer"]
        }
        for role, future in futures.items():
            results[role] = future.result()
            print(f"{role} 完了")

    # フェーズ3: 統合エージェントが結果をまとめる
    print("統合中...")
    final = run_agent(
        role="integrator",
        system_prompt="あなたは複数の専門家の意見を統合して最終成果物を作るディレクターです。",
        task=f"""以下の各専門家の作業結果を統合して、最終成果物を作成してください。

元のタスク: {task}

リサーチ結果:
{results['researcher']}

ライター原稿:
{results['writer']}

レビューコメント:
{results['reviewer']}"""
    )

    return final

# 実行例
result = multi_agent_pipeline("AIエージェントフレームワーク比較記事の初稿を作成してください")
print(result)

TypeScript での実装例(LangChain.js 風)

import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic();

interface AgentResult {
  role: string;
  output: string;
}

async function runWorker(role: string, systemPrompt: string, task: string): Promise<AgentResult> {
  const response = await client.messages.create({
    model: "claude-opus-4-5",
    max_tokens: 512,
    system: systemPrompt,
    messages: [{ role: "user", content: task }],
  });

  return {
    role,
    output: (response.content[0] as { text: string }).text,
  };
}

async function reviewPipeline(draft: string): Promise<string> {
  // 2つのレビュアーが並列でレビュー
  const [technicalReview, styleReview] = await Promise.all([
    runWorker(
      "technical_reviewer",
      "技術的な正確性のみを評価してください。",
      `以下の文章の技術的な問題点を指摘してください:\n${draft}`
    ),
    runWorker(
      "style_reviewer",
      "文章のわかりやすさと構造のみを評価してください。",
      `以下の文章の読みやすさを評価してください:\n${draft}`
    ),
  ]);

  // 最終統合
  const final = await runWorker(
    "editor",
    "レビューコメントを反映して文章を改善してください。",
    `原文:\n${draft}\n\n技術レビュー:\n${technicalReview.output}\n\nスタイルレビュー:\n${styleReview.output}`
  );

  return final.output;
}

利点と欠点

利点

  • 専門化によって各エージェントの精度が上がる
  • 並列実行で処理時間を短縮できる
  • 役割ごとに異なるモデルやパラメータを使い分けられる

欠点

  • 実装の複雑さが増す
  • API コストが単体エージェントより大幅に増加する
  • エージェント間の連携ミスが発生しやすい(特に出力フォーマットの不一致)

3パターンの比較

観点ReActPlan-and-ExecuteMulti-Agent
実装の複雑さ
適したタスク規模小〜中中〜大
API コスト低〜中
デバッグのしやすさ
並列実行不可一部可
動的な計画変更得意苦手(Replanner必要)
向いているユースケース検索・調査・Q&Aレポート作成・マルチステップ作業大規模コンテンツ生成・複雑な分析

どれを選ぶか

実際のプロジェクトでの判断基準を整理する。

ReAct を選ぶとき

  • タスクが3〜7ステップで完結する
  • 途中の情報に応じて柔軟に動いてほしい
  • まず動くものを作りたい

Plan-and-Execute を選ぶとき

  • タスクが複雑だが、処理の流れは事前にある程度読める
  • ユーザーに計画を見せてから実行したい
  • 失敗ステップだけ再実行したい

Multi-Agent を選ぶとき

  • タスクに明確に分離できる専門領域がある
  • 並列処理でスピードを上げたい
  • レビューや対話によって品質を上げたい

まとめ

設計パターンはあくまで出発点だ。実際のプロダクトでは、これらを組み合わせることも多い。例えば、Plan-and-Execute の各ステップ実行に ReAct を使ったり、Multi-Agent の一部ワーカーを Plan-and-Execute で動かしたりする構成は実用的でよく機能する。

まずは ReAct で動くものを作り、複雑さに応じて Plan-and-Execute や Multi-Agent に移行していく段階的なアプローチが、個人開発では現実的だ。コストと複雑さを常に意識しながら、必要最小限の設計を選ぶことが長続きの秘訣になる。

関連記事

A

Agentive 編集部

AIエージェントを実際に使い倒す個人開発者。サイト制作の自動化を実践しながら、その知見を発信しています。