AgentKit 2.0 マルチエージェント協調の失敗パターンと本番復旧

# デッドロック検出と強制解放
def resolve_deadlock(session_id: str, deadlocked_agents: list[str]):
    session = ag.Session.load(session_id)
 
    # 1. デッドロックしているエージェントを特定して強制停止
    for agent_name in deadlocked_agents:
        agent = session.agents[agent_name]
        agent.force_stop(reason="deadlock_detected")
        print(f"✓ {agent_name} を強制停止しました")
 
    # 2. 共有リソースのロックを解放
    session.release_all_locks()
    print("✓ 全ロックを解放しました")
 
    # 3. タスクを再スケジュール（先に完了できる方を優先）
    # 依存関係グラフを確認して非循環な順序で再実行
    task_graph = session.get_task_dependency_graph()
    restart_order = task_graph.topological_sort()
 
    for task in restart_order:
        if task.name in deadlocked_agents:
            print(f"→ {task.name} を再起動します...")
            task.restart(clear_waiting_state=True)

def fix_orchestrator_loop(session_id: str, orchestrator_name: str):
    session = ag.Session.load(session_id)
    orchestrator = session.agents[orchestrator_name]
 
    # ループの原因を診断
    recent_actions = orchestrator.get_recent_actions(n=20)
    action_types = [a.type for a in recent_actions]
 
    # 同じアクションが繰り返されているか確認
    from collections import Counter
    action_counts = Counter(action_types)
    print("直近20アクションの分布:", action_counts)
 
    # ループ検出: 同じアクションが10回以上
    if action_counts.most_common(1)[0][1] >= 10:
        print("⚠️ ループを検出しました。コンテキストをクリアして再起動します...")
        orchestrator.clear_context(preserve_initial_instructions=True)
        orchestrator.restart_with_new_context(
            additional_context="前のタスク割り当ては一旦リセットされました。"
                               "現在完了済みのタスクを確認して、未完了のタスクから再開してください。"
        )

def fix_context_contamination(session_id: str):
    session = ag.Session.load(session_id)
 
    # 1. 全エージェントを一時停止
    for agent in session.agents.values():
        agent.pause()
 
    # 2. 共有コンテキストのスナップショットを取得
    snapshots = session.shared_context.get_write_history()
    print(f"最後の書き込み履歴 ({len(snapshots)}件):")
    for snap in snapshots[-10:]:
        print(f"  {snap.timestamp} | {snap.agent} → {snap.key}: {str(snap.value)[:50]}")
 
    # 3. 最後に正常だったスナップショットを特定して復元
    last_good_snapshot = snapshots[-5]  # 5操作前が安全か確認
    session.shared_context.restore(last_good_snapshot.id)
    print(f"✓ コンテキストを {last_good_snapshot.timestamp} 時点に復元しました")
 
    # 4. 排他制御を追加してエージェントを再起動
    session.shared_context.enable_write_locking()
    for agent in session.agents.values():
        agent.resume()

# 依存関係の循環を事前に防ぐ設計パターン
class DeadlockSafeOrchestrator:
    def __init__(self):
        self.task_dependencies = {}  # タスク → 依存タスクのマッピング
        self.task_timeout = 180  # 秒
 
    def assign_task(self, agent_name: str, task: dict, depends_on: list = None):
        """循環依存チェック付きタスク割り当て"""
        if depends_on:
            # 循環依存の検出
            if self._creates_cycle(task["id"], depends_on):
                raise ValueError(
                    f"循環依存を検出しました: {task['id']} → {depends_on}"
                )
            self.task_dependencies[task["id"]] = depends_on
 
        # タイムアウト付きでタスクを割り当て
        return ag.assign_task(
            agent=agent_name,
            task=task,
            timeout_seconds=self.task_timeout,
            on_timeout="fail_gracefully"  # タイムアウト時に失敗として扱う
        )
 
    def _creates_cycle(self, task_id: str, depends_on: list) -> bool:
        """DFS でサイクル検出"""
        visited = set()
        def dfs(node):
            if node in visited:
                return True
            visited.add(node)
            for dep in self.task_dependencies.get(node, []):
                if dfs(dep):
                    return True
            return False
        return any(dfs(dep) for dep in depends_on)

# 全エージェントにタイムアウトとフォールバックを設定する
AGENT_CONFIG = {
    "data_collector": {
        "timeout_seconds": 120,
        "retry_count": 2,
        "on_failure": "skip_and_continue",  # 失敗しても次のステップへ
        "fallback_value": {"status": "skipped", "data": []}
    },
    "analyzer": {
        "timeout_seconds": 300,
        "retry_count": 1,
        "on_failure": "escalate_to_human",  # 人間に通知
        "fallback_value": None
    },
    "reporter": {
        "timeout_seconds": 60,
        "retry_count": 3,
        "on_failure": "use_template",  # テンプレート出力を使う
        "fallback_value": "分析エラーのため結果を生成できませんでした。"
    }
}

class ContextManager:
    MAX_CONTEXT_LENGTH = 100_000  # トークン数の目安
 
    def __init__(self, agent):
        self.agent = agent
        self.message_count = 0
 
    def add_message(self, message: str):
        self.agent.context.append(message)
        self.message_count += 1
 
        # 一定間隔でコンテキストを圧縮
        if self.message_count % 20 == 0:
            self._compress_context()
 
    def _compress_context(self):
        """古いメッセージを要約して保持量を削減"""
        current_length = self.agent.estimate_context_tokens()
 
        if current_length > self.MAX_CONTEXT_LENGTH * 0.8:
            # 古い半分のメッセージを要約
            old_messages = self.agent.context[:-10]  # 直近10件は保持
            summary = self.agent.summarize(old_messages)
            self.agent.context = [
                f"[以前の作業の要約] {summary}"
            ] + self.agent.context[-10:]
            print(f"コンテキストを圧縮: {len(old_messages)} → 1メッセージ")