最近一两年大模型应用圈子里最难绕开的技术热词Multi-Agent绝对算一个。不管是做AI客服、代码生成还是做企业级的复杂流程自动化大家都在尝试把单一的LLM调用拆成多个角色分工协作的Agent系统。但我也见过太多人把Multi-Agent想得很玄觉得就是多调几次Prompt、多写几个子Agent再拼到一起。实际上真正的Multi-Agent系统需要你从架构层面想清楚状态怎么管理、消息怎么流转、各个Agent之间怎么协作更别提后面还有部署到生产环境的稳定性问题。这篇文章会基于我实际做过的项目经验把Multi-Agent系统从架构设计、LangGraph实现到生产级部署的完整链路讲透。内容不搞纸上谈兵全部是能直接落地的东西尤其适合那些已经熟悉LangChain基础、想往Multi-Agent方向进阶的开发者以及正在选型或改造现有LLM应用的技术负责人。读完你至少能搞清楚什么时候真的需要Multi-Agent怎么设计Actor之间的协作结构怎么用LangGraph把设计变成可运行的代码以及上了生产环境之后哪些坑最容易爆。1. 先想清楚为什么你的应用真的需要Multi-Agent1.1 单体Agent的性能上限在哪在动手写代码之前建议先冷静评估一下你的场景到底需不需要多Agent。我见过不少项目上去就搞了五六个Agent互相调用最后效果还不如一个精心设计的单Agent流程。单体Agent在任务相对简单、链路固定、上下文可控的场景下其实非常稳定。比如说做一个固定模板的摘要生成、简单问答、格式化输出一个Agent配上好Prompt绰绰有余。真正让单体方案失效的是两类情况。第一类是任务复杂度高比如你要做一份行业调研报告涉及资料搜集、数据分析、观点提炼、文字润色、格式排版所有逻辑都塞进一个Agent里上下文窗口会被撑到极限模型在长上下文里往往会丢失早期的关键信息输出质量肉眼可见地下降。第二类是需要多角色约束同一个模型在同一轮对话中既要扮演提问者又要扮演批判者很容易角色混淆最后给你来一个既不够专业又不够客观的四不像答案。1.2 拆分的核心不是Agent数量是职责边界我个人的判断标准很简单如果任务可以清晰地拆成几个子任务每个子任务有独立的输入输出边界且子任务之间存在信息传递或汇合关系那么这就是一个适合Multi-Agent的场景。拆分的粒度要遵循高内聚低耦合每个Agent只负责一件事不要让一个Agent既做数据提取又做最终决策。举例来说我在做一个工单处理系统时把流程拆成意图分类Agent、信息提取Agent、工单处理Agent、质检Agent和最终回复Agent。意图分类Agent只负责判断用户是想咨询订单、申请退款还是投诉物流不做任何后续处理信息提取Agent只负责从用户消息中抽取订单号和诉求关键词后续的工单处理Agent再根据前两者输出做业务查询。这样每个Agent的System Prompt都很简洁模型不需要在角色之间反复切换出错率会显著降低。1.3 什么时候应该放弃Multi-Agent这里必须说点泼冷水的话。Multi-Agent不是银弹它带来的额外延迟、状态管理复杂度和调试成本都是实打实的。如果你的任务链路固定比如永远是先做A再做B再做C那用LangGraph的线性流程就够了没必要让多个Agent互相调用。只有当你需要动态决策下一步交给谁做、不同输入走不同分支的时候多Agent的网络拓扑才能体现出价值。另外如果你的LLM调用延迟本身就很高叠加多Agent的多次串行调用会让用户体验雪崩这时候要先考虑优化模型响应速度而不是盲目上多智能体。2. 架构设计多Agent协作模式与状态管理的底层逻辑2.1 三种主流协作架构模式怎么选多Agent系统的架构模式大致可以归为三类每类的适用场景差异很大选错了后面会走很多弯路。路由模式是最容易理解和落地的一种。入口Agent接收到用户请求后根据意图分发到不同的专业Agent处理每个专业Agent只处理自己领域内的任务处理完直接返回结果。这种模式适合意图明确、领域边界清晰的场景比如客服系统按售前、售后、投诉分线处理。优点是实现简单、职责清晰、易扩展缺点是入口Agent容易成为瓶颈而且跨领域任务需要多次路由。协作模式是多个Agent地位对等地处理同一个任务各负责一个子模块最后汇总结果。典型例子是写文章调研Agent负责搜集素材写作Agent负责撰写正文审校Agent负责检查质量。这种模式适合需要多角度综合处理的复杂任务但你需要格外注意Agent之间的信息共享是否完整否则容易遗漏关键信息。层级模式是上级Agent负责任务分解和结果验收下级Agent负责具体执行类似项目经理和组员的关系。这种模式适合结构化强、可分解性高的复杂任务也是最接近真实团队协作的模式。层级模式的难点在于上级Agent的任务分解能力直接影响整体效果如果分解不合理下面做得再好也白搭。架构模式适用场景优势劣势路由模式意图明确、领域边界清晰实现简单、易扩展入口易成瓶颈、跨域任务弱协作模式多角度综合处理的复杂任务并行度高、覆盖全面信息共享难、汇总易遗漏层级模式结构化强、需要任务分解职责分明、可控性强上级分解质量决定上限2.2 共享状态设计一切协作的基础Multi-Agent系统里最容易被忽视但最关键的就是状态设计。你可以把整个系统理解成一个状态机每个Agent的输入是当前状态输出是更新后的状态Agent之间不直接传递消息而是通过共享状态进行间接通信。这种模式天然适合用LangGraph来实现因为LangGraph的StateGraph本质上就是这个思路。状态设计的第一原则是结构化而不是随意塞字段。我见过有人把所有Agent的输出都塞进一个messages字段里这会导致状态越来越臃肿。更好的做法是区分短期消息数据和长期业务数据。比如在工单系统里messages字段保存原始的对话记录intent字段保存分类结果order_info字段保存提取出来的结构化订单信息final_answer字段保存最终回复。这样每个Agent只读取自己关心的字段不至于被无关数据干扰。第二原则是考虑并发写入的冲突问题。如果有多个Agent要同时往同一个字段追加内容LangGraph提供了add_messages之类的Reducer机制来处理合并逻辑如果你是自定义字段需要自己定义Reducer函数或者用operator.add之类的操作符。不处理并发冲突的后果就是状态丢失部分Agent的输出被静默覆盖排查起来相当痛苦。2.3 通信机制让Agent之间的协作有迹可循Agent之间的通信方式直接影响系统的可观测性。现实项目里我不推荐让Agent之间直接进行自由的自然语言对话因为这种通信方式难以追溯、难以测试一个Agent说错一句话就会污染下游所有Agent的判断。更稳妥的做法是让Agent只输出结构化数据比如JSON格式的结果下游Agent先解析JSON再做决策。这里要特别提醒一个常见误区不要以为两个Agent互相发消息才算多Agent。在LangGraph里所谓的Agent通信本质上是状态字段的流转和更新。Agent A的返回结果写入stateAgent B从state里读取结果作为自己的输入。理解这一点你的系统设计会清晰很多也方便后续做单元测试和异常定位。3. LangGraph实现从零搭建一个可运行的工单处理系统3.1 LangGraph核心概念五分钟速览LangGraph是LangChain团队推出的一个专门构建有状态Agent应用的开源框架它的核心抽象就是StateGraph。整个框架的概念其实不多搞懂这几个你就能上手写代码了。State是承载所有共享数据的对象通常用TypedDict或Pydantic BaseModel定义。Node是图中的计算单元每个Node接收State作为输入处理完成后返回一个包含更新字段的字典LangGraph会自动把这些更新合并回State中。Edge定义了Node之间的跳转关系分为普通Edge和条件Edge普通Edge表示固定顺序条件Edge则根据State中的某个字段动态决定下一跳。你还可以在Node里使用一个特殊的Send API实现动态并行分支让一个Node的多次执行同时进行。整体使用流程是先定义State结构再实现各个Node函数然后创建StateGraph实例添加Node和Edge最后调用compile方法编译成可执行对象。编译之后的对象支持invoke同步调用、ainvoke异步调用还有stream流式输出可以按需选择。3.2 实战搭建一个客服工单处理系统的完整代码下面这段代码是我在实际项目中验证过的工单处理系统简化版。它包含了意图分类、信息提取、工单处理、转人工升级和回复生成五个核心节点。先定义State结构from typing import TypedDict, List, Dict, Any, Annotated from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langchain_openai import ChatOpenAI class AgentState(TypedDict): messages: Annotated[List[Dict[str, Any]], add_messages] intent: str order_info: Dict[str, Any] requires_human: bool final_answer: strState定义里最值得注意的就是messages字段的Annotated类型标注。add_messages是一个消息合并器当多个Node返回的消息都写到messages字段时它会自动按时间顺序追加合并而不是直接覆盖。这是LangGraph处理多Agent并发写消息的标准做法。接着实现节点函数。意图分类节点负责判断用户意图只返回一个JSON字符串def classify_intent(state: AgentState): llm ChatOpenAI(modelgpt-4o-mini, temperature0) last_message state[messages][-1][content] prompt f判断以下用户消息属于哪个意图类别只从以下选项中选择一项 - order_query: 订单查询 - return_request: 退换货申请 - complaint: 投诉 - other: 其他 用户消息{last_message} 只要输出一个类别词。 intent llm.invoke(prompt).content.strip().lower() return {intent: intent, requires_human: False}信息提取节点的逻辑是解析用户消息里的订单号和商品信息。这个节点做了容错处理提取不到关键信息时把requires_human标记为True后续会走人工升级分支def extract_order_info(state: AgentState): llm ChatOpenAI(modelgpt-4o-mini, temperature0) last_message state[messages][-1][content] prompt f从用户消息中提取订单信息。 用户消息{last_message} 请以JSON格式输出字段包括 order_id订单号、item_name商品名、issue_desc问题描述。 如果某字段无法提取输出 null。 import json raw llm.invoke(prompt).content try: info json.loads(raw) except Exception: info {} requires_human not info.get(order_id) return {order_info: info, requires_human: requires_human}业务处理节点根据意图做分发。为了演示条件路由我把订单查询和退换货申请分别定义为独立节点投诉也单独处理def handle_order_query(state: AgentState): # 实际项目里这里会调用订单系统API order_id state[order_info].get(order_id, N/A) return {final_answer: f订单 {order_id} 当前状态为已发货预计48小时内送达。} def handle_return_request(state: AgentState): # 调用退换货服务API order_id state[order_info].get(order_id, N/A) return {final_answer: f已为您登记订单 {order_id} 的退货申请退款会在7个工作日内原路返回。} def handle_complaint(state: AgentState): return {final_answer: 我们已经记录您的投诉专人将在24小时内联系您核实情况。} def escalate_to_human(state: AgentState): return {final_answer: 抱歉没能完全理解您的需求已为您转接人工客服请稍候。}节点函数写完之后就到了构建图的核心部分。这里用add_conditional_edges实现基于intent字段的条件路由from langgraph.graph import StateGraph, END def build_graph(): graph StateGraph(AgentState) graph.add_node(classify, classify_intent) graph.add_node(extract, extract_order_info) graph.add_node(order_query, handle_order_query) graph.add_node(return_request, handle_return_request) graph.add_node(complaint, handle_complaint) graph.add_node(human, escalate_to_human) graph.add_node(respond, lambda state: state) # 入口边classify - extractextract 后做条件判断 graph.set_entry_point(classify) graph.add_edge(classify, extract) # extract 之后判断是否需要人工升级 def route_after_extract(state: AgentState): if state.get(requires_human): return human return route_by_intent def route_by_intent(state: AgentState): intent state.get(intent, ) if intent order_query: return order_query elif intent return_request: return return_request elif intent complaint: return complaint else: return human graph.add_conditional_edges(extract, route_after_extract, {human: human, route_by_intent: route_by_intent}) # route_by_intent 是一个透明节点不实际做处理只做路由 graph.add_node(route_by_intent, lambda state: state) graph.add_conditional_edges(route_by_intent, route_by_intent, {order_query: order_query, return_request: return_request, complaint: complaint, human: human}) # 汇总输出 for node in [order_query, return_request, complaint, human]: graph.add_edge(node, respond) graph.add_edge(respond, END) return graph.compile()这段代码里有一个我特意留的细节extract之后的condition路由返回了route_by_intent而这个标签对应一个透明节点在透明节点上再做真正的意图路由。这是为了保持图拓扑的可读性避免在一条条件边上同时承担是否转人工和按意图分发两层判断逻辑。你完全可以把两层判断合在一个函数里但拆开后后续加路由分支会轻松很多。3.3 让Agent真正具备自主决策能力的进阶实现上面这个例子其实是半决策性的路由规则是我们硬编码的。真正的Agent系统应该让模型自己决定下一步调用哪个工具、访问哪个Agent。LangGraph为此提供了预构建的ReAct Agent实现核心是构建一个Agent节点并绑定工具列表from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langgraph.prebuilt import create_react_agent tool def query_order_status(order_id: str) - str: 根据订单号查询订单状态 # 调用后端API return 已发货 tool def create_return_request(order_id: str, reason: str) - str: 创建退货申请 return 退货申请已创建 tools [query_order_status, create_return_request] model ChatOpenAI(modelgpt-4o, temperature0) agent create_react_agent(model, tools)create_react_agent会返回一个编译好的LangGraph图它内部实现了ReAct循环模型推理决策 → 调用工具 → 观察结果 → 再决策直到不需要调用工具为止。这个实现的好处是Agent有了自主调用外部工具的能力配合我们自定义的其他Agent节点可以组合出更灵活的工作流。不过要提醒的是ReAct循环会带来额外的多次LLM调用你要在灵活性、延迟和成本之间做权衡而且要务必设置最大循环次数避免模型陷入工具调用的死循环。4. 生产级部署从能跑到稳定运行的完整链路4.1 容器化部署与依赖锁定开发环境里跑通的LangGraph代码部署到生产环境完全是另一回事。第一步就是容器化。我习惯用Docker基础镜像选择Python 3.10以上的官方slim版本而不是自带一堆无关依赖的机器学习镜像。依赖管理用requirements.txt或poetry锁定所有包的版本尤其要锁定langgraph和langchain-core的版本这两个库的API更新频繁小版本之间都可能出现破坏性变更。还要注意环境变量管理。API密钥、数据库连接串、业务系统接口地址这些敏感信息绝对不能写进镜像或代码库里。用环境变量注入或者在Kubernetes里用Secret管理。我见过有人图省事把密钥硬编码进Dockerfile结果镜像推到私有仓库后被扫描出来那真是欲哭无泪。4.2 并发架构如何扛住真实业务流量单实例串行处理Agent请求的吞吐量是非常低的。一个复杂工单流程可能要访问3次LLM API每次耗时2~5秒单实例并发只能撑到个位数QPS。生产环境必须考虑并发架构。LangGraph的invoke是同步阻塞的在一个进程内同时处理多个请求时需要开线程池。更推荐的方式是使用异步接口ainvoke配合asyncio实现高并发。如果业务量更大就把请求打到消息队列如RabbitMQ或Kafka由Worker节点消费队列消息并异步执行Agent流程这样可以实现水平扩展。有一个关键教训LLM API供应商通常有每分钟Token数TPM限制和每分钟请求数RPM限制。即使你的服务可以高并发调用上游API的限流也会成为瓶颈。务必在系统里做好请求速率限制和退避重试机制比如用tenacity库给LLM调用加上重试逻辑。from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10)) def call_llm_with_retry(prompt: str) - str: return llm.invoke(prompt)4.3 可观测性没有监控的Agent系统等于定时炸弹Multi-Agent系统的可观测性比普通Web应用更复杂因为一个用户请求会经过多个Agent节点每个节点都可能调用LLM、访问外部API任意一个环节出错都会影响最终结果。我在项目里做的第一件事就是全链路日志追踪。具体做法是生成一个request_id贯穿整个Agent流程在每个Node的入口和出口打上结构化日志包括当前节点名称、输入状态摘要、输出状态摘要、调用LLM耗时和Token消耗。日志统一输出为JSON格式方便采集到ELK或Loki里做检索分析。LangChain生态自带的LangSmith可以自动完成这些追踪它能可视化Agent执行过程中的每一步还能查看每一步的Token消耗和耗时调试和性能分析都很有用。4.4 多环境隔离与发布回滚生产级部署一定要区分开发、测试、灰度、生产环境每个环境使用独立的API密钥和数据库。系统配置也要支持多环境覆盖。我推荐用类似于pydantic-settings的方案根据环境变量自动加载对应的配置。发布流程上构建好的镜像先跑一遍测试环境的完整回归再灰度发布到生产环境观察一段时间的错误率和耗时指标确认稳定后全量切换。这里要特别强调LLM输出的不确定性意味着不同版本的Prompt或模型参数可能带来难以预料的输出变化灰度发布前一定要准备好在故障时快速回滚到上一个版本的方案。4.5 成本控制Token消耗是最大的隐形支出Multi-Agent系统的成本常常在账单出来那一刻让人惊掉下巴。N个Agent串行访问LLMToken消耗是单次的N倍还不算重复调用和失败重试。我在生产环境里做了三件事来控制成本。第一给每个Agent设置独立的模型规格简单分类任务用便宜的小模型复杂推理任务才用大模型这是最立竿见影的降本手段。第二引入上下文压缩机制对进入长链路Agent的历史消息做摘要只保留关键信息。第三监控每个Agent的Token消耗定期分析哪个Agent的Prompt太长或调用频率过高针对性优化。5. 常见问题与排查实录5.1 状态被静默覆盖我遇到过最隐蔽的问题就是状态字段被覆盖。多个Agent并发返回同一个字段的更新时如果不使用ReducerLangGraph会用最后一次返回的结果覆盖前面的结果。这段问题往往表现为明明Agent A已经提取到了订单号Agent B执行后状态里的订单号变成了空值。排查方法是在日志里跟踪每个Node返回前state中关键字段的值。解决方案就是前面提到的对所有可能被多次分配的字段使用Annotated[type, operation]声明合并策略。对消息列表使用add_messages对累加类数据使用operator.add对业务字段尽量保持单一写入者。经验法则一个字段只让一个Agent负责写入避免多Agent竞争写权限。5.2 Agent陷入无限循环条件路由写得不好很容易出现A转B、B又转回A、A再转B的无限循环。ReAct Agent更是容易出现模型反复调用同一工具的bug。生产系统里我强制做了三个保护措施。一是给图编译时传入recursion_limit参数限制单次执行的节点上限超过直接终止并报错。二是给所有可逆边设置深度阈值当某个节点被访问超过N次就触发备用路径。三是在ReAct Agent的工具描述里明确告知如果已经掌握所需信息就停止调用工具从Prompt层面减少无效循环。5.3 LLM输出不稳定导致路由错误条件路由完全依赖LLM的输出而LLM是有随机性的。即使temperature设为0也不保证同一个Prompt每次都输出完全相同的结果。我在工单系统里就遇到过用户改了俩词意图分类就从order_query飘到了complaint后续整个处理链路全错了。对抗这个问题的关键手段是设计规范化。让LLM只输出预定义枚举值不要输出自由文本用few-shot示例强化格式约束条件路由函数里做模糊匹配和默认回退。比如判断意图时不是精确匹配字符串而是检查包含关系无法匹配时回退到人工处理分支宁慢勿错。5.4 问题速查表症状可能原因排查方向某个字段丢失多Agent并发覆盖检查Reducer合并策略调用超时LLM API限流或网络抖动加退避重试、检查并发配额Token消耗异常飙升循环调用或Prompt过长追踪日志、压缩上下文路由判断错误LLM输出不规范强化枚举约束、增加默认回退结果时好时坏缺少state信息检查是否有Agent依赖了未准备的字段系统吞吐量低单实例串行调用改异步、加队列Worker5.5 测试与回归多Agent系统的质量防线多Agent系统比普通应用更需要测试。传统的单元测试只能验证函数逻辑但Agent系统里很多行为是非确定性的LLM的每次输出都可能不同。我实践下来最有效的方法是录制回放测试在测试环境里把LLM的返回结果录制下来存成Fixture文件在CI流程里用录制的响应替换真实的LLM调用验证整个Agent图的状态流是否正确流转。这样能保证代码重构和依赖升级不会破坏已有的流程逻辑。真实的LLM行为留到集成测试阶段再验证数量不需要多但每个主流程都要覆盖到。写在最后的实操体会多Agent系统做了几个项目之后我个人最深的体会有三点。第一架构设计永远是第一位先画清楚状态和节点关系再写代码比上来就在流程图上糊一堆Agent要靠谱得多。第二可观测性建设要前置不要等出了问题才想到加日志一个跑在生产环境的Agent系统如果没有全链路追踪排查问题会像大海捞针。第三多Agent的价值在于处理真正需要分工协作的复杂任务如果一个流程用单Agent加简单工具就能解决就不要为了炫技而搞复杂架构。最后再分享一个小技巧本地调试多Agent系统的时候建议在LangGraph图里给每个Node的入口加一个断点或者debug print打印此时State的关键字段。这样你能直观看到每一步的信息流转很多看似玄学的问题其实一眼就能定位。这个习惯帮我省了无数排查时间希望对你也一样管用。