AIエージェント・オーケストレーション：マルチエージェントシステムの設計と実装

from anthropic import Anthropic
import json
 
client = Anthropic()
 
def decompose_task(task: str, available_agents: list[dict]) -> list[dict]:
    """
    LLMを使ってタスクを分解し、各サブタスクをエージェントに割り当てる
    """
    decompose_prompt = f"""
    あなたはタスク分解の専門家です。以下のタスクを、利用可能なエージェントで実行可能な
    サブタスクに分解してください。
    
    メインタスク: {task}
    
    利用可能なエージェント:
    {json.dumps(available_agents, ensure_ascii=False, indent=2)}
    
    以下のJSON形式で回答してください:
    {{
        "subtasks": [
            {{
                "id": "task_1",
                "description": "サブタスクの説明",
                "assigned_agent": "エージェントID",
                "depends_on": [],
                "can_parallel": true
            }}
        ],
        "execution_order": [["task_1", "task_2"], ["task_3"]]
    }}
    
    execution_orderは並列実行可能なグループを表す二次元配列です。
    """
    
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=4096,
        messages=[{"role": "user", "content": decompose_prompt}]
    )
    
    result = json.loads(response.content[0].text)
    return result
 
def execute_task_plan(plan: dict, agents: dict) -> dict:
    """実行計画に従ってタスクを順次・並列実行"""
    results = {}
    
    for parallel_group in plan["execution_order"]:
        if len(parallel_group) == 1:
            # シーケンシャル実行
            task_id = parallel_group[0]
            subtask = next(t for t in plan["subtasks"] if t["id"] == task_id)
            agent = agents[subtask["assigned_agent"]]
            
            # 依存タスクの結果を注入
            context = {dep: results[dep] for dep in subtask["depends_on"]}
            results[task_id] = agent.execute(subtask["description"], context)
        else:
            # 並列実行
            import concurrent.futures
            with concurrent.futures.ThreadPoolExecutor() as executor:
                futures = {}
                for task_id in parallel_group:
                    subtask = next(t for t in plan["subtasks"] if t["id"] == task_id)
                    agent = agents[subtask["assigned_agent"]]
                    context = {dep: results[dep] for dep in subtask["depends_on"]}
                    futures[task_id] = executor.submit(
                        agent.execute, subtask["description"], context
                    )
                
                for task_id, future in futures.items():
                    results[task_id] = future.result()
    
    return results

from dataclasses import dataclass, field
from typing import Any, Dict
import threading
 
@dataclass
class SharedWorkspace:
    """エージェント間で共有される作業スペース"""
    artifacts: Dict[str, Any] = field(default_factory=dict)
    messages: list = field(default_factory=list)
    _lock: threading.RLock = field(default_factory=threading.RLock)
    
    def write_artifact(self, key: str, value: Any, agent_id: str):
        """アーティファクトの書き込み（スレッドセーフ）"""
        with self._lock:
            self.artifacts[key] = {
                "value": value,
                "written_by": agent_id,
                "timestamp": __import__("datetime").datetime.utcnow().isoformat()
            }
    
    def read_artifact(self, key: str) -> Any:
        """アーティファクトの読み取り"""
        with self._lock:
            item = self.artifacts.get(key)
            return item["value"] if item else None
    
    def post_message(self, from_agent: str, to_agent: str, content: str):
        """エージェント間メッセージング"""
        with self._lock:
            self.messages.append({
                "from": from_agent,
                "to": to_agent,
                "content": content,
                "timestamp": __import__("datetime").datetime.utcnow().isoformat()
            })
    
    def get_messages_for(self, agent_id: str) -> list:
        """特定エージェント宛のメッセージを取得"""
        with self._lock:
            return [m for m in self.messages if m["to"] == agent_id]

