ARTICLE DETAIL

资讯详情

深耕网站建设、视觉设计与SEO优化的一线实战洞察。

多智能体系统设计实战:角色分工与协作机制避坑指南

多智能体系统设计实战:角色分工与协作机制避坑指南 多智能体系统这两年从论文里的概念快速落到了工程实践里但真正动手搭过的人都知道从单智能体切到多智能体代码量的增长远小于心智负担的增长。单智能体的时候你只需要关心一个提示词写得好不好、工具调用对不对一旦拆成多个角色问题就变成了谁该干什么、干完之后交给谁、交接的时候传什么、出了错算谁的。这些问题的本质不是技术问题而是协作设计问题。我前后用 LangGraph 搭过几套多智能体流程也踩过不少看起来分工清晰、跑起来一团乱麻的坑这篇就把角色分工和协作机制这两件事拆开讲透从为什么要拆、怎么拆、拆完怎么连、连完怎么稳一步步说清楚。不管你是刚接触 Agent 开发的新手还是已经写过单智能体想往多智能体演进的开发者都能从里面找到可以直接抄的配置思路和避坑经验。1. 从单智能体到多智能体的分水岭在哪里1.1 单智能体不是能力不够而是职责过载很多人以为多智能体是为了让系统更聪明其实更准确的说法是单智能体在职责变多之后会变得不稳定。我最早做的一个单智能体项目是帮用户做技术文档问答一开始只做检索加回答提示词两百字就够了效果很稳。后来需求加上了先判断问题类型、再决定要不要查资料、查完还要做一轮事实校验、最后按固定格式输出提示词膨胀到一千多字结果开始出现各种诡异现象该查资料的时候不查该校验的时候直接跳过输出格式时对时错。这不是模型变笨了而是单个上下文里塞了太多互相冲突的目标。模型在每一步都要同时权衡我要不要检索我要不要校验我要怎么排版注意力被稀释任何一环的指令遵循度都会下降。这就像让一个人同时当客服、当审核、当排版他不是做不了而是每件事都只能做到六十分。判断要不要拆的第一个信号就是你的提示词里出现了大量如果……就……否则……的分支逻辑。当分支超过三四个且每个分支对应的是不同性质的认知任务时就该考虑拆了。1.2 拆分的收益和代价必须一起算拆分带来的收益很直观每个智能体的提示词可以很短、很聚焦职责单一之后指令遵循度明显提升调试的时候也能单独定位是哪个环节出了问题。但代价同样真实而且往往被低估。第一个代价是延迟。单智能体一次调用能搞定的事拆成三个角色就是三次甚至更多次模型调用串行执行的话响应时间直接翻倍。第二个代价是信息损耗。角色之间传递的不是完整的上下文而是上一个角色加工过的摘要摘要写得不好下游角色就会缺信息。第三个代价是错误传播。上游角色判断错了下游角色基于错误前提继续干活最后错得很离谱而且很难回溯是哪一步开始歪的。所以我的经验是只有当单智能体的职责过载已经严重影响到稳定性且拆分后每个角色的边界足够清晰时拆分才划算。如果只是为了看起来架构先进而拆大概率是给自己找麻烦。1.3 一个实用的拆分判断清单在动手之前我一般会用下面这几个问题过一遍只要有三条以上命中就说明该拆了判断维度命中信号说明提示词长度超过 800 字且分支多单角色已经装不下任务性质混合了检索、推理、生成、校验认知模式差异大工具数量单个智能体挂了 5 个以上工具工具选择容易出错调试难度出错后无法定位是哪一步黑盒太大复用需求某段逻辑在多个流程里都要用适合抽成独立角色并行可能有多个子任务互不依赖拆分后能并行提速这张表不是硬标准但它能帮你把感觉该拆变成有依据地拆。我见过太多项目是拍脑袋拆的拆完发现角色之间来回甩锅最后又合回去了。2. 角色分工怎么切才是真的切干净了2.1 按认知模式切而不是按业务步骤切这是我最想强调的一点。很多人的第一反应是按业务流程切第一步做 A 的角色、第二步做 B 的角色、第三步做 C 的角色。这样切出来的角色本质上还是流水线上的工人每个角色都在做混合认知的工作只是把长流程切短了而已。更好的切法是按认知模式切。所谓认知模式指的是这个角色主要在做哪一类思考规划型拆解任务、决定下一步做什么比如 Planner检索型从外部获取信息比如 Retriever、Researcher推理型基于已有信息做判断和推导比如 Analyzer、Critic生成型产出最终内容比如 Writer、Formatter执行型调用工具产生副作用比如 Executor、Tool Runner同一个认知模式的角色提示词结构、工具集、输出格式都可以高度统一维护起来也轻松。而按业务步骤切出来的角色每个都混着好几种认知模式等于把单智能体的问题复制了 N 份。举个具体例子。做竞品分析报告这个任务按业务步骤切可能是收集资料的角色、分析资料的角色、写报告的角色。但收集资料里其实混了规划和检索分析资料里混了推理和校验。按认知模式切应该是Planner 负责拆出要分析哪几个维度Researcher 负责按维度去检索Critic 负责对检索结果做可信度判断Writer 负责成文。这样每个角色的提示词都能写得很短很准。2.2 角色数量控制在 3 到 5 个是甜点区角色不是越多越好。我实测下来3 到 5 个角色是大多数任务的甜点区。少于 3 个拆分收益不明显多于 5 个协调成本急剧上升而且角色之间的边界会开始模糊出现这个活到底该谁干的扯皮。如果任务确实复杂需要更多角色我的建议是分层而不是平铺。也就是先分成几个大组每个组内部再分角色。比如一个内容生产系统顶层是策划组生产组质检组生产组内部再分写手配图排版。这样每一层的角色数量都控制在可控范围内层与层之间通过明确的接口通信。角色数量上去之后还有一个隐形成本每个角色都要维护自己的提示词、工具集、输出 schema。五个角色就是五套配置改一个需求可能要动三四个角色的提示词回归测试的工作量是线性增长的。所以能少则少能用配置解决的不要用新角色解决。2.3 每个角色必须回答清楚的四个问题定义一个角色的时候我要求自己必须能回答清楚这四个问题答不清楚就说明这个角色还没设计好它的唯一职责是什么用一句话说清楚如果一句话说不清说明职责不单一它的输入是什么从哪个角色来、格式是什么、包含哪些字段它的输出是什么给哪个角色、格式是什么、必须包含哪些字段它什么时候不该被调用边界条件是什么什么情况下应该跳过它第四个问题最容易被忽略但恰恰最重要。比如一个 Critic 角色如果上游给的内容本身就是空的它该不该被调用如果调用了它输出什么这些边界不定义清楚流程跑到异常分支就会卡死或者产生垃圾输出。我一般会把这些定义写成一个角色卡片格式大概是这样role: researcher responsibility: 根据规划出的分析维度检索并整理原始资料 input: from: planner fields: [dimensions, constraints] output: to: critic fields: [raw_findings, sources, confidence] skip_when: - dimensions 为空 - 所有维度都已在缓存中命中这张卡片既是给团队看的文档也可以直接转成代码里的状态定义和路由条件一举两得。2.4 角色之间的信息接口比角色本身更重要我踩过最大的一个坑就是花了很多时间打磨每个角色的提示词却没认真设计角色之间的信息接口结果每个角色单看都挺好连起来就出问题。信息接口的核心是状态结构。在多智能体系统里角色之间不是直接对话而是通过一个共享的状态对象来传递信息。这个状态对象长什么样决定了整个系统的信息流转质量。以 LangGraph 为例状态通常是一个 TypedDict每个角色读取自己需要的字段写入自己负责的字段。设计状态的时候有几个原则字段命名要体现语义不要用data1、result这种含糊的名字用raw_findings、critique_notes这种一看就懂的每个字段要有明确的写入者避免多个角色往同一个字段写那样会互相覆盖保留原始信息不要过早做摘要让下游角色自己决定要不要压缩加上元信息比如confidence、source、timestamp方便下游判断可信度一个典型的状态定义大概长这样from typing import TypedDict, Annotated from operator import add class AgentState(TypedDict): task: str dimensions: list[str] raw_findings: Annotated[list, add] critique_notes: list[str] final_report: str next_step: str注意raw_findings用了Annotated[list, add]这是 LangGraph 里的 reducer 写法表示多个角色写入时做追加而不是覆盖。这种细节不处理好并行执行的时候数据就会丢。3. 协作机制角色之间到底怎么对话3.1 三种主流协作拓扑各有各的适用场景角色分好之后接下来要决定它们怎么连。我总结下来常用的有三种拓扑各有适用场景第一种是流水线式Pipeline。角色按固定顺序串行执行A 做完给 BB 做完给 C。这种最简单、最可控适合步骤明确、依赖关系强的任务。缺点是灵活性差中间某一步结果不理想也没法回头。第二种是主管式Supervisor。有一个主管角色负责调度它决定下一步该叫哪个下属角色干活下属干完把结果交回主管主管再决定下一步。这种灵活性高能根据中间结果动态调整路径是 LangGraph 里最常用的多智能体模式。缺点是主管角色本身可能成为瓶颈而且主管的判断质量直接决定整个系统的表现。第三种是群聊式Group Chat / Swarm。角色之间可以自由对话没有固定的调度者谁觉得该发言就发言。这种最灵活但也最难控制容易出现角色之间无限对话、跑题、或者互相等待的情况。适合探索性任务不适合生产环境。我的建议是生产环境优先用主管式简单任务用流水线群聊式只在实验阶段用。主管式兼顾了灵活性和可控性而且主管的调度逻辑可以单独调试和优化。3.2 主管式协作的调度逻辑怎么写才靠谱主管式的核心是主管角色的调度逻辑。这个逻辑可以很简单也可以很复杂取决于任务需要多细的控制。最基础的调度逻辑是让主管用模型来判断下一步。主管的提示词里列出所有下属角色的职责然后让它根据当前状态输出下一个该调用的角色名。这种方式的优点是灵活缺点是模型可能判断错而且每次调度都要花一次模型调用。更稳的做法是混合调度能用规则判断的用规则规则覆盖不到的再交给模型。比如如果 raw_findings 为空下一步一定是 researcher这种就是硬规则不需要模型判断而资料够了没有、要不要再补一轮检索这种模糊判断才交给模型。在 LangGraph 里主管的调度通常用一个条件边来实现def route_next(state: AgentState) - str: if not state[raw_findings]: return researcher if not state[critique_notes]: return critic if state[critique_notes] and not state[final_report]: return writer return end graph.add_conditional_edges(supervisor, route_next, { researcher: researcher, critic: critic, writer: writer, end: END, })这段代码里route_next就是调度逻辑。注意它先处理了硬规则字段为空的情况最后才返回 end。这种写法比纯模型调度稳定得多而且调试的时候一眼就能看出流程走向。3.3 交接时传什么摘要、原文还是引用角色之间交接的时候传什么内容是个大学问。传全文上下文会爆炸传摘要信息会损耗。我的经验是分层传递关键决策信息必须传原文比如用户的原始需求、硬性约束条件中间产物传结构化摘要比如检索结果传标题 关键结论 来源链接而不是整篇网页大块原始数据传引用比如把完整文档存到外部存储状态里只放一个 ID需要的时候再取这样既控制了上下文长度又保证了关键信息不丢。我见过一个项目所有角色之间都传完整上下文跑到第三个角色的时候上下文已经两万多 token成本和延迟都受不了而且模型开始忘记前面的关键约束。还有一个细节交接的时候要带上为什么。不只是传结果还要传这个结果是怎么来的、有什么不确定性。比如 Researcher 交给 Critic 的时候除了 findings还要带上哪些结论是强证据支持的、哪些是推测。这样 Critic 才能有针对性地做校验而不是从头再判断一遍。3.4 循环和终止条件多智能体最容易失控的地方多智能体系统最容易出问题的地方就是循环。角色 A 觉得信息不够让 B 再查一遍B 查完给 AA 还是觉得不够又让 B 查……如果没有终止条件这个循环能跑到天荒地老烧钱又烧时间。终止条件我一般设三层第一层是硬性步数上限。整个流程最多执行 N 步超过就强制结束。这个 N 根据任务复杂度设一般 10 到 20 步。这是最后的保险防止任何意外情况下的无限循环。第二层是角色调用次数上限。每个角色最多被调用 M 次比如 Researcher 最多查三轮。超过就说明这个任务在当前信息条件下做不下去了应该降级处理或者报错。第三层是质量收敛判断。让 Critic 角色判断当前结果是否已经达到可接受标准达到就结束。这一层最理想但也最依赖模型判断的准确性所以要和前两层配合使用。在 LangGraph 里这些终止条件通常写在条件边里或者用一个专门的计数器字段来跟踪。我习惯在状态里加一个step_count字段每次路由的时候检查一下def route_with_limit(state: AgentState) - str: if state.get(step_count, 0) 15: return force_end # 其他路由逻辑 ...这个force_end分支通常会接一个兜底角色负责把当前已有的信息整理成一个未完成的结果返回而不是直接报错。用户体验上一个不完美但有用的结果比一个错误提示要好得多。4. 用 LangGraph 落地状态、节点、边的具体写法4.1 为什么选 LangGraph 而不是自己写调度多智能体的调度逻辑理论上自己用 Python 写个 while 循环也能实现。但实际做下来LangGraph 这类框架帮你省掉的是状态管理和流程可视化这两块最烦的工作。自己写调度的话状态传递、条件分支、循环控制、断点续跑这些都要手写而且很难调试。LangGraph 把这些抽象成了状态图节点是角色边是流转关系条件边是分支逻辑。跑起来之后还能生成流程图一眼就能看出整个系统的结构排查问题的时候特别有用。另外 LangGraph 支持 checkpoint也就是把每一步的状态存下来。这意味着流程跑到一半崩了可以从上一个 checkpoint 恢复不用从头再来。对于长流程的多智能体系统这个特性太重要了。4.2 状态定义整个系统的共享内存前面提过状态定义的重要性这里展开讲具体怎么写。LangGraph 的状态本质上是一个 TypedDict每个节点读取和写入其中的字段。设计状态的时候我遵循几个原则原则一字段按角色划分归属。每个字段明确是哪个角色负责写的其他角色只读。这样避免并发写入冲突也让代码更容易理解。原则二用 reducer 处理需要累积的字段。像 findings 这种多个角色都可能追加内容的字段要用Annotated[list, add]声明 reducer否则后写的会覆盖先写的。原则三状态里不放超大对象。完整文档、图片这些放外部存储状态里只放引用。一个完整一点的状态定义示例from typing import TypedDict, Annotated, Literal from operator import add class ResearchState(TypedDict): # 输入 user_query: str constraints: list[str] # Planner 写入 plan: list[str] current_step: int # Researcher 写入 findings: Annotated[list[dict], add] # Critic 写入 critique: list[str] quality_score: float # Writer 写入 draft: str final_output: str # 全局控制 step_count: int next_action: Literal[plan, research, critique, write, end]这个状态定义里每个字段的归属都很清楚注释里标了谁写谁读。团队协作的时候这份定义就是接口文档。4.3 节点实现每个角色就是一个函数在 LangGraph 里每个角色对应一个节点函数。这个函数接收当前状态返回要更新的字段。写法上有个关键点节点函数只更新自己负责的字段不要动别人的字段。一个 Researcher 节点的实现大概是这样def researcher_node(state: ResearchState) - dict: plan state[plan] current state[current_step] if current len(plan): return {next_action: critique} dimension plan[current] # 调用检索工具 results search_tool.invoke(dimension) findings [{ dimension: dimension, content: results, source: web_search, confidence: 0.8 }] return { findings: findings, current_step: current 1, next_action: research # 继续下一个维度 }注意返回的字典里只包含这个节点负责的字段。findings因为有 reducer会被追加到已有列表里current_step是覆盖更新next_action用来告诉路由函数下一步去哪。这种写法的一个好处是每个节点都可以单独测试。给一个构造好的状态看它返回什么就能验证逻辑对不对不需要跑整个图。4.4 条件边把调度逻辑和角色实现解耦条件边是 LangGraph 里实现动态调度的关键。它接收当前状态返回下一个节点的名字。把调度逻辑放在条件边里而不是塞进节点函数里好处是调度逻辑可以集中管理、单独调试。一个典型的多角色路由函数def supervisor_route(state: ResearchState) - str: action state.get(next_action, plan) if state[step_count] 20: return force_end if action plan: return planner if action research: return researcher if action critique: return critic if action write: return writer return end然后在构建图的时候把这些边连起来from langgraph.graph import StateGraph, END workflow StateGraph(ResearchState) workflow.add_node(planner, planner_node) workflow.add_node(researcher, researcher_node) workflow.add_node(critic, critic_node) workflow.add_node(writer, writer_node) workflow.set_entry_point(planner) workflow.add_conditional_edges(planner, supervisor_route) workflow.add_conditional_edges(researcher, supervisor_route) workflow.add_conditional_edges(critic, supervisor_route) workflow.add_conditional_edges(writer, supervisor_route) app workflow.compile()这样每个角色节点执行完都会经过supervisor_route决定下一步。调度逻辑集中在一个函数里改起来方便也容易加日志。4.5 加一个人类介入的断点有些关键决策点让模型自己判断风险太大这时候可以加人类介入。LangGraph 支持在节点执行前或执行后中断等人类确认后再继续。比如在 Writer 生成最终报告之前先中断让人类看一眼草稿确认没问题再继续。实现上可以用interrupt_before参数app workflow.compile( checkpointermemory, interrupt_before[writer] )这样流程跑到 writer 之前会停下来人类可以通过 API 查看当前状态、修改字段、然后恢复执行。这个特性在需要审核的场景里特别有用比如内容发布、代码提交这些不能全自动的操作。5. 实测中那些文档不会告诉你的坑5.1 角色提示词里的隐性冲突多智能体系统里每个角色的提示词单独看都没问题但合起来可能互相打架。我遇到过一个典型案例Planner 的提示词里写尽可能拆出详细的步骤Critic 的提示词里写严格审查每一步的必要性。结果 Planner 拆出十个步骤Critic 每个都质疑流程在两者之间来回拉锯半天出不来结果。这类冲突的根源是角色目标没有对齐。解决办法是在设计阶段就把所有角色的提示词放在一起过一遍检查有没有互相矛盾的目标。更根本的办法是给整个系统定一个全局目标每个角色的提示词里都引用这个全局目标让它们知道自己是为什么服务的。我现在的做法是在状态里加一个global_goal字段每个角色的提示词开头都带上它。这样即使角色之间有分歧也能回到全局目标上做判断。5.2 上下文在角色间传递时的信息衰减前面提过交接要传摘要但摘要做不好会导致信息衰减。我实测过一个流程原始检索结果有五千字Researcher 摘要成五百字给 CriticCritic 再摘要成两百字给 Writer最后 Writer 拿到的信息已经面目全非写出来的东西跟原始资料对不上。解决这个问题的关键是区分决策信息和素材信息。决策信息比如这个维度资料充足可以层层摘要素材信息比如具体的引用、数据、案例必须原样传递不能摘要。我的做法是在状态里分开存这两类信息决策信息走摘要链素材信息走引用链。还有一个技巧是让下游角色能回查上游的原始数据。状态里存一个指向原始数据的引用下游角色需要细节的时候可以主动去取而不是被动接收摘要。这样既控制了默认上下文长度又保留了获取完整信息的能力。5.3 并行执行时的状态竞争当多个角色并行执行的时候状态写入会出问题。比如两个 Researcher 同时往findings里写如果没有 reducer后写的会覆盖先写的。LangGraph 里解决这个问题靠 reducer。Annotated[list, add]表示这个字段的更新方式是追加。但 reducer 只解决了列表追加的问题如果两个角色写的是同一个字典的不同 key还是可能冲突。我的经验是并行执行的节点尽量让它们写不同的字段。如果实在要写同一个字段用 reducer 并且确保 reducer 的逻辑是幂等的。另外并行节点的输出最好带上来源标识方便后续区分是谁写的。5.4 调试多智能体流程的正确姿势多智能体流程出问题的时候最难的是定位是哪一步出的错。我的调试方法分三步第一步看流程图。LangGraph 编译后可以生成流程图先看流程走向对不对有没有意外的循环或者跳转。第二步看状态快照。用 checkpoint 把每一步的状态存下来出问题的时候回放看是哪个字段从哪一步开始不对。第三步单节点复现。把出问题的节点单独拿出来用当时的状态跑一遍看它的输出是否符合预期。如果单节点没问题那就是交接或者调度的问题。这三步下来基本能定位到具体是哪个环节。我强烈建议在开发阶段就把 checkpoint 和详细日志打开虽然会慢一点但排查问题的时候能省大量时间。5.5 成本控制的几个实操手段多智能体系统的成本是单智能体的好几倍因为模型调用次数多了。控制成本有几个手段小模型干粗活规划、路由、简单判断这些用便宜的小模型只有生成和复杂推理用大模型缓存重复调用同样的输入不要重复调用模型加一层缓存限制循环次数前面说的三层终止条件本质上也是成本控制按需调用不是每个任务都要走完整流程简单任务走短路径我实测下来合理配置之后多智能体的成本能控制在单智能体的两到三倍而不是想象中的十倍。关键是把便宜的角色用便宜模型贵的角色只在必要时调用。6. 从能跑到好用多智能体系统的演进路径6.1 第一阶段跑通最小闭环刚开始不要追求完美先跑通一个最小的闭环。我的建议是从两个角色开始一个负责干活一个负责检查。比如 Researcher 加 CriticResearcher 查资料Critic 判断够不够。这个最小系统能跑通说明状态传递、条件路由、终止条件这些基础设施都没问题再往上加角色就顺了。这个阶段的目标是验证架构不是追求效果。提示词可以粗糙工具可以简单重点是流程能跑通、状态能正确传递、异常能正常处理。6.2 第二阶段优化每个角色的提示词架构跑通之后再逐个优化角色的提示词。这时候因为每个角色职责单一提示词可以写得很精准。优化的重点是输出格式的稳定性让每个角色的输出严格符合状态定义的 schema这样下游角色才能可靠地解析。我一般会在这个阶段给每个角色加上 few-shot 示例特别是输出格式的示例。多智能体系统里格式不对齐是导致流程失败的高频原因加几个示例能大幅降低出错率。6.3 第三阶段加监控和评估系统能稳定跑之后就要考虑怎么知道它跑得好不好。多智能体的评估比单智能体复杂因为要评估的维度多了每个角色的输出质量、角色之间的协作效率、整体任务的完成度。我的做法是分两层评估角色级评估看每个角色的输出是否符合预期系统级评估看最终结果的质量和整个流程的效率步数、耗时、成本。角色级评估用规则加模型判断系统级评估用人工抽检加自动指标。LangGraph 有专门的评估工具可以记录每次运行的轨迹方便做回归测试。每次改提示词或者改调度逻辑之后跑一遍评估集看指标有没有下降这是保证系统不退化的重要手段。6.4 第四阶段处理边界和异常生产环境里异常情况比正常情况还多。多智能体系统要处理的异常包括某个角色输出格式错误、工具调用失败、循环超限、状态字段缺失等等。我的处理原则是每一层都要有兜底。角色内部出错返回一个标记为失败的输出而不是抛异常调度层发现异常走降级路径最外层有全局超时和步数限制。这样即使某个环节出问题整个系统也能给出一个部分完成的结果而不是直接崩溃。兜底逻辑写起来不复杂但一定要在开发阶段就写好不要等到线上出问题才补。我见过太多项目是上线之后才发现异常处理没做结果一个小错误导致整个流程卡死。6.5 一个我常用的演进检查清单每次系统迭代之后我会用下面这个清单过一遍确保没有退化检查项合格标准检查方式角色边界每个角色职责能用一句话说清人工 review 角色卡片状态完整性每个字段都有明确写入者检查状态定义终止条件三层终止条件都生效构造超限场景测试异常兜底每个节点都有失败返回注入错误测试成本可控单次任务成本在预算内统计 token 消耗可观测性能回放任意一次运行检查 checkpoint 和日志这张表看起来简单但每次迭代都过一遍能避免很多低级问题。特别是状态完整性和终止条件这两项出问题的时候往往很隐蔽等到线上才发现就晚了。多智能体系统的设计说到底是在分工带来的清晰和协作带来的复杂之间找平衡。角色切得太粗等于没拆切得太细协调成本吃掉所有收益。我自己的体会是先把单智能体的职责过载问题想清楚再决定拆不拆、怎么拆比一上来就追求多智能体架构要靠谱得多。真正跑起来之后你会发现大部分问题不在模型能力上而在状态设计、交接协议、终止条件这些工程细节上把这些打磨好系统自然就稳了。
返回列表