ClawTeamのマルチエージェント群集オーケストレーションとOpenAI関数呼び出しを実装したコーディング実演

この包括的なチュートリアルでは、HKUDSによって開発されたオープンソースのAgent Swarm IntelligenceフレームワークであるClawTeamのコア・アーキテクチャを紹介します。ClawTeamを強力にする基本概念を実装します。具体的には、複雑な目標をサブタスクに分解するリーダーエージェント、それらのタスクを自律的に実行する専門ワーカーエージェント、自動依存関係解決機能を持つ共有タスクボード、そしてリアルタイム調整を可能にするエージェント間メッセージングシステムです。このチュートリアルはColabでシームレスに実行できるように設計されており、OpenAI APIキーのみが必要です。これにより、元のClawTeam CLIが必要とするtmux、git worktrees、ファイルシステムベースのメッセージキューなどのローカルインフラストラクチャを設定することなく、誰でもマルチエージェント・オーケストレーションを体験できます。 必要なパッケージであるOpenAIとrichをインストールし、Colab Secretsまたはターミナル入力を通じてOpenAI APIキーを安全に収集することから始めます。次に、並行処理のためのthreading、クリーンなデータモデリングのためのdataclasses、美しいターミナル出力のためのRichなど、チュートリアル全体で必要なすべてのライブラリをインポートします。グローバルなOpenAIクライアント、Richコンソールを初期化し、コスト効率の良いスウォーム実行のためにgpt-4o-miniをデフォルトモデルとして設定します。 ClawTeamのファイルシステムベースのアーキテクチャを反映する3つの基盤システムを構築します。完全なステータス・ライフサイクルと自動依存関係ブロック解除機能を持つTaskBoard、エージェント間のポイントツーポイントおよびブロードキャストメッセージングをサポートするInboxSystem、そして各エージェントの役割、ステータス、完了タスク数を追跡するTeamRegistryです。複数のエージェントが同時に読み書きできるように、これら3つのシステムすべてにスレッドセーフ・ロックを実装します。ブロックされたタスクがその前提条件完了時に保留中に自動的に移行する、blocked_by依存関係チェーンでタスクをモデル化します。これは、ClawTeamのCLIがタスク依存関係を解決する方法とまったく同じです。 SWARM_TOOLS = [ { "type": "function", "function": { "name": "task_update", "description": "タスクのステータスを更新します。開始時は'in_progress'、完了時は'completed'を使用してください。", "parameters": { "type": "object", "properties": { "task_id": {"type": "string", "description": "タスクID"}, "status": {"type": "string", "enum": ["in_progress", "completed", "failed"]}, "result": {"type": "string", "description": "タスクの結果または出力"}, }, "required": ["task_id", "status"], }, }, }, { "type": "function", "function": { "name": "inbox_send", "description": "他のエージェント（例：'leader'またはワーカー名）にメッセージを送信します。", "parameters": { "type": "object", "properties": { "to": {"type": "string", "description": "受信者エージェント名"}, "message": {"type": "string", "description": "メッセージ内容"}, }, "required": ["to", "message"], }, }, }, { "type": "function", "function": { "name": "inbox_receive", "description": "受信トレイのすべてのメッセージを確認して消費します。", "parameters": { "type": "object", "properties": {}, }, }, }, { "type": "function", "function": { "name": "task_list", "description": "自分に割り当てられたタスクまたはチーム全体のタスクを一覧表示します。", "parameters": { "type": "object", "properties": { "owner": {"type": "string", "description": "所有者名でフィルタリング（オプション）"}, }, }, }, }, ] class SwarmAgent: def __init__( self, name: str, role: str, system_prompt: str, task_board: TaskBoard, inbox: InboxSystem, registry: TeamRegistry, ): self.name = name self.role = role self.system_prompt = system_prompt self.task_board = task_board self.inbox = inbox self.registry = registry self.conversation_history: list[dict] = [] self.inbox.register(name) self.registry.register(name, role) def _build_system_prompt(self) -> str: coord_protocol = f""" ## 調整プロトコル（自動挿入 — あなたはエージェント'{self.name}'です） あなたはAIエージェント・スウォームの一員です。あなたの役割: {self.role} あなたの名前: {self.name} 利用可能なツール（ClawTeam CLIと同等）: - task_list: 割り当てられたタスクを確認（`clawteam task list`と同様） - task_update: タスクステータスをin_progress/completed/failedに更新（`clawteam task update`と同様） - inbox_send: 他のエージェントにメッセージを送信（`clawteam inbox send`と同様） - inbox_receive: 受信トレイのメッセージを確認（`clawteam inbox receive`と同様） ワークフロー: 1. task_listでタスクを確認 2. 開始時にタスクをin_progressとしてマーク 3. 作業を実行（思考、分析、出力生成） 4. 結果とともにタスクをcompletedとしてマーク 5. 完了時に'leader'に要約メッセージを送信 """ return self.system_prompt + "\n" + coord_protocol def _handle_tool_call(self, tool_name: str, args: dict) -> str: if tool_name == "task_update": status = TaskStatus(args["status"]) result = args.get("result", "") self.task_board.update_status(args["task_id"], status, result) if status == TaskStatus.COMPLETED: self.registry.increment_completed(self.name) return json.dumps({"ok": True, "task_id": args["task_id"], "new_status": args["status"]}) elif tool_name == "inbox_send": self.inbox.send(self.name, args["to"], args["message"]) return json.dumps({"ok": True, "sent_to": args["to"]}) elif tool_name == "inbox_receive": msgs = self.inbox.receive(self.name) if not msgs: return json.dumps({"messages": [], "note": "新しいメッセージはありません"}) return json.dumps({ "messages": [ {"from": m.sender, "content": m.content, "time": m.timestamp} for m in msgs ] }) elif tool_name == "task_list": owner = args.get("owner", self.name) tasks = self.task_board.get_tasks(owner=owner) return json.dumps({"tasks": [t.to_dict() for t in tasks]}) return json.dumps({"error": f"不明なツール: {tool_name}"}) def run(self, user_message: str, max_iterations: int = 6) -> str: self.conversation_history.append({"role": "user", "content": user_message}) for iteration in range(max_iterations): try: response = client.chat.completions.create( model=MODEL, messages=[ {"role": "system", "content": self._build_system_prompt()}, *self.conversation_history, ], tools=SWARM_TOOLS, tool_choice="auto", temperature=0.4, ) except Exception as e: return f"[APIエラー] {e}" choice = response.choices[0] msg = choice.message assistant_msg = {"role": "assistant", "content": msg.content or ""} if msg.tool_calls: assistant_msg["tool_calls"] = [ { "id": tc.id, "type": "function", "function": {"name": tc.function.name, "arguments": tc.function.arguments}, } for tc in msg.tool_calls ] self.conversation_history.append(assistant_msg) if not msg.tool_calls: return msg.content or "（応答なし）" for tc in msg.tool_calls: fn_name = tc.function.name fn_args = json.loads(tc.function.arguments) result = self._handle_tool_call(fn_name, fn_args) self.conversation_history.append({ "role": "tool", "tool_call_id": tc.id, "content": result, }) return "（エージェントが最大反復回数に達しました）" 4つのOpenAI Function Callingツール、task_update、inbox_send、inbox_receive、task_listを定義します。これらはエージェントのClawTeam CLIコマンドと同等の機能を提供します。SwarmAgentクラスを構築し、これはLLMの推論ループをラップして、エージェントが最終的な回答を生成する前に複数回ツールを呼び出すことを可能にします。これは、実際のClawTeamエージェントが反復的にタスクを確認し、作業を実行し、結果を報告する方法を模倣しています。すべてのエージェントのシステムプロンプトに調整プロトコルを自動挿入し、その名前、役割、および従わなければならない正確なワークフローを認識させます。これはClawTeamがすべての生成エージェントに指示を挿入するのと同様です。 LEADER_SYSTEM_PROMPT = """あなたはAIエージェント・スウォーム（ClawTeamスタイル）のリーダーです。 あなたの仕事: 1. 人間から高レベルの目標を受け取る 2. 明確な成果物を持つ具体的なサブタスクに分解する 3. 専門的なワーカーエージェントにタスクを割り当てる 4. 進捗を監視し、最終結果を統合する あなたは戦略的に考え、効果的に委任します。詳細な作業は自分では行いません — スウォームを調整します。 タスク計画を作成する際は、この正確な構造を持つ有効なJSONを出力してください: { "team_description": "このチームが行うことの簡潔な説明", "tasks": [ { "subject": "短いタスクタイトル", "description": "ワーカーへの詳細な指示", "worker_role": "必要な専門家の役割（例：'市場調査アナリスト'）", "worker_name": "短いスネークケース名（例：'market_analyst'）", "blocked_by_indices": [] } ] } blocked_by_indicesは、先に完了しなければならない他のタスクの0ベースのインデックスのリストです。 3-5個のタスクを作成してください。各タスクは専門家によって独立して実行可能であるべきです。 """ def leader_plan_tasks(goal: str) -> dict: console.print(Panel( f"[bold cyan]![image](https://s.w.org/images/core/emoji/17.0.2/72x72/1f99e.png) リーダーエージェント起動[/bold cyan]\n\n" f"[white]目標:[/white] {goal}\n" f"[dim]タスク分解を計画中...[/dim]", border_style="cyan", )) response = client.chat.completions.create( model=MODEL, messages=[ {"role": "system", "content": LEADER_SYSTEM_PROMPT}, {"role": "user", "content": f"この目標をAIエージェント・スウォームのサブタスクに分解してください:\n\n{goal}"}, ], temperature=0.3, response_format={"type": "json_object"}, ) plan_text = response.choices[0].message.content plan = json.loads(plan_text) return plan def display_task_board(task_board: TaskBoard, registry: TeamRegistry): table = Table(title="![image](https://s.w.org/images/core/emoji/17.0.2/72x72/1f4cb.png) タスクボード（かんばん）", show_lines=True, expand=True) table.add_column("ID", style="dim", width=8) table.add_column("件名", style="bold") table.add_column("所有者", style="cyan") table.add_column("ステータス", width=14) table.add_column("結果プレビュー", max_width=50) status_colors = { TaskStatus.PENDING: "yellow", TaskStatus.IN_PROGRESS: "blue", TaskStatus.COMPLETED: "green", TaskStatus.BLOCKED: "red", TaskStatus.FAILED: "magenta", } for t in task_board.get_tasks(): color = status_colors.get(t.status, "white") status_str = f"[{color}]●[/{color}] {t.status.value}" result_preview = (t.result[:47] + "...") if len(t.result) > 50 else t.result table.add_row(t.id, t.subject, t.owner, status_str, result_preview) console.print(table) agent_table = Table(title="![image](https://s.w.org/images/core/emoji/17.0.2/72x72/1f916.png) エージェント名簿", show_lines=True) agent_table.add_column("エージェント", style="bold cyan") agent_table.add_column("役割") agent_table.add_column("ステータス") agent_table.add_column("完了タスク", justify="center") for a in registry.agents.values(): status_emoji = {"active": "![image](https://s.w.org/images/core/emoji/17.0.2/72x72/1f504.png)", "idle": "![image](https://s.w.org/images/core/emoji/17.0.2/72x72/1f634.png)", "completed": "![image](https://s.w.org/images/core/emoji/17.0.2/72x72/2705.png)", "failed": "![image](https://s.w.org/images/core/emoji/17.0.2/72x72/274c.png)"}.get(a.status, "![image](https://s.w.org/images/core/emoji/17.0.2/72x72/2753.png)") agent_table.add_row(a.name, a.role, f"{status_emoji} {a.status}", str(a.tasks_completed)) console.print(agent_table) def run_swarm(goal: str): console.print(Panel( "[bold white on blue] ![image](https://s.w.org/images/core/emoji/17.0.2/72x72/1f99e.png) ClawTeam スウォーム・インテリジェンス — チュートリアル実行 [/bold white on blue]", expand=True, )) console.print(f"\n[bold]人間の目標:[/bold] {goal}\n") console.rule("[bold cyan]フェーズ1: リーダーが目標をタスクに分解") plan = leader_plan_tasks(goal) console.print(f"\n[bold green]チーム:[/bold green] {plan.get('team_description', 'N/A')}") console.print(f"[bold green]計画されたタスク:[/bold green] {len(plan

ClawTeamのマルチエージェント群集オーケストレーションとOpenAI関数呼び出しを実装したコーディング実演

背景や根拠まで確認しますか？

関連記事

調べる

選ぶ

サイト