【Ping-Pong Loop の無限地獄】
[Coder] ---> "修正しました" ---> [Reviewer]
   ^                                  |
   └-------- "ここも直してください" ------┘ (無限にループ)

# 動作確認済みライブラリバージョン: openai>=1.30.0, pydantic>=2.7.0
import os
import sys
from typing import List, Literal, Optional, Dict, Any
from pydantic import BaseModel, Field
from openai import OpenAI

# 1. 共有ステート（State）と構造化ルーティングモデルの定義

class AgentState(BaseModel):
    """マルチエージェント全体で共有する状態オブジェクト"""
    task_description: str = Field(description="元のユーザー要求仕様")
    current_code: str = Field(default="", description="作成された最新のPythonコード")
    review_feedback: str = Field(default="", description="Reviewerからの指摘事項")
    is_approved: bool = Field(default=False, description="Reviewerが合格を出したか")
    step_count: int = Field(default=0, description="累計状態遷移ステップ数")


class SupervisorDecision(BaseModel):
    """Supervisor が決定する次の遷移先エージェント"""
    reasoning: str = Field(description="なぜこのエージェントを指名したかの決定理由")
    next_agent: Literal["coder", "reviewer", "END"] = Field(
        description="次に実行すべきノード: 'coder', 'reviewer', または 'END'"
    )


class CoderOutput(BaseModel):
    """Coder Agent の出力結果"""
    code: str = Field(description="作成・修正された完全なPythonコード")
    change_summary: str = Field(description="修正・変更点の概要")


class ReviewerOutput(BaseModel):
    """Reviewer Agent の評価結果"""
    passed: bool = Field(description="コードが要求仕様を満たし合格しているか")
    feedback: str = Field(description="改善すべき指摘事項（合格の場合は称賛）")


# =====================================================================
# マルチエージェント オーケストレーター クラスの実装
# =====================================================================

class SafeMultiAgentOrchestrator:
    def __init__(self, model_name: str = "gpt-4o", max_steps: int = 10):
        self.client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY", "dummy_key"))
        self.model_name = model_name
        self.max_steps = max_steps  # サーキットブレーカーの上限

    def run_supervisor(self, state: AgentState) -> SupervisorDecision:
        """1. Supervisor: 共有Stateを元に次のノードを型安全に判定します"""
        print("🧠 [Supervisor] 次のステップを判定中...")
        
        prompt = f"""あなたはプロジェクトマネージャーです。現在のステートを確認し、次にどのエージェントを作動させるか決定してください。

【タスク仕様】: {state.task_description}
【現在のコード存在】: {'あり' if state.current_code else 'なし (未作成)'}
【Reviewer合格フラグ】: {state.is_approved}
【直前のフィードバック】: {state.review_feedback if state.review_feedback else 'なし'}

【判定基準】:
- まだコードがない、またはフィードバックがあり未合格の場合 ➡ 'coder'
- コードが作成された直後で、まだレビューを受けていない場合 ➡ 'reviewer'
- レビューが合格 (passed=True) した場合 ➡ 'END'
"""
        response = self.client.beta.chat.completions.parse(
            model=self.model_name,
            messages=[{"role": "user", "content": prompt}],
            response_format=SupervisorDecision,
            temperature=0.0
        )
        decision = response.choices[0].message.parsed
        print(f"🎯 [Supervisor決定]: 次ノード = '{decision.next_agent.upper()}' (理由: {decision.reasoning})")
        return decision

    def run_coder(self, state: AgentState) -> AgentState:
        """2. Coder Agent: コードの新規作成またはフィードバックに基づく修正"""
        print("💻 [Coder Agent] コード作成/修正作業を開始...")
        
        if not state.current_code:
            prompt = f"以下の要求仕様を満たす完全なPythonコードを作成してください。\n要求: {state.task_description}"
        else:
            prompt = f"""Reviewerからの指摘に基づきコードを修正してください。

【要求】: {state.task_description}
【前回のコード】:
```python
{state.current_code}

    response = self.client.beta.chat.completions.parse(
        model=self.model_name,
        messages=[{"role": "user", "content": prompt}],
        response_format=CoderOutput,
        temperature=0.1
    )
    res = response.choices[0].message.parsed
    
    # Stateの更新
    state.current_code = res.code
    print(f"✅ [Coder Agent] コード作成完了 (変更点: {res.change_summary})")
    return state

def run_reviewer(self, state: AgentState) -> AgentState:
    """3. Reviewer Agent: コードの品質・仕様適合性を厳格評価"""
    print("🔍 [Reviewer Agent] コードのレビューを開始...")
    
    prompt = f"""以下のPythonコードが要求仕様を完璧に満たしているか評価してください。