def quality_gate(output: str, criteria: dict) -> dict:
    """
    エージェントの出力品質を評価するクオリティゲート
    """
    evaluation_prompt = f"""
    以下の出力を評価してください。
    
    出力:
    {output}
    
    評価基準:
    {json.dumps(criteria, ensure_ascii=False)}
    
    JSON形式で回答:
    {{
        "passed": true/false,
        "score": 0-100,
        "issues": ["問題点1", "問題点2"],
        "improvement_suggestions": ["改善案1", "改善案2"]
    }}
    """
    
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": evaluation_prompt}]
    )
    
    return json.loads(response.content[0].text)
 
def iterative_refinement(
    task: str,
    agent,
    criteria: dict,
    max_iterations: int = 3
) -> str:
    """品質基準を満たすまで反復的に改善"""
    
    result = agent.execute(task)
    
    for i in range(max_iterations):
        quality = quality_gate(result, criteria)
        
        if quality["passed"]:
            return result
        
        # 改善フィードバックを含めて再実行
        improvement_task = f"""
        元のタスク: {task}
        
        前回の出力:
        {result}
        
        改善すべき点:
        {json.dumps(quality["improvement_suggestions"], ensure_ascii=False)}
        
        上記の改善点を踏まえて、より良い出力を生成してください。
        """
        result = agent.execute(improvement_task)
    
    return result  # 最大試行回数後の最良の結果を返す

class ResearchAgent:
    def __init__(self, search_tools: list):
        self.tools = search_tools
        self.system_prompt = """あなたは優れたリサーチャーです。
        指定されたトピックについて、信頼性の高い情報源から情報を収集し、
        構造化された形式でまとめてください。"""
    
    def execute(self, topic: str, context: dict = None) -> dict:
        research_prompt = f"以下のトピックについて詳しく調査してください: {topic}"
        
        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=8192,
            system=self.system_prompt,
            tools=self.tools,
            messages=[{"role": "user", "content": research_prompt}]
        )
        
        return {
            "findings": response.content[0].text,
            "sources": self._extract_sources(response)
        }
 
class WritingAgent:
    def __init__(self):
        self.system_prompt = """あなたは経験豊富なテクニカルライターです。
        リサーチ結果を基に、読みやすく情報量の豊富な記事を執筆してください。
        見出しには具体的で魅力的なタイトルを使い、コード例は実際に動作するものを提供してください。"""
    
    def execute(self, outline: str, research_data: dict) -> str:
        writing_prompt = f"""
        以下のアウトラインと調査データを基に、記事本文を執筆してください。
        
        アウトライン:
        {outline}
        
        調査データ:
        {json.dumps(research_data, ensure_ascii=False)}
        
        要件:
        - 各セクションを十分な深さで解説する
        - 実践的なコード例を含める
        - 読者が実際に活用できる知識を提供する
        """
        
        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=16384,
            system=self.system_prompt,
            messages=[{"role": "user", "content": writing_prompt}]
        )
        
        return response.content[0].text

class BudgetAwareAgent:
    def __init__(self, max_tokens: int = 4096, model: str = "claude-sonnet-4-6"):
        self.max_tokens = max_tokens
        self.model = model
        self.tokens_used = 0
    
    def execute(self, task: str) -> str:
        if self.tokens_used > self.max_tokens * 0.9:
            # 予算残量が10%を切ったら軽量モデルにフォールバック
            self.model = "claude-haiku-4-5"
        
        response = client.messages.create(
            model=self.model,
            max_tokens=min(2048, self.max_tokens - self.tokens_used),
            messages=[{"role": "user", "content": task}]
        )
        
        self.tokens_used += response.usage.input_tokens + response.usage.output_tokens
        return response.content[0].text

import asyncio
 
async def execute_with_timeout(agent, task: str, timeout: float = 30.0) -> str:
    """タイムアウト付きエージェント実行"""
    try:
        result = await asyncio.wait_for(
            asyncio.to_thread(agent.execute, task),
            timeout=timeout
        )
        return result
    except asyncio.TimeoutError:
        return f"ERROR: Agent timed out after {timeout}s"