多Agent协作架构用LangGraph构建多智能体系统上一篇讲了条件分支和循环。单个Agent能做的事有限任务一复杂就容易翻车。这一篇聊多Agent协作看看怎么把多个Agent组织起来一起干活。为什么要用多Agent我之前做过一个需求一个Agent又要查资料、又要写代码、还要做审查。prompt塞了一大堆角色描述模型经常搞混该查资料的时候开始写代码该审查的时候又跑去查资料。调了好几天prompt都不太行。后来拆成三个Agent各管一摊效果好了一大截。单个Agent像一个人什么都干容易顾此失彼。多个Agent分工以后每个只管自己擅长的事整体表现稳得多。LangGraph做这件事有天然优势。它本身就是图结构加个Agent就是加个节点连接关系随便改组织起来很灵活。不过也别动不动就上多Agent。简单任务一个Agent就能搞定拆成多个反而增加复杂度调试都费劲。判断标准很简单如果一个Agent的prompt里要塞三种以上不同的角色指令就考虑拆分。Supervisor模式最常用的多Agent架构是Supervisor模式。一个主管Agent负责分配任务下面的工作Agent各自干活。主管看完任务决定交给谁等那个Agent干完汇报回来再决定下一步。来看代码。fromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.graph.messageimportadd_messagesfromtypingimportAnnotated,TypedDictfromlangchain_openaiimportChatOpenAIclassTeamState(TypedDict):messages:Annotated[list,add_messages]next:strllmChatOpenAI(modelgpt-4o,temperature0)defsupervisor(state:TeamState)-dict:last_msgstate[messages][-1].contentif搜索inlast_msgor查找inlast_msg:gotoresearcherelif写代码inlast_msgor实现inlast_msg:gotocoderelse:gotowriterreturn{next:goto}defresearcher(state:TeamState)-dict:responsellm.invoke(state[messages][{role:user,content:搜索相关资料}])return{messages:[response]}defcoder(state:TeamState)-dict:responsellm.invoke(state[messages][{role:user,content:编写代码}])return{messages:[response]}defwriter(state:TeamState)-dict:responsellm.invoke(state[messages][{role:user,content:撰写文案}])return{messages:[response]}defroute(state:TeamState)-str:laststate[messages][-1].contentif任务完成inlast:returnENDreturnstate[next]workflowStateGraph(TeamState)workflow.add_node(supervisor,supervisor)workflow.add_node(researcher,researcher)workflow.add_node(coder,coder)workflow.add_node(writer,writer)workflow.add_edge(START,supervisor)workflow.add_conditional_edges(supervisor,route)fornodein[researcher,coder,writer]:workflow.add_edge(node,supervisor)appworkflow.compile()每个工作Agent干完活消息回到supervisor。supervisor根据最新消息决定下一步去哪。任务在Agent之间传递直到supervisor认为可以结束。踩过一个坑。supervisor的路由逻辑只看关键词用户问帮我搜索一下怎么写代码supervisor看到搜索就交给researcher但用户实际想要代码。后来改成让模型自己判断该交给谁准确率高了很多。LangGraph官方也有个langgraph-supervisor包封装了这套逻辑不用自己手写路由。Hierarchical模式任务特别多的时候一个supervisor管不过来。这时候可以用Hierarchical模式supervisor上面再套supervisor。顶层supervisor管几个中层supervisor每个中层supervisor再管几个工作Agent像公司组织架构一样层层往下。比如做内容生产顶层supervisor分管研究和写作两组。研究组管搜索Agent和分析Agent写作组管初稿Agent和润色Agent。顶层把任务分给对应的组组内supervisor再具体分配。这种模式适合任务能清晰分层的场景。任务类型不多的话一层supervisor就够了别为了显得复杂硬套多层结构。我见过有人三个Agent也要搞两层supervisor纯属给自己找麻烦。层级越多调试越痛苦出了问题你得一层层往上查哪个supervisor路由错了都不好定位。Agent之间怎么传递消息多Agent系统里最关键的问题是信息怎么在Agent之间流动。LangGraph的做法是通过共享状态。最常见的是用消息列表。所有Agent读写同一个messages字段用add_messages reducer自动追加。每个Agent干完活把结果追加进去下一个Agent就能看到前面所有人的输出。也可以用专门的字段传递结构化数据。比如planner输出任务清单存到plan字段executor从plan字段读取来执行。这比把所有信息塞进消息列表里清晰得多。有个坑要注意。Agent多了以后共享状态会膨胀每个Agent往里塞消息跑几轮下来messages列表就很长token消耗上去了模型还可能被无关信息干扰。解决办法是定期做摘要压缩或者让每个Agent只读自己需要的那部分状态别把整个历史都喂给它。fromtypingimportAnnotated,TypedDictfromlanggraph.graph.messageimportadd_messagesclassCollabState(TypedDict):messages:Annotated[list,add_messages]plan:strresult:strreview:striteration:int完整示例规划加执行加审查把前面的东西串起来。三个Agent协作规划Agent拆解任务执行Agent干活审查Agent检查质量。审查不过就打回去重做。fromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.graph.messageimportadd_messagesfromlanggraph.typesimportCommandfromtypingimportAnnotated,TypedDictfromlangchain_openaiimportChatOpenAIclassCollabState(TypedDict):messages:Annotated[list,add_messages]task:strplan:strresult:strreview:strapproved:booliteration:intllmChatOpenAI(modelgpt-4o,temperature0)defplanner(state:CollabState)-dict:promptf为以下任务制定执行计划\n任务:{state[task]}responsellm.invoke([{role:user,content:prompt}])return{plan:response.content}defexecutor(state:CollabState)-dict:promptf按照计划执行\n计划:{state[plan]}responsellm.invoke([{role:user,content:prompt}])return{result:response.content}defreviewer(state:CollabState)-Command:promptf审查结果是否合格\n结果:{state[result]}responsellm.invoke([{role:user,content:prompt}])approved合格inresponse.content iterationstate.get(iteration,0)1ifapprovedoriteration3:returnCommand(gotoEND,update{review:response.content,approved:approved,iteration:iteration})returnCommand(gotoexecutor,update{review:response.content,iteration:iteration},)graphStateGraph(CollabState)graph.add_node(planner,planner)graph.add_node(executor,executor)graph.add_node(reviewer,reviewer)graph.add_edge(START,planner)graph.add_edge(planner,executor)graph.add_edge(executor,reviewer)appgraph.compile()resultapp.invoke({task:写一篇Python异步编程的技术文章,iteration:0},config{recursion_limit:15},)print(f审查通过:{result.get(approved)})print(f迭代次数:{result.get(iteration)})流程很清楚。planner出计划executor按计划执行reviewer审查结果。审查通过就结束没通过就回executor重做最多重试三次。实际跑起来有个问题要注意。reviewer的判断逻辑要写好。我一开始用合格这个词来判断结果模型有时候说基本合格但不完美没包含合格两个字就误判成不通过。后来改成让模型输出JSON里面放个布尔字段表示是否通过稳定多了。写在最后多Agent协作的核心就这些。选好架构模式设计好状态结构让信息在Agent之间顺畅流动。Supervisor模式适合大多数场景任务特别复杂再考虑分层。下一篇开始LangGraph项目实战把前面学的状态管理、分支循环、多Agent协作全部串到一个完整项目里从零搭建一个能跑的Agent应用。
