第二章 自律型AIエージェントの「思考・行動ループ（Agentic Loop）」

自律型AIエージェントの「思考・行動ループ（Agentic Loop）」

自律型AIエージェントの核心は、静的なプロンプト入出力（一問一答）ではなく、状況に応じて動的に推論と行動を繰り返す**「ループ（循環構造）」**にあります。LLM（大規模言語モデル）に自律的な振る舞いを与えるためには、このループをどのように設計し、制御するかが極めて重要になります。

本章では、エージェントを自律的に動かすための代表的な制御パターンである「ReAct」や「Plan-Execute-Evaluate」、エラーを自己解決する「自己修復（Self-Correction）」、無限ループを防ぐ「停止条件」、状態遷移（State Machine）による複雑なループワークフローの構築について、具体的なPythonコードやMermaid図を交えて体系的に解説します。

---

2.1 ReAct（Reasoning and Acting）パターンの基本設計

自律型エージェントの最も基本かつ強力なループ設計パターンが ReAct（Reasoning and Acting） です（Yao et al., 2022）。

ReActの基本概念

従来のプロンプト手法では、LLMに「思考（Reasoning）」だけを行わせるか（Chain-of-Thoughtなど）、「行動（Acting：APIの呼び出しなど）」だけを行わせるかのどちらかでした。
ReActはこれらを融合し、「思考（Thought） → 行動（Action） → 観察（Observation）」 という3ステップのループを回します。

• Thought (思考): 現在の状況を分析し、ゴールに向けて次に何をするべきかを論理的に考える。
• Action (行動): 思考に基づいて、外部ツール（検索API、データベース、電卓など）を呼び出すための具体的なクエリを生成する。
• Observation (観察): ツールの実行結果（外部からのフィードバック）を受け取り、それをプロンプトに追記して次のループへ繋げる。

LLMが「思考」を挟むことで、次に実行すべきツールの選択ミスが減り、さらに「観察」の結果を受けて自身の推論を動的に軌道修正できるようになります。

PythonによるReActパターンのミニマム実装

フレームワーク（LangChainやLangGraphなど）の内部で何が起きているかを理解するために、外部ライブラリを使わず、プレーンなPythonでReActループを実装してみましょう。

ここでは、LLM APIの呼び出しをモック関数で擬似的に再現し、実際に「Web検索ツール」を叩きながら回答を導くフローを示します。

import re
from typing import Callable, Dict

# =====================================================================
# 1. 外部ツール（Tools）の定義
# =====================================================================
def web_search(query: str) -> str:
    """簡易的な検索ツールのモック"""
    print(f"  [Tool] Web検索を実行中: '{query}'")
    if "東京の天気" in query:
        return "東京の天気は曇りのち雨、気温は24度です。"
    elif "大阪の天気" in query:
        return "大阪の天気は晴れ、気温は29度です。"
    return "該当する検索結果が見つかりませんでした。"

# 利用可能なツールのマップ
TOOLS: Dict[str, Callable[[str], str]] = {
    "web_search": web_search
}

# =====================================================================
# 2. LLMのモック（ReActのフォーマットに従って回答する）
# =====================================================================
class MockLLM:
    def __init__(self):
        self.turn = 0

    def generate(self, prompt: str) -> str:
        """会話のターン数に応じて、ReActプロンプトを模した返答をするモック"""
        self.turn += 1
        
        if self.turn == 1:
            return (
                "Thought: 東京と大阪の天気を調べて比較する必要がある。まずは東京の天気を調べよう。\n"
                "Action: web_search[東京の天気]"
            )
        elif self.turn == 2:
            return (
                "Thought: 東京の天気は分かった（曇りのち雨、24度）。次は大阪の天気を調べよう。\n"
                "Action: web_search[大阪の天気]"
            )
        elif self.turn == 3:
            return (
                "Thought: 両方の天気が判明した。東京は曇りのち雨（24度）、大阪は晴れ（29度）なので、大阪の方が気温が高く天気が良い。\n"
                "Final Answer: 東京は曇りのち雨（24度）ですが、大阪は晴れており気温も29度と、大阪の方が天気が良く暖かいです。"
            )
        return "Final Answer: すでに回答は完了しています。"

