构建自己的 AI Agent · 第五篇：多 Agent 协作 — 从单打独斗到团队协作的数学原理

import json
from openai import OpenAI

class GeneratorVerifierTeam:
    """Generator-Verifier 双 Agent 协作"""

    def __init__(self, max_revisions: int = 3):
        self.client = OpenAI()
        self.max_revisions = max_revisions

    def solve(self, task: str, test_cases: list[dict] = None) -> dict:
        """
        双 Agent 协作求解任务。

        Args:
            task: 任务描述
            test_cases: [{"input": ..., "expected": ...}]
        Returns:
            {"solution": ..., "iterations": N, "passed": bool}
        """
        # Phase 1: Generator 生成初始解
        code = self._generate(task)
        revision_history = [code]

        for iteration in range(self.max_revisions):
            # Phase 2: Verifier 审查
            review = self._verify(task, code, test_cases)

            if review["approved"]:
                return {
                    "solution": code,
                    "iterations": iteration + 1,
                    "revisions": revision_history,
                    "passed": True,
                    "review_comments": review["comments"]
                }

            # Phase 3: Generator 根据反馈修改
            code = self._revise(task, code, review["feedback"])
            revision_history.append(code)

        return {
            "solution": code,
            "iterations": self.max_revisions,
            "revisions": revision_history,
            "passed": False,
            "review_comments": review.get("feedback", "达到最大修改次数")
        }

    def _generate(self, task: str) -> str:
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "system",
                "content": ("你是一个资深 Python 开发者。根据任务描述编写高质量代码。"
                           "包含类型注解、docstring 和边界条件处理。只输出代码，"
                           "不要额外解释。")
            }, {
                "role": "user",
                "content": task
            }],
            temperature=0.2
        )
        return response.choices[0].message.content

    def _verify(self, task: str, code: str,
                test_cases: list[dict] = None) -> dict:
        """Verifier: 审查代码质量"""
        prompt = f"""审查以下代码。原始任务：{task}

```python
{code}

from concurrent.futures import ThreadPoolExecutor, as_completed
import concurrent.futures

class OrchestratorWorkers:
    def __init__(self, registry: ToolRegistry):
        self.client = OpenAI()
        self.registry = registry
        self.executor = ThreadPoolExecutor(max_workers=4)

    def solve(self, task: str) -> str:
        # Step 1: Orchestrator 分解任务
        subtasks = self._decompose(task)
        print(f"[Orchestrator] 分解为 {len(subtasks)} 个子任务")

        # Step 2: 并行执行子任务
        futures = {}
        for i, subtask in enumerate(subtasks):
            future = self.executor.submit(
                self._execute_subtask, subtask
            )
            futures[future] = i

        # Step 3: 收集结果
        results = [None] * len(subtasks)
        for future in as_completed(futures):
            idx = futures[future]
            results[idx] = future.result()
            print(f"[Worker {idx}] 完成")

        # Step 4: 汇总合成
        return self._synthesize(task, subtasks, results)

    def _decompose(self, task: str) -> list[str]:
        """Orchestrator 将复杂任务分解为独立子任务"""
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "system",
                "content": (
                    "将复杂任务分解为 3-5 个相互独立的子任务。"
                    "每个子任务应能独立完成，不需要其他子任务的结果。"
                    "输出 JSON 数组，每个元素是子任务描述字符串。"
                )
            }, {
                "role": "user",
                "content": task
            }],
            temperature=0.1
        )
        return json.loads(response.choices[0].message.content)

    def _execute_subtask(self, subtask: str) -> str:
        """Worker 独立执行子任务"""
        # 每个 Worker 是一个独立的 SimpleAgent 实例
        agent = SimpleAgent(
            system_prompt="你是一个高效的执行者。完成任务并返回结果。",
            registry=self.registry
        )
        return agent.run(subtask)

    def _synthesize(self, task: str, subtasks: list[str],
                    results: list[str]) -> str:
        """Orchestrator 汇总结果"""
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "system",
                "content": "将子任务的结果合成为一个完整、连贯的回答。"
            }, {
                "role": "user",
                "content": (
                    f"原始任务：{task}\n\n"
                    + "\n\n".join(
                        f"子任务 {i}: {s}\n结果: {r}"
                        for i, (s, r) in enumerate(
                            zip(subtasks, results))
                    )
                )
            }],
            temperature=0.3
        )
        return response.choices[0].message.content

class AgentTeam:
    """多 Agent 团队——角色扮演协作"""

    def __init__(self, members: list[dict]):
        """
        members = [
            {"name": "产品经理", "role": "定义需求和验收标准"},
            {"name": "架构师", "role": "设计系统架构"},
            {"name": "开发者", "role": "实现代码"},
            {"name": "QA", "role": "设计和运行测试"}
        ]
        """
        self.members = members
        self.history: list[dict] = []

    def discuss(self, topic: str, rounds: int = 3) -> str:
        """团队成员轮流发言讨论"""
        context = f"团队讨论主题：{topic}\n\n团队成员：\n"
        for m in self.members:
            context += f"- {m['name']}: {m['role']}\n"

        for round_num in range(rounds):
            for member in self.members:
                response = self._speak(member, context)
                self.history.append({
                    "round": round_num,
                    "speaker": member["name"],
                    "content": response
                })
                context += f"\n[{member['name']}]: {response}\n"

        # 最终综合
        return self._summarize(topic, context)

    def _speak(self, member: dict, context: str) -> str:
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "system",
                "content": (
                    f"你是团队的{member['name']}。你的职责：{member['role']}。"
                    f"基于讨论上下文，从你的专业角度发表意见。"
                    f"可以赞同、反对或补充其他成员的观点。"
                )
            }, {
                "role": "user",
                "content": (
                    f"当前讨论上下文：\n{context}\n\n"
                    f"请以{member['name']}的身份发言。"
                )
            }],
            temperature=0.7  # 更高温度鼓励多样化观点
        )
        return response.choices[0].message.content