LangGraph多Agent系统从架构设计到生产部署全实践-平芜编程栈

聊到Multi-Agent，大部分人的第一反应是：“不就是多写几个Agent，轮流调一下大模型吗？”我一开始也是这么想的，直到在模拟项目X里栽了跟头，才意识到架构设计、任务拆分、状态管理、部署运维，每一步都没法糊弄。这篇我用LangGraph把一套多Agent系统从架构设计到生产部署完整走了一遍，把每个阶段的关键决策和踩坑点都写出来。内容不追求概念堆砌，适合正在用LangGraph搭多Agent系统、或者准备把Demo推向生产的开发者参考。

1. 先说结论：Multi-Agent 为什么值得折腾

半年前我接手了一个内部工具类项目，需求很朴素：用户丢进来一批零散文档，系统自动输出三份东西——进度报告、风险清单、趋势分析。听起来不复杂，但第一版我图省事，用一个大模型Agent + 超长Prompt硬扛。结果Prompt写了三千多字，模型还是经常犯糊涂，摘要写到一半跑题，格式也是三天两头变。最难受的是上下文被各种中间结果越撑越满，单次调用的Token成本肉眼可见地翻倍。

后来我换个思路：把任务拆开，让不同的Agent各管一摊。文件解析归解析Agent，风险点提取归风险Agent，趋势分析归分析Agent，最后再上一个调度节点把它们的产出汇总成统一格式。效果可以说立竿见影，单次任务的返工率降了不少，格式也稳定了。这就是Multi-Agent系统真正的价值：它把一个语义模糊的大任务，拆成多个边界清晰的小任务，每个子任务由最合适它的模型角色完成，再通过约定的接口把结果拼起来。

这篇文章就是按照我当时从零到上线的过程整理的，覆盖三块：第一块是架构设计，讲清楚多Agent系统到底怎么拆、怎么连；第二块是LangGraph实现，直接给能跑的代码骨架；第三块是生产级部署，把流式输出、可观测性、并发控制这些“上了线才知道重要”的细节一起说透。读完不一定能让你立刻成为专家，但至少能少走几周弯路。

2. 架构设计：动手写代码前，先把四件事想清楚

2.1 别把多个Agent当成堆数量

刚开始做多Agent的人最容易犯的错，就是觉得“Agent越多越智能”。我见过一个Demo，拆了八个Agent，每个Agent就是一句“你是专家”的Prompt，结果互相抢活、重复处理，效果反而不如一个Agent。所以先搞清楚常见的三种协作形态，再决定怎么搭。

第一种是流水线式（Pipeline）。Agent按固定顺序执行，前一个的输出是后一个的输入。比如：解析Agent -> 分类Agent -> 摘要Agent。好处是流程简单、结果可预期，适合任务阶段非常固定的场景。缺点是链路里任何一环出错，后面全白做，而且顺序一旦固定就很难动态调整。

第二种是中心调度式（Supervisor）。一个调度Agent不干活，只负责看任务、拆任务、把子任务派给合适的执行Agent，然后再收集结果。这是目前生产环境里最常用、也最不容易失控的形态，因为决策权和执行权分离了。调度Agent的判断即使出错，也只影响派单，不会污染实际产出。

第三种是网状式（Mesh）。Agent之间互相自由通信，谁有空谁接活。这种形态论文里很漂亮，但在真实业务里非常难控制，容易出现“两个Agent反复讨论同一件事”的无效循环，Token消耗巨大。我一般只在探索性原型里用，不敢直接上生产。

协作形态	优点	缺点	适合场景
流水线式	流程清晰、易调试	不够灵活、单点故障	固定流程的任务
中心调度式	分工明确、可控性强	调度节点可能成为瓶颈	多角色协作、任务动态路由
网状式	灵活度最高	容易失控、难追踪	研究探索或小规模原型

2.2 任务拆解与职责边界

定好协作形态后，最花时间的是任务拆解。拆得好不好，直接决定后面的Prompt和工具设计是否顺利。我常用的拆法有三个维度。

按工具边界拆。如果系统里已经有搜索工具、计算器工具、数据库查询工具，那就让每个Agent只负责和自己工具有关的活。负责搜索的Agent加载搜索结果，负责计算的Agent处理数值推导。这样做的好处是Agent上下文干净，该带的工具都能带上，不会因为工具太多导致模型选错。

按专业领域拆。业务上天然有专业分野，就按领域拆。比如做文档分析时，风险识别Agent专门看合同和协议条款，财务分析Agent只看报表和数字。领域单一，Prompt可以写得很深，输出质量也会更稳。