# =====================================================================
# 3. ReAct ループコントローラー
# =====================================================================
def run_react_loop(question: str, max_turns: int = 5):
    llm = MockLLM()
    
    # システムプロンプトでReActの動作ルールを規定
    system_prompt = (
        "あなたは自律型アシスタントです。以下のルールに従って思考し行動してください。\n"
        "回答プロセスは以下のフォーマットで出力してください：\n"
        "Thought: [あなたの思考]\n"
        "Action: tool_name[tool_input]\n"
        "Observation: [ツールの実行結果]\n\n"
        "最後に、回答が得られたら以下のフォーマットで終了してください：\n"
        "Final Answer: [最終的な回答]\n\n"
        "利用可能なツール:\n"
        "- web_search[クエリ]: Webから最新の情報を検索します。\n"
    )
    
    context = f"{system_prompt}\nUser Question: {question}\n"
    print(f"=== ReAct ループ開始: {question} ===\n")

    for turn in range(1, max_turns + 1):
        print(f"--- Turn {turn} ---")
        
        # LLMに現在のコンテキスト（これまでの履歴すべて）を投げて出力を得る
        response = llm.generate(context)
        print(response)
        
        # コンテキストにLLMの出力を追記
        context += response + "\n"
        
        # Final Answer が出たらループ終了
        if "Final Answer:" in response:
            print("\n=== ループ正常終了 ===")
            break
            
        # Action をパースする (例: Action: web_search[東京の天気])
        action_match = re.search(r"Action:\s*(\w+)\[(.*?)\]", response)
        if action_match:
            tool_name = action_match.group(1)
            tool_input = action_match.group(2)
            
            if tool_name in TOOLS:
                # ツールを実行してObservation（観察結果）を取得
                observation = TOOLS[tool_name](tool_input)
                obs_text = f"Observation: {observation}"
                print(obs_text)
                # 観察結果をコンテキストに追記して次のループへ
                context += obs_text + "\n"
            else:
                error_text = f"Observation: エラー: ツール '{tool_name}' は存在しません。"
                print(error_text)
                context += error_text + "\n"
        else:
            print("Actionのパースに失敗しました。リトライします。")
            context += "Observation: エラー: Actionフォーマットが不正です。Action: tool_name[input] の形式で出力してください。\n"

if __name__ == "__main__":
    run_react_loop("東京と大阪の天気を比較して、どちらが天気が良いか教えてください。")

---

2.2 プランニング・実行・自己評価（Plan-Execute-Evaluate）の制御サイクル

ReActパターンは「一歩進んで一歩考える」という局所的なアドホック処理には適していますが、複雑で大きな目標（例：「あるWebサイトの全APIドキュメントを読み込み、Pythonのモッククライアントコードを生成し、テストを実行してバグを修正する」）に対しては、大局的な視点を失って無限ループに陥りやすいという弱点があります。

そこで重要となるのが、より大局的なマクロループである Plan-Execute-Evaluate（プラン・実行・評価） サイクルです。

各フェーズの設計ポイント

1. プランニング（Plan）:
   • LLMに大目標を与え、それを具体的な依存関係を持つ「サブタスクのリスト（有向非巡回グラフ: DAG）」に分解させます。
   • プランは静的なものではなく、途中でいつでも修正可能な構造（可変配列やキュー）にしておきます。
2. 実行（Execute）:
   • プランで定義された現在のサブタスクを、専用のエージェントやモジュール（Executor）に割り当てて実行させます。
   • Executorは、前述の「ReAct」などのミクロループでツールを駆使して成果物を生成します。
3. 自己評価（Evaluate / Critique）:
   • 生成された成果物が、当初定義されたサブタスクの「終了要件（Definition of Done）」を満たしているかを別のLLM（またはルールベースの評価ロジック）でテスト・評価します。
   • 評価結果がNGであれば、何が足りないかをフィードバックし、プランを再構成（Re-Plan）します。

---

2.3 エラー検知と「自己修復・再試行（Self-Correction / Retry）」ループ

AIエージェントの運用において、最も頻発するトラブルは「ツールの呼び出しエラー」や「LLMの出力フォーマットエラー（JSONパースエラーなど）」です。これらが発生した際にシステム全体をクラッシュさせるのではなく、エラー自体をLLMへの入力（Observation）としてフィードバックし、LLM自身に自己修正させるのが自己修復ループです。

典型的な自己修復フロー

1. LLMが不正なJSONを出力する。
2. システム側でパースを試みるが、json.JSONDecodeError が発生する。
3. 例外をキャッチし、「あなたが先ほど出力したテキストはJSONとして無効です。エラー内容: JSONDecodeError: Expecting ',' delimiter。以下の正しいフォーマットでもう一度出力してください」 というプロンプトを作成し、LLMに再度投げかける。

