适合前后端/测试等有编程基础的同学手把手带你构建企业级多智能体协作系统前言通过上一节课的学习我们掌握了多Agent协作的三大基础模式——委派、投票、辩论以及AutoGen框架的用法。你现在已经能让多个Agent“聊起来”了。但“能聊天”和“能高效协作完成复杂任务”之间还隔着一道鸿沟。想象一下真实的软件开发团队产品经理提出需求程序员编写代码测试工程师验证质量技术总监协调全局这不是简单的“轮流说话”而是一个有层级、有分工、有状态同步的复杂协作系统。今天这节课我们就要迈入这个更高的层次——多Agent复杂编排。一句话定义多Agent复杂编排是通过层级化架构、标准化通信协议和共享状态管理让多个专业Agent像人类团队一样有序协作的系统化方法。全文约8000字包含完整可运行的代码。一、从“群聊”到“团队”为什么需要复杂编排1.1 基础模式的三个天花板上一节课学的三大基础模式虽然强大但面对真实企业级场景时会遇到三个根本性的局限局限说明后果缺乏层级所有Agent平起平坐没有指挥链协调混乱决策效率低状态碎片化每个Agent各自为政信息孤岛上下文丢失重复劳动流程不可控Agent自由对话无法精确控制容易跑题难以审计1.2 复杂编排的三大武器复杂编排通过三个核心机制解决以上问题武器作用类比层级化架构Hierarchy建立指挥链Supervisor协调Worker项目经理带队干活共享状态Shared State所有Agent读写同一份状态数据团队共享的项目看板标准化通信ProtocolAgent间通过规范协议传递信息团队用统一格式写周报核心思想让AI Agent像人类专业团队一样有层级、有分工、有共享信息、有标准流程。二、层级式多Agent系统Supervisor-Worker架构2.1 什么是Supervisor-Worker架构Supervisor-Worker监督者-工作者是多Agent系统中最核心、最常用的层级架构模式。核心设计┌─────────────────────────────────────┐ │ Supervisor │ │ 监督者/协调者 │ │ 职责任务分配、进度追踪、结果汇总 │ └───────────────┬─────────────────────┘ │ ┌───────────────────────┼───────────────────────┐ ↓ ↓ ↓ ┌───────────────┐ ┌───────────────┐ ┌───────────────┐ │ Worker 1 │ │ Worker 2 │ │ Worker 3 │ │ 专家A │ │ 专家B │ │ 专家C │ │ 职责需求分析│ │ 职责代码开发│ │ 职责测试验证│ └───────────────┘ └───────────────┘ └───────────────┘三个核心角色角色职责决策权Supervisor监督者理解任务→分解子任务→分配给Worker→汇总结果最高决定“谁做什么”Worker工作者接收子任务→调用工具/LLM执行→返回结果有限在自己的领域内自主状态管理器维护共享状态确保信息同步无决策权纯基础设施2.2 Supervisor-Worker vs 基础模式维度GroupChat基础模式Supervisor-Worker层级模式决策方式自由发言轮流或LLM选下一个Supervisor集中决策Worker执行信息流动网状谁都可以对谁说星型Supervisor是中心可控性低容易跑题高流程清晰可审计扩展性差Agent多了混乱好加Worker不影响整体适用场景探索性讨论、头脑风暴流程化任务、软件开发三、Agent间通信协议与状态同步3.1 为什么需要标准化通信在多Agent系统中Agent之间需要传递信息。如果每个Agent都用自己定义的格式系统很快就会变成一团乱麻。常见问题Agent A输出JSONAgent B期望纯文本 → 解析失败状态分散在各个Agent的“记忆”中 → 信息不一致无法追踪“谁在什么时候说了什么” → 调试困难3.2 LangGraph的共享状态模式LangGraph通过“共享状态Shared State”来解决通信问题。核心机制所有Agent节点读写同一个State对象而不是直接互相通信。┌─────────────────────────────────────────────────────────────┐ │ 共享状态Shared State │ │ ┌───────────────────────────────────────────────────────┐ │ │ │ messages: [对话历史] │ │ │ │ requirements: 用户需求文档 │ │ │ │ code: 生成的代码 │ │ │ │ test_results: 测试结果 │ │ │ │ status: 当前状态 │ │ │ │ ... │ │ │ └───────────────────────────────────────────────────────┘ │ │ ↑ ↑ ↑ │ │ ┌────┴────┐ ┌────┴────┐ ┌────┴────┐ │ │ │ Agent A │ │ Agent B │ │ Agent C │ │ │ │ (读取/写入)│ │ (读取/写入)│ │ (读取/写入)│ │ │ └─────────┘ └─────────┘ └─────────┘ │ └─────────────────────────────────────────────────────────────┘这种设计的好处解耦Agent之间不直接依赖只依赖State可追踪所有状态变化都有记录可恢复State可以持久化支持断点续传可观测随时查看完整状态3.3 A2A协议与MCP协议2026年趋势在更大型的多Agent系统中Agent可能运行在不同的服务甚至不同的组织中。这时需要标准化的跨Agent通信协议协议全称定位提出方A2AAgent-to-AgentAgent间的发现与通信GoogleMCPModel Context ProtocolAgent与工具/数据源的标准化连接Anthropic在实际应用中A2A负责Agent之间的“握手”与任务委托MCP负责Agent与外部工具的标准化交互。在LangGraph中Agent之间通过共享状态传递数据而MCP则用于Agent调用外部工具。四、实战三Agent软件开发团队现在我们用LangGraph构建一个完整的“产品经理 程序员 测试”三Agent协作开发流程。4.1 完整代码fromtypingimportTypedDict,Annotated,Literalfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.graph.messageimportadd_messagesfromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportHumanMessage,AIMessage,SystemMessageimportjson# 1. 配置LLM llmChatOpenAI(modeldeepseek-v4-flash,api_key你的API_Key,base_urlhttps://api.deepseek.com,temperature0.3)# 2. 定义共享状态 classSoftwareTeamState(TypedDict):三Agent开发团队的共享状态messages:Annotated[list,add_messages]# 完整对话历史feature_request:str# 用户需求requirements:str# 产品经理输出的PRDcode:str# 程序员输出的代码test_results:str# 测试工程师输出的测试结果review_status:Literal[pending,approved,rejected]# 审核状态iteration:int# 当前迭代轮次max_iterations:int# 最大迭代次数# 3. 节点1产品经理需求分析 defproduct_manager_node(state:SoftwareTeamState)-dict:产品经理Agent分析需求输出PRDprint(\n [产品经理] 正在分析需求...)promptf 你是一位资深产品经理。请根据以下用户需求输出一份清晰的产品需求文档PRD。 用户需求{state[feature_request]}请按以下格式输出PRD 1. 功能概述一句话说明这个功能是做什么的 2. 核心功能点列出3-5个核心功能 3. 验收标准列出3-5条验收标准 4. 优先级高/中/低 要求PRD要清晰、具体、可执行。 responsellm.invoke([HumanMessage(contentprompt)])prdresponse.contentprint(f✅ [产品经理] PRD已生成\n{prd[:200]}...)return{requirements:prd,messages:[AIMessage(contentf[产品经理] 需求分析完成\n{prd})]}# 4. 节点2程序员代码实现 defdeveloper_node(state:SoftwareTeamState)-dict:程序员Agent根据PRD编写代码print(\n [程序员] 正在编写代码...)promptf 你是一位资深程序员。请根据以下PRD编写代码实现。 PRD{state[requirements]}要求 1. 使用Python编写 2. 代码要规范、有注释 3. 包含必要的错误处理 4. 输出完整可运行的代码 只输出代码不要其他解释。 responsellm.invoke([HumanMessage(contentprompt)])coderesponse.contentprint(f✅ [程序员] 代码已生成\n{code[:200]}...)return{code:code,messages:[AIMessage(contentf[程序员] 代码实现完成\n{code})]}# 5. 节点3测试工程师质量验证 deftester_node(state:SoftwareTeamState)-dict:测试工程师Agent验证代码质量print(\n [测试工程师] 正在执行测试...)promptf 你是一位资深测试工程师。请对以下代码进行测试和审查。 代码{state[code]}PRD用于对照验收{state[requirements]}请输出测试报告包含 1. 代码审查结果是否有明显问题 2. 测试用例设计至少3个测试用例 3. 验收标准对照PRD中的验收标准是否全部满足 4. 总体评价通过/不通过 5. 如果不通过请说明具体问题 responsellm.invoke([HumanMessage(contentprompt)])test_resultsresponse.content# 判断是否通过is_passed通过intest_resultsand不通过notintest_results[:50]print(f✅ [测试工程师] 测试完成{通过 ✅ifis_passedelse不通过 ❌})return{test_results:test_results,review_status:approvedifis_passedelserejected,messages:[AIMessage(contentf[测试工程师] 测试完成\n{test_results})]}# 6. 路由函数 defshould_continue(state:SoftwareTeamState)-Literal[developer,finalize]:决定是否继续迭代# 检查是否达到最大迭代次数ifstate.get(iteration,0)state.get(max_iterations,3):print(\n⚠️ 达到最大迭代次数强制结束)returnfinalize# 如果测试通过结束ifstate.get(review_status)approved:print(\n✅ 测试通过项目完成)returnfinalize# 否则让程序员修改print(\n 测试未通过程序员返工修改...)returndeveloper# 7. Finalizer节点汇总输出 deffinalizer_node(state:SoftwareTeamState)-dict:汇总最终结果print(\n [汇总] 生成最终报告...)summaryf ╔══════════════════════════════════════════════════════════╗ ║ 软件开发完成报告 ║ ╠══════════════════════════════════════════════════════════╣ ║ 需求{state[feature_request][:50]}... ╠══════════════════════════════════════════════════════════╣ ║ PRD{state[requirements][:100]}... ╠══════════════════════════════════════════════════════════╣ ║ 代码{state[code][:100]}... ╠══════════════════════════════════════════════════════════╣ ║ 测试结果{state[test_results][:100]}... ╠══════════════════════════════════════════════════════════╣ ║ 最终状态{✅ 通过ifstate.get(review_status)approvedelse❌ 需改进}║ 迭代次数{state.get(iteration,0)}╚══════════════════════════════════════════════════════════╝ return{messages:[AIMessage(contentsummary)]}# 8. 构建图 # 创建状态图graphStateGraph(SoftwareTeamState)# 添加节点graph.add_node(product_manager,product_manager_node)graph.add_node(developer,developer_node)graph.add_node(tester,tester_node)graph.add_node(finalizer,finalizer_node)# 添加边PM → Developer → Testergraph.add_edge(START,product_manager)graph.add_edge(product_manager,developer)graph.add_edge(developer,tester)# 条件边根据测试结果决定下一步graph.add_conditional_edges(tester,should_continue,{developer:developer,# 测试不通过 → 程序员返工finalize:finalizer# 测试通过 → 结束})graph.add_edge(finalizer,END)# 编译appgraph.compile()# 9. 执行 defrun_development_team(feature_request:str,max_iterations:int3):print(f\n 启动三Agent开发团队)print(f 需求{feature_request})print(*60)resultapp.invoke({messages:[],feature_request:feature_request,requirements:,code:,test_results:,review_status:pending,iteration:0,max_iterations:max_iterations})# 打印最终报告print(\n*60)formsginresult[messages]:ifhasattr(msg,content):print(msg.content)returnresult# 10. 测试 if__name____main__:run_development_team(开发一个Python函数计算斐波那契数列的第n项要求支持大数计算n最大1000。)4.2 执行流程图┌─────────────────────────────────────┐ │ START │ └──────────────────┬──────────────────┘ ↓ ┌─────────────────────────────────────┐ │ product_manager │ │ 需求分析 → 输出PRD │ └──────────────────┬──────────────────┘ ↓ ┌─────────────────────────────────────┐ │ developer │ │ 编写代码 → 输出代码 │ └──────────────────┬──────────────────┘ ↓ ┌─────────────────────────────────────┐ │ tester │ │ 测试验证 → 输出报告 │ └──────────────────┬──────────────────┘ ↓ ┌──────┴──────┐ │ 测试通过 │ └──────┬──────┘ ↙ ↘ 通过 不通过 ↓ ↓ ┌─────────────────┐ ┌─────────────────┐ │ finalizer │ │ developer │ │ 汇总输出 │ │ 返工修改 │ └────────┬────────┘ └────────┬────────┘ ↓ │ ┌─────────────────┐ │ │ END │←───────────┘ └─────────────────┘4.3 运行效果 启动三Agent开发团队 需求开发一个Python函数计算斐波那契数列的第n项要求支持大数计算n最大1000。 [产品经理] 正在分析需求... ✅ [产品经理] PRD已生成 功能概述实现一个计算斐波那契数列第n项的函数... 核心功能点1. 支持n0到1000... 验收标准1. n0返回0... [程序员] 正在编写代码... ✅ [程序员] 代码已生成 def fibonacci(n: int) - int: 计算斐波那契数列的第n项 if n 0: raise ValueError(n必须是非负整数) ... [测试工程师] 正在执行测试... ✅ [测试工程师] 测试完成通过 ✅ ✅ 测试通过项目完成 [汇总] 生成最终报告... ╔══════════════════════════════════════════════════════════╗ ║ 软件开发完成报告 ║ ╠══════════════════════════════════════════════════════════╣ ║ 需求开发一个Python函数计算斐波那契数列的第n项... ╠══════════════════════════════════════════════════════════╣ ║ 最终状态✅ 通过 ║ 迭代次数1 ╚══════════════════════════════════════════════════════════╝五、扩展为不同Agent配置不同模型在真实项目中不同Agent可能需要不同能力的模型Agent推荐模型原因产品经理GPT-4 / DeepSeek-V4需要强推理和规划能力程序员DeepSeek-V4 / Claude需要代码生成能力测试工程师GPT-4o-mini / 通义千问需要逻辑分析成本敏感实现方式# 为每个Agent使用不同的模型pm_llmChatOpenAI(modelgpt-4,api_key...)dev_llmChatOpenAI(modeldeepseek-v4-flash,api_key...,base_url...)tester_llmChatOpenAI(modelgpt-4o-mini,api_key...)defproduct_manager_node(state):# 使用 pm_llmresponsepm_llm.invoke([...])...defdeveloper_node(state):# 使用 dev_llmresponsedev_llm.invoke([...])...六、LangGraph vs AutoGen for 复杂编排维度LangGraphAutoGen编排方式图结构节点边精确控制流程对话驱动GroupChatAgent自由发言状态管理显式共享State所有节点读写隐式通过消息传递流程控制条件边、循环、中断精确可控轮询/LLM选下一个发言者适用场景需要精确控制流程的企业级应用探索性多Agent对话学习曲线陡峭中等选择建议需要精确控制每一步→ LangGraph需要Agent自由对话、探索性协作→ AutoGen两者可以混合使用AutoGen做对话层LangGraph做编排层七、真实案例从30人到6人效率提升10倍Amazon Kiro案例原本需要30人18个月完成的项目使用多Agent协作后仅需6人76天完成效率提升10倍以上。RepoPilot案例基于LangGraph构建的多智能体软件公司模拟产品经理、项目经理、架构师、开发、测试、DevOps、安全审查和交付团队从需求沟通到ZIP交付形成完整闭环。这些案例证明了多Agent复杂编排在真实企业场景中的巨大价值。八、实战小练习作业练习扩展三Agent团队为“四Agent团队”需求在现有的PM → Developer → Tester流程中增加一个CodeReviewer代码审查者节点新流程PM → Developer → CodeReviewer → TesterCodeReviewer的职责在代码交给测试之前先进行代码规范、安全性、性能方面的审查如果CodeReviewer不通过回到Developer修改如果CodeReviewer通过才交给Tester提示代码框架defcode_reviewer_node(state:SoftwareTeamState)-dict:代码审查者审查代码质量和安全性promptf 请审查以下代码从以下维度评估 1. 代码规范是否符合PEP8 2. 安全性有无安全漏洞 3. 性能有无性能问题 4. 可维护性代码是否清晰易懂 代码{state[code]}输出通过/不通过 具体意见 responsellm.invoke([HumanMessage(contentprompt)])review_resultresponse.content is_approved通过inreview_resultand不通过notinreview_result[:30]return{code_review:review_result,review_status:approvedifis_approvedelserejected}# 修改路由逻辑增加code_reviewer分支结语今天这节课我们全面学习了多Agent复杂编排知识点核心内容为什么需要复杂编排基础模式缺乏层级、状态碎片化、流程不可控Supervisor-Worker架构监督者协调工作者执行层级化团队协作共享状态通信LangGraph的State模式Agent间通过状态传递数据通信协议A2AAgent间发现 MCP工具标准化三Agent实战产品经理 程序员 测试的完整开发流程多模型配置不同Agent使用不同能力的模型LangGraph vs AutoGen精确控制 vs 自由对话根据场景选择至此模块四进阶篇的第13-16节已全部完成课时内容第13节ReAct架构深度解析第14节Plan-and-Execute架构第15节多Agent协作上— 基础模式第16节多Agent协作下— 复杂编排 ✅下节课第17节我们将进入Agent的高级记忆系统——向量记忆、图谱记忆、关系型记忆以及如何构建“长期记忆短期记忆”的双层系统如果觉得有帮助欢迎点赞、收藏、评论三连我们下节课见 本文是《AI Agent开发实战》课程第16节的完整内容系列文章持续更新中关注我不迷路