按流程阶段拆。比如“先解析，再结构化，最后生成报告”，每个阶段一个Agent，职责明确，排查问题也方便，谁出问题找谁。

注意，不管按哪种方式拆，都要把职责边界写清楚。我之前吃过亏：两个Agent都负责“提取问题”，结果同一个问题被提了两遍，最后汇总时出现重复内容。后来在每个Agent的系统提示词里明确写“你只负责X，遇到Y内容一律忽略并交给负责Y的Agent”，重复率才降下来。

2.3 状态共享：多Agent的通信方案

多Agent系统里，Agent之间的通信不是“互相发消息”那么简单，它背后是状态管理的问题。LangGraph里每个Agent节点都会接收一个全局State，节点里的处理结果会更新State，再把新的State传给下游节点。做架构设计时就要想清楚：哪些字段是全程共享的，哪些字段只是局部传递的。

我的经验是，共享字段越少越好。全程共享的通常只有任务ID、原始输入、最终结果这几个核心字段。中间过程字段如果也全局共享，会带来两个问题：一是State越滚越大，每次传给大模型的Token越来越多；二是下游Agent容易被一堆和自己无关的中间内容干扰。

比较好的做法，是让每个Agent的输入输出结构保持一致，比如统一用messages保存对话历史，用artifacts保存中间产物。这个约定的设计要和代码结构一起定下来，不然后期改会牵扯到所有节点。

2.4 容错与兜底：先把失败想明白

多Agent系统比单Agent多了一个可怕的故障源：模型调用失败之外，还有路由失败、工具调用失败、Agent之间循环调用。上线前如果不把失败场景走一遍，生产环境里一定会出事。

我的做法是给每个节点配置三件套：超时时间、重试策略、降级方案。超时时间不能设太长，LLM在复杂任务上普遍反应慢，但也不能无限等，一般外部模型建议30秒到60秒，内部小模型可以短一些。重试要考虑幂等性，尤其工具调用，重复执行会不会带来副作用，这一点必须确认。降级方案更是关键，比如调度Agent无法识别该派给谁时，宁可返回“无法处理”让用户介入，也不要随便乱派。

这里有个容易被忽略的点：超时和重试必须在LangGraph的节点逻辑外面包一层通用组件，而不是在每个Agent里各写一套。不然代码里到处都是重复的错误处理逻辑，维护起来非常痛。

3. LangGraph实现：从零搭一个能跑的多Agent协作系统

3.1 核心概念速览

LangGraph用起来其实不复杂，核心就是把一个复杂工作流定义成一个“图”，图上有很多节点，节点之间通过边相连，再靠一个全局状态对象来传递数据。几个关键词先熟悉一下：

State：全局状态，就是一个数据结构，用来承载所有Agent节点之间传递的信息。
Node：节点，每个节点是一段可执行逻辑，通常就是一个Agent的一次调用。
Edge：边，用来定义执行顺序和跳转逻辑，可以是顺序边，也可以是有条件边。
Command：一个动作指令，节点里可以返回Command来告诉图“下一步该去哪个节点”，同时更新状态。

我把这个模型类比成一条生产流水线：State是传送带上的料箱，Node是工位上的工人，Edge是工位之间的传送带。工人干完活把料箱塞回去，流水线根据指令决定下一个送到哪个工位。这样一想，整个多Agent系统的执行逻辑就很好理解了。

3.2 最小骨架：一个会调度的Supervisor

下面给一个最常用的中心调度式最小骨架。它包含两个执行Agent（research_agent和coding_agent），以及一个负责派单的supervisor节点。代码基于当前主流的LangGraph版本，FastAPI接口在0.4.x里已经比较稳定，但导入路径偶尔有变动，具体以你安装版本为准。