自己修復ループの実装例（Python）

以下は、LLMが指定のスキーマに合致しない不正なデータを出力した際に、エラーメッセージをフィードバックして再試行させる実装パターンです。

import json
from typing import Dict, Any

# 期待するスキーマの定義
REQUIRED_KEYS = {"name", "age", "skills"}

def validate_response(raw_text: str) -> Dict[str, Any]:
    """JSONのパースとスキーマ検証を行う関数"""
    # LLMがMarkdownブロックで囲んでくることを考慮
    cleaned_text = raw_text.strip()
    if cleaned_text.startswith("```json"):
        cleaned_text = cleaned_text.split("```json")[1]
    if cleaned_text.endswith("```"):
        cleaned_text = cleaned_text.rsplit("```", 1)[0]
    cleaned_text = cleaned_text.strip()
    
    # JSONパース
    data = json.loads(cleaned_text)
    
    # スキーマチェック
    missing_keys = REQUIRED_KEYS - data.keys()
    if missing_keys:
        raise ValueError(f"必須キーが不足しています: {missing_keys}")
        
    return data

# 不正な出力を連発した後に正しい出力をするLLMのモック
class ErrorProneLLM:
    def __init__(self):
        self.call_count = 0

    def generate(self, prompt: str) -> str:
        self.call_count += 1
        print(f"  [LLM呼び出し {self.call_count}回目]")
        
        if self.call_count == 1:
            # カンマが抜けている不正なJSON
            return '{"name": "Alice", "age": 30 "skills": ["Python"]}'
        elif self.call_count == 2:
            # パースはできるが必須キー(skills)が足りない
            return '{"name": "Alice", "age": 30}'
        else:
            # 正しいJSON
            return '{"name": "Alice", "age": 30, "skills": ["Python", "LangGraph"]}'

def generate_with_self_correction(prompt: str, max_retries: int = 3) -> Dict[str, Any]:
    llm = ErrorProneLLM()
    current_prompt = prompt
    
    for attempt in range(1, max_retries + 1):
        print(f"\n--- 試行 {attempt}/{max_retries} ---")
        try:
            # LLMから出力を取得
            raw_output = llm.generate(current_prompt)
            # 検証を実行
            valid_data = validate_response(raw_output)
            print(">> 成功: 正常なデータを取得しました。")
            return valid_data
            
        except (json.JSONDecodeError, ValueError) as e:
            error_message = str(e)
            print(f">> エラー検知: {error_message}")
            
            if attempt == max_retries:
                raise RuntimeError("最大リトライ回数に達しましたが、正常な出力を得られませんでした。")
            
            # エラー内容をコンテキストに含めてリトライ用プロンプトを構築
            current_prompt = (
                f"{prompt}\n\n"
                f"[過去のあなたの出力]\n{raw_output}\n\n"
                f"[エラー内容]\n{error_message}\n\n"
                f"指示: 上記のエラーを修正し、必ず有効なJSONフォーマット（キー: name, age, skills）のみを出力してください。"
            )

if __name__ == "__main__":
    base_prompt = "ユーザー情報をJSONで出力してください。"
    result = generate_with_self_correction(base_prompt)
    print("結果:", result)

---

2.4 ループの「暴走」を防ぐ停止条件（Termination Conditions）と最大反復回数の設計

自律型エージェント開発において、最も実務的かつ予算管理上重要なのが 「ループの暴走（無限ループ）防止」 です。
LLMの思考が同じ場所をループし始めたり、エラーの自己修復が無限に繰り返されたりすると、APIのトークン課金が数分で数万円に跳ね上がるといったインシデントが発生します。

暴走が起こる典型的なパターン

1. Tool-Use Loop: ツールAを呼んだ結果（エラー）に対し、LLMが「再度同じパラメータでツールAを呼ぶ」という動作を繰り返す。
2. Thought Loop (Hallucination Loop): 「回答を導き出すために情報が足りない。検索しよう」→「情報が見つからない」→「情報が足りない。検索しよう」という自己満足的な探索のループ。
3. Format Loop: LLMが何度修正を指示されても、特定のフォーマットエラー（例: JSONのパースミス）を繰り返し出力する。

堅牢なエージェントシステムに必須の4大停止条件

無限ループを防ぐために、ループ制御ロジックには以下のガードレールを必ず実装してください。

