Python 构建AI多智能体系统：让三个 AI 协作完成复杂任务
单个 AI 能做的事有限，三个 AI 分工协作能做的事远超你想象。本文用纯 Python 从零实现一个多智能体系统，完整代码可直接运行。
一、什么是多智能体系统（Multi-Agent System）？
简单说：多个 AI Agent 各司其职，协作完成一个复杂任务。
类比现实团队：
- 产品经理负责分析需求
- 程序员负责写代码
- 测试工程师负责检查质量
每个 Agent 有自己的职责、工具和思考方式，它们之间通过消息传递来协作。
调研结果
代码输出
反馈意见
需要修改
📋 最终交付
🎯 协调者 Agent
任务拆分与调度
🔍 研究 Agent
信息搜集与分析
💻 编码 Agent
代码生成
🧪 审查 Agent
质量检查
二、为什么不用现成框架？
LangGraph、CrewAI、AutoGen 都很好，但先理解原理再选框架，才是正确的学习路径。
32%28%22%13%5%多智能体开发者的技术选型（2026）自研轻量框架LangGraphCrewAIAutoGen其他
本文先用纯 Python 实现核心逻辑，再对比框架选型。
三、系统架构设计
我们要构建的系统：AI 研发小队
用户输入需求 → 协调者拆分任务 → 三位 Agent 各司其职 → 汇总交付 | 角色 | 职责 | 工具 |
|---|---|---|
| Coordinator（协调者） | 分析需求、分配任务、汇总结果 | 任务拆解、结果整合 |
| Researcher（研究员） | 搜集技术方案、查找最佳实践 | 知识检索、方案分析 |
| Coder（程序员） | 编写代码实现 | 代码生成、文件操作 |
| Reviewer（审查员） | 检查代码质量和安全性 | 代码审查、漏洞扫描 |
四、代码实现
4.1 基础类：定义 Agent
# agent.pyimport json import ollama from dataclasses import dataclass, field from typing import Optional @dataclassclassMessage: role:str# "user" | "agent" | "system" sender:str# 发送者名称 content:str@dataclassclassAgent: name:str role:str system_prompt:str model:str="qwen3:8b" history:list= field(default_factory=list)defthink(self, messages:list[Message])->str:"""思考并回复"""# 构建完整的消息列表 full_messages =[{"role":"system","content": self.system_prompt}]for msg in messages: full_messages.append({"role":"user"if msg.sender != self.name else"assistant","content":f"[来自 {msg.sender}] {msg.content}"})# 调用模型 response = ollama.chat( model=self.model, messages=full_messages )return response["message"]["content"]4.2 定义三个专业 Agent
# agents.pyfrom 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 ## 优点 （列出优点） ## 问题 （列出问题） ## 修改建议 （具体建议）""")4.3 协调者：任务调度中心
# coordinator.pyfrom agent import Agent, Message classCoordinator:"""协调者：负责调度多个 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]+"..."iflen(content)>500else content)defrun(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 =0while 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 >=8or 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 +1return 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):passreturn5# 默认评分4.4 主程序：启动你的 AI 团队
# main.pyfrom agents import researcher, coder, reviewer from coordinator import Coordinator import json defmain():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']}")# 保存结果withopen("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()五、执行流程详解
审查员程序员研究员协调者用户审查员程序员研究员协调者用户第一阶段：调研第二阶段：编码第三阶段：审查评分不足，要求修改提交任务需求需求分析 + 技术方案返回调研报告调研结果 + 编码需求返回代码提交代码审查评分 6/10 + 修改建议审查意见 + 修改要求返回修改后代码再次审查评分 9/10 ✓交付最终代码 + 报告
六、扩展：加入工具能力
让 Agent 不只是"说话"，还能"做事"：
# tools.pyimport os import subprocess defwrite_file(path:str, content:str)->str:"""写入文件""" os.makedirs(os.path.dirname(path)or".", exist_ok=True)withopen(path,"w", encoding="utf-8")as f: f.write(content)returnf"文件已写入: {path}"defrun_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"]}}七、性能对比
在我的测试环境（i7-13700K + 32GB 内存，纯 CPU 模式）下的表现：
40%25%20%12%3%多 Agent 系统各阶段耗时占比技术调研（Researcher）代码生成（Coder）代码审查（Reviewer）修改迭代（Revision）协调调度开销
| 任务复杂度 | Agent 数量 | 平均耗时 | 修改轮数 |
|---|---|---|---|
| 简单（单文件脚本） | 3 | ~2 分钟 | 1 轮 |
| 中等（带参数的工具） | 3 | ~5 分钟 | 1-2 轮 |
| 复杂（多文件项目） | 4-5 | ~10 分钟 | 2-3 轮 |
如有 GPU，速度可提升 3-5 倍。
八、注意事项
- Prompt 是灵魂：每个 Agent 的 System Prompt 决定了它的输出质量，值得反复打磨
- 防止死循环：设置
max_revisions上限，避免审查员和程序员无限互相"踢皮球" - 上下文窗口：长对话会消耗大量 token，注意定期截断历史
- 错误处理：模型偶尔会输出格式不符的内容，需要做容错解析
- 本地模型局限：8B 模型的多步推理能力有限，复杂任务建议用 GPT-5 或 Claude
总结
能力有限
分工明确
互相审查
并行处理
单 Agent
多 Agent 协作
更高质量输出
更少 Bug
更快交付
多智能体系统的核心理念：不是让一个 AI 做所有事，而是让多个专精 AI 各司其职、互相补位。
就像真正的团队一样——没有人是全能的，但一个好的团队可以。
