1. 从一个真实困境说起为什么单靠大模型搞不定复杂任务去年下半年我接手了一个内部知识助手的改造项目。需求听起来不复杂让模型能查数据库、能调内部接口、能读本地文档最后把结果整合成一段人话回复。我一开始的想法很朴素——写几个函数用if/else判断用户意图然后手动把结果拼进 prompt 里。结果第一版上线不到三天就崩了用户问“帮我看看上周那个订单为什么退款失败”模型需要先查订单、再查退款流水、再比对规则表中间任何一步的参数都依赖上一步的返回值。我那套硬编码的分支逻辑写到第七层就彻底失控了代码里全是嵌套判断改一个环节要动五个地方。这个经历让我意识到一个核心问题大模型本身只是一个“会说话的大脑”它没有手也没有脚。你让它推理可以但让它去真实世界里拿数据、改状态、触发动作就必须有一套机制把“思考”和“执行”串起来。这套机制要解决三件事第一模型怎么知道有哪些工具可用第二模型怎么决定调用哪个工具、传什么参数第三工具执行完之后结果怎么回到模型手里继续下一轮推理。早期大家用 Function Calling 硬扛能跑通简单场景但一旦任务变成多步骤、有分支、需要回退重试的流程代码就会迅速膨胀成一团乱麻。LangGraph 就是在这个背景下进入我视野的——它把整个任务流程建模成一张状态图节点是执行单元边是流转条件状态在节点之间传递和累积。而 MCP 协议解决的则是另一个维度的问题工具的标准化接入。以前每接一个外部系统就要写一套适配代码MCP 把这件事抽象成统一的协议工具提供方按协议暴露能力调用方按协议发现和调用双方解耦。把这两者结合起来就是我这篇要聊的核心用 LangGraph 编排流程用 MCP 对接工具搭出一套能真正跑复杂任务的智能体系统。这篇内容适合已经了解大模型基础调用、想往智能体方向深入的同学也适合正在被多步骤任务编排折磨的开发者。我会把踩过的坑、选型的理由、能直接抄的代码结构都摊开讲。2. 拆解 LangGraph 的状态图模型节点、边与状态到底怎么配合2.1 为什么是“图”而不是“链”大多数人接触 LangChain 是从 Chain 开始的LLMChain、SequentialChain这类线性结构直观好用。但线性结构有个致命缺陷它假设流程是一条道走到黑。真实任务里充满了条件分支——“如果查到的订单状态是已发货就走物流查询如果是待付款就走支付引导”。用 Chain 实现这种逻辑你只能在中间塞一个路由函数然后手动管理后续走向本质上还是回到了if/else的老路。LangGraph 的图模型把这件事彻底改观了。一张图由三个要素构成State状态一个贯穿全流程的共享数据结构通常是个字典或 TypedDict。每个节点都能读取它、修改它修改结果会自动合并回全局状态。Node节点一个执行单元本质是个函数接收当前状态返回状态更新。节点里可以放模型调用、工具执行、数据转换任何逻辑。Edge边定义节点之间的流转关系。普通边是固定的 A→B条件边则根据状态里的某个字段动态决定下一步去哪。我习惯用一个类比来解释State 是接力棒Node 是跑道上的一棒选手Edge 是交接规则。接力棒上记录着当前跑到哪了、已经完成了什么、下一步该谁接。选手跑完自己那段把棒子交给下一个交接规则决定了是正常传递还是临时改道。2.2 状态设计里最容易犯的错状态设计是 LangGraph 实战里第一个分水岭。新手最常见的错误是把状态设计得太“胖”——把所有可能用到的字段一股脑塞进去包括临时变量、中间结果、调试信息。这样做的后果是状态膨胀节点之间耦合严重改一个字段要检查所有节点。我的经验是状态里只放“跨节点必须共享”的东西。具体来说分三类输入类用户原始问题、会话标识、上下文参数。累积类已经执行过的步骤记录、已获取的数据片段。这类字段通常用Annotated配合operator.add做追加合并而不是覆盖。控制类当前步骤计数、下一步路由标记、错误信息。举个具体的状态定义例子from typing import TypedDict, Annotated import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] # 对话与工具调用记录追加式 user_query: str # 原始问题 tool_results: Annotated[list, operator.add] # 工具返回结果追加式 next_action: str # 路由控制字段 retry_count: int # 重试计数这里messages和tool_results用了operator.add意味着每次节点返回新列表时是追加而不是替换。这个细节极其关键——如果你忘了加Annotated后一个节点的返回值会直接把前面的记录覆盖掉导致模型丢失上下文表现为“聊着聊着就忘了前面查过什么”。我当初在这个坑里蹲了整整一个下午日志里明明有工具返回模型却像没看见一样。2.3 条件边让流程真正“活”起来条件边是 LangGraph 区别于普通工作流引擎的灵魂。它的写法是给add_conditional_edges传一个路由函数函数读取当前状态返回一个字符串这个字符串对应预先注册好的节点名。def route_after_think(state: AgentState) - str: last_msg state[messages][-1] if last_msg.tool_calls: return execute_tool return finalize graph.add_conditional_edges( think, route_after_think, {execute_tool: tool_node, finalize: end_node} )路由函数的返回值必须是映射表里的键否则运行时会直接报错。我建议路由函数保持纯粹——只读状态、只做判断、不产生副作用。有些人喜欢在路由函数里顺手改状态这是大忌因为路由函数可能被多次调用比如图编译时的校验阶段副作用会导致不可预期的行为。还有一个实战技巧给条件边留一个“兜底分支”。真实任务里模型可能返回意料之外的输出路由函数如果没覆盖到就会抛异常。我通常会在映射表里加一个默认走向比如__default__: error_handler让异常情况有个体面的收场而不是整个流程崩掉。3. MCP 协议到底解决了什么工具接入的标准化之路3.1 没有 MCP 之前工具接入有多痛在 MCP 出现之前我给模型接工具的方式基本是“一事一议”。接数据库要写一套 SQL 执行封装接内部 HTTP 接口要写一套请求构造和响应解析接本地文件系统又要写一套路径处理和内容读取。每个工具的入参格式、返回格式、错误码都不一样模型要在 prompt 里记住每个工具的调用约定稍微复杂一点就开始胡编参数。更麻烦的是工具发现。当工具数量从 3 个涨到 30 个你不可能把所有工具描述都塞进 system prompt——token 成本先不说模型在几十个工具里选对的概率会明显下降。你需要一套机制让模型按需发现工具、按需加载描述。MCPModel Context Protocol的思路很直接定义一套标准协议工具提供方按协议暴露能力调用方按协议发现和调用。它把工具接入拆成三个标准动作发现Discovery调用方询问“你有哪些工具”提供方返回工具清单和每个工具的入参 schema。调用Invocation调用方按 schema 传参提供方执行并返回结构化结果。上下文传递Context工具执行需要的额外上下文比如认证信息、会话状态通过协议字段传递而不是硬编码在工具里。这套协议的价值在于解耦。工具开发者不需要关心调用方是哪个模型、用什么框架调用方也不需要为每个工具写适配层只要按协议说话就行。我把它类比成 USB 接口——以前每个设备一个专用口现在统一成 USB-C插上就能用。3.2 MCP 的通信层与传输方式MCP 协议本身是分层的。上层是语义层定义工具描述、调用请求、返回结果的 JSON 结构下层是传输层负责消息怎么在两端之间传。传输层常见两种模式传输模式适用场景特点标准输入输出本地进程间通信简单直接工具作为子进程启动通过 stdin/stdout 交换 JSON 消息HTTP 流式远程服务调用工具部署在独立服务上通过 HTTP 长连接传输支持流式返回本地开发阶段我强烈建议先用标准输入输出模式。工具作为子进程跑在本机调试方便日志直接打到终端出问题一眼能看到。等工具稳定了、需要多实例共享了再迁移到 HTTP 模式。这个迁移成本不高因为语义层是统一的换的只是传输层。3.3 工具描述的质量决定调用成功率MCP 协议规定了工具描述的格式但描述内容的质量完全取决于工具开发者。我见过太多工具描述写得像天书模型根本看不懂什么时候该用它。一个好的工具描述应该包含四要素功能一句话这个工具干什么用大白话。适用场景什么情况下该调用它最好给一两个例子。参数说明每个参数的类型、含义、是否必填、取值范围。返回说明返回什么结构关键字段是什么。举个例子一个查询订单的工具描述写成“查询订单信息”就是不及格写成“根据订单号查询订单的当前状态、金额和创建时间适用于用户询问订单进度或退款原因的場景参数 order_id 为字符串类型的订单编号”才算合格。模型选工具靠的就是这段描述描述越精准选错的概率越低。4. 把 LangGraph 和 MCP 接起来一套可复用的编排骨架4.1 整体架构三层分离我在多个项目里沉淀下来的架构是三层分离编排层LangGraph负责流程控制决定“先干什么、后干什么、什么条件下改道”。协议层MCP Client负责工具发现和调用把编排层的意图翻译成 MCP 协议消息。工具层MCP Server负责具体执行每个外部系统对应一个 Server按协议暴露能力。这三层之间通过明确定义的接口通信任何一层内部改动不影响其他层。比如我想把订单查询从数据库换成内部 API只需要改工具层的 Server 实现编排层和协议层完全不用动。4.2 核心节点设计思考、执行、收尾一张典型的智能体图包含三个核心节点思考节点think调用大模型把当前状态里的对话历史和工具清单一起发给模型让模型决定下一步是调用工具还是直接回答。这个节点的关键是工具清单要动态注入——从 MCP Client 拉取当前可用的工具列表转成模型能理解的格式塞进请求。async def think_node(state: AgentState): tools await mcp_client.list_tools() model_with_tools model.bind_tools(tools) response await model_with_tools.ainvoke(state[messages]) return {messages: [response]}执行节点tool_node读取模型返回的工具调用请求通过 MCP Client 转发给对应的 Server拿到结果后包装成工具消息追加回状态。async def tool_node(state: AgentState): last_msg state[messages][-1] results [] for call in last_msg.tool_calls: result await mcp_client.call_tool(call[name], call[args]) results.append(ToolMessage(contentstr(result), tool_call_idcall[id])) return {messages: results}收尾节点finalize当模型不再请求工具调用时进入把最终回复整理输出同时可以做日志记录、状态清理。这三个节点加上条件边就构成一个完整的 ReAct 循环思考→执行→再思考→再执行→直到收尾。LangGraph 的图结构天然支持这种循环你只需要把tool_node的出口边指回think就行。4.3 工具发现的缓存策略每次思考都去拉一遍工具清单在工具数量多、调用频繁的场景下会成为性能瓶颈。我的做法是加一层带过期时间的缓存工具清单在内存里存一份设定一个合理的过期时间比如 60 秒过期后重新拉取。这样既保证了工具变更能及时生效又避免了每次调用都走网络。class ToolCache: def __init__(self, client, ttl60): self.client client self.ttl ttl self._cache None self._timestamp 0 async def get_tools(self): now time.time() if self._cache is None or now - self._timestamp self.ttl: self._cache await self.client.list_tools() self._timestamp now return self._cache这个缓存策略在工具数量稳定时几乎零开销在工具动态增减时也能在 TTL 内感知到变化。TTL 设多少取决于你的工具变更频率我一般设 60 秒工具特别稳定的场景可以设到 300 秒。5. 实战中踩出来的坑从参数错乱到死循环5.1 参数类型不匹配模型给的字符串工具要的是整数这是最高频的坑。模型在生成工具调用参数时经常把数字写成字符串。比如订单号12345模型可能返回12345。如果工具的参数 schema 定义的是整数MCP Server 收到字符串就会报类型错误。解决方案有两层第一层是在工具描述里明确标注参数类型让模型知道该传什么第二层是在 MCP Server 入口做类型转换兜底能转就转转不了再报错。我通常在 Server 的参数解析函数里加一段def coerce_param(value, target_type): if target_type int and isinstance(value, str): return int(value) if target_type float and isinstance(value, str): return float(value) return value这个兜底逻辑救了我无数次。模型不是每次都能严格遵守类型约定与其在 prompt 里反复强调不如在工程层面做好容错。5.2 工具调用死循环模型反复调同一个工具另一个经典问题是模型陷入死循环——它调用工具拿到结果后觉得结果不满意又调一次同样的工具如此往复。这种情况通常发生在工具返回空结果或错误信息时模型不知道该怎么处理就本能地重试。我的应对策略是在状态里加一个调用计数器同一个工具连续调用超过阈值就强制中断把控制权交给收尾节点让模型基于已有信息给出回复哪怕回复是“抱歉我没能查到相关信息”。def route_after_tool(state: AgentState) - str: if state[retry_count] 3: return finalize return think阈值设多少要看场景。查询类工具我设 3 次写入类工具我设 1 次——写操作重试风险太大宁可失败也不能重复执行。5.3 上下文丢失工具结果没进对话历史前面提过Annotated的坑这里再展开说一个相关的问题工具返回的结果格式。如果工具返回的是一个巨大的 JSON直接塞进对话历史会迅速吃满上下文窗口。我的做法是在工具节点里做一层结果摘要——只保留关键字段长文本截断结构化数据转成自然语言描述。def summarize_result(raw_result): if isinstance(raw_result, dict): keys [status, amount, created_at] return {k: raw_result.get(k) for k in keys if k in raw_result} return str(raw_result)[:500]这层摘要逻辑要放在工具节点里而不是让模型自己去解析原始结果。模型解析大 JSON 既费 token 又容易出错工程层面先处理好模型拿到的就是干净的关键信息。5.4 错误传播一个工具挂了整个流程崩外部工具不可靠是常态。网络超时、服务重启、认证过期任何一个环节出问题都可能让整个流程抛异常。如果不在图层面做错误处理一次工具失败就会导致整个任务中断。我的做法是给工具节点包一层异常捕获把异常转成结构化的错误消息返回给模型让模型决定是重试、换工具还是放弃。async def tool_node(state: AgentState): try: result await mcp_client.call_tool(name, args) return {messages: [ToolMessage(contentstr(result), tool_call_idcall_id)]} except Exception as e: error_msg f工具 {name} 执行失败{str(e)} return {messages: [ToolMessage(contenterror_msg, tool_call_idcall_id)]}这样模型收到的不是异常而是一条“工具执行失败”的消息它可以基于这条消息做出下一步决策。实测下来模型处理错误消息的能力比我们想象的要强很多时候它能自己找到替代路径。6. 性能与稳定性让这套系统扛住真实流量6.1 并发调用多个工具同时执行当模型一次返回多个工具调用请求时比如同时查订单和查物流串行执行会浪费时间。LangGraph 的节点本身支持异步你可以用asyncio.gather并发执行多个工具调用async def tool_node(state: AgentState): calls state[messages][-1].tool_calls tasks [mcp_client.call_tool(c[name], c[args]) for c in calls] results await asyncio.gather(*tasks, return_exceptionsTrue) # 处理结果...并发执行能把多工具场景的耗时从“累加”降到“取最大值”。但要注意并发安全——如果多个工具操作同一份资源并发可能引发竞态。我的原则是读操作可以并发写操作必须串行。6.2 超时控制给每个工具设一个死线外部工具响应时间不可控必须设超时。我在 MCP Client 层统一加超时控制每个工具调用超过设定时间就主动放弃返回超时错误。超时时间按工具类型区分本地查询类设 5 秒远程接口类设 15 秒批量处理类设 60 秒。async def call_with_timeout(self, name, args, timeout15): try: return await asyncio.wait_for( self._call(name, args), timeouttimeout ) except asyncio.TimeoutError: return {error: f工具 {name} 调用超时}超时控制配合前面的错误处理能让系统在部分工具不可用时依然保持可用。6.3 可观测性日志、追踪与回放智能体系统的调试难度远高于普通程序因为它的行为有随机性。同一套输入模型可能走出不同的路径。没有完善的可观测性出了问题根本无从下手。我在三个层面做记录节点级日志每个节点进出时记录状态快照包括输入状态、输出更新、耗时。工具调用日志每次工具调用记录工具名、参数、返回结果、耗时、是否成功。全链路追踪给每个任务分配一个 trace_id所有日志带上这个 id方便串联。LangGraph 本身支持回调机制可以挂载自定义的日志处理器。我通常会在图编译时传入一个回调配置把关键事件打到结构化日志里。这些日志在排查“为什么模型选了错误的工具”这类问题时是救命稻草。7. 几个值得单独拎出来的经验7.1 工具粒度宁可细一点不要粗一点设计工具时我倾向于把粒度做细。比如“查询订单”和“查询物流”拆成两个工具而不是合成一个“查询订单及物流”。原因有三细粒度工具描述更精准模型选择更准细粒度工具可以独立复用组合灵活细粒度工具出错时影响面小便于定位。当然也不能细到原子操作级别那样工具数量爆炸模型选择困难。我的经验法则是一个工具对应一个完整的业务动作这个动作有明确的输入输出能独立完成一件对用户有意义的事。7.2 状态字段的命名要“自解释”状态字段名不要用data、info、result这种含糊的词。用order_query_result、logistics_status、retry_count这种一看就知道是什么的名字。状态是跨节点共享的命名清晰能大幅降低维护成本。我接手过一个项目状态里有个字段叫temp全组没人知道它是干嘛的最后靠 git blame 才查出来是半年前一个临时需求留下的。7.3 给模型留“退路”不要假设模型每次都能完美决策。在 prompt 里明确告诉它如果信息不足可以请求用户补充如果工具失败可以告知用户并建议替代方案。给模型留退路比逼它硬扛要靠谱得多。我在 system prompt 里通常会加一句“如果连续两次工具调用都没有获得有效信息请直接告知用户当前无法完成并说明原因。”这句话显著降低了死循环的发生率。7.4 版本管理工具 schema 变更要谨慎MCP 工具的 schema 一旦对外暴露就成了一种契约。改 schema 要考虑向后兼容——加字段可以删字段和改类型要慎重。我的做法是给工具 schema 加版本号新版本上线时保留旧版本一段时间等调用方都迁移完再下线。这个策略在多人协作的项目里尤其重要避免你改了 schema 把别人的流程搞崩。8. 写在最后的一点个人体会这套 LangGraph 加 MCP 的组合我从去年底开始在生产环境跑中间迭代了七八个版本。最大的感受是智能体系统的难点不在模型而在工程。模型能力再强如果工具接入不规范、流程编排不清晰、错误处理不到位系统照样跑不起来。LangGraph 解决的是编排的工程问题MCP 解决的是工具接入的工程问题两者结合才让智能体从 demo 走向可用。如果你正准备上手我的建议是从小场景切入——先做一个只有两三个工具、流程简单的智能体把状态设计、条件边、错误处理这些基础打牢再逐步扩展工具数量和流程复杂度。一上来就搞几十个工具、十几层分支大概率会在调试阶段耗尽耐心。另外工具描述和 prompt 的打磨是个持续过程。我到现在还在根据线上日志不断调整工具描述每次调整都能看到调用成功率的提升。这件事没有终点但只要方向对每一分投入都有回报。