from typing import Literal from langchain.chat_models import init_chat_model from langchain_core.messages import HumanMessage, SystemMessage from langgraph.graph import StateGraph, MessagesState, START, END from langgraph.types import Command members = ["research_agent", "coding_agent"] supervisor_prompt = ( "你是一个负责任务调度的主管Agent，你不直接干活。\n" f"你手下的Agent有：{members}\n" "你需要根据用户请求，从成员中选一个最适合的Agent来处理，" "并输出该Agent的名字。如果任务已经完成，输出 FINISH。" ) def supervisor(state: MessagesState) -> Command[Literal["research_agent", "coding_agent", "__end__"]]: # 这里的模型名和provider请换成你自己环境可用的配置 model = init_chat_model(model="your-model-name", model_provider="your-provider") response = model.invoke([ SystemMessage(content=supervisor_prompt), *state["messages"] ]) content = response.content.strip() if "FINISH" in content.upper(): return Command( goto=END, update={"messages": [response]} ) for member in members: if member in content: return Command( goto=member, update={"messages": [response]} ) # 路由不明确时，宁可结束，也不瞎派 return Command( goto=END, update={"messages": [ HumanMessage(content="我无法确定该把任务派给谁，请补充需求。") ]} ) def make_agent(name: str): def agent_node(state: MessagesState) -> Command: model = init_chat_model(model="your-model-name", model_provider="your-provider") system_message = SystemMessage( content=f"你是{name}，你只负责自己职责范围内的任务，完成后把结果交还给主管节点。" ) response = model.invoke([ system_message, *state["messages"] ]) return Command(update={"messages": [response]}) return agent_node graph = StateGraph(MessagesState) graph.add_node("supervisor", supervisor) graph.add_node("research_agent", make_agent("research_agent")) graph.add_node("coding_agent", make_agent("coding_agent")) graph.add_edge(START, "supervisor") graph.add_edge("research_agent", "supervisor") graph.add_edge("coding_agent", "supervisor") app = graph.compile()

这段代码跑通之后，你就有了一个最基础的Multi-Agent系统：用户输入任务，supervisor根据内容决定派给谁，执行完结果回到supervisor，再看任务是否完成、需不需要再派。核心逻辑就这么多，后面所有复杂功能都是在这个骨架上加东西。

3.3 给Agent装上工具，让它真的会干活

只靠大模型嘴上回答肯定不够，生产级系统里Agent必须会调用工具。LangGraph里给Agent加工具的方式有两种，一种是用预置的create_agent把工具绑定好，另一种是自己手写节点逻辑，灵活控制调用过程。我建议先理解手写方式，再回到封装API，这样出了问题才知道怎么查。

from langchain_core.messages import ToolMessage from langchain_core.tools import tool @tool def multiply(a: int, b: int) -> int: """计算两个整数相乘的结果。""" return a * b def coding_agent_node(state: MessagesState) -> Command: model = init_chat_model( model="your-model-name", model_provider="your-provider", temperature=0 ).bind_tools([multiply]) response = model.invoke(state["messages"]) # 如果模型决定调用工具，就执行工具并把结果传回去再调一次 if response.tool_calls: tool_messages = [] for call in response.tool_calls: result = multiply.invoke(call["args"]) tool_messages.append( ToolMessage(content=str(result), tool_call_id=call["id"]) ) second_response = model.invoke([*state["messages"], response, *tool_messages]) return Command(update={"messages": [response, *tool_messages, second_response]}) return Command(update={"messages": [response]})

这里有个非常容易踩的坑：工具调用的结果必须用ToolMessage返回，而且tool_call_id要和模型返回的一致，否则模型无法把工具结果和对应调用关联起来。我见过有人在工具执行完后直接把结果拼在字符串里丢回给模型，结果模型完全不认账，导致二次调用还是错的。

如果你用的是create_agent，这些细节都被封装好了，写起来很舒服：

from langgraph.prebuilt import create_agent research_agent = create_agent( model="your-model-name", model_provider="your-provider", tools=[multiply], system_message="你是负责数值计算的Agent，只处理和你职责相关的问题。" )

封装API把工具循环、消息组装这些都处理好了，新手可以先用它跑通流程，后续追求粒度控制的时候再自己写节点。

3.4 让系统记住上下文：Checkpointer与持久化

多Agent系统上线后马上会遇到一个需求：用户聊到一半，换了个话题，过一会儿又回来追问上一个任务，系统得记得住。LangGraph的解决方案是引入Checkpointer，字面意思是“检查点”，它负责在图的每一步执行之后保存状态。这样每一次执行都能被恢复，历史会话也能被继续。

from langgraph.checkpoint.memory import MemorySaver memory = MemorySaver() app = graph.compile(checkpointer=memory) config = {"configurable": {"thread_id": "session-001"}}

有了checkpointer之后，同一个thread_id下的多轮对话，会共享同一条消息历史。后端需要给每个用户会话维护一个thread_id，在请求里传给LangGraph就行。生产环境里MemorySaver不够用，因为进程一重启状态就丢了，得用Postgres之类的持久化版本。

from langgraph.checkpoint.postgres import PostgresSaver # 仅示意，实际使用时把连接串换成你自己的 with PostgresSaver.from_conn_string("your-connection-string") as checkpointer: app = graph.compile(checkpointer=checkpointer)

持久化这个设计我非常喜欢，等于把多Agent系统的状态管理和业务系统的会话管理解耦了。你不需要自己维护一堆消息表，LangGraph已经帮你把状态序列化好了。

