多 Agent 协作：当一个 Agent 不够用时，如何让多个 Agent 分工合作

import openai
import json
import asyncio

client = openai.AsyncOpenAI()

async def run_worker(
    worker_name: str,
    worker_instructions: str,
    task: str,
    tools: list | None = None,
) -> str:
    """启动一个 Worker Agent 执行子任务"""
    messages = [
        {"role": "system", "content": worker_instructions},
        {"role": "user", "content": task},
    ]

    for _ in range(10):
        response = await client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
        )
        choice = response.choices[0]

        if choice.finish_reason == "stop":
            return choice.message.content

        if choice.finish_reason == "tool_calls":
            messages.append(choice.message)
            for tc in choice.message.tool_calls:
                result = await dispatch_tool(tc.function.name,
                                             json.loads(tc.function.arguments))
                messages.append({
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": str(result),
                })

    return f"[{worker_name}] 执行超时"


async def supervisor_orchestrate(goal: str, workers: list[dict]) -> str:
    """
    Supervisor 模式的核心编排逻辑。

    workers 格式：
    [
        {"name": "researcher", "instructions": "...", "tools": [...]},
        {"name": "analyst", "instructions": "...", "tools": [...]},
    ]
    """

    # 第一步：Supervisor 分解任务
    plan_response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "system",
            "content": (
                "你是一个任务协调者。根据目标和可用的 Worker 列表，"
                "把任务分解成子任务并分配给合适的 Worker。"
                "输出 JSON 格式。"
            ),
        }, {
            "role": "user",
            "content": f"""
目标：{goal}

可用 Workers：
{json.dumps([{"name": w["name"], "capabilities": w["instructions"][:200]}
             for w in workers], ensure_ascii=False, indent=2)}

输出格式：
{{
  "assignments": [
    {{"worker": "worker_name", "task": "具体子任务描述"}},
  ],
  "synthesis_plan": "如何汇总各 Worker 的结果"
}}
""",
        }],
        response_format={"type": "json_object"},
    )

    plan = json.loads(plan_response.choices[0].message.content)
    worker_map = {w["name"]: w for w in workers}

    # 第二步：并行分发子任务给 Workers
    tasks = []
    for assignment in plan["assignments"]:
        w = worker_map[assignment["worker"]]
        tasks.append(run_worker(
            worker_name=w["name"],
            worker_instructions=w["instructions"],
            task=assignment["task"],
            tools=w.get("tools"),
        ))

    results = await asyncio.gather(*tasks)

    # 第三步：Supervisor 汇总结果
    result_summary = "\n\n".join([
        f"=== {plan['assignments'][i]['worker']} ===\n{results[i]}"
        for i in range(len(results))
    ])

    synthesis = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "system",
            "content": "你是一个任务协调者。根据各 Worker 的执行结果，汇总出最终答案。",
        }, {
            "role": "user",
            "content": f"目标：{goal}\n\n各 Worker 结果：\n{result_summary}\n\n汇总计划：{plan['synthesis_plan']}",
        }],
    )

    return synthesis.choices[0].message.content

from dataclasses import dataclass
from typing import Callable

@dataclass
class SpecialistAgent:
    name: str
    description: str
    instructions: str
    tools: list | None = None
    condition: Callable[[str], bool] | None = None


async def handoff_router(
    user_message: str,
    specialists: list[SpecialistAgent],
    conversation_history: list[dict],
) -> tuple[str, str]:
    """
    路由 Agent 决定交给哪个专家处理，然后移交控制权。
    返回 (specialist_name, response)。
    """

    specialist_descriptions = "\n".join([
        f"- {s.name}: {s.description}" for s in specialists
    ])

    routing_response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "system",
            "content": f"""你是一个路由 Agent。根据用户的请求，选择最合适的专家来处理。

可用专家：
{specialist_descriptions}

只输出 JSON：{{"specialist": "专家名称", "reason": "选择原因"}}""",
        }] + conversation_history + [{
            "role": "user",
            "content": user_message,
        }],
        response_format={"type": "json_object"},
    )

    routing = json.loads(routing_response.choices[0].message.content)
    chosen = next(s for s in specialists if s.name == routing["specialist"])

    # 控制权完整移交给专家——专家直接处理，不经过路由中转
    response = await run_worker(
        worker_name=chosen.name,
        worker_instructions=chosen.instructions,
        task=user_message,
        tools=chosen.tools,
    )

    return chosen.name, response

