适合前后端/测试等有编程基础的同学，手把手带你构建企业级多智能体协作系统
前言
通过上一节课的学习，我们掌握了多Agent协作的三大基础模式——委派、投票、辩论，以及AutoGen框架的用法。你现在已经能让多个Agent“聊起来”了。
但“能聊天”和“能高效协作完成复杂任务”之间，还隔着一道鸿沟。
想象一下真实的软件开发团队：
- 产品经理提出需求
- 程序员编写代码
- 测试工程师验证质量
- 技术总监协调全局
这不是简单的“轮流说话”，而是一个有层级、有分工、有状态同步的复杂协作系统。
今天这节课，我们就要迈入这个更高的层次——多Agent复杂编排。
一句话定义：多Agent复杂编排是通过层级化架构、标准化通信协议和共享状态管理，让多个专业Agent像人类团队一样有序协作的系统化方法。
全文约8000字，包含完整可运行的代码。
一、从“群聊”到“团队”：为什么需要复杂编排？
1.1 基础模式的三个天花板
上一节课学的三大基础模式虽然强大，但面对真实企业级场景时，会遇到三个根本性的局限：
| 局限 | 说明 | 后果 |
|---|---|---|
| 缺乏层级 | 所有Agent平起平坐，没有指挥链 | 协调混乱，决策效率低 |
| 状态碎片化 | 每个Agent各自为政，信息孤岛 | 上下文丢失，重复劳动 |
| 流程不可控 | Agent自由对话，无法精确控制 | 容易跑题，难以审计 |
1.2 复杂编排的三大武器
复杂编排通过三个核心机制解决以上问题：
| 武器 | 作用 | 类比 |
|---|---|---|
| 层级化架构（Hierarchy） | 建立指挥链，Supervisor协调Worker | 项目经理带队干活 |
| 共享状态（Shared State） | 所有Agent读写同一份状态数据 | 团队共享的项目看板 |
| 标准化通信（Protocol） | Agent间通过规范协议传递信息 | 团队用统一格式写周报 |
💡核心思想：让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输出JSON，Agent 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通信协议：
| 协议 | 全称 | 定位 | 提出方 |
|---|---|---|---|
| A2A | Agent-to-Agent | Agent间的发现与通信 | |
| MCP | Model Context Protocol | Agent与工具/数据源的标准化连接 | 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 ============llm=ChatOpenAI(model="deepseek-v4-flash",api_key="你的API_Key",base_url="https://api.deepseek.com",temperature=0.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：分析需求，输出PRD"""print("\n📋 [产品经理] 正在分析需求...")prompt=f""" 你是一位资深产品经理。请根据以下用户需求，输出一份清晰的产品需求文档（PRD）。 用户需求：{state['feature_request']}请按以下格式输出PRD： 1. 功能概述：一句话说明这个功能是做什么的 2. 核心功能点：列出3-5个核心功能 3. 验收标准：列出3-5条验收标准 4. 优先级：高/中/低 要求：PRD要清晰、具体、可执行。 """response=llm.invoke([HumanMessage(content=prompt)])prd=response.contentprint(f"✅ [产品经理] PRD已生成：\n{prd[:200]}...")return{"requirements":prd,"messages":[AIMessage(content=f"[产品经理] 需求分析完成：\n{prd}")]}# ============ 4. 节点2：程序员（代码实现） ============defdeveloper_node(state:SoftwareTeamState)->dict:"""程序员Agent：根据PRD编写代码"""print("\n💻 [程序员] 正在编写代码...")prompt=f""" 你是一位资深程序员。请根据以下PRD编写代码实现。 PRD：{state['requirements']}要求： 1. 使用Python编写 2. 代码要规范、有注释 3. 包含必要的错误处理 4. 输出完整可运行的代码 只输出代码，不要其他解释。 """response=llm.invoke([HumanMessage(content=prompt)])code=response.contentprint(f"✅ [程序员] 代码已生成：\n{code[:200]}...")return{"code":code,"messages":[AIMessage(content=f"[程序员] 代码实现完成：\n{code}")]}# ============ 5. 节点3：测试工程师（质量验证） ============deftester_node(state:SoftwareTeamState)->dict:"""测试工程师Agent：验证代码质量"""print("\n🧪 [测试工程师] 正在执行测试...")prompt=f""" 你是一位资深测试工程师。请对以下代码进行测试和审查。 代码：{state['code']}PRD（用于对照验收）：{state['requirements']}请输出测试报告，包含： 1. 代码审查结果：是否有明显问题 2. 测试用例设计：至少3个测试用例 3. 验收标准对照：PRD中的验收标准是否全部满足 4. 总体评价：通过/不通过 5. 如果不通过，请说明具体问题 """response=llm.invoke([HumanMessage(content=prompt)])test_results=response.content# 判断是否通过is_passed="通过"intest_resultsand"不通过"notintest_results[:50]print(f"✅ [测试工程师] 测试完成，{'通过 ✅'ifis_passedelse'不通过 ❌'}")return{"test_results":test_results,"review_status":"approved"ifis_passedelse"rejected","messages":[AIMessage(content=f"[测试工程师] 测试完成：\n{test_results}")]}# ============ 6. 路由函数 ============defshould_continue(state:SoftwareTeamState)->Literal["developer","finalize"]:"""决定是否继续迭代"""# 检查是否达到最大迭代次数ifstate.get("iteration",0)>=state.get("max_iterations",3):print("\n⚠️ 达到最大迭代次数，强制结束")return"finalize"# 如果测试通过，结束ifstate.get("review_status")=="approved":print("\n✅ 测试通过，项目完成！")return"finalize"# 否则让程序员修改print("\n🔄 测试未通过，程序员返工修改...")return"developer"# ============ 7. Finalizer节点（汇总输出） ============deffinalizer_node(state:SoftwareTeamState)->dict:"""汇总最终结果"""print("\n📊 [汇总] 生成最终报告...")summary=f""" ╔══════════════════════════════════════════════════════════╗ ║ 软件开发完成报告 ║ ╠══════════════════════════════════════════════════════════╣ ║ 需求：{state['feature_request'][:50]}... ╠══════════════════════════════════════════════════════════╣ ║ PRD：{state['requirements'][:100]}... ╠══════════════════════════════════════════════════════════╣ ║ 代码：{state['code'][:100]}... ╠══════════════════════════════════════════════════════════╣ ║ 测试结果：{state['test_results'][:100]}... ╠══════════════════════════════════════════════════════════╣ ║ 最终状态：{"✅ 通过"ifstate.get('review_status')=='approved'else'❌ 需改进'}║ 迭代次数：{state.get('iteration',0)}╚══════════════════════════════════════════════════════════╝ """return{"messages":[AIMessage(content=summary)]}# ============ 8. 构建图 ============# 创建状态图graph=StateGraph(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)# 编译app=graph.compile()# ============ 9. 执行 ============defrun_development_team(feature_request:str,max_iterations:int=3):print(f"\n🚀 启动三Agent开发团队")print(f"📝 需求：{feature_request}")print("="*60)result=app.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. 支持n=0到1000... 验收标准：1. n=0返回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_llm=ChatOpenAI(model="gpt-4",api_key="...")dev_llm=ChatOpenAI(model="deepseek-v4-flash",api_key="...",base_url="...")tester_llm=ChatOpenAI(model="gpt-4o-mini",api_key="...")defproduct_manager_node(state):# 使用 pm_llmresponse=pm_llm.invoke([...])...defdeveloper_node(state):# 使用 dev_llmresponse=dev_llm.invoke([...])...六、LangGraph vs AutoGen for 复杂编排
| 维度 | LangGraph | AutoGen |
|---|---|---|
| 编排方式 | 图结构（节点+边），精确控制流程 | 对话驱动（GroupChat），Agent自由发言 |
| 状态管理 | 显式共享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 → Tester
- CodeReviewer的职责：在代码交给测试之前，先进行代码规范、安全性、性能方面的审查
- 如果CodeReviewer不通过，回到Developer修改
- 如果CodeReviewer通过，才交给Tester
提示代码框架：
defcode_reviewer_node(state:SoftwareTeamState)->dict:"""代码审查者：审查代码质量和安全性"""prompt=f""" 请审查以下代码，从以下维度评估： 1. 代码规范：是否符合PEP8 2. 安全性：有无安全漏洞 3. 性能：有无性能问题 4. 可维护性：代码是否清晰易懂 代码：{state['code']}输出：通过/不通过 + 具体意见 """response=llm.invoke([HumanMessage(content=prompt)])review_result=response.content is_approved="通过"inreview_resultand"不通过"notinreview_result[:30]return{"code_review":review_result,"review_status":"approved"ifis_approvedelse"rejected"}# 修改路由逻辑，增加code_reviewer分支结语
今天这节课，我们全面学习了多Agent复杂编排：
| 知识点 | 核心内容 |
|---|---|
| 为什么需要复杂编排 | 基础模式缺乏层级、状态碎片化、流程不可控 |
| Supervisor-Worker架构 | 监督者协调，工作者执行，层级化团队协作 |
| 共享状态通信 | LangGraph的State模式，Agent间通过状态传递数据 |
| 通信协议 | A2A（Agent间发现）+ MCP（工具标准化） |
| 三Agent实战 | 产品经理 + 程序员 + 测试的完整开发流程 |
| 多模型配置 | 不同Agent使用不同能力的模型 |
| LangGraph vs AutoGen | 精确控制 vs 自由对话，根据场景选择 |
至此，模块四（进阶篇）的第13-16节已全部完成：
| 课时 | 内容 |
|---|---|
| 第13节 | ReAct架构深度解析 |
| 第14节 | Plan-and-Execute架构 |
| 第15节 | 多Agent协作（上）— 基础模式 |
| 第16节 | 多Agent协作（下）— 复杂编排 ✅ |
下节课（第17节），我们将进入Agent的高级记忆系统——向量记忆、图谱记忆、关系型记忆，以及如何构建“长期记忆+短期记忆”的双层系统！
如果觉得有帮助，欢迎点赞、收藏、评论三连！我们下节课见！
📌 本文是《AI Agent开发实战》课程第16节的完整内容，系列文章持续更新中，关注我不迷路！