停止条件	概要	実装アプローチ
1. 最大反復回数 (Max Iterations)	ループの総ステップ数を制限する。最も確実な防壁。	ループカウンタによる制御、またはLangGraphの recursion_limit を利用。
2. タイムアウト制限 (Timeout)	実行時間（秒数）を制限する。外部APIのハングやデッドロック対策。	非同期処理の asyncio.wait_for 等でタイムアウトを設定。
3. コスト・トークン制限 (Token Budget)	1セッションで消費可能な累計トークン数（またはドル換算コスト）に上限を設ける。	API呼び出しごとの使用トークン数を追記・加算し、閾値超えでエラー終了。
4. 重複検知 (Duplicate Detection)	過去N回の「Action（ツール呼び出しの引数）」の履歴を記録し、直近の履歴内で同一の実行が重複した場合にループを検知して強制終了する。	過去のActionシグネチャ（ツール名＋引数）のハッシュセットを作成して検知。

# 4. 重複検知（Loop Detection）の簡易ロジック例
class ActionHistoryTracker:
    def __init__(self, window_size: int = 3):
        self.history = []
        self.window_size = window_size

    def check_and_record(self, tool_name: str, tool_input: str) -> bool:
        """直近の履歴ウィンドウ内に完全に一致するActionが重複して存在する場合にTrueを返す（ループ検知）"""
        action_signature = (tool_name, tool_input)
        
        # 直近の履歴に同じアクションがあるか確認
        if action_signature in self.history:
            return True
            
        self.history.append(action_signature)
        if len(self.history) > self.window_size:
            self.history.pop(0)
        return False

---

2.5 状態遷移（State Machine）による複雑なループワークフローのモデリング

単純なスクリプトであれば while ループでエージェントを構築できますが、本番運用のアプリケーション（Webサービスのバックエンドなど）では、単純なループ記述には限界があります。

なぜ while ループでは限界があるのか？

• 非同期とステート管理の難しさ: エージェントの動作中に「ユーザーの割り込み（Human-in-the-Loop）」が発生した場合、ループを一時停止し、状態をデータベースに保存して、数時間後に再開する必要がある。
• 履歴の肥大化と管理: 会話やツールの入出力履歴が複雑に分岐する場合、線形なループコードでは管理が破綻する。
• 並列処理（Fork-Join）の制御: 複数のタスクを並列で実行し、全て終わったら次のステップへ進むような合流制御が困難。

これらを解決するのが 「有限状態機械（Finite State Machine: FSM）」 によるモデリングです。現在普及している主要なエージェントフレームワーク（特に LangGraph）は、このステートマシンモデルをベースに設計されています。

状態遷移モデルの3大要素

1. State（状態）: グラフ全体で共有される読み書き可能なキーバリュー型のデータストア（スレッドセーフに管理される）。
2. Nodes（ノード）: 各ステップの具体的な処理（Python関数）。状態（State）を入力として受け取り、更新された状態（Stateの一部）を出力して書き換える。
3. Edges（エッジ）: ノード間の遷移ルール。通常の「次のノードへ進む（Normal Edge）」と、現在の状態に基づいて次に進むノードを決定する「条件分岐（Conditional Edge）」の2種類がある。

LangGraphの設計思想を模した状態遷移ループの実装例

以下は、LangGraphのアーキテクチャ設計を模した、状態遷移ベースのシンプルなReActエージェントの実装コードです。実用的なグラフ定義の流れを理解できます。

from typing import TypedDict, Annotated, Sequence, Literal
import operator

# =====================================================================
# 1. State（グラフで共有する状態）の定義
# =====================================================================
class AgentState(TypedDict):
    # messagesリストは、新しい追加メッセージを自動的にマージ（append）するように設計
    messages: Annotated[Sequence[dict], operator.add]
    # ループ回数のカウンタ
    loop_count: int

# =====================================================================
# 2. Nodes（各処理の定義）
# =====================================================================
def agent_node(state: AgentState) -> dict:
    """LLMの思考と決定を行うノード"""
    print(f"\n[Node] Agent: 意思決定中... (ループ回数: {state['loop_count']})")
    messages = state["messages"]
    
    # 実際はここでLLM APIを叩く
    # ここでは2回目の呼び出しで終了するようにモック化
    if state["loop_count"] == 0:
        response_message = {
            "role": "assistant",
            "content": "東京の天気を調べます。",
            "tool_calls": [{"name": "get_weather", "args": {"location": "Tokyo"}}]
        }
    else:
        response_message = {
            "role": "assistant",
            "content": "東京の天気は晴れです。回答を完了します。"
        }
        
    return {
        "messages": [response_message],
        "loop_count": state["loop_count"] + 1
    }

