Python 从零构建 AI 多智能体系统：三方协作实现复杂任务

这里要注意 think 方法里的消息构造逻辑。我们保留了对话历史，这样 Agent 能记住上下文，而不是每次只看到单条指令。

4.2 定义三个专业 Agent

接下来，我们为每个角色注入特定的 Prompt，让它们拥有'人设'。

# agents.py
from agent import Agent

# 研究员 Agent：负责技术调研
researcher = Agent(
    name="Researcher",
    role="技术研究员",
    system_prompt="""你是一个技术研究员。你的任务是：
1. 分析用户需求中涉及的技术点
2. 给出最佳实践和技术方案建议
3. 列出关键注意事项和潜在风险
输出格式：
## 技术分析
（分析内容）
## 推荐方案
（方案详情）
## 注意事项
（风险和注意点）"""
)

# 程序员 Agent：负责写代码
coder = Agent(
    name="Coder",
    role="高级程序员",
    system_prompt="""你是一个高级 Python 程序员。你的任务是：
1. 根据需求和调研结果编写高质量的 Python 代码
2. 代码必须包含完整的类型注解和 docstring
3. 包含错误处理和日志记录
4. 输出可直接运行的完整代码
输出格式：
## 实现思路
（简述思路）
## 完整代码
```python
（代码）
```"""
)

# 审查员 Agent：负责代码审查
reviewer = Agent(
    name="Reviewer",
    role="代码审查专家",
    system_prompt="""你是一个严格的代码审查专家。你的任务是：
1. 审查代码的正确性、安全性和性能
2. 检查是否遵循最佳实践
3. 给出修改建议
评分标准（满分 10 分）：
- 正确性：代码逻辑是否正确
- 安全性：是否存在安全漏洞
- 可读性：命名、注释、代码结构
- 性能：是否有明显的性能问题
输出格式：
## 评分：X/10
## 优点
（列出优点）
## 问题
（列出问题）
## 修改建议
（具体建议）"""
)

Prompt 的设计是灵魂所在。比如给审查员的 Prompt 里明确要求了评分标准，这样后续我们才能解析出分数来决定是否进入下一轮迭代。

4.3 协调者：任务调度中心

协调者是整个系统的中枢，负责串联各个 Agent，管理执行流程。

# coordinator.py
from agent import Agent, Message

class Coordinator:
    """协调者：负责调度多个 Agent 协作"""
    def __init__(self, researcher, coder, reviewer):
        self.researcher = researcher
        self.coder = coder
        self.reviewer = reviewer
        self.log = []

    def _log(self, phase: str, content: str):
        """记录执行日志"""
        self.log.append({"phase": phase, "content": content})
        print(f"\n{'='*60}")
        print(f"📋 [{phase}]")
        print(f"{'='*60}")
        print(content[:500] + "..." if len(content) > 500 else content)

    def run(self, task: str, max_revisions: int = 2) -> dict:
        """
        执行完整的多智能体协作流程
        Args:
            task: 用户需求描述
            max_revisions: 最大修改轮数
        Returns:
            包含所有阶段输出的字典
        """
        results = {"task": task, "phases": []}

        # ========== 第一阶段：需求分析 + 技术调研 ==========
        print("\n🔍 第一阶段：技术调研")
        research_msg = Message(
            role="user",
            sender="Coordinator",
            content=f"请分析以下需求的技术方案：\n{task}"
        )
        research_result = self.researcher.think([research_msg])
        self._log("技术调研", research_result)
        results["phases"].append({"phase": "research", "output": research_result})

        # ========== 第二阶段：代码生成 ==========
        print("\n💻 第二阶段：代码生成")
        code_msg = Message(
            role="user",
            sender="Coordinator",
            content=f"""请根据以下需求和调研结果编写代码：
## 需求
{task}
## 技术调研结果
{research_result}
请直接输出完整可运行的 Python 代码。"""
        )
        code_result = self.coder.think([code_msg])
        self._log("代码生成", code_result)
        results["phases"].append({"phase": "code", "output": code_result})

        # ========== 第三阶段：代码审查（可循环） ==========
        current_code = code_result
        revision = 0
        while revision <= max_revisions:
            print(f"\n🧪 第三阶段：代码审查（第 {revision + 1} 次）")
            review_msg = Message(
                role="user",
                sender="Coordinator",
                content=f"请审查以下代码：\n{current_code}"
            )
            review_result = self.reviewer.think([review_msg])
            self._log(f"代码审查（第{revision+1}次）", review_result)

            # 检查评分（简单解析）
            score = self._extract_score(review_result)
            print(f"\n📊 当前评分：{score}/10")

            if score >= 8 or revision == max_revisions:
                results["phases"].append({"phase": "review", "output": review_result, "score": score})
                break

            # 分数不够，让程序员修改
            print(f"\n🔄 评分不足，进入修改...")
            fix_msg = Message(
                role="user",
                sender="Coordinator",
                content=f"""请根据审查意见修改代码：
## 审查意见
{review_result}
## 原始代码
{current_code}
请输出修改后的完整代码。"""
            )
            current_code = self.coder.think([fix_msg])
            self._log(f"代码修改（第{revision+1}次）", current_code)
            revision += 1

        results["final_code"] = current_code
        results["revisions"] = revision + 1
        return results

    def _extract_score(self, review_text: str) -> int:
        """从审查结果中提取评分"""
        for line in review_text.split("\n"):
            if "评分" in line and "/" in line:
                try:
                    score = int(line.split(":")[-1].strip().split("/")[0])
                    return score
                except (ValueError, IndexError):
                    pass
        return 5  # 默认评分

