Skip to content
Published on

LLM Agent設計パターン完全ガイド:ReActからマルチエージェントまで

Authors

AIエージェントを初めて構築するとき、最もよくある失敗は「とりあえずGPT-4を繋いでみよう」というアプローチです。ところが実際に作り始めると、同じツールを無限ループで呼び続けたり、自信満々に誤った答えを返したり、コンテキストウィンドウが溢れてクラッシュするといった状況に直面します。

この記事では、現場で実証済みのLLM Agent設計パターン5種類を解説します。各パターンの原理、動作するコード、そして正直なトレードオフまで。

Agentとは何か?(実用的な定義)

様々な定義がありますが、最も実用的なものはこれです:

"An agent is an LLM that can take actions, observe results, and decide what to do next — in a loop."

エージェント = フィードバックループの中で行動するLLM。それだけです。

Perception(入力受け取り)
Reasoning(LLMが思考)
Action(ツール使用)
Observation(結果確認)
    → back to Reasoning(再び思考へ)

このループが核心です。単純なLLM呼び出しとエージェントの違いはフィードバックループの有無です。

Pattern 1: ReAct(Reason + Act)

2022年にYao et al.が発表したパターンで、現在最も広く使われているエージェント構造です。名前の通り、**Reason(推論)Act(行動)**を交互に実行します。

動作原理

LLMに特定のフォーマットで思考するよう強制します:

Thought: 何をすべきかの推論
Action: tool_name(params)
Observation: [ツール実行結果]
Thought: 結果を見て次の行動を決定
...(繰り返し)
Final Answer: 最終回答

実際の実装

system_prompt = """
You have access to these tools: [search, calculator, code_executor]

Always follow this exact format:
Thought: I need to...
Action: tool_name(params)
Observation: [tool result]
... (repeat as needed)
Final Answer: ...

Never skip the Thought step. Never fabricate Observations.
"""

# 実際のトレース例:
# Question: "2023年の韓国GDPを人口で割ると?"
# Thought: 韓国のGDPと人口を調べる必要がある
# Action: search("South Korea GDP 2023")
# Observation: "South Korea GDP 2023: $1.71 trillion USD"
# Thought: 次に人口が必要
# Action: search("South Korea population 2024")
# Observation: "51.7 million"
# Thought: 計算する
# Action: calculator("1710000000000 / 51700000")
# Observation: "33075"
# Final Answer: 約$33,075(一人当たりGDP)

ツール定義とパースロジックを含む本番用実装:

import re
from typing import Callable

class ReActAgent:
    def __init__(self, llm, tools: dict[str, Callable], max_iterations=10):
        self.llm = llm
        self.tools = tools
        self.max_iterations = max_iterations

    def run(self, question: str) -> str:
        messages = [
            {"role": "system", "content": self._build_system_prompt()},
            {"role": "user", "content": question}
        ]

        for i in range(self.max_iterations):
            response = self.llm.invoke(messages)
            content = response.content

            # Final Answerが出たら終了
            if "Final Answer:" in content:
                return content.split("Final Answer:")[-1].strip()

            # Actionをパース
            action_match = re.search(r"Action: (\w+)\((.*?)\)", content)
            if not action_match:
                return content  # フォーマット失敗時はそのまま返す

            tool_name = action_match.group(1)
            tool_args = action_match.group(2)

            # ツール実行
            if tool_name not in self.tools:
                observation = f"Error: tool '{tool_name}' not found"
            else:
                try:
                    observation = self.tools[tool_name](tool_args)
                except Exception as e:
                    observation = f"Error executing tool: {str(e)}"

            # Observationをメッセージ履歴に追加
            messages.append({"role": "assistant", "content": content})
            messages.append({"role": "user", "content": f"Observation: {observation}"})

        return "Max iterations reached without final answer"

    def _build_system_prompt(self):
        tool_names = list(self.tools.keys())
        return f"""You have access to these tools: {tool_names}
Always use this format:
Thought: [your reasoning]
Action: tool_name(params)
Observation: [will be filled in]
Final Answer: [when done]"""

ReActの長所と短所

長所: 推論プロセスが透明に見える — デバッグが容易。ほとんどのユースケースに対応。

短所: トークン消費が多い。同じThoughtパターンが繰り返されることがある(幻覚リスク)。

Pattern 2: Chain of Thought(CoT)

外部ツールなしで複雑な推論が必要なときに使うパターンです。

Zero-shot CoT

最も単純な形 — プロンプトに「Let's think step by step」を付けるだけ:

