AIエージェントのオーケストレーション設計 — タスク分解・ハンドオフ・ループ制御の実践

import asyncio
from anthropic import Anthropic
 
client = Anthropic()
 
async def run_agent(task: str, context: str) -> str:
    """単一タスクを実行するサブエージェント"""
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        system="あなたは与えられたタスクを完結させる専門エージェントです。",
        messages=[{"role": "user", "content": f"コンテキスト: {context}
 
タスク: {task}"}]
    )
    return response.content[0].text
 
async def parallel_research(topics: list[str], shared_context: str) -> list[str]:
    """複数トピックを並列調査"""
    tasks = [run_agent(f"{topic}について調査してまとめて", shared_context) for topic in topics]
    return await asyncio.gather(*tasks)

def create_handoff_context(
    goal: str,
    current_state: dict,
    success_criteria: str,
    failure_policy: str = "report and stop"
) -> str:
    """サブエージェントへの引き渡しコンテキストを構造化する"""
    return f"""
## あなたのミッション
{goal}
 
## 現在わかっていること
{json.dumps(current_state, ensure_ascii=False, indent=2)}
 
## 完了条件
{success_criteria}
 
## 失敗時の対応
{failure_policy}
"""

class AgentLoop:
    def __init__(self, max_iterations: int = 10, timeout_seconds: int = 300):
        self.max_iterations = max_iterations
        self.timeout_seconds = timeout_seconds
        self.iteration = 0
        self.start_time = None
 
    def should_continue(self, agent_says_done: bool, result: dict) -> bool:
        """三層の終了判定"""
        # Layer 1: ハードリミット（最優先）
        if self.iteration >= self.max_iterations:
            return False
        import time
        if time.time() - self.start_time > self.timeout_seconds:
            return False
        # Layer 2: エージェントの自己申告
        if agent_says_done:
            # Layer 3: 結果の検証（自己申告を鵜呑みにしない）
            return not self._validate_completion(result)
        return True
 
    def _validate_completion(self, result: dict) -> bool:
        """完了条件を独立して検証する"""
        required_keys = ["summary", "confidence", "next_actions"]
        return all(k in result and result[k] for k in required_keys)

from enum import Enum
 
class ErrorType(Enum):
    TRANSIENT = "transient"    # リトライ可能
    TOOL = "tool"              # ツール変更でリトライ
    LOGIC = "logic"            # エスカレーション
 
def classify_error(error: Exception) -> ErrorType:
    error_str = str(error).lower()
    if "timeout" in error_str or "rate limit" in error_str:
        return ErrorType.TRANSIENT
    if "not found" in error_str or "permission" in error_str:
        return ErrorType.TOOL
    return ErrorType.LOGIC