多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:strllm=ChatOpenAI(model="gpt-4o",temperature=0)defsupervisor(state:TeamState)->dict:last_msg=state["messages"][-1].contentif"搜索"inlast_msgor"查找"inlast_msg:goto="researcher"elif"写代码"inlast_msgor"实现"inlast_msg:goto="coder"else:goto="writer"return{"next":goto}defresearcher(state:TeamState)->dict:response=llm.invoke(state["messages"]+[{"role":"user","content":"搜索相关资料"}])return{"messages":[response]}defcoder(state:TeamState)->dict:response=llm.invoke(state["messages"]+[{"role":"user","content":"编写代码"}])return{"messages":[response]}defwriter(state:TeamState)->dict:response=llm.invoke(state["messages"]+[{"role":"user","content":"撰写文案"}])return{"messages":[response]}defroute(state:TeamState)->str:last=state["messages"][-1].contentif"任务完成"inlast:returnENDreturnstate["next"]workflow=StateGraph(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")app=workflow.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:intllm=ChatOpenAI(model="gpt-4o",temperature=0)defplanner(state:CollabState)->dict:prompt=f"为以下任务制定执行计划\n任务:{state['task']}"response=llm.invoke([{"role":"user","content":prompt}])return{"plan":response.content}defexecutor(state:CollabState)->dict:prompt=f"按照计划执行\n计划:{state['plan']}"response=llm.invoke([{"role":"user","content":prompt}])return{"result":response.content}defreviewer(state:CollabState)->Command:prompt=f"审查结果是否合格\n结果:{state['result']}"response=llm.invoke([{"role":"user","content":prompt}])approved="合格"inresponse.content iteration=state.get("iteration",0)+1ifapprovedoriteration>=3:returnCommand(goto=END,update={"review":response.content,"approved":approved,"iteration":iteration})returnCommand(goto="executor",update={"review":response.content,"iteration":iteration},)graph=StateGraph(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")app=graph.compile()result=app.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应用。