response = llm.invoke(
    "Q: バットとボールで合わせて110円。バットはボールより100円高い。"
    "ボールはいくら?\n\nLet's think step by step."
)
# LLMが段階的に推論するようになる

Few-shot CoT

例を提供して推論フォーマットを示します:

few_shot_examples = """
Q: 5台の機械が5分で5個のウィジェットを作る場合、
   100台の機械で100個作るには何分かかる?
A:
Step 1: 1台の機械は1個を5分で作る。
Step 2: 100台が同時に各1個を作る = 5分。
Answer: 5分。

Q: [新しい問題]
A:
"""

ReAct vs CoT:判断基準

  • 外部情報や計算不要の純粋な推論 → CoT
  • リアルタイム情報、計算、コード実行が必要 → ReAct
  • 単純なQ&A → どちらも不要、普通のLLM呼び出しで十分

Pattern 3: Plan-and-Execute

問題を事前に分解できる複雑な長期タスクに適したパターンです。

class PlanAndExecuteAgent:
    def __init__(self, planner_llm, executor_llm, tools):
        self.planner = planner_llm      # 高性能、高コストなモデル
        self.executor = executor_llm    # 安価、高速なモデル
        self.tools = tools

    def run(self, goal: str) -> str:
        # Phase 1: 計画立案 — 最良のLLMを使用
        plan_prompt = f"""
        Goal: {goal}

        このゴールを達成するためのステップバイステップの計画を作成してください。
        具体的で実行可能なステップの番号付きリストを返してください。
        各ステップは独立して実行できるくらい具体的にしてください。
        """
        plan_response = self.planner.invoke(plan_prompt)
        steps = self._parse_plan(plan_response)

        print(f"計画作成完了: {len(steps)}ステップ")

        # Phase 2: 実行 — 安価なモデルを使用可能
        results = []
        for i, step in enumerate(steps):
            print(f"ステップ{i+1}実行中: {step}")

            result = self.executor.invoke(
                f"このステップを実行してください: {step}\n\n"
                f"前のステップの文脈: {results}\n\n"
                f"利用可能なツール: {list(self.tools.keys())}"
            )
            results.append({"step": step, "result": result})

        # Phase 3: まとめ
        summary = self.planner.invoke(
            f"Goal: {goal}\n\nResults: {results}\n\n最終結果をまとめてください。"
        )
        return summary

主要な利点: 計画フェーズと実行フェーズに異なるモデルを使えます。GPT-4で計画、GPT-3.5で実行 — 長いワークフローでコスト削減になります。

主要な弱点: 初期計画が誤っていると、下流の全てが誤りになります。実行失敗時に再計画するロジックが必要です。

Pattern 4: ReflectionとSelf-Critique

エージェントが自身の出力を批判し、改善するパターンです。

class ReflectionAgent:
    def __init__(self, llm):
        self.llm = llm

    def run(self, question: str, num_reflections: int = 2) -> str:
        # 初期回答生成
        answer = self.llm.invoke(f"この質問に答えてください: {question}")

        for i in range(num_reflections):
            # 自己批判フェーズ
            critique = self.llm.invoke(f"""
            質問: {question}
            回答: {answer}

            この回答を批判的に評価してください:
            1. 事実として正確か?
            2. 完全か?何が欠けているか?
            3. 明確に説明されているか?
            4. 論理的なエラーはあるか?

            具体的かつ建設的に評価してください。
            """)

            print(f"批判{i+1}: {critique[:200]}...")

            # 改善フェーズ
            answer = self.llm.invoke(f"""
            元の質問: {question}
            以前の回答: {answer}
            批判: {critique}

            批判の全ポイントに対処した改善された回答を提供してください。
            """)

        return answer

シンプルに見えますが、実際に品質向上の効果は大きいです。特にコード生成、技術文書作成、複雑な分析で有効です。

注意: Reflectionを繰り返しすぎると過剰な自己批判が生まれ、逆に品質が下がることがあります。通常2〜3回が適切です。

Pattern 5: Tree of Thoughts(ToT)

2023年に発表されたパターンで、複数の推論経路を同時に探索します。チェスエンジンが複数の手を先読みするようなアプローチです。

from typing import List
import asyncio

