做多代理Multi-Agent系统最头疼的往往不是单个Agent的Prompt写不好而是多个Agent一起跑的时候怎么协作不失控。我最初做LangGraph多代理时用过一种“野生”方案建一个中心节点把所有人的上下文塞到一起靠一个LLM在聊天记录里“悟”出下一步该调用谁结果上下文越来越长输出越来越飘最后整条链路完全不可控。后来切到LangGraph的Supervisor模式才真正把多代理的编排问题理顺——它不搞玄学而是用图结构把“决策”和“执行”拆开让LLM只负责出决策路由逻辑交给图本身去完成。这篇文章我会从原理到代码把Supervisor模式的完整链路讲一遍为什么多代理需要Supervisor、它在LangGraph里到底是怎么运作的、以及一份可以直接抄作业的实现代码。适合正在用LangChain/LangGraph搭多代理但对“到底怎么调度多个Agent”还比较模糊的同学。看完你会发现Supervisor并没有多神秘本质就是给图加了一个带结构性输出的“调度员”节点。1. 多代理架构是怎样一步步走到Supervisor的1.1 单代理的瓶颈在哪里先说清楚一个前提为什么一开始要上多代理单代理不是不能干活但在真实场景里你会遇到几个绕不开的墙。第一是上下文窗口的墙。一个Agent如果同时要处理检索、计算、写文档、做总结Prompt必然越拼越长。长Prompt不仅贵而且模型注意力会涣散越靠后的工具结果越容易被忽略。我试过把检索到的十几段资料全部塞进一个Agent的上下文结果它写到一半开始自由发挥引用了根本不存在的来源。第二是职责边界。一个Agent既要有创造力又要精确计算这本身就是互相打架的目标。你让同一个模型既当会计又当诗人它大概率两个都干不好。第三是故障隔离。单代理链路里只要某一个环节出错整个任务就崩了你也没有办法单独去替换或调试“某一种能力”。多代理的思路就是拆模块。一个Agent只干一件事干好一件彼此通过状态和消息协作。听起来很美但拆开之后紧接着就是一道送命题谁来调度1.2 多代理的核心矛盾没有主编的编辑部会怎样你可以把多代理想象成一个编辑部。有记者、有编辑、有校对每个人能力都OK但如果没有人决定“现在该让谁写稿、稿子写完给谁审”大家就会一起卡在同一个环节或者互相改来改去形成死循环。多代理的核心问题就三个路由下一步该谁上、状态共享上一个人产出的东西怎么传给下一个人、终止条件什么时候活算干完了。这三个问题不解决多代理就是空壳。早期有人直接在LangChain的Agent外层套一个循环靠AgentExecutor反复调用但这种方式有两个致命缺陷一是路由决策完全隐藏在LLM的thought/action输出里你没法精确干预二是状态管理很弱所有历史消息一股脑扔给模型上一轮谁说了什么根本分不清。LangGraph能解决这个问题是因为它把多代理变成了一张有向图。每个Agent是一个节点节点之间用边连接边的走向可以由代码条件控制也可以由LLM动态决定。节点与节点之间通过一个共享的State对象传数据这样你就有了明确的“主编”——一个专门负责路由的Supervisor节点。1.3 常见的多代理组织模式对比在LangGraph里多代理的组织模式一般分三种。我把它们的核心逻辑和使用场景整理成了一张表方便对比。模式结构优点缺点典型场景Network网络式所有Agent两两互相可见自由通信信息流通充分适合头脑风暴状态图混乱容易死循环调试难小规模协作探索型任务Supervisor中心化所有子Agent只和Supervisor通信路由清晰状态集中可控性强所有决策都在Supervisor可能成为瓶颈大多数实际业务凡是需要明确调度顺序的Hierarchical层级式多个Supervisor组成树状上层管下层可扩展性强适合超大规模实现复杂状态穿透困难大型系统团队/部门级多代理从我的经验来看大多数项目从Supervisor模式起步是性价比最高的。因为它把“决策”收拢到一个节点里出问题你只需要盯住Supervisor一个节点而不是去跟踪一堆Agent之间的网状对话。Hierarchical模式本质上就是Supervisor的递归套娃等系统规模大到单个Supervisor的上下文塞不下时再升级也不迟。这里顺便回应一个热词LangGraph和LangChain到底什么区别。LangChain提供的是组件模型封装、工具、检索器AgentExecutor那套在老版本里也能做循环但图状态和条件分支能力很弱LangGraph才是真正把 LangChain 的世界和“状态机”结合起来的产物强调节点、边、状态、检查点这些图论概念。你完全可以把LangGraph理解为“给LangChain补上了流程控制这一课”。2. Supervisor原理深度拆解2.1 Supervisor的本质是路由器不是人很多人一听“Supervisor”以为它应该是个更聪明的LLM能回答所有问题。大错特错。Supervisor的核心定位是一台“路由分发器”它不负责具体干活只负责从当前状态里判断“下一步应该调用哪个子代理”。在LangGraph里Supervisor通常是一个普通节点但它返回的不是自然语言回答而是一个结构化指令。最常用的结构就是一个字段next表示下一个要执行的节点名称。比如{next: researcher}就表示“接下来让研究员上场”{next: FINISH}表示“任务结束”。这种设计有两大好处。第一路由逻辑可以被程序直接读取不会因为LLM偶尔多说了几句废话而崩掉。第二你可以用代码强制约束路由空间。比如用Literal类型把next限制成只能取“researcher”、“writer”、“FINISH”中的一种从根上杜绝模型自己编节点名。我见过有人让Supervisor自由输出文本然后靠字符串匹配决定下一个节点结果模型输出个“Research Agent”导致完全匹配不上白白浪费一次调用。结构化输出就是把这种不确定性提前扼杀掉。2.2 状态与消息传递所有Agent都在写同一份会议纪要Supervisor要正确路由前提是它能看清全局状态。在LangGraph里状态State是一个可共享的字典类型一般用TypedDict定义。每个节点执行完后返回一个字典LangGraph会把这字典里的字段合并到全局State里。关键点是消息的合并方式。如果你直接定义state[messages]为一个普通list那么每个节点返回的新消息会直接覆盖掉旧消息——这显然不行多代理需要不断地把新产出追加到会话历史里。LangGraph对此提供了reducer机制最常见的是add_messages它能把新消息和旧消息合并成一个列表。class AgentState(TypedDict): messages: Annotated[list, add_messages] next: str这段代码的意思是messages字段每一次节点返回后都会自动做“追加合并”而不仅仅是覆盖next字段是普通更新每次由Supervisor覆盖。你可以把这想象成一份会议纪要每个子代理开了个小会说了一段话说完后把发言记录贴到会议纪要末尾Supervisor读纪要决定下一个谁发言下一个Agent发言完毕后又把纪要末尾追加。整个过程顺序推进谁也不会丢上下文。2.3 条件边与send函数一个管“下一步”一个管“分叉”Supervisor的路由在图上是用条件边实现的。你可以给某个节点加一个条件函数函数的返回值决定走哪条边。比如def route_next(state): if state[next] FINISH: return END return state[next]这个函数读Supervisor写入的next字段返回节点名字或END。用add_conditional_edges挂到Supervisor的边后面。这种模式适合“逐个轮流执行”的场景——每个时刻只有一个节点在跑。但多代理还有一种场景一个任务需要拆成多个独立子任务并行执行。这时候条件边就不够用了因为条件边只能选一个方向走。LangGraph提供了一个更底层的原语send(node_name, state)。它和普通边的区别在于不经过父节点直接把一个独立的State实例发送给目标节点你可以循环调用多次Send从而在同一轮里扇出N个并行子代理。from langgraph.types import Send def fan_out(state): tasks state[tasks] return [ Send(process_node, {task: t}) for t in tasks ]这个模式俗称Map-Reduce它和Supervisor并不冲突。Supervisor管的是“路由到不同能力的Agent”send管的是“对同一批数据批量分发”。我在做批量审核时就常用send把上千条文本切成片段分给多个Agent并行处理再汇总结果。如果你一直没搞懂send(node_name, state)记住一句话它就像一个复印机可以把同样一份工作单复制成多份送给多个工人同时开工。3. 代码实践从零实现一个Supervisor系统下面进入正题我们用LangGraph写一个能跑的Supervisor多代理示例。这个示例的业务场景是“写一篇科普短文”研究员Agent负责搜集并整理资料作家Agent负责把资料润色成文Supervisor负责判断当前该让谁干活以及什么时候收工。3.1 环境准备与依赖安装先安装依赖注意版本别太老。我这边测试用的是 langgraph 0.2.x 和 langchain-openai 0.1.x太老的版本API差异比较大建议直接装最新。pip install langgraph langchain-openai python-dotenv然后设置环境变量。我习惯在项目根目录放一个.env文件OPENAI_API_KEYsk-xxx用python-dotenv加载。这里要提醒一句不要硬编码密钥我见过有人把API Key直接写进代码接着往GitHub上一推十几分钟就被别人刷了几百美元的费用。3.2 定义子代理节点Researcher和Writer为了不依赖额外的工具调用每个子代理节点我会直接调用LLM但给不同的system提示词。这样既能展示多代理协作又不会让示例代码膨胀到不可读。import os from typing import TypedDict, Annotated, Literal from dotenv import load_dotenv from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage load_dotenv() llm ChatOpenAI(modelgpt-4o-mini, temperature0) class AgentState(TypedDict): messages: Annotated[list, add_messages] next: str def researcher_node(state): system_prompt 你是一位严谨的研究员。请根据已有信息尽量客观地列出关键事实和数据不要做过多修饰控制在200字以内。 user_input state[messages][-1].content resp llm.invoke([ {role: system, content: system_prompt}, {role: user, content: f请研究所要写的主题{user_input}} ]) return {messages: [AIMessage(contentf[研究员产出] {resp.content})]} def writer_node(state): system_prompt 你是一位科普作家。请基于已有研究材料写出一篇结构清晰、通俗易懂的短文控制在400字以内。 # 把除最后一条外的研究材料都送进去 history_text \n.join(m.content for m in state[messages] if isinstance(m, AIMessage)) resp llm.invoke([ {role: system, content: system_prompt}, {role: user, content: f研究材料如下\n{history_text}\n请成文。} ]) return {messages: [AIMessage(contentf[作家产出] {resp.content})]}你可能会问为什么子代理节点返回时要把结果包一层AIMessage因为我们在State里用了add_messages它会自动把AIMessage追加到消息列表末尾这样Supervisor才能在下一轮看到历史记录。如果你直接返回一个纯字符串add_messages会因类型不对而报错这是LangGraph新手最常见的坑之一。3.3 定义Supervisor节点让LLM做路由决策Supervisor节点不能直接让LLM自由发挥必须用结构化输出把路由空间限定死。我用with_structured_output让模型输出一个固定结构的字典next只能是researcher、writer、FINISH三者之一。from typing_extensions import TypedDict class RouteResponse(TypedDict): next: Literal[researcher, writer, FINISH] supervisor_llm llm.with_structured_output(RouteResponse) def supervisor_node(state): system_prompt ( 你是一个多代理系统的Supervisor。你的职责是判断下一步应该由哪个子代理工作。\n 可选节点\n - researcher负责查找和整理事实材料。当内容缺少信息时优先调用它。\n - writer负责撰写、润色最终文章。\n - FINISH当最终文章已经完成或者任务已经达成时输出FINISH。\n 决策规则\n 1. 如果还没有完成研究的材料请选择researcher。\n 2. 如果有足够材料但还没有成文请选择writer。\n 3. 如果writer已经产出了最终文章输出FINISH。\n ) messages state[messages] message_list [{role: system, content: system_prompt}] for m in messages: if isinstance(m, HumanMessage): message_list.append({role: user, content: m.content}) elif isinstance(m, AIMessage): message_list.append({role: assistant, content: m.content}) resp supervisor_llm.invoke(message_list) return {next: resp[next]}这里有两个细节决定成败。第一system prompt里必须给每个子代理一个“能力说明”和一个“触发条件”。Supervisor本质上还是LLM不是真正的代码调度器你不把每个Agent的能力边界讲清楚它就会乱点兵。第二不要一上来就多加节点我建议先给两三个子代理跑通再扩。节点一多LLM的决策准确率会明显下降这时候与其堆更多提示词不如考虑分层Supervisor。3.4 组装图并运行完整流程定义好三个节点后剩下的工作就是把它们用边连起来形成一个循环Supervisor - 子代理 - Supervisor - 子代理 - ... - 直到输出FINISH。from langgraph.graph import StateGraph, START, END def route_next(state): if state[next] FINISH: return END return state[next] builder StateGraph(AgentState) builder.add_node(supervisor, supervisor_node) builder.add_node(researcher, researcher_node) builder.add_node(writer, writer_node) builder.add_edge(START, supervisor) builder.add_conditional_edges( supervisor, route_next, { researcher: researcher, writer: writer, END: END, } ) builder.add_edge(researcher, supervisor) builder.add_edge(writer, supervisor) graph builder.compile()然后我们就可以直接调用图了。result graph.invoke({ messages: [HumanMessage(content请写一篇关于太阳能电池原理的科普短文)] }) for m in result[messages]: print(type(m).__name__, :, m.content) print(---)你会看到整个执行过程大概是这样的Supervisor先做第一次判断通常会让researcher先干研究完回到SupervisorSupervisor看到有了材料但没有成文会转给writerwriter写完回来后Supervisor看到消息里已经包含“作家产出”结尾的内容决定输出FINISH。整套流程不需要你在外部写任何循环LangGraph会在图上自动走过来。这里顺便说一下我建议你在跑真实验证前先把这个例子跑通。我经常看到有人一上来就搬官方复杂的多代理工具箱prebuilt create_agent遇到问题反而搞不清是哪一层出错。自己手写一遍原语才能真正理解LangGraph的数据流和路由机制。4. 常见问题与排查技巧实录4.1 结构化输出返回格式出错使用with_structured_output时偶尔会遇到模型返回的JSON缺字段或者枚举值超出了Literal限制。我碰到最多的是旧版langchain中使用pydantic v1和v2不兼容导致校验一直报错。解决方案是固定langchain-core版本或者把模型换成gpt-4o级别以上的模型。还有一个土办法在Prompt里写死“只输出一个JSON对象字段为next只能取值researcher、writer、FINISH”然后用JsonOutputParser自己解析虽然麻烦点但可控性很强。4.2 子代理拿到的是最新消息而不是全局上下文这个问题非常隐蔽。researcher节点如果只读state[messages][-1]拿到的是Supervisor上一轮输出的东西可能是路由结果也可能是用户输入未必是子代理真正需要的信息。我在做writer节点时特意把历史AIMessage全部拼接起来作为上下文就是防止这个问题。如果你想更精细控制可以把每个子代理“需要参考的字段”单独写到State里不要统一放在messages里比如加一个research_notes字段只有researcher能更新writer和Supervisor只读。4.3 任务一直不结束陷入死循环这是Supervisor模式最经典的问题。原因通常很单一Supervisor的Prompt没有定义清楚“什么时候算完”。例如没有writer这条路径整个图只剩researcher那模型就永远在“研究-研究-研究”。我处理这个问题有两个手段第一是在系统Prompt里写“如果重复调用同一个子代理超过两次请输出FINISH”第二是在图编译时加执行上限LangGraph的invoke可以传config{recursion_limit: 10}超过10步自动抛异常避免烧钱跑飞。try: result graph.invoke(initial_state, config{recursion_limit: 10}) except Exception as e: print(执行次数超限请检查路由逻辑, e)4.4 上下文越来越长最后Prompt超限多代理跑几轮后messages列表会积累大量中间产物。Supervisor每次路由都要读一遍全部历史成本很高也容易超上下文窗口。解决办法我常用两条路径一是用langchain_core.messages里的trim_messages按最新N条或最大token数裁剪消息把研究过程的冗长内容转成摘要再传下去二是把“中间产物”放在独立的state字段里不塞进最终给LLM看的messages只保留必要结果。后者更干净代码也更清晰但需要你一开始就设计好状态结构。4.5 常见问题速查表症状可能原因解决办法Supervisor总是选同一个子代理Prompt没有定义“完成条件”增加更明确的触发规则加入次数限制模型输出next是“文学创作”没限制Literal取值范围用with_structured_output并指定Literal枚举子代理返回string导致报错add_messages只接受消息对象返回AIMessage(content...)状态覆盖而不是追加没有用Annotated reducermessages字段加Annotated[list, add_messages]长任务跑到一半超限recursion_limit太小适当调大或用循环内提前结束策略不同子代理互相串台所有节点共用一个messages历史增加独立字段按节点隔离数据5. 从Supervisor继续向前的三个方向代码跑通只是第一步。我在实际项目里踩过不少坑也积累了一些经验最后分享三个值得继续深挖的方向。**第一给图加检查点持久化。**LangGraph的MemorySaver或PostgresSaver可以保存每一步的状态让系统能中断、恢复、甚至在失败后从断点续跑。多代理一旦超过五六步任何一步网络抖动都可能导致前面全白跑检查点机制能让你不用重新生成历史结果。**第二分层Supervisor。**不要让一个Supervisor一次性管十几个子代理。推理模型的决策精度会随着候选集变大而降得很快。我更建议把大系统拆成多个“小组”每个小组有自己的Supervisor上层还有一个总Supervisor就形成了层级结构。虽然代码变复杂但每个Supervisor只需要在五六个选项里做选择准确率会明显高很多。**第三自己做一套可视化调试工具。**LangGraph自带的graph.get_graph().draw_mermaid_png()可以生图但到多代理复杂循环时图会乱成一团。我更推荐在每次invoke之后打印每个节点进出时的状态快照自己log一份“决策轨迹”。这样你才能在Supervisor做出错误路由时快速定位是Prompt问题还是状态数据问题。多代理的Supervisor模式归根到底还是一句话把决策权和执行权分开让图结构来控制流程让LLM只做判断题。这个思路看着简单但一旦跑通了你搭任何复杂Agent系统都会觉得顺手很多。