4. 生产级部署：从Demo到能扛压的完整清单

4.1 先选托管Graph还是自建服务

做生产级部署前，先回答一个选择题：用LangGraph的托管平台，还是自建服务。这道题没有标准答案，我两个都试过，各自的坑很清楚。

托管方式的优势是省心，部署、扩容、日志追踪都有人管，团队可以专注业务逻辑。成本是绑定平台，一旦你的部署环境和平台约束有冲突，比如需要内网调用公司内部的模型服务，托管方案会变得不顺手，而且单个请求的延时在跳板之后会明显增加。

自建方式的优势是完全可控，可以自己定制并发模型、对接内部服务，成本也更容易控制。代价是运维压力全在自己身上：要做容器化部署、状态存储、监控告警，每一样都要花时间。我的建议是：如果你的系统要和公司内部系统深度耦合、数据不出内网是硬需求，那直接自建。如果只是做一个轻量级内部工具，愿意接受托管带来的省心，那优先考虑托管。

4.2 流式输出不能省

多Agent系统的响应时间普遍比单Agent长，因为内部要走多次模型调用和工具调用。如果前端是等全部跑完才一次性返回结果，用户那边的体验会非常差，一个任务转十秒以上，基本就没人愿意用了。

LangGraph原生支持流式返回，代码里用astream替代invoke就行。它能按节点粒度把结果实时吐出来，前端拿到后可以做成“当前正在由哪个Agent处理中”的状态展示。用户看到系统在动，哪怕总耗时没变，体验也会好很多。

async for chunk in app.astream( {"messages": [HumanMessage(content="请整理这两份文档并生成风险清单")]}, config=config, ): # 每个chunk里包含当前节点的执行信息 print(chunk)

想更进一步，可以用stream_mode参数控制流式粒度，比如messages模式能直接拿到Token级别的增量。但Token级流式对前端架构要求更高，我通常是先做到节点级流式，等用户反馈吞吐瓶颈了再升级。

4.3 可观测性：不要等出了错再翻日志

多Agent系统的链路比普通接口长得多，一个请求可能经历：调度 -> AgentA -> 调用工具 -> AgentB -> 汇总。任何一环出问题，都可能导致最终结果错误。上线后如果只靠看业务日志排障，会非常痛苦，因为你很难还原某一次请求内部到底发生了什么。

我的做法是把四类信息全部记录下来：请求入参和最终结果、每个节点的输入输出摘要、模型调用耗时与Token数、工具调用参数与返回结果。LangGraph本身会生成执行轨迹，可以把这些轨迹通过LangSmith之类的工具做可视化追踪。如果团队不用第三方工具，就自己把结构化日志打出来，每条日志带上request_id，用链路ID串起来。

{ "request_id": "abc-123", "node": "research_agent", "model_call_ms": 2340, "prompt_tokens": 312, "completion_tokens": 156, "tool_result": "multiply(3, 5) -> 15" }

有了这样的日志，排查一个“为什么某个Agent回得不对”的问题，就可以直接从链路里找那一次的模型输入输出，而不是靠猜。

4.4 回归测试：多Agent系统最容易无声变差

普通接口上线后好不好用，看错误率就行。多Agent系统不是这样，它可能所有调用都成功，但回答质量悄悄变差。原因往往是某个模型升级了、工具返回的内容变化了、或者某个Prompt里的措辞被改动了一个字。这种“质量退化”没有明确的Error信号，只能靠回归测试去兜住。

我做过几类测试，最实用的其实是“黄金用例回归”：找十到二十个历史上最典型的任务，比如以模拟项目X里的“风险提取”为例，把它们的真实输出保存下来，代码每次改动后跑一遍，人工扫一遍结果差异。不需要自动打分，只要人眼扫一遍，就能发现大部分明显退化。

如果团队有条件，再叠加一层评估脚本：定义一个检查项清单，比如“输出是否包含风险等级”“风险描述是否超过30个字”，用规则或另一个强模型去判断达标率。这样每次发布前都有量化指标可以看，不会盲目上线。

4.5 并发、限流与成本控制

多Agent系统上线后，最大的两个运维课题是并发和成本，而且这俩是紧密绑在一起的。每个用户的一次请求背后可能是3到5次模型调用，假设模型单次调用1秒，一个请求需要4秒，那么50个并发用户至少需要每秒处理50个模型请求的吞吐能力。不控制好并发，下游模型服务能被直接打爆。