这里有一个关键点：max_revisions 参数。为了防止审查员和程序员无限互相踢皮球，必须设置上限。同时，评分提取逻辑做了容错处理，避免因为模型输出格式微调导致程序崩溃。

4.4 主程序：启动你的 AI 团队

最后，把上面的模块组装起来，跑通整个流程。

# main.py
from agents import researcher, coder, reviewer
from coordinator import Coordinator
import json

def main():
    print("🚀 AI 研发小队已就绪！")
    print("=" * 60)

    # 初始化协调者
    coordinator = Coordinator(researcher, coder, reviewer)

    # 示例任务
    task = "用 Python 写一个文件批量重命名工具，支持按日期、序号、正则匹配重命名，需要命令行参数解析和日志记录"

    # 执行任务
    results = coordinator.run(task, max_revisions=2)

    # 输出最终结果
    print("\n" + "=" * 60)
    print("✅ 任务完成！")
    print("=" * 60)
    print(f"\n📊 总修改轮数：{results['revisions']}")
    print(f"\n💻 最终代码：\n{results['final_code']}")

    # 保存结果
    with open("output.json", "w", encoding="utf-8") as f:
        json.dump(results, f, ensure_ascii=False, indent=2)
    print("\n💾 完整结果已保存到 output.json")

if __name__ == "__main__":
    main()

五、执行流程详解

整个协作过程可以看作一个闭环：

1. 用户提交需求：协调者接收原始任务。
2. 第一阶段：调研：研究员分析技术可行性，输出调研报告。
3. 第二阶段：编码：程序员结合需求和报告，生成初版代码。
4. 第三阶段：审查：审查员打分。如果低于阈值（如 8 分），则触发修改循环。
5. 迭代或交付：若分数达标或达到最大修改次数，则输出最终代码。

这种机制模拟了真实开发中的 Code Review 流程，能有效减少低级错误。

六、扩展：加入工具能力

让 Agent 不只是'说话'，还能'做事'：

# tools.py
import os
import subprocess

def write_file(path: str, content: str) -> str:
    """写入文件"""
    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        f.write(content)
    return f"文件已写入：{path}"

def run_python(filepath: str) -> str:
    """运行 Python 文件并返回结果"""
    try:
        result = subprocess.run(["python", filepath], capture_output=True, text=True, timeout=30)
        output = result.stdout
        if result.returncode != 0:
            output += f"\n❌ 错误:\n{result.stderr}"
        return output
    except subprocess.TimeoutExpired:
        return "执行超时（30 秒）"

# 将工具注册到 Agent
AVAILABLE_TOOLS = {
    "write_file": {
        "func": write_file,
        "description": "将内容写入指定文件",
        "params": ["path", "content"]
    },
    "run_python": {
        "func": run_python,
        "description": "运行 Python 文件",
        "params": ["filepath"]
    }
}

在实际应用中，你可以将这些工具函数挂载到 Agent 的能力列表中，让它具备读写文件或执行命令的权限，但这需要严格的安全沙箱控制。

七、性能对比

在我的测试环境（i7-13700K + 32GB 内存，纯 CPU 模式）下的表现：

如有 GPU 加速，速度可提升 3-5 倍。

八、注意事项

1. Prompt 是灵魂：每个 Agent 的 System Prompt 决定了它的输出质量，值得反复打磨。
2. 防止死循环：设置 max_revisions 上限，避免审查员和程序员无限互相'踢皮球'。
3. 上下文窗口：长对话会消耗大量 token，注意定期截断历史。
4. 错误处理：模型偶尔会输出格式不符的内容，需要做容错解析。
5. 本地模型局限：8B 模型的多步推理能力有限，复杂任务建议使用 GPT-4 或 Claude 等主流商业模型。

总结

多智能体系统的核心理念：不是让一个 AI 做所有事，而是让多个专精 AI 各司其职、互相补位。

就像真正的团队一样——没有人是全能的，但一个好的团队可以。