エージェント・アリーナを構築：マルチエージェントAIの“神経系”としてValkeyを使う

ほとんどのAIエージェントのデモは知能を証明します。しかし、協調（コーディネーション）を証明できるものはごくわずかです。

このプロジェクトでは、4つの自律エージェントが1つの共有基盤で協力する、リアルタイムのマルチプレイヤーゲーム Agent Arena: Fact or Fake を構築しました。その共有基盤は Valkey です。

この記事では、再利用できるアーキテクチャ、実装の詳細、トレードオフ、そして開発者向けのパターンを順を追って説明します。

問題設定

LLMはコンテンツを生成できます。しかし、本番レベルのマルチエージェントシステムにはそれ以上が必要です：

• 独立したワーカー間での共有状態
• 密結合なしに行うイベント駆動型の引き継ぎ
• 将来のふるまいに影響する長期メモリ
• 障害発生時の観測性とリカバリ

これらがなければ、エージェントシステムは脆いAPIコールの連鎖になってしまいます。

システム概要

このアプリにおけるエージェント：

• Researcher：事実／誤解を招く主張の候補を生成（Ollama）
• Writer：主張をプレイヤー向けの質問に書き換え（Ollama）
• Editor：真偽＋確信度を検証（OpenAI）
• Game Master：タイミング付きのラウンドを統括し、スコアリングとリーダーボードを管理

プレイヤーはWebSocket経由で参加し、リアルタイムに FACT または FAKE に回答します。

中核となる設計原則

エージェント同士の直接呼び出しをしない。

すべての引き継ぎはValkeyを通して行います：

• 状態 -> JSONキー
• オーケストレーション -> Pub/Subイベント
• 長期の想起 -> ベクターインデックス（FT.CREATE / FT.SEARCH）

アーキテクチャ図

Players (WS) -> FastAPI -> Valkey (JSON + Pub/Sub + Vector)
                                |      |           |
                                v      v           v
                           Researcher Writer     Editor
                                   \      |      /
                                    \     |     /
                                     -> Game Master

実装の内訳

1) 共有状態（Valkey JSON）

エージェントの出力とゲーム状態を名前空間付きのキーに書き込みます：

• game:state:{room_id}
• agent:researcher:output:{room}
• agent:writer:draft:{room}
• agent:editor:review:{room}
• game:round:{room}:{round}

# backend/services/state_store.py
async def set_game_state(self, room_id: str, state: dict[str, Any]) -> None:
    await self.valkey.set_json(f'game:state:{room_id}', state)

set_json/get_json の層は、RedisJSONが利用できない場合に SET/GET のJSON文字列へフォールバックするため、ローカルデモでも堅牢さを保てます。

2) イベント駆動型オーケストレーション（Valkey Pub/Sub）

ワークフローの各遷移は、イベントのエンベロープを公開します：

# backend/services/event_bus.py
await self.valkey.publish(channel, envelope.model_dump_json())

各エージェントは、自分が関心を持つチャネルだけを購読し、イベントに反応します。

これにより：

• 疎結合なスケーリング
• 独立したプロセスの再起動
• 明確な失敗境界

3) 長期メモリ（ValkeySearchのベクター）

質問は埋め込みとしてメモリドキュメントに格納します：

# backend/services/vector_memory.py
await self.valkey.set_json(f'memory:question:{round_id}', {
  'question': question,
  'topic': topic,
  'difficulty': difficulty,
  'player_accuracy': player_accuracy,
  'embedding': emb,
})

ベクター検索により、類似した過去の質問が取得されるため、繰り返しを減らし、トピックの進行を改善できます。

ラウンドのライフサイクル

ラウンドごとのイベントチェーン：

1. GAME_START / START_ROUND -> Researcherが RESEARCH_DONE を出力
2. Writerが反応 -> DRAFT_READY を出力
3. Editorが反応 -> VALIDATION_COMPLETE を出力
4. Game Masterがラウンドを開始 -> NEW_QUESTION を出力
5. プレイヤーがWebSocketで回答
6. Game Masterが ROUND_RESULT + LEADERBOARD_UPDATE + ROUND_COMPLETE を出力

さらに、古い処理や重複した下流処理を防ぐために、cycle_id の伝播も追加しました。

追加した信頼性向上策

• 再接続時にプレイヤースコアを保持（HSETNX）
• 同じ部屋で新しいゲーム開始時にスコアをリセット
• イベントハンドラの安全性：イベントごとの例外が、エージェントのループ全体を殺さない
• WebSocketペイロードのバリデーション（invalid_json、invalid_round_id、round_not_active）
• ヘルスエンドポイントでValkeyの到達可能性＋JSON/Search機能を確認

運用時のチェック

デモの前にこれを使ってください：

curl -s http://127.0.0.1:8000/health | jq

次が表示されます：

• reachable
返却形式: {"translated": "翻訳されたHTML"}
• json_module
• search_module

環境管理（Varlock-first）

このアプリはプロセスの環境変数から設定を読み込みます。これにより、Varlockで管理される秘密情報／設定に適した構成になります。

実行例：

varlock run -- uvicorn main:app --reload --port 8000
varlock run -- python scripts/run_agents.py --agents researcher writer editor game_master

テスト戦略

最小限の統合テストが含まれています：

pytest -q tests/test_integration_round.py

検証する内容：ゲーム開始 -> 最初のラウンド -> リーダーボード更新。

なぜこのパターンが重要なのか

エージェント間のダイレクトなAPIチェイニングと比べて、この設計により次が得られます：

• より優れたフォールト分離
• より優れた可観測性
• 水平方向のスケーリングが容易
• 分散ワークフローに対する理解（メンタルモデル）がシンプル

次に改善するべき点

• オーケストレーションをPub/SubからValkey Streamsへ移行（耐久配信）
• イベントの冪等性ストア + デッドレター処理を追加
• イベントのライフサイクル用にOpenTelemetryトレースを追加
• 契約テスト + 信頼性テストのためのCIパイプラインを追加

LLMは推論を提供しますが、調整（コーディネーション）がシステムを信頼できるものにします。

マルチエージェントのワークフローを構築するなら、Valkeyを単なるキャッシュではなく、あなたの共有された思考基盤（shared cognition fabric）として扱ってください。

スクリーンショット

*Github: *https://github.com/harishkotra/neuroloop