多智能体系统实战:角色分工、协作机制与LangGraph编排经验
发布时间:2026/9/23 4:17:13 作者:尧图编辑部 阅读量:1,286

1. 从单兵作战到团队协同为什么单智能体撑不住复杂任务我最早接触 Agent 开发的时候和大多数人一样都是从单智能体起步的。一个 LLM 加上几个工具函数套一个 ReAct 循环能查天气、能算数学、能搜网页感觉已经很厉害了。但真正把它放到一个稍微复杂一点的业务场景里问题就全暴露出来了。最典型的场景是你让一个 Agent 帮你完成调研某个技术方案并输出一份对比报告这样的任务。它需要先搜索资料然后筛选信息接着做技术对比最后组织语言输出报告。单智能体做这件事的时候你会发现它在搜索阶段表现还不错但一旦进入分析阶段它就开始忘记之前搜到的关键信息或者在输出阶段它把前面辛苦收集的数据格式搞乱了。这不是模型能力不够而是上下文窗口的容量和注意力分配在作祟。单智能体的本质问题是它把规划、执行、验证、修正全部压在一个上下文里完成。这就像让一个人同时扮演项目经理、程序员、测试工程师和文档撰写者而且这个人还没有笔记本全靠脑子记。任务一复杂信息量一上来必然丢三落四。多智能体系统Multi-Agent SystemMAS要解决的核心问题就是这个。它的思路很朴素既然一个 Agent 扛不住那就拆成多个 Agent每个负责一块通过协作机制把它们串起来。这跟软件工程里的微服务拆分是一个道理——单体应用臃肿了就拆成多个服务各司其职通过 API 通信。但多智能体不是简单地把单智能体复制几份就完事了。它引入了一个全新的复杂度维度协作。多个 Agent 之间怎么分工谁来决定谁做什么信息怎么传递出现冲突怎么办这些问题如果没想清楚多智能体系统会比单智能体更混乱——你会看到几个 Agent 互相踢皮球或者重复做同一件事甚至陷入无限循环。这篇文章要聊的就是我在实际项目中踩过的坑和总结出来的经验。从单智能体到多智能体的演进路径、角色分工的设计原则、协作机制的实现方式以及用 LangGraph 这类编排框架落地时的具体细节。适合已经写过单智能体、想往多智能体方向进阶的开发者也适合正在选型多智能体框架的技术负责人。2. 角色分工的设计不是切得越细越好2.1 常见的角色划分模式与适用边界多智能体系统里角色分工是第一步也是最容易做错的一步。我见过不少团队一上来就切出七八个 Agent搜索 Agent、分析 Agent、写作 Agent、审核 Agent、格式化 Agent、翻译 Agent……看起来很专业实际跑起来一团糟。角色划分的核心原则是按能力边界切而不是按任务步骤切。这两者的区别很关键。按任务步骤切就是第一步做 A 的 Agent、第二步做 B 的 Agent。这种切法的问题是步骤之间的耦合度很高前一步的输出格式稍微变一下后一步就崩了。而且步骤是线性的一旦某个步骤需要回退或迭代整个流程就卡住了。按能力边界切是负责信息检索的 Agent、负责逻辑推理的 Agent、负责内容生成的 Agent。每个 Agent 的能力是独立的输入输出接口可以标准化组合方式更灵活。我在实际项目中总结了几种常见的角色划分模式各有适用场景划分模式典型角色适用场景不适用场景按职能划分检索员、分析师、写手、审核员内容生产、报告生成实时交互、低延迟场景按领域划分法律 Agent、财务 Agent、技术 Agent多领域咨询、跨学科问答单一领域深度任务按层级划分管理者、执行者、验证者复杂规划、长链路任务简单线性任务按对抗划分生成者、批判者质量优化、方案打磨需要快速收敛的场景我个人的经验是刚开始做多智能体角色数量控制在 3 到 5 个之间。少于 3 个协作的价值体现不出来多于 5 个协调成本会指数级上升。等你对协作机制足够熟悉了再考虑扩展到更多角色。2.2 角色定义的三要素职责、输入输出、边界确定要几个角色之后接下来是给每个角色写清楚定义。我发现很多人写角色定义就一句话你是一个搜索助手负责搜索信息。这种定义在实际运行中会导致大量问题——Agent 不知道自己该输出什么格式、不知道什么情况下该停下来、不知道哪些事不该自己做。一个合格的角色定义至少包含三个要素第一职责描述。用自然语言说清楚这个 Agent 负责什么。但要注意职责描述不是越详细越好而是要精确到可判断的程度。比如负责搜索与用户问题相关的技术资料优先选择近两年的权威来源就比负责搜索信息要好得多。第二输入输出规范。这是最容易被忽略但最关键的部分。你需要明确告诉每个 Agent你会收到什么格式的输入你必须输出什么格式的结果。在多智能体系统里Agent 之间的通信全靠这些结构化数据格式一乱整个链路就断了。# 一个典型的角色定义示例以 LangGraph 的节点函数为例 def researcher_node(state: AgentState) - dict: 检索 Agent负责根据研究问题搜索相关资料 输入state[research_question] - 待研究的问题 输出state[search_results] - 搜索结果列表每项包含 title/source/summary question state[research_question] # 调用搜索工具 results search_tool.invoke(question) # 强制结构化输出 formatted [ {title: r.title, source: r.url, summary: r.snippet} for r in results[:5] ] return {search_results: formatted}第三边界条件。明确告诉 Agent 什么情况下应该停止、什么情况下应该转交给其他 Agent、什么情况下应该报错。比如检索 Agent 的边界可以是如果搜索结果少于 3 条在输出中标记 low_confidence 为 True由后续的分析 Agent 决定是否重新检索。2.3 一个反直觉的经验角色越少系统越稳这里分享一个可能跟直觉相反的经验在大多数业务场景下角色越少系统反而越稳定。我做过一个对比实验。同一个技术方案调研报告任务分别用 3 个 Agent检索、分析、写作和 6 个 Agent检索、筛选、分析、对比、写作、审核来实现。结果 3 个 Agent 的版本完成率是 92%6 个 Agent 的版本只有 67%。失败的原因主要是Agent 之间的信息传递丢失、某个环节的输出格式不符合下游预期、以及最要命的——循环等待A 等 B 的输出B 等 C 的输出C 又在等 A。为什么会这样因为每增加一个 Agent就增加了一组接口约定和一次信息转换。每次转换都可能引入误差误差累积到一定程度系统就崩了。这跟传话游戏是一个道理传的人越多最后的话越离谱。所以我的建议是先用最少的角色跑通流程确认协作机制没问题之后再根据实际瓶颈决定是否增加角色。增加角色的理由应该是某个环节确实是性能瓶颈或某类任务确实需要专门的能力而不是看起来更专业。3. 协作机制让多个 Agent 真正配合起来角色分好了接下来是让它们协作。这是多智能体系统里技术含量最高的部分也是最容易出问题的地方。协作机制的核心要解决三个问题谁来决定下一步做什么、信息怎么在 Agent 之间流转、出现异常怎么处理。3.1 三种主流协作拓扑顺序、层级、对等从拓扑结构上看多智能体的协作方式主要有三种顺序式Sequential是最简单的Agent 按固定顺序依次执行前一个的输出是后一个的输入。这种模式适合流程固定的任务比如检索→分析→写作→审核。优点是可控性强缺点是灵活性差无法处理需要回退或分支的情况。层级式Hierarchical是有一个管理者 Agent负责调度它根据当前状态决定调用哪个执行者 Agent。这种模式适合任务步骤不固定的场景管理者可以根据中间结果动态调整策略。LangGraph 里的 Supervisor 模式就是典型的层级式。对等式Peer-to-Peer是 Agent 之间可以互相通信、协商没有中心调度者。这种模式最灵活但也最难控制容易出现无限对话或死锁。实际项目中用得比较少更多出现在研究性质的系统里。我个人的选型建议是80% 的场景用顺序式就够了15% 的场景需要层级式只有 5% 的极端复杂场景才需要考虑对等式。不要一上来就追求最灵活的架构先用最简单的把业务跑通。3.2 状态传递共享内存还是消息传递协作机制里最核心的技术决策是Agent 之间怎么共享信息。目前主流有两种方案共享状态Shared State是所有 Agent 读写同一个状态对象。LangGraph 就是这种模式它维护一个全局的 State每个节点函数接收 State 并返回要更新的字段。这种方案的好处是信息透明任何 Agent 都能看到完整上下文坏处是状态会越来越大而且容易出现并发写冲突。消息传递Message Passing是 Agent 之间通过发送消息来通信每个 Agent 有自己的私有状态。这种方案更接近人类团队的协作方式但实现复杂度更高需要处理消息路由、序列化、超时等问题。在 LangGraph 里共享状态是默认方案也是我推荐大多数项目采用的方案。它的 State 定义通常长这样from typing import TypedDict, Annotated from operator import add class AgentState(TypedDict): # 原始任务 task: str # 检索结果用 add 表示追加而非覆盖 search_results: Annotated[list, add] # 分析结论 analysis: str # 最终输出 final_output: str # 迭代计数防止无限循环 iteration: int这里有个细节值得展开Annotated[list, add]这个写法。它告诉 LangGraph当多个节点都往search_results里写数据时用追加而不是覆盖的方式合并。这个机制在多智能体场景下非常重要因为经常会有多个 Agent 往同一个字段里贡献信息。注意共享状态虽然方便但一定要设置迭代上限。我见过太多项目因为忘了加iteration计数导致 Agent 之间无限循环API 费用一夜之间跑掉几百块。3.3 用 LangGraph 搭建协作流程的实操细节LangGraph 是目前做多智能体编排最顺手的框架之一。它的核心概念是图——节点是 Agent边是流转逻辑。相比 LangChain 的 ChainLangGraph 支持循环、分支、条件跳转这些正是多智能体协作必需的。先说说 LangChain 和 LangGraph 的区别这是很多人问的问题。简单说LangChain 适合线性流程LangGraph 适合有状态、有循环的复杂流程。单智能体用 LangChain 的 AgentExecutor 就够了但多智能体协作几乎一定要上 LangGraph。一个典型的多智能体图结构是这样的from langgraph.graph import StateGraph, END # 创建图 workflow StateGraph(AgentState) # 添加节点每个节点是一个 Agent workflow.add_node(researcher, researcher_node) workflow.add_node(analyst, analyst_node) workflow.add_node(writer, writer_node) workflow.add_node(reviewer, reviewer_node) # 设置入口 workflow.set_entry_point(researcher) # 添加边定义流转逻辑 workflow.add_edge(researcher, analyst) workflow.add_edge(analyst, writer) workflow.add_edge(writer, reviewer) # 条件边审核通过则结束不通过则回到写作 workflow.add_conditional_edges( reviewer, lambda state: pass if state[review_passed] else revise, {pass: END, revise: writer} ) app workflow.compile()这段代码里最关键的是add_conditional_edges。它让流程有了判断能力——审核 Agent 觉得不行就打回重写。这就是多智能体相比单智能体的核心优势有了自我修正的闭环。但这里有个坑我要提醒条件边一定要有明确的终止条件。上面这个例子如果review_passed永远为 False就会无限循环。实际项目中我会在 reviewer 节点里加一个计数器超过 3 次就强制通过并标记需人工复核。3.4 踩坑实录Agent 之间踢皮球的排查过程说一个我实际踩过的坑。有一次我搭了一个检索→分析→写作的三 Agent 流程测试的时候发现任务经常卡住不动。日志显示检索 Agent 正常返回了结果但分析 Agent 一直不执行。排查过程是这样的第一步我检查了图的结构确认add_edge(researcher, analyst)确实存在。结构没问题。第二步我在 researcher 节点里加了日志打印它返回的字典。发现它返回的是{search_results: [...]}字段名没错。第三步我在 analyst 节点入口打印 state发现search_results是空的。问题定位到了——数据没传过去。第四步我仔细看了 State 定义发现search_results字段我写的是Annotated[list, add]但 researcher 返回的是一个普通 list。问题在于当使用add作为 reducer 时返回的必须是可迭代对象而且 LangGraph 会把它和已有值合并。我当时的写法在某种边界情况下触发了合并逻辑的异常导致数据被吃掉了。修复方案很简单把 researcher 的返回值改成{search_results: [formatted]}确保返回的是一个列表的列表让 reducer 正确追加。或者更稳妥的做法是对于不需要追加的字段直接用普通类型不加Annotated。这个坑的教训是LangGraph 的 State reducer 机制很强大但也很容易用错。我的建议是除非确实需要多个节点往同一字段追加数据否则不要用Annotated[list, add]用普通类型更安全。4. 通信协议与信息压缩多智能体系统的隐形瓶颈4.1 Agent 之间到底该传什么多智能体系统跑起来之后你会发现一个隐形但致命的问题Agent 之间传递的信息量直接决定了系统的成本和稳定性。最朴素的做法是每个 Agent 把自己的完整输出传给下一个 Agent。检索 Agent 把 10 篇文档的全文传下去分析 Agent 把 5000 字的分析传下去写作 Agent 再基于这些写。听起来没问题但实际跑起来token 消耗会爆炸而且下游 Agent 会被大量无关信息干扰。我的经验是Agent 之间传递的应该是结论和必要的证据而不是原始素材。具体来说检索 Agent 传给分析 Agent 的应该是每篇文档的标题、来源、核心摘要100 字以内而不是全文分析 Agent 传给写作 Agent 的应该是结构化的分析结论要点列表而不是分析过程的完整推理写作 Agent 传给审核 Agent 的应该是成稿加上写作时的不确定点标注而不是写作过程的草稿这个原则说起来简单做起来需要你在每个 Agent 的输出端做信息压缩。压缩的方式可以是让 Agent 自己总结也可以是用规则提取关键字段。我倾向于前者因为 LLM 的总结能力已经足够好而且能保留语义信息。4.2 上下文窗口的分配策略多智能体系统里每个 Agent 都有自己的上下文窗口。怎么分配这些窗口是个需要认真设计的问题。我的做法是给每个 Agent 的上下文分三个区固定区角色定义、输出格式要求、few-shot 示例。这部分每个 Agent 都一样占用的 token 是固定的。输入区上游 Agent 传来的信息。这部分要严格控制大小我一般限制在 2000 token 以内。工作区Agent 自己的推理过程。这部分是动态的但也要设上限防止某个 Agent 陷入长推理。一个实用的技巧是在角色定义里明确告诉 Agent你的输入不会超过 X 字请在这个范围内工作。这样 Agent 会主动做信息筛选而不是试图处理所有内容。4.3 结构化输出让 Agent 说机器话多智能体协作的另一个关键点是Agent 的输出必须是结构化的。如果检索 Agent 返回一段自然语言分析 Agent 就得先理解这段话再提取信息这个过程中很容易出错。我的做法是强制所有 Agent 输出 JSON 格式并在角色定义里给出 schema。比如检索 Agent 的输出 schema{ results: [ { title: 文档标题, source: 来源URL, summary: 100字以内的摘要, relevance: 0.85 } ], confidence: high/medium/low, need_more_search: true }有了这个 schema下游 Agent 可以直接解析字段不用做自然语言理解。而且confidence和need_more_search这两个字段给了下游 Agent 判断是否要回退的依据。提示强制 JSON 输出时一定要在 prompt 里加上只输出 JSON不要有任何其他文字。否则 LLM 经常会在 JSON 前后加一句好的以下是结果导致解析失败。更稳妥的做法是用 LangChain 的with_structured_output方法让框架帮你处理格式约束。5. 异常处理与循环控制让系统跑得久而不崩5.1 多智能体系统最常见的四种故障多智能体系统跑起来之后故障模式比单智能体复杂得多。我总结下来最常见的四种故障是第一种无限循环。A 让 B 做事B 觉得不对让 A 重做A 又让 B 重做……这种在生成-审核模式里特别常见。防御方法是设置全局迭代上限超过就强制终止并输出当前最佳结果。第二种信息丢失。某个 Agent 的输出格式不符合下游预期下游 Agent 解析失败但又不报错直接用空值继续跑最后输出一个看似完整实则空洞的结果。防御方法是每个 Agent 入口做输入校验格式不对就抛异常。第三种责任扩散。多个 Agent 都觉得某件事不是自己的职责结果没人做。这在角色边界模糊的时候特别容易发生。防御方法是明确每个 Agent 的必须完成项并在流程末尾加一个完整性检查节点。第四种成本失控。某个 Agent 因为 prompt 设计问题每次调用都消耗大量 token或者循环次数过多导致 API 费用飙升。防御方法是给每个 Agent 设置 token 预算超了就降级处理。5.2 用检查点机制实现断点续跑LangGraph 提供了一个非常实用的功能Checkpointer检查点。它可以把每一步的状态保存下来如果系统在中途崩溃可以从最后一个检查点恢复而不用从头跑。这在多智能体场景下价值巨大因为多智能体流程通常很长一次跑完可能要几分钟甚至更久。如果跑到一半因为网络问题失败没有检查点就得全部重来成本很高。from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() app workflow.compile(checkpointermemory) # 运行时指定 thread_id同一个 thread 可以恢复 config {configurable: {thread_id: task-001}} result app.invoke({task: 调研 LangGraph}, config)生产环境建议用持久化的 checkpointer比如基于数据库的实现这样即使服务重启也能恢复。5.3 人工介入Human-in-the-Loop的接入点设计多智能体系统再智能也需要人工介入的入口。LangGraph 支持在任意节点前后插入人工确认这在关键决策点非常有用。我的经验是人工介入点应该设在不可逆操作之前。比如发送邮件之前、写入数据库之前、调用付费 API 之前。这些操作一旦执行就无法撤销必须让人确认。# 在关键节点前中断等待人工确认 app workflow.compile( checkpointermemory, interrupt_before[send_email] ) # 人工确认后继续 app.invoke(None, config) # 传入 None 表示从断点继续这个机制在实际项目中救过我好几次。有一次一个 Agent 因为理解偏差准备给客户发一封措辞不当的邮件幸好设了人工确认点被拦下来了。6. 从能跑到好用多智能体系统的调优经验6.1 评估体系怎么判断多智能体比单智能体好搭好多智能体系统之后你需要回答一个问题它真的比单智能体好吗这个问题不能靠感觉要有评估体系。我一般从四个维度评估维度评估方法合格标准任务完成率跑 50 个测试任务统计成功比例比单智能体高 15% 以上输出质量人工评分或 LLM 评分平均分不低于单智能体成本统计 token 消耗和 API 调用次数不超过单智能体的 3 倍稳定性连续跑 100 次统计失败率失败率低于 5%如果多智能体在这四个维度上没有明显优势那说明你的角色划分或协作机制有问题需要重新设计。我见过不少项目多智能体跑出来的结果还不如单智能体原因往往是角色切得太细信息在传递过程中丢失了。6.2 提示词工程在多智能体场景下的特殊考量多智能体场景下的 prompt 设计和单智能体有很大不同。单智能体的 prompt 可以很详细因为只有一个 Agent 要看。多智能体的 prompt 要考虑接口——你的输出是给另一个 Agent 看的不是给人看的。我的经验是多智能体的 prompt 要遵循三个原则简洁优先。每个 Agent 的 prompt 只保留它完成任务必需的信息不要把所有背景都塞进去。背景信息应该通过 State 传递而不是写在 prompt 里。格式明确。输出格式的要求要放在 prompt 的最后并且用醒目的方式标注。LLM 对 prompt 末尾的内容注意力更高。示例具体。给每个 Agent 配 1 到 2 个输入输出的完整示例比写一堆描述有效得多。示例要覆盖正常情况和边界情况。6.3 我踩过的三个印象最深的坑最后分享三个我踩过的、印象最深的坑都是文档里不会写的第一个坑Agent 名字会影响行为。我给一个负责批判性审核的 Agent 起名叫 critic结果它变得过度挑剔什么方案都能挑出毛病导致流程反复回退。后来改名叫 reviewer并在 prompt 里强调在指出问题的同时也要认可做得好的部分行为就正常多了。LLM 对角色名称的语义是有感知的起名要慎重。第二个坑State 字段太多会拖慢速度。我一开始把所有中间结果都塞进 State结果 State 越来越大每次节点执行都要序列化整个 State速度越来越慢。后来改成只保留必要的字段中间结果用完就清理速度提升了 40%。第三个坑并行节点不一定更快。LangGraph 支持并行执行节点我一度以为把所有能并行的都并行起来会更快。但实际上并行节点会同时消耗 API 配额如果配额有限反而会因为限流而变慢。而且并行节点的结果合并逻辑很复杂容易出 bug。我的建议是只有在节点之间确实没有依赖、且 API 配额充足时才考虑并行。多智能体系统不是银弹它解决的是单智能体上下文不够用的问题但引入了协作复杂度的新问题。什么时候该用多智能体什么时候单智能体就够了这个判断需要基于具体业务场景。我的经验是当你的单智能体 prompt 超过 2000 字、或者需要处理超过 3 个独立子任务时就该考虑多智能体了。否则把单智能体打磨好往往比仓促上多智能体更划算。