def action_node(state: AgentState) -> dict:
    """ツール（Action）を実行するノード"""
    print("[Node] Action: ツールを実行中...")
    last_message = state["messages"][-1]
    tool_call = last_message["tool_calls"][0]
    
    # ツールの実行シミュレーション
    tool_result = f"{tool_call['args']['location']}の天気は『快晴』、気温は28度です。"
    print(f"  -> ツール結果: {tool_result}")
    
    return {
        "messages": [{
            "role": "tool",
            "content": tool_result,
            "name": tool_call["name"]
        }]
    }

# =====================================================================
# 3. Edges（条件分岐ルール）の定義
# =====================================================================
def should_continue(state: AgentState) -> Literal["continue", "end"]:
    """次にどのノードに進むべきかを状態から判定するエッジ"""
    last_message = state["messages"][-1]
    
    # 無限ループ防止用のガードレール
    if state["loop_count"] >= 5:
        print("[Edge] 警告: 最大ループ回数に達したため強制終了します。")
        return "end"
        
    # LLMがツール実行（tool_calls）を要求しているかどうかで分岐
    if "tool_calls" in last_message and last_message["tool_calls"]:
        print("[Edge] 条件合致: ツール実行へ遷移します。")
        return "continue"
        
    print("[Edge] 条件合致: 終了ノードへ遷移します。")
    return "end"

# =====================================================================
# 4. グラフ（State Machine）の組み立てと実行シミュレーション
# =====================================================================
class SimpleGraphRunner:
    """LangGraphのコンパイルと実行フローを再現する簡易ランナー"""
    def __init__(self):
        self.nodes = {}
        self.entry_point = None

    def add_node(self, name: str, func):
        self.nodes[name] = func

    def set_entry_point(self, name: str):
        self.entry_point = name

    def run(self, initial_state: AgentState) -> AgentState:
        # 呼び出し元が渡した状態（辞書）の破壊的変更を防ぐため、コピーを作成
        state = dict(initial_state)
        current_node = self.entry_point
        
        while current_node:
            # ノードの処理を実行し、状態をアップデート
            node_output = self.nodes[current_node](state)
            
            # 状態の更新
            state["loop_count"] = node_output.get("loop_count", state["loop_count"])
            if "messages" in node_output:
                state["messages"] = list(state["messages"]) + list(node_output["messages"])
                
            # 条件分岐（エッジの評価）
            if current_node == "agent":
                decision = should_continue(state)
                if decision == "continue":
                    current_node = "action"
                else:
                    current_node = None # 終了
            elif current_node == "action":
                # アクション実行後は必ずエージェントへ戻る
                current_node = "agent"
                
        return state

if __name__ == "__main__":
    # グラフの構築
    workflow = SimpleGraphRunner()
    workflow.add_node("agent", agent_node)
    workflow.add_node("action", action_node)
    workflow.set_entry_point("agent")

    # 初期状態の設定
    init_state: AgentState = {
        "messages": [{"role": "user", "content": "東京の天気を調べて教えて。"}],
        "loop_count": 0
    }

    # 実行
    final_state = workflow.run(init_state)
    print("\n=== 実行完了後の最終メッセージ ===")
    print(final_state["messages"][-1]["content"])

[!NOTE]
上記コードの AgentState で定義している Annotated[Sequence[dict], operator.add] は、実際のLangGraphにおいて「ノードが返した新しいメッセージを、既存のメッセージリストに自動的にマージ（追加）する」ための特別なアノテーション（Reducer）です。
本モック実装（SimpleGraphRunner）では、アノテーションを解析する代わりに、ランナー内部で明示的にリストをマージ（list(state["messages"]) + list(node_output["messages"])）することで、この挙動を再現しています。

このように、ループを「状態（State）」と「処理（Nodes）」に完全に切り離してモデリングすることで、以下のような本番運用のメリットが得られます。

• 各ノードが独立した純粋関数に近くなるため、ユニットテストが極めて容易になる。
• エッジを変更するだけで、プログラマティックに「特定の条件下で人間に確認（Approve）を求める」といったループ制御の拡張ができる。
• ステートが永続化DBに記録されていれば、数日間にわたる長期の非同期ワークフローも安全にハンドリング可能になる。