class TreeOfThoughts:
    def __init__(self, llm, branching_factor=3, max_depth=4):
        self.llm = llm
        self.branching_factor = branching_factor
        self.max_depth = max_depth

    async def solve(self, problem: str) -> str:
        root = {"thought": problem, "score": 1.0, "children": []}

        # BFS探索
        queue = [root]
        best_leaf = None
        best_score = 0

        for depth in range(self.max_depth):
            next_queue = []

            for node in queue:
                # 各ノードから複数の次の思考を生成
                children = await self._generate_thoughts(node["thought"])

                for child_thought in children:
                    # 各経路をスコアリング
                    score = await self._evaluate_thought(child_thought, problem)
                    child_node = {
                        "thought": child_thought,
                        "score": score,
                        "children": []
                    }
                    node["children"].append(child_node)
                    next_queue.append(child_node)

                    if score > best_score:
                        best_score = score
                        best_leaf = child_node

            # 上位k件の経路のみ継続探索(ビームサーチ)
            queue = sorted(next_queue, key=lambda x: x["score"], reverse=True)[:self.branching_factor]

        # 最良経路で最終回答生成
        return await self._generate_final_answer(best_leaf["thought"], problem)

    async def _generate_thoughts(self, current_thought: str) -> List[str]:
        response = await self.llm.ainvoke(f"""
        現在の推論: {current_thought}
        この推論を続ける{self.branching_factor}つの異なる方法を生成してください。
        番号付きリストで返してください。
        """)
        return self._parse_numbered_list(response)

    async def _evaluate_thought(self, thought: str, problem: str) -> float:
        response = await self.llm.ainvoke(f"""
        問題: {problem}
        推論経路: {thought}
        この推論経路を0.0から1.0で評価してください。
        数字だけ返してください。
        """)
        try:
            return float(response.strip())
        except:
            return 0.5

ToTを使うべき場合:

  • 複数の解法アプローチを探索する必要がある数学問題
  • 多くの可能な戦略がある複雑な計画立案
  • 発散的な方向性を探りたいクリエイティブなタスク

使うべきでない場合:

  • 単純な情報検索
  • 速度が重要な場合(APIコストとレイテンシが非常に高い)
  • 明確に正しい線形的な解法がある問題

パターン選択ガイド

状況推奨パターン
ツール使用が必要(検索、計算など)ReAct
純粋な推論、ツール不要Chain of Thought
長い多段階プロジェクトPlan-and-Execute
品質重視のドキュメントやコード生成Reflection
複雑な数学/計画、コスト問わずTree of Thoughts
複数の専門家が必要なタスクMulti-Agent(次の記事参照)

よくある間違いと解決法

本番環境で実際に直面する問題です。

1. ツール呼び出しの無限ループ

エージェントが同じツールを繰り返し呼び続けます。主にエラー処理の欠如やObservationの無視が原因です。

MAX_ITERATIONS = 15
seen_actions = set()

for i in range(MAX_ITERATIONS):
    action = agent.get_next_action()
    action_key = f"{action.tool}:{action.args}"

    if action_key in seen_actions:
        return "エラー: エージェントがループに入りました。強制終了します。"
    seen_actions.add(action_key)

2. コンテキストウィンドウのオーバーフロー

長いセッションで蓄積されたObservationがコンテキストウィンドウを超えます。

def trim_messages_to_fit(messages, max_tokens=100000):
    """古いObservationから削除、システムメッセージは維持"""
    while count_tokens(messages) > max_tokens:
        for i, msg in enumerate(messages[1:], 1):
            if "Observation:" in msg.get("content", ""):
                messages.pop(i)
                messages.pop(i - 1)  # 対応するActionも削除
                break
    return messages

3. ツールエラーの握りつぶし

# 悪い例 — サイレントな失敗
result = tool.run(args)

# 良い例 — LLMにエラーを伝えて対応させる
try:
    result = tool.run(args)
except Exception as e:
    result = f"ツールエラー: {str(e)}。別のアプローチを試してください。"

4. ツール呼び出しのタイムアウトなし

外部APIが応答しない場合、エージェントが永遠に待ち続けます。

import asyncio

async def run_tool_with_timeout(tool, args, timeout=30):
    try:
        return await asyncio.wait_for(tool.arun(args), timeout=timeout)
    except asyncio.TimeoutError:
        return f"ツールが{timeout}秒後にタイムアウトしました。別のアプローチを試してください。"

まとめ

パターンを知ることより、いつどのパターンを使うかを知ることの方が重要です。まずReActから始めてください — ほとんどのケースをカバーします。複雑さが増すにつれてPlan-and-Execute、Reflectionを追加し、品質が重要な場合にのみToTを検討してください。

次の記事では、これらのパターンをつなぐ**MCP(Model Context Protocol)**を解説します — AIと外部世界を繋ぐAnthropicの新しいオープン標準です。