Python 构建AI多智能体系统：让三个 AI 协作完成复杂任务

优质文章学习记录

— 9 min read
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 倍。

八、注意事项

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

总结

能力有限

分工明确

互相审查

并行处理

单 Agent

多 Agent 协作

更高质量输出

更少 Bug

更快交付

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

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

Read more

InstructPix2Pix效果实测：结构保留能力 vs Stable Diffusion 图生图对比

InstructPix2Pix效果实测：结构保留能力 vs Stable Diffusion 图生图对比 1. 为什么说InstructPix2Pix是真正的“魔法修图师” 你有没有过这样的经历：想把一张照片里的白天改成夜晚，或者给朋友P一副墨镜，又或者让一张普通街景变成雨天氛围——但打开PS，面对层层叠叠的图层和蒙版，最后只留下满屏困惑？传统图像编辑工具需要你懂色彩曲线、图层混合模式、甚至手绘遮罩；而Stable Diffusion这类图生图模型，又常常让人陷入“写对Prompt像解谜”的困境：多加一个词，画面就崩掉；少写一个细节，AI就自由发挥到千里之外。 InstructPix2Pix不一样。它不把你当设计师，也不把你当咒语学徒，而是直接把你当“导演”——你只需要用日常英语说出想法，它就照着执行，而且几乎不会跑偏。 这不是滤镜，不是风格迁移，更不是粗暴重绘。它像一位经验丰富的修图老手，先仔仔细细看清原图里每一条轮廓线、每一个人物姿态、每一处光影关系，再只动你点名要改的那一小块。你让它“add sunglasses”，它不会顺手把人脸拉长、把背景重画一遍；你让它“
企微群机器人发markdown消息支持表格

企微群机器人发markdown消息支持表格

结论 1.V1接口可以圈人，但是无法正确展示表格的markdown语法 2.V2接口可以展示表格的markdown语法，但是无法圈人 3.企微消息有长度限制 前言 今天是日本投降日，写篇技术文档。 企业微信机器人发markdown表格信息+如何艾特人 企微机器人发消息通知，目标是生成数据对比表格，然后艾特到具体的人来跟进事物的变化 1、成果收益 发表格数据，圈人 2、背景 目前机器人通知的内容太单调了，无法满足告警提醒的作用，需要罗列表格进行对比，需要艾特到具体人 3、解决方案 如何支持markdown表格类型 1.企业微信从4.1.38开始支持markdown表格的语法了。可以参看官方文档4.1.38版本新功能介绍 所以企业客户端要升级 2.我们历史使用的是msgtype：markdown，这个还是不支持的 { "msgtype": "markdown", "markdown&
混合知识库搭建：本地Docker部署Neo4j图数据库与Milvus向量库

混合知识库搭建：本地Docker部署Neo4j图数据库与Milvus向量库

混合知识库搭建：本地Docker部署Neo4j图数据库与Milvus向量库 前言 在多代理混合RAG系统中，知识库是“知识储备核心”，直接决定了代理检索的精准度与响应质量。上一篇我们解析了5个子代理的执行逻辑，而这些代理能高效完成知识检索任务，背后依赖“Neo4j图知识库+Milvus向量库”的混合支撑——图知识库擅长挖掘实体关系，向量库精准匹配语义细节，二者互补形成全场景知识覆盖。 本文作为系列博客的第三篇，将聚焦混合知识库的落地实现：从本地Docker部署、数据建模、索引构建，到双库协同逻辑，手把手带你搭建高可用的混合知识库，让你掌握“关系型知识+语义型知识”的全链路管理技巧。 1 混合知识库的设计逻辑：为什么需要“图+向量”双引擎？ 1.1 单一知识库的局限性 * 纯图数据库：擅长实体关系查询（如“小米的合作品牌”），但无法高效处理细粒度文本检索（如“苹果的环保目标细节”）； * 纯向量数据库：擅长语义相似性检索（如“查找与5G技术相关的内容”），但难以挖掘实体间的复杂关联（如“华为-开发-鸿蒙-适配-智能设备”