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 成本先不说，模型在几十个工具里选对的概率会明显下降。你需要一套机制让模型按需发现工具、按需加载描述。
MCP（Model 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(content=str(result), tool_call_id=call["id"])) return {"messages": results}收尾节点（finalize）：当模型不再请求工具调用时进入，把最终回复整理输出，同时可以做日志记录、状态清理。
这三个节点加上条件边，就构成一个完整的 ReAct 循环：思考→执行→再思考→再执行→直到收尾。LangGraph 的图结构天然支持这种循环，你只需要把tool_node的出口边指回think就行。
4.3 工具发现的缓存策略
每次思考都去拉一遍工具清单，在工具数量多、调用频繁的场景下会成为性能瓶颈。我的做法是加一层带过期时间的缓存：工具清单在内存里存一份，设定一个合理的过期时间（比如 60 秒），过期后重新拉取。这样既保证了工具变更能及时生效，又避免了每次调用都走网络。
class ToolCache: def __init__(self, client, ttl=60): 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(content=str(result), tool_call_id=call_id)]} except Exception as e: error_msg = f"工具 {name} 执行失败：{str(e)}" return {"messages": [ToolMessage(content=error_msg, tool_call_id=call_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_exceptions=True) # 处理结果...并发执行能把多工具场景的耗时从“累加”降到“取最大值”。但要注意并发安全——如果多个工具操作同一份资源，并发可能引发竞态。我的原则是：读操作可以并发，写操作必须串行。
6.2 超时控制：给每个工具设一个死线
外部工具响应时间不可控，必须设超时。我在 MCP Client 层统一加超时控制，每个工具调用超过设定时间就主动放弃，返回超时错误。超时时间按工具类型区分：本地查询类设 5 秒，远程接口类设 15 秒，批量处理类设 60 秒。
async def call_with_timeout(self, name, args, timeout=15): try: return await asyncio.wait_for( self._call(name, args), timeout=timeout ) 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 的打磨是个持续过程。我到现在还在根据线上日志不断调整工具描述，每次调整都能看到调用成功率的提升。这件事没有终点，但只要方向对，每一分投入都有回报。