限流是必须做的。最简单的方案就是在网关层做令牌桶限流，按用户维度和总维度各设上限。用户维度防止单个用户把资源占满，总维度防止系统被打爆。LangGraph自建服务的话，还需要考虑线程池大小，直接控制同时进入图的请求数。

成本控制方面，我有三个经验。第一，给不同任务选择不同档位的模型，简单的分类任务别用超大模型，只有最后汇总、生成结构化报告这种复杂任务，才需要上强模型。第二，把Agent之间的中间内容做压缩，保留关键字段，而不是把完整对话历史一直往下游传。第三，对高频且结果可复用的调用来做缓存，比如同样的文档解析结果，同一个用户短期内重复请求可以直接命中缓存。

5. 常见问题与排查实录

5.1 死循环：Agent反复把任务交给彼此

多Agent系统最典型的故障，就是两个Agent在互相交接任务中陷入循环。现象是日志里出现大量supervisor -> AgentA -> supervisor -> AgentA的记录，直到超时被掐断。我在模拟项目X里就踩过一次，原因是supervisor指令里没加限制，它觉得任务没完成，就反复派给同一个Agent，而那个Agent每次只做一半就交回来。

解决办法有两个方向。一个是给supervisor的指令明确写清楚“同一个Agent最多只能被调用一次”，并加上计数逻辑。另一个是LangGraph层面直接限制图的执行步数，超过设定步数就强制结束并提示用户。

app = graph.compile(checkpointer=memory) # 在运行时给调用加上recursion_limit限制 result = app.invoke(input_data, {"recursion_limit": 15})

设置了这个限制后，即使路由逻辑出bug，最多转15步就会被中断，不会让系统卡死。

5.2 上下文爆炸：Token消耗成倍增长

多Agent系统比单Agent更容易吃Token，因为同一个中间结果会被多个Agent反复看到。尤其是消息历史里累积了每次调用的完整输入输出后，后面的Agent每次都要重新“读”一遍前面所有的历史，开销越来越大。

我排查Token消耗异常时，第一步就是看每个节点的输入消息数。如果发现某个Node收到的消息列表一直涨，那说明上游节点在往State里塞了太多东西。解决办法是在节点返回前做字段裁剪，只保留对后续节点有意义的结果摘要，或者用一个专门的状态字段存放“最终答案”，不把工具调用细节带下去。

另一个常见做法是定期“重置上下文”：如果对话轮次太多，就在达到N轮后，把前面的历史压缩成一段总结，只保留重要结论，后续节点只读总结不再读全量历史。

5.3 Agent之间“传谣”：错误信息的级联放大

多Agent系统里有个隐蔽问题：AgentA产生了一个不太准确的中间结果，AgentB把它当成事实，继续加工并放大，最后汇总时这个错误已经变得非常严重。单个Agent犯错影响范围有限，多个Agent串联之后，错误会被级联放大。

应对措施是让每个对结果负责的节点都加上“事实性校验”环节。比如风险Agent输出风险清单后，用一个轻量校验节点检查每个条目是否能在原始文档里找到出处，找不到的标成“无法溯源”。另一个思路是条件路由：低置信度的结果不走自动汇总链路，而是进入人工复核流程。多Agent系统的优势就是可以把这种判断逻辑做成节点，而不是靠纠结一句话。

5.4 路由不稳定的应对方案

调度Agent判断派给谁偶尔出错，这个情况很常见。排查这类问题时，我建议先把调度Agent的决策过程记录挂出来，看看它到底依据哪些关键词做判断。很多时候不是模型不行，而是成员命名和调度反馈的提示词太模糊。如果成员名叫 “AgentA”“AgentB”，模型自己也不知道该派给谁；改成 “risk_analyzer_agent”“report_generator_agent”之后，路由准确率会明显提升。

还有一个比较稳的技巧：把路由从“让模型自由选”改成“给定选项选”。意思是在调度Prompt里列出所有可选成员的完整描述和适用场景，让模型做选择题而不是填空题。我实测过，改成选择题之后，路由准确率能提升不少，因为模型的决策空间被收敛了。

常见问题	根因	排查方式	解决建议
Agent死循环	路由逻辑缺少终止条件	观察链路日志，统计节点调用次数	设置recursion_limit，限制单Agent调用次数
Token消耗爆炸	全局State塞入过多中间内容	检查节点输入消息长度、Token用量	节点返回前裁剪字段，定期压缩历史
错误信息级联放大	中间结果不被验证	人工抽查链路中间输出	增加结果校验节点，低置信度走人工
路由判断不稳定	成员命名模糊、Prompt约束弱	查看调度Agent的决策日志	改用选项式路由，细化成员说明