LangGraph实战：多智能体分层团队与主管监督模式设计

做多智能体系统也有小两年了从最初的硬编码调度到接入 LangGraph 后再也没回去手搓状态机这个框架确实把以前很多绕弯子的写法给捋直了。今天就拿我最近在做的“分层团队 主管监督”这套协作模式聊聊我在实战中的设计与踩坑记录。如果你正在规划多智能体MAS协作或者已经上手了 LangGraph 但困在怎么编排多个智能体这篇文章应该能给你一些直接能抄作业的参考。先说结论LangGraph 真正解决的不是“怎么让智能体说话”而是“怎么让智能体协作”。而协作的关键在于状态设计和控制流编排。分层团队与主管监督模式就是把一群人合作干活的组织方式搬进代码我用了一个模拟项目 X内部代号“协作中枢”完整跑通了这套方案下面全程围绕这份实战经验展开。1. 整体设计与思路拆解1.1 为什么需要多智能体协作而不是一个超级智能体很多人一开始的思路是“把所有功能塞进一个提示词里让一个智能体把所有活都干了”。这个方案在 demo 阶段很香但一旦进入真实业务场景马上会碰到三个让人头疼的问题上下文窗口撑不住。让一个智能体同时处理“用户意图识别 → 信息检索 → 业务推理 → 结果格式化 → 情绪安抚”这一整套链路提示词越叠越长输出质量越来越不稳定偶尔还会丢中间步骤。职责边界模糊返工率高。一个智能体既要懂技术又要懂话术一旦中间某个环节出现偏差你很难定位到底是在哪一步出了问题。可维护性差。改一个模块的逻辑得把整个提示词捞出来重新捋一遍而且极容易影响其他模块的表现。多智能体协作本质上是把“一个全能的智能体”拆成“一组各司其职的智能体”每个智能体只负责一块相对独立的子任务再由协调层决定任务的分配、流转和汇总。这件事的好处不言而喻每个智能体都能把提示词做深做专出问题可单独调试新业务接入时也能按需增删成员。1.2 LangGraph 的编排单位节点、边与状态图在动手设计之前先对齐一个基础认知LangGraph 的一切都是围绕“状态图”来的。你定义一个状态的数据结构然后建一批节点Node节点之间用边Edge连接数据在节点间以状态State的形式传递。它的运行方式非常像流水线每个节点处理完状态后返回增量更新图引擎负责调度下一个节点。这个设计最让我喜欢的一点是所有控制流都是显式的。比如你有一个分支判断不是写在某个智能体的提示词里让“它自己看着办”而是通过条件边Conditional Edge在引擎层面做路由。这种“控制权在框架、业务权在智能体”的隔离正是多智能体协作能够稳定跑起来的基础。1.3 分层团队与主管监督模式的基本结构这套协作模式借鉴了公司组织架构里的“项目管理”思路。整体分两层主管层负责接收任务、拆解任务、分派任务、汇总结果并且对中间产物做质量把控。执行层一组具备特定技能的专家智能体研究员、代码开发、文档撰写、测试验证等各自执行子任务并回报结果。主管监督模式的精妙之处在于“监督”两个字。它不是简单的分派后等着收结果而是会在每个关键节点审视产出条件不满足就驳回重做满足则继续流转。所以整体控制流是主管先判断任务类型路由给对应执行者执行者干完后把产物交回主管主管审核通过则进入下一步不通过则退回重做。这套模式和“平级自治模式”相比胜在可控和“单一主管全权代理模式”相比胜在均衡。项目里多个智能体同时协作时有一个明确的监督角色能让系统的行为可预期、可追踪、可干预这也是我为什么在实战里首选这套方案。2. 核心细节解析与实操要点2.1 状态结构的统一设计多智能体协作的第一个硬门槛是状态结构的设计。这不是随便定义一个 dict 就完事我们要考虑的是这个状态要能被多个智能体共用、能被图引擎正确识别增量更新、能保留运行过程中的历史足迹。我常用的状态定义是这样设计的from typing import TypedDict, Annotated import operator class MASState(TypedDict): task: str task_type: str plan: list subtask_results: Annotated[dict, operator.or_] current_agent: str final_answer: str history: Annotated[list, operator.add]这里我先说明几个关键字段的设计意图task保存原始任务描述所有智能体都能读取保证上下文一致。task_type是主管的第一个产出用于路由分发。把它单独拎出来是为了让后续条件边判断时不必再做语义解析。subtask_results用operator.or_注解表示各智能体写入的结果按 key 合并而不是互相覆盖。这是 LangGraph 里一个非常重要的细节不同的智能体执行不同的子任务结果放在同一个字典的不同 key 下。history用operator.add注解每次追加操作都会累积成列表方便调试和追踪调用链路。注意Annotated的 reducer 函数决定节点返回的增量如何合并到全局状态。如果拿不准尽量用operator.or_处理字典、operator.add处理列表避免踩到覆盖更新的坑。2.2 主管智能体的职责边界与提示词设计主管是整个协作模式的核心节点但它不是万能的。我给它设定的职责只有四件事拆解任务、分派任务、审核结果、决定下一步。禁止它亲自下场写具体内容。在实践中我把主管的提示词按这四个职责逐一写清楚并且加了严格的输出格式约束。主管的每次输出都要求是一个结构化的 JSON这样后续解析和条件判断都极其顺畅。下面是一段模板参考SUPERVISOR_PROMPT 你是整个团队的主管。你的职责是 1. 分析用户任务判断任务类型 2. 将任务拆解为可执行的子任务 3. 选择合适的执行者来完成每个子任务 4. 审核执行者提交的成果决定是否通过或退回重做。 可用的执行者 - researcher负责信息检索、资料收集与验证 - coder负责代码编写与调试 - writer负责文档撰写与技术说明。 输出格式JSON { thought: 你的推理过程, task_type: research | coding | mixed | other, plan: [子任务1, 子任务2, ...], assignments: [{agent: researcher, task: 具体任务描述}], review: {approved: true/false, comment: 审核意见或下一步指示} } 这段设计的要点在于“把流程从提示词里剥出来”。主管不需要自己思考“我该按什么顺序干活”它的思考被框架的边和路由给定死了它只需要在给定的几步里做选择题和判断题。实测下来这种设计极大提升了系统的稳定性因为业务逻辑在代码里而不是语言模型随机的推理路径里。2.3 执行智能体的接入方式共用一套工具还是独立工具集执行层的智能体设计相对简单但这里有一个容易纠结的点每个执行智能体要不要配一套独立的工具集我的建议是看工具是否和智能体职责强绑定。对于一个研究员智能体它需要检索工具对于一个代码开发智能体它需要代码执行工具。工具如果被多个不同的智能体共用那就抽出来作为共享工具如果某个工具只服务于某个智能体的能力域就直接绑定在智能体内部。为了避免主管在分派时“乱点鸳鸯谱”我还给每个执行智能体加了一个capabilities描述挂在主管提示词的可用执行者列表里。这样一来主管在做任务分配时是基于显式能力描述来选择而不是靠猜。RESEARCHER_AGENT { name: researcher, capabilities: 信息检索与资料验证, tools: [search_tool, web_fetch_tool] } CODER_AGENT { name: coder, capabilities: 代码编写与运行调试, tools: [python_repl_tool] }用这种方式组织后续新加一个智能体时只需要在能力清单里追加一行主管的路由逻辑不需要动扩展成本非常低。2.4 条件边的路由策略避免让大模型做他不擅长的事LangGraph 的条件边既是最大的利器也是最大的坑。新手最容易犯的错误是让条件边调用语言模型去判断该走哪条路。举个例子有人会写一个“意图分类器节点”直接调用模型判断当前任务类型然后根据返回结果走不同分支。这个做法不是不行但你要承担模型判断不稳定的风险。在我的实战方案里路由的依据不是模型输出而是主管节点返回的结构化 JSON 中的字段。主管返回task_type之后条件边直接读取该字段进行 switch完全不依赖二次模型判断。这样做的理由是主管已经完成了语义理解再把语义理解的结果转成结构化数据接下来交给代码做确定性的路由每一步都是可控的。下面是条件边定义的关键代码def route_after_supervisor(state: MASState) - str: if state[task_type] research: return researcher elif state[task_type] coding: return coder elif state[task_type] mixed: return coordinator # 混合任务走协调节点 else: return writer提示路由函数里尽量只做纯逻辑判断不调用模型、不读取复杂语义。把有歧义的判断放在主管节点里完成把确定性的分发放在边里完成整套系统能稳定好几个量级。3. 实操过程与核心环节实现3.1 构建最小可用的主管-执行者图先给你一个最小可用的实现这段代码我建议你直接跑通之后再逐步加复杂度。from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages def supervisor_node(state: MASState): # 调用模型生成拆解与分配结果示意 response call_model(SUPERVISOR_PROMPT, state[task]) assignment parse_llm_json(response) return { task_type: assignment[task_type], plan: assignment[plan], history: [f主管拆解任务: {assignment[plan]}] } def researcher_node(state: MASState): subtask get_assignment_for_agent(state, researcher) result research(subtask) return {subtask_results: {researcher: result}, history: [研究员完成资料检索]} def coder_node(state: MASState): subtask get_assignment_for_agent(state, coder) result code_and_run(subtask) return {subtask_results: {coder: result}, history: [开发智能体完成代码实现]} def writer_node(state: MASState): subtask get_assignment_for_agent(state, writer) result write_doc(subtask) return {final_answer: result, history: [文档智能体完成撰写]} # 图的定义 graph StateGraph(MASState) graph.add_node(supervisor, supervisor_node) graph.add_node(researcher, researcher_node) graph.add_node(coder, coder_node) graph.add_node(writer, writer_node) graph.add_edge(START, supervisor) graph.add_conditional_edges( supervisor, route_after_supervisor, {researcher: researcher, coder: coder, writer: writer} ) graph.add_edge(researcher, supervisor) graph.add_edge(coder, supervisor) graph.add_edge(writer, END) app graph.compile()这段代码的运行逻辑是任务进入主管主管拆解并分派执行者干完活回到主管主管再次决定是继续分派、审核通过还是结束。这里有一个我特别想强调的设计执行者的边都指向主管而不是直接指向结束。这意味着每个执行者的产出都会过一遍主管的审核这就把“主管监督”从口号变成了实际的控制流。只有主管判断任务已经完成才会通过 writer 产出最终结果并走向 END。3.2 让主管节点支持“退回重做”的循环控制上面这段代码还缺一个关键的“退回重做”能力。真实场景里执行者的产出很少一锤定音大部分情况需要两三轮修正。所以我把主管节点的逻辑改成了这样def supervisor_node(state: MASState): # 先检查是否有执行结果需要审核 if state.get(subtask_results): review review_subtask_results(state) if not review[approved]: return { history: [f主管退回: {review[comment]}], assignments: [{agent: review[rework_agent], task: review[rework_instruction]}], task_type: rework } # 否则进入正常的拆解分派流程 ...同时条件边也新增了一个分支def route_after_supervisor(state: MASState) - str: if state[task_type] rework: return state[assignments][0][agent] ...这样就能形成一个“分派 → 执行 → 审核 → 退回/通过”的闭环。从图结构上看这是一个带回边back edge的循环LangGraph 对这样的循环执行支持的很好而且每次循环都能在history里留下可追踪的记录调试时非常方便。3.3 场景案例混合任务全流程追踪为了让你看得更直观我用一个具体的任务来完整走一遍这套流程。假设用户的任务是“调研当前主流的本地知识库方案并选一个最合适的写一份技术选型报告附一段验证代码。”第一步任务进入supervisor。主管拆解后返回的 JSON 大致是task_type: mixedplan: [调研主流本地知识库方案, 对比分析优劣势, 实现一个最小验证用例, 撰写选型报告]assignments: [{agent: researcher, task: 调研...}, {agent: coder, task: 写代码...}, {agent: writer, task: 写报告...}]第二步条件边看到task_type mixed路由到coordinator。这里我加了一个协调节点它的职责是确保多个子任务之间有正确的先后依赖。如果这个项目里研究员的结果会影响代码方案的选型协调节点会把研究结果注入到代码智能体的上下文里再派发下去。第三步研究智能体完成任务后回到主管主管审核研究成果发现缺了几个对比维度于是退回重做。研究智能体根据主管的退回意见补充资料再次交回。第四步审核通过后轮到代码智能体执行产出验证代码。代码智能体返回代码片段后主管可能还会调用一个“代码检查”逻辑可以是一个规则校验也可以是另一个轻量智能体来确认代码能跑通然后才放行。第五步最终由文档智能体整合所有产出生成完整报告写入final_answer图走到 END。整个过程里任何一个子任务的中间状态都挂在MASState上所有节点都可以读取、修改这在调试时几乎就是“开卷考试”哪里出了问题一目了然。3.4 用图可视化辅助设计而不是依赖它排障LangGraph 自带图可视化能输出图的 PNG 或 mermaid 格式预览。我的使用习惯是在设计阶段用可视化确认整体结构是否符合预期但在运行排障阶段我更依赖history列表和节点返回的即时日志。具体来说我在每个节点都顺手往history里追加一条人类可读的记录比如“主管拆解任务”、“研究员完成任务结果通过审核”。跑完一轮之后直接打印history整个任务的流转链路就复原出来了。这比盯着可视化图找问题高效得多。4. 常见问题与排查技巧实录4.1 主管重复分派同一个任务进入死循环这是我最早踩到的一个坑。主管接到执行者返回的结果后如果审核逻辑不够严谨很容易出现“同一个子任务被反复分派执行者反复执行”的循环。根因是主管的提示词里没有说明“如果该子任务已有产出并且已经过审核则不应再次分派”。解决方式有两个层面提示词层面明确要求主管“不要重复分派已经在subtask_results中存在的任务”。代码层面在主管节点里加一个去重检查如果新的assignments中的任务和已有subtask_results的 key 重复则自动跳过。我建议两层都做。模型层面的约束再强也架不住偶尔的“叛逆”代码层面的兜底才是真正的安全网。4.2 状态更新互相覆盖执行者结果丢失这是一个典型的 reducer 误用问题。如果你的subtask_results字段没有加operator.or_注解那么每个执行者返回时都会把整个subtask_results覆盖成自己那一份前一个执行者的成果直接没了。这个坑排查起来其实不难只要在每次节点返回后打印一下状态就能发现。但预防更简单定义状态时所有需要多个智能体共享写入的字段一律加上合适的 reducer。字典用operator.or_列表用operator.add单个值字段不需要加。4.3 模型输出结构不稳定JSON 解析频繁失败大语言模型输出 JSON 偶尔会带有额外的解释文字或者括号不闭合。在 LangGraph 的节点函数里解析失败会导致整个链路中断。我用的方案是双重保险第一层提示词里强约束只输出 JSON禁止任何解释。第二层写一个健壮的解析函数能容忍少量噪声字符必要时用正则提取 JSON 块。import json, re def parse_llm_json(text: str) - dict: text text.strip() # 去掉可能的代码块标记 text re.sub(r^(?:json)?|$, , text, flagsre.MULTILINE).strip() try: return json.loads(text) except json.JSONDecodeError: match re.search(r\{.*\}, text, re.DOTALL) if match: return json.loads(match.group()) raise这个函数帮我省下了无数个凌晨三点的崩溃时刻。建议你直接把这段复制进工具模块。4.4 主管监督导致整体耗时过长怎么优化主管审核每个执行者的产出设计上带来了可控性但也不可避免地增加了调用次数和链路时长。一个混合任务的完整流程可能需要五六轮 LLM 调用整体耗时可能在几十秒到数分钟不等。我的优化策略有三个并行化对于互相没有依赖的子任务使用 LangGraph 的“扇出-扇入”模式让多个执行者同步执行而不是串行等待。轻量审核对于某些低风险子任务比如资料检索主管的审核可以用规则匹配替代完整 LLM 调用大幅降低耗时。提前结束机制如果主管判断某个分支的结果已经可以满足最终需求就允许跳过后续分支直接走打包输出路径。4.5 主管监督模式的实际排障心法最后分享一个综合性的排障心法先查历史记录再查图结构最后才查模型输出。多智能体系统的复杂之处在于问题可能出在任意一层。按照这个顺序排查能减少大量无头苍蝇式的时间浪费查看history确认任务流转是否符合预期卡在了哪个节点。查看当前状态里的关键字段确认路由判断依据是否正确。查看模型原始输出确认是提示词问题还是解析问题。如果是模型输出问题再回头调整提示词或改用结构化输出方法。我个人的经验是大约七成的问题出在“状态设计和控制逻辑”不到三成是模型本身的输出质量问题。不要一上来就怀疑模型不行先把自己的代码逻辑捋顺。5. 从单团队到多团队这套模式的扩展边界5.1 多主管并行协作的组织方式单个主管团队跑通之后最自然的扩展方向是“多主管并行”。比如一个项目中将“技术调研”和“视觉设计”两条线拆成两个团队每个团队各有一个主管两个主管之间通过一个总协调器通信。在 LangGraph 里这种结构完全可以在同一个状态图里实现总协调器节点作为顶层入口分派给两个子团队主管节点每个子团队主管内部再走上面说的“分派-执行-审核”循环。这种扩展写起来并不复杂但有一个新的挑战主管之间如何共享信息我的做法是给状态增加一个shared_context字段用法和subtask_results类似任何团队都可以把阶段性成果写入这个公共区域供其他团队读取。这样一来团队之间的信息传递不过度耦合彼此的独立性也保持住了。5.2 新增执行者时的最小改动随着项目推进你几乎一定会面临“需要新增一个执行智能体”的场景。按照上面的设计你需要做的改动只有三步定义新智能体的名称和能力描述挂到主管提示词的可用执行者列表里实现该智能体的节点函数并在图里add_node在条件边的路由表里增加一个目标分支。主管的提示词逻辑和图的结构都不需要大改。这套设计的扩展成本基本被压到了最低这也是我为什么推荐保持“控制流显式化”的原因。如果你把任务分派逻辑写在模型提示词里而没有相应的代码分支新增一个执行者就会很痛苦因为你需要让模型自己“学会”使用新角色——这显然不是一个稳定的方案。5.3 什么时候不要用这个模式说了这么多好处也该泼点冷水。主管监督模式并不适用于所有场景。如果你的任务足够简单一个智能体加几个工具就能解决那就不要硬搭团队。多一层主管就意味着多一轮模型调用、多一个失败点、多一份提示词维护成本。另外如果你的核心诉求是“开放式探索”没有明确的任务可拆解比如“给我一些创意灵感”那么主管监督的强流程感反而会限制系统的发散性。这种场景更适合走单智能体交互或多智能体自由讨论的模式而不是主管分派模式。这一点想明白能让你避免为了技术而技术。写在最后的实操体会这套模式跑通到现在我最大的体会是多智能体协作的难点从来不在单个智能体的“聪明程度”而在系统层面的“秩序设计”。LangGraph 的价值在于把调度、路由、状态流转这些和智能体能力无关的脏活累活接管过去让你能把全部精力放在提示词打磨和业务逻辑上。如果你正准备入坑我的建议是先不要追求复杂的图结构用手上的任务跑通一个“主管 两个执行者”的最小闭环把状态定义、条件路由、退回重做这三件事吃透再逐步往上加复杂度。踩过一轮坑之后你自然会知道下一步该往哪个方向扩展。