{state.current_code}

    # Stateの更新
    state.is_approved = res.passed
    state.review_feedback = res.feedback
    print(f"📊 [Reviewer Agent] 判定: {'合格 (PASS)' if res.passed else '不合格 (FAIL)'}")
    print(f"   フィードバック: {res.feedback}")
    return state

def execute_workflow(self, task_description: str) -> AgentState:
    """ステートグラフの循環実行とサーキットブレーカー防衛線の駆動"""
    print(f"\n🚀 === マルチエージェント・オーケストレーター起動: 『{task_description}』 ===")
    
    # 初期状態の作成
    state = AgentState(task_description=task_description)
    
    while True:
        # 防衛線3: サーキットブレーカーによるステップ数判定
        state.step_count += 1
        print(f"\n--- [ステート遷移 #{state.step_count} / 上限 {self.max_steps}] ---")
        
        if state.step_count > self.max_steps:
            print(f"🚨 [サーキットブレーカー発動]: 最大遷移数({self.max_steps})を超過したため、デッドロック防止のため緊急遮断しました。")
            break
            
        # 防衛線1 & 2: Supervisor によるルーティング
        decision = self.run_supervisor(state)
        
        if decision.next_agent == "END":
            print("\n🎉 [全工程完了]: Supervisor が終了指示 (END) を出しました。")
            break
        elif decision.next_agent == "coder":
            state = self.run_coder(state)
        elif decision.next_agent == "reviewer":
            state = self.run_reviewer(state)
            
    return state

orchestrator = SafeMultiAgentOrchestrator(model_name="gpt-4o", max_steps=8)

task = "与えられた数値リストから素数のみを抽出し、降順にソートして返す型ヒント・Docstring付きの Python 関数 `filter_primes` を作成してください。"

final_state = orchestrator.execute_workflow(task)

print("\n==================================================")
print("                 最終完成成果物                    ")
print("==================================================")
print(final_state.current_code)

---

## 5. コードの行別・ロジック詳細解説

実装した `SafeMultiAgentOrchestrator` のコア設計を解説します。

### 1. `AgentState` による集中型メモリ管理
- 各エージェント（Coder / Reviewer）は独自の内部メモリを持たず、全情報が単一の `AgentState`（Pydanticインスタンス）に集約されます。
- これにより、エージェント間で同じ会話履歴が二重・三重に重複して膨らむ「コンテキスト爆発」を完全に抑え込んでいます。

### 2. `SupervisorDecision` による非決定論的分岐の排除
- `Literal["coder", "reviewer", "END"]` という Pydantic 列挙型を用いることで、LLMが「次はデザイナーに振ります」といった未定義の行動を起こすスクリプトエラーを防いでいます。

### 3. `step_count` とサーキットブレーカーの防衛線
- `while True:` ループの直頭で `state.step_count += 1` をインクリメントし、`self.max_steps`（例: 8〜10回）を超えた瞬間に即座に `break` します。
- 万が一 Supervisor や Reviewer が永久に不合格を出し続ける「Ping-Pong Loop」に陥った場合でも、課金爆発やプロセス固着を物理的に100%回避できます。

---

## 6. プロダクション導入・LangGraph 連携・Human-in-the-Loop

本手法を実務プロダクションへ展開する際のアドバンスドな設計について解説します。

### 1. LangGraph における `recursion_limit` の設定
LangGraph フレームワークを利用する場合、同様のサーキットブレーカー機能が `recursion_limit` として組み込まれています。

```python
# LangGraph での制限設定と実行
app = workflow.compile()
config = {"recursion_limit": 10}  # デフォルトは25。実務では10〜15が推奨
result = app.invoke(initial_state, config=config)

[Supervisor] ---> `action == 'deploy'` ---> [Human Approval Wait] 
                                                │
                                    ├── (承認) ──> [Deploy Agent]
                                    └── (拒否) ──> [State修正 / キャンセル]