def agent_as_tool(
    name: str,
    description: str,
    agent_instructions: str,
    agent_tools: list | None = None,
) -> dict:
    """把一个 Agent 包装成 Tool 定义"""
    return {
        "type": "function",
        "function": {
            "name": name,
            "description": description,
            "parameters": {
                "type": "object",
                "properties": {
                    "task": {
                        "type": "string",
                        "description": "要交给这个 Agent 完成的任务描述",
                    },
                },
                "required": ["task"],
            },
        },
        "_agent_config": {
            "instructions": agent_instructions,
            "tools": agent_tools,
        },
    }


async def dispatch_tool(name: str, args: dict) -> str:
    """工具调度——如果是 Agent Tool，启动子 Agent 执行"""
    tool_registry = get_tool_registry()
    tool_def = tool_registry[name]

    if "_agent_config" in tool_def:
        config = tool_def["_agent_config"]
        return await run_worker(
            worker_name=name,
            worker_instructions=config["instructions"],
            task=args["task"],
            tools=config.get("tools"),
        )

    return await call_regular_tool(name, args)

import time
from dataclasses import dataclass, field

@dataclass
class SharedTaskBoard:
    """Agent 之间的共享任务板——协调的核心数据结构"""
    tasks: list[dict] = field(default_factory=list)
    messages: list[dict] = field(default_factory=list)
    findings: list[dict] = field(default_factory=list)

    def add_task(self, task_id: str, description: str, depends_on: list[str] | None = None):
        self.tasks.append({
            "id": task_id,
            "description": description,
            "status": "pending",
            "claimed_by": None,
            "depends_on": depends_on or [],
        })

    def claim_task(self, task_id: str, agent_name: str) -> bool:
        for t in self.tasks:
            if t["id"] == task_id and t["status"] == "pending":
                deps_done = all(
                    any(d["id"] == dep and d["status"] == "completed" for d in self.tasks)
                    for dep in t["depends_on"]
                )
                if not deps_done:
                    return False
                t["status"] = "in_progress"
                t["claimed_by"] = agent_name
                return True
        return False

    def complete_task(self, task_id: str, result: str):
        for t in self.tasks:
            if t["id"] == task_id:
                t["status"] = "completed"
                t["result"] = result
                break

    def post_message(self, sender: str, recipient: str | None, content: str):
        """发消息：recipient=None 表示广播"""
        self.messages.append({
            "sender": sender,
            "recipient": recipient,
            "content": content,
            "timestamp": time.time(),
        })

    def get_messages_for(self, agent_name: str) -> list[dict]:
        return [
            m for m in self.messages
            if m["recipient"] is None or m["recipient"] == agent_name
        ]


async def peer_agent_loop(
    agent_name: str,
    instructions: str,
    board: SharedTaskBoard,
    tools: list | None = None,
):
    """单个 Peer Agent 的执行循环"""
    while True:
        available = [
            t for t in board.tasks
            if t["status"] == "pending" and not t["claimed_by"]
        ]
        if not available and all(t["status"] == "completed" for t in board.tasks):
            break

        new_messages = board.get_messages_for(agent_name)
        context = build_peer_context(agent_name, board, new_messages)

        response = await client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": instructions},
                {"role": "user", "content": context},
            ],
            response_format={"type": "json_object"},
        )

        action = json.loads(response.choices[0].message.content)

        if action.get("claim_task"):
            board.claim_task(action["claim_task"], agent_name)
        elif action.get("complete_task"):
            board.complete_task(action["complete_task"], action["result"])
        elif action.get("send_message"):
            board.post_message(agent_name, action["send_message"]["to"],
                               action["send_message"]["content"])
        elif action.get("post_finding"):
            board.findings.append({
                "agent": agent_name,
                "finding": action["post_finding"],
            })

# 反面示例：子 Agent 缺少关键上下文
await run_worker(
    worker_name="reviewer",
    worker_instructions="审查代码质量",
    task="审查 src/auth/ 模块",  # 没有说明项目的认证方案和技术栈
)

# 正面示例：传递必要上下文
await run_worker(
    worker_name="reviewer",
    worker_instructions="审查代码质量",
    task=(
        "审查 src/auth/ 模块的安全性。"
        "项目使用 JWT + httpOnly cookie 方案，"
        "后端是 Express.js，数据库是 PostgreSQL。"
        "重点关注 token 过期处理和 CSRF 防护。"
    ),
)