在实际开发 Agent 应用时很多人会遇到一个分水岭用 LangChain 写几个链式调用很容易一旦开始处理循环、条件分支、状态回写、多轮对话记忆代码就会变得难以维护。LangGraph 就是为了解决这类问题而出现的编排框架它把 Agent 的每一步都建模成图上的节点和边开发者可以精确控制状态如何流转、决策如何产生、工具如何被调用。而 MCPModel Context Protocol出现之后LangGraph 与 MCP 的组合又成了新的关注点一个负责 Agent 的运行编排一个负责接入外部工具和数据源。这篇文章会从 LangGraph 的核心机制讲起搭配可运行的最小示例再引入 MCP 工具集成最后给出常见报错、排查路径和生产落地建议帮助你从能跑通 Demo 走向能设计真实 Agent。在开始之前先明确适用范围。本文适合已经接触过 LangChain 基础调用但还没系统学习过 LangGraph 的开发者也适合已经写过简单 ReAct Agent但是对循环检测、条件路由、人类干预、状态管理缺乏完整认知的读者。文章中的代码以 Python 和 LangGraph 官方 API 为基础版本相关部分会给出验证方式落地时请以你本地的实际依赖版本为准。1. 先理解 LangGraph 到底解决了什么问题1.1 Agent 应用从链式调用走向图编排的必然原因在 LangChain 早期版本中最常见的 Agent 写法是链条结构大模型接收 prompt解析输出决定是否调用工具然后把工具结果拼回去再次调用大模型。这种写法在小场景下够用但一旦出现“先查库存、再判断是否需要采购、如果价格波动过大则进入人工审批、无论是否审批都需要发通知”这类真实业务链条表达就非常吃力。原因在于真实流程往往有分支、循环和异步等待不是线性顺序。LangGraph 的出现改变了这个建模方式。它借鉴了图数据库和状态机的设计思想把 Agent 运行过程拆成三部分State全局状态所有节点共享节点可以读取和修改。Node具体的处理步骤每个节点是一个函数或可调用对象。Edge节点之间的连接关系分为普通边、条件边、循环边和入口边。这种设计带来的直接好处是每一步都会修改状态而状态的变更可以通过图定义完整追踪不会出现“这个变量到底被谁改了”的混乱。LangGraph 也因此能够天然支持循环这对 Agent 特别重要因为大模型经常需要“思考 - 调用工具 - 观察结果 - 再思考”的多轮迭代如果只允许线性结构这个循环只能靠外部 while 模拟图内无法表达。1.2 LangGraph、LangChain、LangSmith 和 MCP 的边界很多初学者分不清这几个项目的关系。LangGraph 是一个独立的 Agent 编排运行时它不依赖 LangChain 也能使用但 LangChain 提供的模型封装、Prompt 模板、输出解析器在 LangGraph 节点内非常好用所以两者经常搭配。如果把 Agent 比作一个公司LangGraph 是业务流程和行政审批系统LangChain 是各个部门提供的标准工具包LangSmith 是监控和日志平台MCP 则是统一插线板让外部工具以标准协议接入。需要特别注意的是LangGraph 不等于 Mcp。LangGraph 解决“流程怎么编排”MCP 解决“工具怎么接入”。一个 Agent 可以完全不用 MCP而是在节点内部写死一个函数调用也可以用 MCP 把 GitHub、数据库、本地文件系统、设计稿等工具统一暴露给模型。两者可以同时使用LangGraph 节点内通过 MCP Client 调用远程工具然后把结果写入 State继续驱动图流转。组件核心职责典型使用方式LangGraphAgent 状态、节点、边、循环、条件分支构建可执行的状态图LangChain模型封装、Prompt、输出解析、工具封装在 LangGraph 节点内调用LangSmith链路追踪、评估、监控观察 LangGraph 每一次运行MCP标准化外部工具接入协议通过 MCP Client 调用远程工具1.3 用一句话理解 StateGraph 的运行机制LangGraph 中最核心的类是 StateGraph。开发者定义一个 State 类型然后往图里添加节点和边最后编译成可执行的 App。调用 App 时传入初始状态从入口节点开始执行每经过一个节点更新 State遇到条件边时根据状态决定走哪条分支遇到循环边时回到指定节点继续执行直到满足结束条件。这种机制和消息队列、工作流引擎的最大区别在于状态就是图本身的一部分而不是图外部维护的临时变量。节点函数唯一需要关心的就是输入 State 和输出 StateLangGraph 会根据图定义决定把哪个字段传给哪个节点从而避免开发者在函数间手动传递参数的麻烦。后面实现章节会通过代码把这一点体现出来。2. 环境准备和依赖版本对齐2.1 安装 LangGraph 和必要的周边库开始前先确认 Python 版本。LangGraph 官方长期支持 3.9 以上版本但一些新特性在 3.10 和 3.11 上表现更稳定。建议使用 3.10 或 3.11虚拟环境隔离是推荐做法。python -m venv .venv source .venv/bin/activate pip install --upgrade pip pip install langgraph langchain-openai如果后续需要接入 MCP 工具还需要安装 MCP 客户端依赖pip install mcp安装完成后先验证版本。这一步很容易被跳过但版本不匹配是 LangGraph 常见报错来源。python -c import langgraph; print(langgraph.__version__) python -c import langchain_openai; print(langchain_openai.__version__)如果你的项目原来已经安装过 langchain 和 langgraph升级前要确认依赖树是否冲突。LangGraph 的 API 在 0.1.x 到 0.2.x 之间有较明显调整尤其 StateGraph 的创建形式、add_conditional_edges 的传参方式、Command 对象的用法不同版本代码风格差异很大。不要盲目照搬网上旧帖子的代码先确认版本再改代码。2.2 配置大模型访问凭证LangGraph 本身不含大模型需要对接一个模型服务。推荐学习阶段使用 OpenAI 兼容接口因为你可能已经在使用各种国内或海外模型服务它们多数提供了 OpenAI 兼容的 Base URL。export OPENAI_API_KEYyour-api-key-here如果使用的是兼容接口在代码中显式指定 base_urlfrom langchain_openai import ChatOpenAI model ChatOpenAI( modelyour-model-name, base_urlhttps://api.example.com/v1, api_keyyour-key, temperature0, )生产环境不要把 API Key 写在代码里使用环境变量或密钥管理服务。学习阶段可以用环境变量但要注意不要提交到 Git 仓库。2.3 确认 LangGraph 官方文档版本和示例代码风格LangGraph 官方文档更新速度很快网上很多教程写的是旧版本 API。当你发现某段代码无法运行时第一反应不应该是改代码去适配而是去官方文档查看当前版本的 StateGraph 和 add_node 签名。这里有一个比较稳妥的排错思路用 print 输出当前 langgraph 版本。在官方文档导航中切换对应版本。对比节点函数签名和add_conditional_edges传参方式。如果代码中用了StateGraph(State)确认新版本是否推荐from typing import TypedDict或dataclass方式定义状态。官方示例代码通常是最小可运行的先在官方示例基础上跑通再替换成自己的业务逻辑比直接套用网上长篇大论更高效。3. 从零构建一个可运行的 LangGraph 最小图3.1 定义 State用 TypedDict 还是 PydanticState 是 LangGraph 中所有节点的共享数据容器。我们可以使用 Python 的 TypedDict也可以使用 Pydantic BaseModel。学习阶段用 TypedDict 最直接因为它类型标注清晰、运行开销小足够满足大多数场景。from typing import TypedDict, Annotated, List class AgentState(TypedDict): messages: List[str] steps: int finished: bool字段含义messages保存对话历史或中间过程消息。steps记录已经执行了多少步用于循环控制。finished标记当前流程是否结束。这里要注意LangGraph 的 State 更新默认是覆盖式替换。如果节点返回了{messages: [new]}旧 messages 会被整体覆盖。如果希望追加而不是覆盖可以通过Annotated加上 reducer 函数。这个知识点是 LangGraph 最容易踩坑的地方之一后面会详细讲解。3.2 编写节点函数和条件边节点函数的核心签名是(state) - dict输入当前完整 State输出一个部分更新的字典。LangGraph 会把输出合并回 State。下面实现一个简单循环节点 A 生成一条消息条件边判断 steps 是否达到阈值没有达到就回到 A达到就进入节点 B。from langgraph.graph import StateGraph, START, END def node_a(state: AgentState): print(fnode_a 执行当前 steps{state[steps]}) return { messages: state[messages] [fstep-{state[steps]}], steps: state[steps] 1, } def node_b(state: AgentState): print(node_b 执行流程结束) return {finished: True, messages: state[messages] [done]} def should_continue(state: AgentState): if state[steps] 3: return a # 返回节点名 return b然后构图graph StateGraph(AgentState) graph.add_node(a, node_a) graph.add_node(b, node_b) graph.add_edge(START, a) graph.add_conditional_edges(a, should_continue, {a: a, b: b}) graph.add_edge(b, END) app graph.compile()运行initial_state {messages: [], steps: 0, finished: False} result app.invoke(initial_state) print(result)预期输出中 node_a 会被执行三次steps 递增最终进入 node_bfinished 为 True。3.3 关键点条件边返回的是节点名还是 Next 对象在 LangGraph 新版本中条件边的目标映射可以是一个字典{a: a, b: b}表示当 should_continue 返回a时跳转到节点a返回b时跳转到节点b。如果条件函数返回的字符串不在映射字典中运行时会报错。另一个常见写法是使用Command对象比如节点内部直接决定下一步。这个写法在更复杂的场景下很有用因为节点可以不经过条件边直接跳转但会提高理解成本。初学阶段先掌握外部条件边写法等理解了调用流程再使用 Command。这里也要区分普通边和条件边。普通边是无条件跳转graph.add_edge(b, END)。条件边是根据 State 动态判断跳转目标graph.add_conditional_edges(a, should_continue, mapping)。循环的本质就是条件边中有一条分支指向了自己或前面的节点。3.4 用 Annotated reducer 实现消息追加默认覆盖式更新在真实场景中很难用因为我们通常希望把新消息追加到历史列表而不是每次都覆盖。LangGraph 提供了一种机制叫 reducer定义在 State 字段上节点返回新值时会自动调用 reducer 合并。from typing import Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[List[str], add_messages]langgraph.graph.message中内置了add_messages这个 reducer。它的行为是如果新消息有 id 且与旧消息冲突就覆盖否则追加。对文本字符串列表它会按顺序合并。这里有一个常见误区如果多次运行同一个 app 并传入相同初始状态State 中的消息并不会自动清空因为 LangGraph 的 State 是一次 invoke 内独立的。每次调用传新状态即可。若要实现多轮对话持久化后面可以引入 Checkpointer 或外部存储。4. 条件路由与分支控制从简单循环到真实 Agent4.1 条件路由的典型业务场景在数学计算应用里模型根据用户提问决定调用哪类工具比如查天气、算积分、查数据库、搜索文档。如果用线性链模型必须先做意图判断然后走入一个固定分支。问题是真实意图可能组合出现比如“帮我计算北京今天气温的 3 倍”需要先调用天气工具再调用计算器这依赖多次工具调用。LangGraph 的节点设计非常适合这种情况。假设有两个工具节点tool_weather和tool_calculator模型节点agent负责判断下一步调用哪个工具条件边从agent出发映射到不同工具节点工具节点执行后回到agent形成循环。def agent_node(state: AgentState): # 这里简化为规则判断实际项目中用大模型选择工具 user_input state[messages][-1].lower() if 天气 in user_input: return {next_tool: weather} elif 计算 in user_input: return {next_tool: calculator} return {next_tool: end} def weather_node(state: AgentState): return {messages: [天气工具返回晴25度]} def calculator_node(state: AgentState): return {messages: [计算器返回42]} def route_after_agent(state: AgentState): return state[next_tool]这种“分类 - 执行 - 回到入口再分类”的模式是所有 ReAct Agent 的基础。真实写法中agent_node内部会调用大模型根据模型输出判断需要调用哪个工具并把工具名和参数写入状态。4.2 循环检测为什么 Agent 会卡死以及 LangGraph 怎么防止循环在 LangGraph 中是核心能力但无限循环是生产环境的灾难。虽然你可以在节点里通过 steps 字段手动控制轮次但更规范的做法是设置recursion_limit。在 invoke 时指定result app.invoke(initial_state, config{recursion_limit: 10})当执行步数超过限制时LangGraph 会抛出异常。这个限制是一种安全网但依赖它作为唯一保护并不够因为业务逻辑可能根本不需要那么多步或者某些循环是合法的多轮工具调用。更合理的做法是设计最终节点判断逻辑当模型返回结束标记时条件边走 END当 steps 达到阈值时强制走结束分支。有人会把无限循环归咎于 LangGraph 框架其实问题往往出在状态设计上。如果条件边的判断字段一直不更新那么无论循环多少次都会走同一条边。所以在实际代码中要特别关注节点的输出是否真的修改了条件边读取的字段以及修改后的值是否导致分支条件发生变化。4.3 子图的拆分与复用当图变得很大所有节点堆在一起就很难维护。LangGraph 支持把一个编译好的图作为另一个图的节点称为子图。这种模式和函数拆分类似目的是复用和隔离。比如先做一个“工具调用子图”包含模型判断、工具执行、结果返回再做一个“总流程图”包含数据校验、调用子图、汇总输出。tool_subgraph graph.compile() class MainState(TypedDict): query: str result: str steps: int def run_tool_subgraph(state: MainState): sub_result tool_subgraph.invoke({ messages: [state[query]], steps: 0, finished: False, }) return {result: sub_result[messages][-1]}子图内部的状态和外层状态是隔离的需要通过输入输出字段做映射。这个特性适合把多个通用能力封装成独立模块让主图只编排大流程。5. 将 MCP 集成到 LangGraph 节点中5.1 MCP 是一个协议不是某一个具体的库很多人在搜索时会把 MCP 和特定插件绑定其实 MCPModel Context Protocol是一个开放协议定义了大模型应用如何发现外部工具、调用外部工具、读取资源。它由协议层、客户端层、服务端层组成。服务端可以是一个本地进程、一个远程 HTTP 服务甚至是一个 IDE 插件。架构主体如下MCP Server托管工具对外暴露工具列表和调用接口。MCP Client负责发现工具列表调用工具并返回结果。Agent 应用在 LangGraph 节点内作为 Client 发起调用。LangGraph 本身不内置 MCP但可以在节点函数里使用 MCP SDK 连接 Server。LangChain 生态提供了一些集成工具也可以用原生 SDK 手动调用这样更可控。5.2 用一个本地 MCP 服务做最小集成下面使用 Python 版 MCP SDK 实现一个简单的数学工具服务端然后用 LangGraph 节点调用。先创建服务端文件math_server.pyfrom mcp.server.fastmcp import FastMCP mcp FastMCP(MathServer) mcp.tool() def add(a: int, b: int) - int: 计算两个整数之和 return a b if __name__ __main__: mcp.run()启动服务python math_server.py如果在进程内测试LangGraph 节点中可以直接声明连接from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client import asyncio server_params StdioServerParameters( commandpython, args[math_server.py], ) async def call_math_add(a: int, b: int) - int: async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() result await session.call_tool(add, {a: a, b: b}) return result.content[0].text然后在 LangGraph 节点内部调用异步函数。需要注意的是LangGraph 的节点可以是同步函数也可以是异步函数。如果用了异步节点整个链路最好都使用ainvoke。async def math_tool_node(state: AgentState): a state.get(a, 1) b state.get(b, 2) value await call_math_add(a, b) return {messages: state[messages] [fadd result: {value}]}这样 MCP 就打通了Agent 在 LangGraph 中运行节点节点通过 MCP Client 调用服务端工具工具结果写回 State继续驱动后续节点。5.3 官方 Remote MCP Server 或社区封装怎么选在搜索材料中可以看到大量 MCP 相关词汇比如 Figma MCP、Playwright MCP、Chat2DB MCP、GitHub MCP 等。这些本质都是“特定工具对外提供 MCP Server 接口”。使用它们的方式有两种直接通过官方或社区提供的 MCP Server 进程启动。在 LangGraph 节点中连接该 Server像调用本地工具一样调用。对于生产项目建议先确认 Server 是否维护活跃、协议版本是否与 MCP SDK 兼容、是否需要提供长期运行的服务地址。某些 MCP Server 适合本地开发比如 Playwright MCP 用于浏览器自动化某些适合部署为远程服务比如内部数据库 MCP。选型时应考虑部署复杂度、权限模型和数据安全。5.4 使用 MCP 时最常见的几个坑第一个坑手动调用session.call_tool时参数格式错误。MCP 协议要求参数是 JSON 对象有些工具声明为必填字符串传成数字时会报 schema 校验失败。建议在实现前先打印服务端工具 schema确认参数类型和必填字段。第二个坑异步环境没有初始化asyncio.run导致报错。在同步节点内运行异步客户端时很多人直接调用asyncio.run()嵌套在事件循环中会报错。推荐把整个 LangGraph 应用写成异步方式使用ainvoke避免嵌套事件循环。第三个坑MCP Server 进程没有启动或端口不通。本地 stdio 模式会由客户端拉起进程但如果服务端文件路径、命令参数不对会出现连接建立失败或初始化超时。先用官方 SDK 的最小示例把 MCP Server 单独跑通再接入 LangGraph这样排错范围可以缩小到一层。下表整理了常见现象和处理优先级问题现象常见原因检查顺序调用工具返回 schema 错误参数名或类型不匹配打印工具 schema对比调用参数工具调用超时Server 进程未启动或网络不通先单独启动 Server 并用 CLI 测试事件循环乱同步节点嵌套运行异步函数改为异步节点和 ainvoke工具结果丢失节点未把工具结果写入 State检查节点返回字典中的 key6. 用 LangGraph 实现一个能实际运行的 ReAct Agent6.1 ReAct 范式在 LangGraph 中的映射ReAct 的核心循环是 Thought思考- Action行动- Observation观察。在 LangGraph 中这个循环可以映射为四个节点和一条条件边agent节点接收用户输入和工具结果让大模型决定下一步。如果决定结束则输出最终答案如果需要调用工具则输出工具名和参数。tools节点根据 agent 节点的输出执行实际工具函数或调用 MCP Server。conditional_edges根据 agent 节点输出判断下一步去tools还是END。工具结果作为新的消息追加到 State 的 messages 中重新进入agent节点。这种实现和传统 LangChain AgentExecutor 的区别在于LangGraph 中的每一步都是显式节点开发者可以插入日志、权限校验、人工审批、重试逻辑和限流而不用修改整个执行器。6.2 手写一个可运行的 ReAct Agent 代码为了减少依赖版本干扰下面用 LangChain 的消息类型保存历史用两个简单工具函数模拟工具调用。真实项目可以把这两个工具替换成 MCP Client 调用。import json from typing import Annotated, TypedDict from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langchain_core.messages import HumanMessage, AIMessage, ToolMessage class ReActState(TypedDict): messages: Annotated[list, add_messages] def fake_weather_tool(city: str) - str: return f{city} 今天晴气温 25 度 def fake_calculator_tool(expression: str) - str: # 这里仅做演示生产环境要用安全表达式解析 return f计算结果{eval(expression)} TOOLS { weather: fake_weather_tool, calculator: fake_calculator_tool, } model ChatOpenAI(modelgpt-4o-mini, temperature0) tools_desc [ {type: function, function: { name: weather, description: 查询城市天气, parameters: { type: object, properties: {city: {type: string}}, required: [city], }, }}, {type: function, function: { name: calculator, description: 计算数学表达式, parameters: { type: object, properties: {expression: {type: string}}, required: [expression], }, }}, ] model_with_tools model.bind_tools(tools_desc) def agent_node(state: ReActState): response model_with_tools.invoke(state[messages]) if response.tool_calls: return {messages: [response]} return {messages: [response]} def tools_node(state: ReActState): last_message state[messages][-1] tool_messages [] for tool_call in last_message.tool_calls: tool_name tool_call[name] tool_args tool_call[args] tool_func TOOLS.get(tool_name) if tool_func: result tool_func(**tool_args) else: result f未找到工具: {tool_name} tool_messages.append(ToolMessage(contentstr(result), tool_call_idtool_call[id])) return {messages: tool_messages} def should_continue(state: ReActState): last_message state[messages][-1] if getattr(last_message, tool_calls, None): return tools return end graph StateGraph(ReActState) graph.add_node(agent, agent_node) graph.add_node(tools, tools_node) graph.add_edge(START, agent) graph.add_conditional_edges(agent, should_continue, {tools: tools, end: END}) graph.add_edge(tools, agent) app graph.compile()运行测试result app.invoke({ messages: [HumanMessage(content北京今天天气怎么样)] }) print(result[messages][-1].content)大模型会先输出 tool_call 调用 weather 工具然后 tools 节点执行 fake_weather_tool 返回工具结果agent 节点再次运行最终输出自然语言答案。6.3 如何加入人类审批节点真实生产环境经常会遇到“机器决定改数据库”无论模型多可靠都应该先暂停让人类确认。LangGraph 可以轻松插入 interrupt 或设计一个等待人类确认的节点。最简单的方式是使用interrupt_before配置在进入tools节点之前暂停。config {recursion_limit: 20, interrupt_before: [tools]}这样在工具调用前执行会暂停需要人工确认后继续。LangGraph 还支持持久化 Checkpointer配合恢复接口实现真正的“等待人工确认 - 继续执行”。这个机制特别适合审批流场景也是 LangGraph 相比普通函数调用链的优势。6.4 输出验证和结果分析验证一个 Agent 不能只看最终是否输出了答案。需要确认以下信息模型是否做出了正确的工具调用工具名和参数是否合理。工具执行结果是否正确回写到 messages。条件边是否正确回到 agent 或走向结束。多轮工具调用时消息顺序是否清晰有没有出现消息错乱。整体是否存在多余的工具调用或超时。建议在开发阶段开启调试输出至少打印每一步的节点名和关键状态变化。LangSmith 是最完整的观测方案但如果只需要本地快速调试直接 print 也能解决大部分问题。app graph.compile() for event in app.stream(initial_state, configconfig): for node_name, node_state in event.items(): print(f节点 {node_name} 已执行) print(node_state)stream返回的每个事件包含节点名和该节点执行后的状态快照。这对理解 LangGraph 的执行流程非常有帮助。7. 状态持久化、记忆和 Checkpointer7.1 为什么 Agent 需要状态持久化Web 应用通常有会话机制Agent 应用也一样。用户说“刚才那个城市再查一遍”如果 Agent 不保存上一轮状态就不知道该查哪个城市。LangGraph 的 State 在一次 invoke 内有效多次对话之间默认不共享。要实现真正的多轮记忆需要把 State 保存到外部存储每次调用前加载上次的 State。LangGraph 的解决方案是 Checkpointer。它负责在每次节点执行后保存状态快照并支持根据 thread_id 恢复。对于内存学习可以使用内存版 Checkpointer但生产环境建议使用 Postgres 或 Redis 等持久化方案。7.2 使用 Checkpointer 实现多轮对话先安装必要的依赖pip install langgraph-checkpoint使用内存版 Checkpointerfrom langgraph.checkpoint.memory import MemorySaver memory MemorySaver() app graph.compile(checkpointermemory) config {configurable: {thread_id: user-123}} result app.invoke( {messages: [HumanMessage(content帮我查北京天气)]}, configconfig, )第二次提问时使用相同 thread_idAgent 会自动加载之前的 messages 作为上下文result app.invoke( {messages: [HumanMessage(content上海呢)]}, configconfig, )模型会根据历史判断“上海呢”是继续查天气从而自动调用 weather 工具。没有记忆时这种省略主语的能力会完全失效。7.3 Checkpointer 和数据库事务的取舍生产环境要注意 Checkpointer 本身也是一类写操作。如果把 Checkpointer 配置到业务数据库要考虑每次节点执行都写入状态快照带来的压力。某些高并发场景可以缩短保存周期或者在关键节点才开启持久化而不是每个节点都写库。LangGraph 的持久化能力是把双刃剑使用方便但如果状态里包含敏感字段保存快照时要注意脱敏。不要把用户密码、Token、完整原始请求全部写入 Checkpointer应该只保存记忆所需的最小字段。8. LangGraph 与 LangChain 的分工和常见选型问题8.1 LangChain 是否过时LangGraph 会替代它吗从技术演进看LangGraph 和 LangChain 不是替代关系而是不同抽象层级。LangChain 提供的是“组件库”模型封装、Prompt 模板、输出解析器、向量存储封装、工具定义。LangGraph 提供的是“运行时”状态、节点、边、循环、分支、持久化。一个可运行的 Agent 通常两者都用。在某些新项目中LangGraph 已经被当成 LangChain 生态的更高层框架LangChain AgentExecutor 则逐渐退居后台。但如果任务只是简单的 RAG 问答链使用 LangChain 的 LCEL 表达链就够了不必非要上 LangGraph。所有流程都会先从简单开始当出现多轮工具调用、条件分支、人工审批时再迁移到 LangGraph 也不迟。8.2 Agent Skill 和 MCP 的区别搜索热词中多次出现“Agent Skill 和 MCP 有什么区别”。这一组概念容易混淆。MCP 是协议解决的是“工具如何被外部发现和调用”Skill 是一种能力封装解决的是“Agent 如何组织和复用一个完整的任务流程”。类比来说MCP 定义插头规格Skill 定义插电后运行的设备逻辑。一个 Skill 内部可以依赖多个 MCP 工具MCP 工具也可以被多个 Skill 复用。在 LangGraph 中Skill 通常体现为子图或节点组合MCP 工具则通过 Client 在节点内调用。两者配合时最清晰的边界是Skill 负责决策和编排MCP 负责执行具体动作。8.3 选型建议什么时候用 LCEL什么时候用 LangGraph场景推荐方案原因一次模型调用加格式化输出LangChain LCEL简单心智负担低连续多次调用模型并传递结果LCEL 也可以但状态复杂时更难维护可先用 LCEL复杂时再迁移工具调用循环、条件分支、多轮协商LangGraph图模型天然支持循环和分支需要人工审批、暂停恢复、持久化会话LangGraphCheckpointer 和 interrupt 机制完善高频低延迟的规则引擎不要用 Agent 框架直接用代码避免模型开销9. 生产环境落地 LangGraph 的排查清单和最佳实践9.1 最常见的报错现象与处理路径LangGraph 的报错信息通常不算难懂但因为涉及异步、状态更新和条件映射新手容易在几个固定位置卡住。下面整理一套按优先级执行的排查流程。首先要看是否是版本问题。报错中包含StateGraph、add_conditional_edges或Command的 AttributeError 或 TypeError 时大概率是代码照搬了旧版本教程。先升级或降级 langgraph 到官方示例对应版本再重新运行。其次要看 State 更新是否符合预期。节点返回的字典 key 必须存在于 State 定义中。如果返回了没有定义的字段LangGraph 默认会忽略或报错。建议定义 State 时把所有可能需要写回 State 的字段都列出来避免遗漏。再看条件边的映射。should_continue返回的字符串必须存在于映射字典中。很多报错信息会直接提示没有找到 path 或 node这时要打印条件边的返回值确认是否写错节点名。还要看异步事件循环。若同时在 Jupyter Notebook 或 IPython 环境中运行 async 代码注意asyncio.run冲突。建议总是使用await app.ainvoke(...)不要在一个已经有事件循环的环境中嵌套调用asyncio.run。最后看工具调用返回格式。如果大模型返回了 tool_calls但tools_node中没有正确生成 ToolMessageAgent 会一直卡在 agent 和 tools 之间循环直到 recursion_limit 报错。检查每个 tool_call 的 id 是否在 ToolMessage 中回传LangGraph 依赖 tool_call_id 把工具结果和模型请求对齐。9.2 开发环境与生产环境的配置差异开发环境可以关掉大部分安全限制方便调试生产环境需要默认启用防护。下面的表格给出了两者的主要差异。项目开发环境生产环境Prompt 调试直接 print 日志使用 LangSmith 或自定义链路日志API Key环境变量本地配置密钥管理服务禁止进代码库模型调用失败直接抛出重试熔断、降级、告警、错误队列Checkpointer内存版Postgres 或 Redis 持久化工具执行允许任意本地函数白名单工具、参数校验、权限控制日志级别DEBUGINFO 或 WARNING敏感信息脱敏不要在生产环境使用eval这类动态执行代码的工具。ReAct 示例中用了 eval 只是为了演示真实项目应该使用安全的数学表达式库或解析器避免提示注入造成代码执行风险。9.3 提示注入和工具权限控制Agent 一旦接入工具攻击面就扩大了。理论上模型输出完全受用户输入影响如果节点内部用模型输出直接拼接 shell 命令或 SQL就是严重安全漏洞。至少要做到以下几点工具参数必须做类型和取值范围校验。数据库查询建议使用参数化查询或预先定义 SQL 模板。高权限工具必须有人工审批节点不能允许模型直接执行。所有外部工具的调用日志要完整保留便于安全审计。MCP Server 的凭证要独立于模型凭证遵循最小权限原则。9.4 可复用的 Agent 发布前检查清单下面是一份可以打印出来对照的检查清单适合在每次发布前快速走查。检查每个节点是否有异常处理模型调用失败时是否有重试或降级。检查 State 中的字段是否都需要写回敏感字段是否在日志中打印。检查条件边所有可能返回值是否都有目标节点。检查 recursion_limit 是否合理循环保护是否生效。检查工具调用是否记录完整 tool_call_idToolMessage 是否配对。检查 MCP Server 连接方式在目标环境是否可访问、超时设置是否合理。检查 Checkpointer 是否配置thread_id 是否由业务上下文正确传递。检查是否在应用层限流防止模型和工具被高频调用。检查日志是否包含节点名、状态变化、工具输入输出摘要且不含敏感信息。检查是否有人工审批的逃生通道高权限操作不会被模型直接触发。9.5 针对 LangGraph 的学习路径建议LangGraph 的学习不应该一上来就追求复杂。建议按这个顺序推进先跑通 TODO 型最小图两个节点加一条边。加上条件边实现一个简单的循环计数。把循环和 ReAct 结合起来让模型决定是否调用工具。加入 Checkpointer实现多轮对话记忆。接入 MCP Server把真实外部工具嵌入节点。引入子图把可复用的 Agent 能力拆成模块。加入人工审批用 interrupt 或者状态机模拟审批流。最后接入 LangSmith对运行质量做可观测性评估。每一步都要在真实场景里重复三到五次直到你能在不看文档的情况下画出状态流转图才算真正理解。否则只是把代码复制到本地遇到报错还是无法判断问题出在节点、边、状态还是工具上。10. 扩展方向从 LangGraph 走向完整 Agent 工程LangGraph 只是 Agent 运行时的底座一个完整的 Agent 工程还需要处理数据接入、知识库、评估、监控、模型成本和版本迭代。在搜索材料中出现的 Fusion MCP、Playwright MCP、数据库 MCP、蓝湖 MCP、Figma MCP 都是一个方向让 Agent 能触达更多外部系统。真正要关注的问题不是数量而是稳定性和权限可控性。如果团队已经在使用 LangChain 生态LangGraph 是继续深挖 Agent 编排的最自然选择。如果团队从零开始也可以只用 LangGraph 加原生模型 SDK不引入 LangChain保持更轻的依赖。两种路线都成立关键看团队的技术栈、模型供应商和运维能力。对大多数开发者而言最有效的投资是把官方文档中的对话式 ReAct Agent、多节点协作、持久化对话这几个示例吃透再结合自己的业务场景做一版带状态流转图的 Agent。当你能够在纸上画出节点、条件边和状态字段再回看 LangGraph 代码会发现它不过是一套非常朴素的图执行引擎真正复杂的永远是业务规则和模型能力边界。
LangGraph与MCP集成：构建可控的AI Agent编排工作流
在实际开发 Agent 应用时很多人会遇到一个分水岭用 LangChain 写几个链式调用很容易一旦开始处理循环、条件分支、状态回写、多轮对话记忆代码就会变得难以维护。LangGraph 就是为了解决这类问题而出现的编排框架它把 Agent 的每一步都建模成图上的节点和边开发者可以精确控制状态如何流转、决策如何产生、工具如何被调用。而 MCPModel Context Protocol出现之后LangGraph 与 MCP 的组合又成了新的关注点一个负责 Agent 的运行编排一个负责接入外部工具和数据源。这篇文章会从 LangGraph 的核心机制讲起搭配可运行的最小示例再引入 MCP 工具集成最后给出常见报错、排查路径和生产落地建议帮助你从能跑通 Demo 走向能设计真实 Agent。在开始之前先明确适用范围。本文适合已经接触过 LangChain 基础调用但还没系统学习过 LangGraph 的开发者也适合已经写过简单 ReAct Agent但是对循环检测、条件路由、人类干预、状态管理缺乏完整认知的读者。文章中的代码以 Python 和 LangGraph 官方 API 为基础版本相关部分会给出验证方式落地时请以你本地的实际依赖版本为准。1. 先理解 LangGraph 到底解决了什么问题1.1 Agent 应用从链式调用走向图编排的必然原因在 LangChain 早期版本中最常见的 Agent 写法是链条结构大模型接收 prompt解析输出决定是否调用工具然后把工具结果拼回去再次调用大模型。这种写法在小场景下够用但一旦出现“先查库存、再判断是否需要采购、如果价格波动过大则进入人工审批、无论是否审批都需要发通知”这类真实业务链条表达就非常吃力。原因在于真实流程往往有分支、循环和异步等待不是线性顺序。LangGraph 的出现改变了这个建模方式。它借鉴了图数据库和状态机的设计思想把 Agent 运行过程拆成三部分State全局状态所有节点共享节点可以读取和修改。Node具体的处理步骤每个节点是一个函数或可调用对象。Edge节点之间的连接关系分为普通边、条件边、循环边和入口边。这种设计带来的直接好处是每一步都会修改状态而状态的变更可以通过图定义完整追踪不会出现“这个变量到底被谁改了”的混乱。LangGraph 也因此能够天然支持循环这对 Agent 特别重要因为大模型经常需要“思考 - 调用工具 - 观察结果 - 再思考”的多轮迭代如果只允许线性结构这个循环只能靠外部 while 模拟图内无法表达。1.2 LangGraph、LangChain、LangSmith 和 MCP 的边界很多初学者分不清这几个项目的关系。LangGraph 是一个独立的 Agent 编排运行时它不依赖 LangChain 也能使用但 LangChain 提供的模型封装、Prompt 模板、输出解析器在 LangGraph 节点内非常好用所以两者经常搭配。如果把 Agent 比作一个公司LangGraph 是业务流程和行政审批系统LangChain 是各个部门提供的标准工具包LangSmith 是监控和日志平台MCP 则是统一插线板让外部工具以标准协议接入。需要特别注意的是LangGraph 不等于 Mcp。LangGraph 解决“流程怎么编排”MCP 解决“工具怎么接入”。一个 Agent 可以完全不用 MCP而是在节点内部写死一个函数调用也可以用 MCP 把 GitHub、数据库、本地文件系统、设计稿等工具统一暴露给模型。两者可以同时使用LangGraph 节点内通过 MCP Client 调用远程工具然后把结果写入 State继续驱动图流转。组件核心职责典型使用方式LangGraphAgent 状态、节点、边、循环、条件分支构建可执行的状态图LangChain模型封装、Prompt、输出解析、工具封装在 LangGraph 节点内调用LangSmith链路追踪、评估、监控观察 LangGraph 每一次运行MCP标准化外部工具接入协议通过 MCP Client 调用远程工具1.3 用一句话理解 StateGraph 的运行机制LangGraph 中最核心的类是 StateGraph。开发者定义一个 State 类型然后往图里添加节点和边最后编译成可执行的 App。调用 App 时传入初始状态从入口节点开始执行每经过一个节点更新 State遇到条件边时根据状态决定走哪条分支遇到循环边时回到指定节点继续执行直到满足结束条件。这种机制和消息队列、工作流引擎的最大区别在于状态就是图本身的一部分而不是图外部维护的临时变量。节点函数唯一需要关心的就是输入 State 和输出 StateLangGraph 会根据图定义决定把哪个字段传给哪个节点从而避免开发者在函数间手动传递参数的麻烦。后面实现章节会通过代码把这一点体现出来。2. 环境准备和依赖版本对齐2.1 安装 LangGraph 和必要的周边库开始前先确认 Python 版本。LangGraph 官方长期支持 3.9 以上版本但一些新特性在 3.10 和 3.11 上表现更稳定。建议使用 3.10 或 3.11虚拟环境隔离是推荐做法。python -m venv .venv source .venv/bin/activate pip install --upgrade pip pip install langgraph langchain-openai如果后续需要接入 MCP 工具还需要安装 MCP 客户端依赖pip install mcp安装完成后先验证版本。这一步很容易被跳过但版本不匹配是 LangGraph 常见报错来源。python -c import langgraph; print(langgraph.__version__) python -c import langchain_openai; print(langchain_openai.__version__)如果你的项目原来已经安装过 langchain 和 langgraph升级前要确认依赖树是否冲突。LangGraph 的 API 在 0.1.x 到 0.2.x 之间有较明显调整尤其 StateGraph 的创建形式、add_conditional_edges 的传参方式、Command 对象的用法不同版本代码风格差异很大。不要盲目照搬网上旧帖子的代码先确认版本再改代码。2.2 配置大模型访问凭证LangGraph 本身不含大模型需要对接一个模型服务。推荐学习阶段使用 OpenAI 兼容接口因为你可能已经在使用各种国内或海外模型服务它们多数提供了 OpenAI 兼容的 Base URL。export OPENAI_API_KEYyour-api-key-here如果使用的是兼容接口在代码中显式指定 base_urlfrom langchain_openai import ChatOpenAI model ChatOpenAI( modelyour-model-name, base_urlhttps://api.example.com/v1, api_keyyour-key, temperature0, )生产环境不要把 API Key 写在代码里使用环境变量或密钥管理服务。学习阶段可以用环境变量但要注意不要提交到 Git 仓库。2.3 确认 LangGraph 官方文档版本和示例代码风格LangGraph 官方文档更新速度很快网上很多教程写的是旧版本 API。当你发现某段代码无法运行时第一反应不应该是改代码去适配而是去官方文档查看当前版本的 StateGraph 和 add_node 签名。这里有一个比较稳妥的排错思路用 print 输出当前 langgraph 版本。在官方文档导航中切换对应版本。对比节点函数签名和add_conditional_edges传参方式。如果代码中用了StateGraph(State)确认新版本是否推荐from typing import TypedDict或dataclass方式定义状态。官方示例代码通常是最小可运行的先在官方示例基础上跑通再替换成自己的业务逻辑比直接套用网上长篇大论更高效。3. 从零构建一个可运行的 LangGraph 最小图3.1 定义 State用 TypedDict 还是 PydanticState 是 LangGraph 中所有节点的共享数据容器。我们可以使用 Python 的 TypedDict也可以使用 Pydantic BaseModel。学习阶段用 TypedDict 最直接因为它类型标注清晰、运行开销小足够满足大多数场景。from typing import TypedDict, Annotated, List class AgentState(TypedDict): messages: List[str] steps: int finished: bool字段含义messages保存对话历史或中间过程消息。steps记录已经执行了多少步用于循环控制。finished标记当前流程是否结束。这里要注意LangGraph 的 State 更新默认是覆盖式替换。如果节点返回了{messages: [new]}旧 messages 会被整体覆盖。如果希望追加而不是覆盖可以通过Annotated加上 reducer 函数。这个知识点是 LangGraph 最容易踩坑的地方之一后面会详细讲解。3.2 编写节点函数和条件边节点函数的核心签名是(state) - dict输入当前完整 State输出一个部分更新的字典。LangGraph 会把输出合并回 State。下面实现一个简单循环节点 A 生成一条消息条件边判断 steps 是否达到阈值没有达到就回到 A达到就进入节点 B。from langgraph.graph import StateGraph, START, END def node_a(state: AgentState): print(fnode_a 执行当前 steps{state[steps]}) return { messages: state[messages] [fstep-{state[steps]}], steps: state[steps] 1, } def node_b(state: AgentState): print(node_b 执行流程结束) return {finished: True, messages: state[messages] [done]} def should_continue(state: AgentState): if state[steps] 3: return a # 返回节点名 return b然后构图graph StateGraph(AgentState) graph.add_node(a, node_a) graph.add_node(b, node_b) graph.add_edge(START, a) graph.add_conditional_edges(a, should_continue, {a: a, b: b}) graph.add_edge(b, END) app graph.compile()运行initial_state {messages: [], steps: 0, finished: False} result app.invoke(initial_state) print(result)预期输出中 node_a 会被执行三次steps 递增最终进入 node_bfinished 为 True。3.3 关键点条件边返回的是节点名还是 Next 对象在 LangGraph 新版本中条件边的目标映射可以是一个字典{a: a, b: b}表示当 should_continue 返回a时跳转到节点a返回b时跳转到节点b。如果条件函数返回的字符串不在映射字典中运行时会报错。另一个常见写法是使用Command对象比如节点内部直接决定下一步。这个写法在更复杂的场景下很有用因为节点可以不经过条件边直接跳转但会提高理解成本。初学阶段先掌握外部条件边写法等理解了调用流程再使用 Command。这里也要区分普通边和条件边。普通边是无条件跳转graph.add_edge(b, END)。条件边是根据 State 动态判断跳转目标graph.add_conditional_edges(a, should_continue, mapping)。循环的本质就是条件边中有一条分支指向了自己或前面的节点。3.4 用 Annotated reducer 实现消息追加默认覆盖式更新在真实场景中很难用因为我们通常希望把新消息追加到历史列表而不是每次都覆盖。LangGraph 提供了一种机制叫 reducer定义在 State 字段上节点返回新值时会自动调用 reducer 合并。from typing import Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[List[str], add_messages]langgraph.graph.message中内置了add_messages这个 reducer。它的行为是如果新消息有 id 且与旧消息冲突就覆盖否则追加。对文本字符串列表它会按顺序合并。这里有一个常见误区如果多次运行同一个 app 并传入相同初始状态State 中的消息并不会自动清空因为 LangGraph 的 State 是一次 invoke 内独立的。每次调用传新状态即可。若要实现多轮对话持久化后面可以引入 Checkpointer 或外部存储。4. 条件路由与分支控制从简单循环到真实 Agent4.1 条件路由的典型业务场景在数学计算应用里模型根据用户提问决定调用哪类工具比如查天气、算积分、查数据库、搜索文档。如果用线性链模型必须先做意图判断然后走入一个固定分支。问题是真实意图可能组合出现比如“帮我计算北京今天气温的 3 倍”需要先调用天气工具再调用计算器这依赖多次工具调用。LangGraph 的节点设计非常适合这种情况。假设有两个工具节点tool_weather和tool_calculator模型节点agent负责判断下一步调用哪个工具条件边从agent出发映射到不同工具节点工具节点执行后回到agent形成循环。def agent_node(state: AgentState): # 这里简化为规则判断实际项目中用大模型选择工具 user_input state[messages][-1].lower() if 天气 in user_input: return {next_tool: weather} elif 计算 in user_input: return {next_tool: calculator} return {next_tool: end} def weather_node(state: AgentState): return {messages: [天气工具返回晴25度]} def calculator_node(state: AgentState): return {messages: [计算器返回42]} def route_after_agent(state: AgentState): return state[next_tool]这种“分类 - 执行 - 回到入口再分类”的模式是所有 ReAct Agent 的基础。真实写法中agent_node内部会调用大模型根据模型输出判断需要调用哪个工具并把工具名和参数写入状态。4.2 循环检测为什么 Agent 会卡死以及 LangGraph 怎么防止循环在 LangGraph 中是核心能力但无限循环是生产环境的灾难。虽然你可以在节点里通过 steps 字段手动控制轮次但更规范的做法是设置recursion_limit。在 invoke 时指定result app.invoke(initial_state, config{recursion_limit: 10})当执行步数超过限制时LangGraph 会抛出异常。这个限制是一种安全网但依赖它作为唯一保护并不够因为业务逻辑可能根本不需要那么多步或者某些循环是合法的多轮工具调用。更合理的做法是设计最终节点判断逻辑当模型返回结束标记时条件边走 END当 steps 达到阈值时强制走结束分支。有人会把无限循环归咎于 LangGraph 框架其实问题往往出在状态设计上。如果条件边的判断字段一直不更新那么无论循环多少次都会走同一条边。所以在实际代码中要特别关注节点的输出是否真的修改了条件边读取的字段以及修改后的值是否导致分支条件发生变化。4.3 子图的拆分与复用当图变得很大所有节点堆在一起就很难维护。LangGraph 支持把一个编译好的图作为另一个图的节点称为子图。这种模式和函数拆分类似目的是复用和隔离。比如先做一个“工具调用子图”包含模型判断、工具执行、结果返回再做一个“总流程图”包含数据校验、调用子图、汇总输出。tool_subgraph graph.compile() class MainState(TypedDict): query: str result: str steps: int def run_tool_subgraph(state: MainState): sub_result tool_subgraph.invoke({ messages: [state[query]], steps: 0, finished: False, }) return {result: sub_result[messages][-1]}子图内部的状态和外层状态是隔离的需要通过输入输出字段做映射。这个特性适合把多个通用能力封装成独立模块让主图只编排大流程。5. 将 MCP 集成到 LangGraph 节点中5.1 MCP 是一个协议不是某一个具体的库很多人在搜索时会把 MCP 和特定插件绑定其实 MCPModel Context Protocol是一个开放协议定义了大模型应用如何发现外部工具、调用外部工具、读取资源。它由协议层、客户端层、服务端层组成。服务端可以是一个本地进程、一个远程 HTTP 服务甚至是一个 IDE 插件。架构主体如下MCP Server托管工具对外暴露工具列表和调用接口。MCP Client负责发现工具列表调用工具并返回结果。Agent 应用在 LangGraph 节点内作为 Client 发起调用。LangGraph 本身不内置 MCP但可以在节点函数里使用 MCP SDK 连接 Server。LangChain 生态提供了一些集成工具也可以用原生 SDK 手动调用这样更可控。5.2 用一个本地 MCP 服务做最小集成下面使用 Python 版 MCP SDK 实现一个简单的数学工具服务端然后用 LangGraph 节点调用。先创建服务端文件math_server.pyfrom mcp.server.fastmcp import FastMCP mcp FastMCP(MathServer) mcp.tool() def add(a: int, b: int) - int: 计算两个整数之和 return a b if __name__ __main__: mcp.run()启动服务python math_server.py如果在进程内测试LangGraph 节点中可以直接声明连接from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client import asyncio server_params StdioServerParameters( commandpython, args[math_server.py], ) async def call_math_add(a: int, b: int) - int: async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() result await session.call_tool(add, {a: a, b: b}) return result.content[0].text然后在 LangGraph 节点内部调用异步函数。需要注意的是LangGraph 的节点可以是同步函数也可以是异步函数。如果用了异步节点整个链路最好都使用ainvoke。async def math_tool_node(state: AgentState): a state.get(a, 1) b state.get(b, 2) value await call_math_add(a, b) return {messages: state[messages] [fadd result: {value}]}这样 MCP 就打通了Agent 在 LangGraph 中运行节点节点通过 MCP Client 调用服务端工具工具结果写回 State继续驱动后续节点。5.3 官方 Remote MCP Server 或社区封装怎么选在搜索材料中可以看到大量 MCP 相关词汇比如 Figma MCP、Playwright MCP、Chat2DB MCP、GitHub MCP 等。这些本质都是“特定工具对外提供 MCP Server 接口”。使用它们的方式有两种直接通过官方或社区提供的 MCP Server 进程启动。在 LangGraph 节点中连接该 Server像调用本地工具一样调用。对于生产项目建议先确认 Server 是否维护活跃、协议版本是否与 MCP SDK 兼容、是否需要提供长期运行的服务地址。某些 MCP Server 适合本地开发比如 Playwright MCP 用于浏览器自动化某些适合部署为远程服务比如内部数据库 MCP。选型时应考虑部署复杂度、权限模型和数据安全。5.4 使用 MCP 时最常见的几个坑第一个坑手动调用session.call_tool时参数格式错误。MCP 协议要求参数是 JSON 对象有些工具声明为必填字符串传成数字时会报 schema 校验失败。建议在实现前先打印服务端工具 schema确认参数类型和必填字段。第二个坑异步环境没有初始化asyncio.run导致报错。在同步节点内运行异步客户端时很多人直接调用asyncio.run()嵌套在事件循环中会报错。推荐把整个 LangGraph 应用写成异步方式使用ainvoke避免嵌套事件循环。第三个坑MCP Server 进程没有启动或端口不通。本地 stdio 模式会由客户端拉起进程但如果服务端文件路径、命令参数不对会出现连接建立失败或初始化超时。先用官方 SDK 的最小示例把 MCP Server 单独跑通再接入 LangGraph这样排错范围可以缩小到一层。下表整理了常见现象和处理优先级问题现象常见原因检查顺序调用工具返回 schema 错误参数名或类型不匹配打印工具 schema对比调用参数工具调用超时Server 进程未启动或网络不通先单独启动 Server 并用 CLI 测试事件循环乱同步节点嵌套运行异步函数改为异步节点和 ainvoke工具结果丢失节点未把工具结果写入 State检查节点返回字典中的 key6. 用 LangGraph 实现一个能实际运行的 ReAct Agent6.1 ReAct 范式在 LangGraph 中的映射ReAct 的核心循环是 Thought思考- Action行动- Observation观察。在 LangGraph 中这个循环可以映射为四个节点和一条条件边agent节点接收用户输入和工具结果让大模型决定下一步。如果决定结束则输出最终答案如果需要调用工具则输出工具名和参数。tools节点根据 agent 节点的输出执行实际工具函数或调用 MCP Server。conditional_edges根据 agent 节点输出判断下一步去tools还是END。工具结果作为新的消息追加到 State 的 messages 中重新进入agent节点。这种实现和传统 LangChain AgentExecutor 的区别在于LangGraph 中的每一步都是显式节点开发者可以插入日志、权限校验、人工审批、重试逻辑和限流而不用修改整个执行器。6.2 手写一个可运行的 ReAct Agent 代码为了减少依赖版本干扰下面用 LangChain 的消息类型保存历史用两个简单工具函数模拟工具调用。真实项目可以把这两个工具替换成 MCP Client 调用。import json from typing import Annotated, TypedDict from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langchain_core.messages import HumanMessage, AIMessage, ToolMessage class ReActState(TypedDict): messages: Annotated[list, add_messages] def fake_weather_tool(city: str) - str: return f{city} 今天晴气温 25 度 def fake_calculator_tool(expression: str) - str: # 这里仅做演示生产环境要用安全表达式解析 return f计算结果{eval(expression)} TOOLS { weather: fake_weather_tool, calculator: fake_calculator_tool, } model ChatOpenAI(modelgpt-4o-mini, temperature0) tools_desc [ {type: function, function: { name: weather, description: 查询城市天气, parameters: { type: object, properties: {city: {type: string}}, required: [city], }, }}, {type: function, function: { name: calculator, description: 计算数学表达式, parameters: { type: object, properties: {expression: {type: string}}, required: [expression], }, }}, ] model_with_tools model.bind_tools(tools_desc) def agent_node(state: ReActState): response model_with_tools.invoke(state[messages]) if response.tool_calls: return {messages: [response]} return {messages: [response]} def tools_node(state: ReActState): last_message state[messages][-1] tool_messages [] for tool_call in last_message.tool_calls: tool_name tool_call[name] tool_args tool_call[args] tool_func TOOLS.get(tool_name) if tool_func: result tool_func(**tool_args) else: result f未找到工具: {tool_name} tool_messages.append(ToolMessage(contentstr(result), tool_call_idtool_call[id])) return {messages: tool_messages} def should_continue(state: ReActState): last_message state[messages][-1] if getattr(last_message, tool_calls, None): return tools return end graph StateGraph(ReActState) graph.add_node(agent, agent_node) graph.add_node(tools, tools_node) graph.add_edge(START, agent) graph.add_conditional_edges(agent, should_continue, {tools: tools, end: END}) graph.add_edge(tools, agent) app graph.compile()运行测试result app.invoke({ messages: [HumanMessage(content北京今天天气怎么样)] }) print(result[messages][-1].content)大模型会先输出 tool_call 调用 weather 工具然后 tools 节点执行 fake_weather_tool 返回工具结果agent 节点再次运行最终输出自然语言答案。6.3 如何加入人类审批节点真实生产环境经常会遇到“机器决定改数据库”无论模型多可靠都应该先暂停让人类确认。LangGraph 可以轻松插入 interrupt 或设计一个等待人类确认的节点。最简单的方式是使用interrupt_before配置在进入tools节点之前暂停。config {recursion_limit: 20, interrupt_before: [tools]}这样在工具调用前执行会暂停需要人工确认后继续。LangGraph 还支持持久化 Checkpointer配合恢复接口实现真正的“等待人工确认 - 继续执行”。这个机制特别适合审批流场景也是 LangGraph 相比普通函数调用链的优势。6.4 输出验证和结果分析验证一个 Agent 不能只看最终是否输出了答案。需要确认以下信息模型是否做出了正确的工具调用工具名和参数是否合理。工具执行结果是否正确回写到 messages。条件边是否正确回到 agent 或走向结束。多轮工具调用时消息顺序是否清晰有没有出现消息错乱。整体是否存在多余的工具调用或超时。建议在开发阶段开启调试输出至少打印每一步的节点名和关键状态变化。LangSmith 是最完整的观测方案但如果只需要本地快速调试直接 print 也能解决大部分问题。app graph.compile() for event in app.stream(initial_state, configconfig): for node_name, node_state in event.items(): print(f节点 {node_name} 已执行) print(node_state)stream返回的每个事件包含节点名和该节点执行后的状态快照。这对理解 LangGraph 的执行流程非常有帮助。7. 状态持久化、记忆和 Checkpointer7.1 为什么 Agent 需要状态持久化Web 应用通常有会话机制Agent 应用也一样。用户说“刚才那个城市再查一遍”如果 Agent 不保存上一轮状态就不知道该查哪个城市。LangGraph 的 State 在一次 invoke 内有效多次对话之间默认不共享。要实现真正的多轮记忆需要把 State 保存到外部存储每次调用前加载上次的 State。LangGraph 的解决方案是 Checkpointer。它负责在每次节点执行后保存状态快照并支持根据 thread_id 恢复。对于内存学习可以使用内存版 Checkpointer但生产环境建议使用 Postgres 或 Redis 等持久化方案。7.2 使用 Checkpointer 实现多轮对话先安装必要的依赖pip install langgraph-checkpoint使用内存版 Checkpointerfrom langgraph.checkpoint.memory import MemorySaver memory MemorySaver() app graph.compile(checkpointermemory) config {configurable: {thread_id: user-123}} result app.invoke( {messages: [HumanMessage(content帮我查北京天气)]}, configconfig, )第二次提问时使用相同 thread_idAgent 会自动加载之前的 messages 作为上下文result app.invoke( {messages: [HumanMessage(content上海呢)]}, configconfig, )模型会根据历史判断“上海呢”是继续查天气从而自动调用 weather 工具。没有记忆时这种省略主语的能力会完全失效。7.3 Checkpointer 和数据库事务的取舍生产环境要注意 Checkpointer 本身也是一类写操作。如果把 Checkpointer 配置到业务数据库要考虑每次节点执行都写入状态快照带来的压力。某些高并发场景可以缩短保存周期或者在关键节点才开启持久化而不是每个节点都写库。LangGraph 的持久化能力是把双刃剑使用方便但如果状态里包含敏感字段保存快照时要注意脱敏。不要把用户密码、Token、完整原始请求全部写入 Checkpointer应该只保存记忆所需的最小字段。8. LangGraph 与 LangChain 的分工和常见选型问题8.1 LangChain 是否过时LangGraph 会替代它吗从技术演进看LangGraph 和 LangChain 不是替代关系而是不同抽象层级。LangChain 提供的是“组件库”模型封装、Prompt 模板、输出解析器、向量存储封装、工具定义。LangGraph 提供的是“运行时”状态、节点、边、循环、分支、持久化。一个可运行的 Agent 通常两者都用。在某些新项目中LangGraph 已经被当成 LangChain 生态的更高层框架LangChain AgentExecutor 则逐渐退居后台。但如果任务只是简单的 RAG 问答链使用 LangChain 的 LCEL 表达链就够了不必非要上 LangGraph。所有流程都会先从简单开始当出现多轮工具调用、条件分支、人工审批时再迁移到 LangGraph 也不迟。8.2 Agent Skill 和 MCP 的区别搜索热词中多次出现“Agent Skill 和 MCP 有什么区别”。这一组概念容易混淆。MCP 是协议解决的是“工具如何被外部发现和调用”Skill 是一种能力封装解决的是“Agent 如何组织和复用一个完整的任务流程”。类比来说MCP 定义插头规格Skill 定义插电后运行的设备逻辑。一个 Skill 内部可以依赖多个 MCP 工具MCP 工具也可以被多个 Skill 复用。在 LangGraph 中Skill 通常体现为子图或节点组合MCP 工具则通过 Client 在节点内调用。两者配合时最清晰的边界是Skill 负责决策和编排MCP 负责执行具体动作。8.3 选型建议什么时候用 LCEL什么时候用 LangGraph场景推荐方案原因一次模型调用加格式化输出LangChain LCEL简单心智负担低连续多次调用模型并传递结果LCEL 也可以但状态复杂时更难维护可先用 LCEL复杂时再迁移工具调用循环、条件分支、多轮协商LangGraph图模型天然支持循环和分支需要人工审批、暂停恢复、持久化会话LangGraphCheckpointer 和 interrupt 机制完善高频低延迟的规则引擎不要用 Agent 框架直接用代码避免模型开销9. 生产环境落地 LangGraph 的排查清单和最佳实践9.1 最常见的报错现象与处理路径LangGraph 的报错信息通常不算难懂但因为涉及异步、状态更新和条件映射新手容易在几个固定位置卡住。下面整理一套按优先级执行的排查流程。首先要看是否是版本问题。报错中包含StateGraph、add_conditional_edges或Command的 AttributeError 或 TypeError 时大概率是代码照搬了旧版本教程。先升级或降级 langgraph 到官方示例对应版本再重新运行。其次要看 State 更新是否符合预期。节点返回的字典 key 必须存在于 State 定义中。如果返回了没有定义的字段LangGraph 默认会忽略或报错。建议定义 State 时把所有可能需要写回 State 的字段都列出来避免遗漏。再看条件边的映射。should_continue返回的字符串必须存在于映射字典中。很多报错信息会直接提示没有找到 path 或 node这时要打印条件边的返回值确认是否写错节点名。还要看异步事件循环。若同时在 Jupyter Notebook 或 IPython 环境中运行 async 代码注意asyncio.run冲突。建议总是使用await app.ainvoke(...)不要在一个已经有事件循环的环境中嵌套调用asyncio.run。最后看工具调用返回格式。如果大模型返回了 tool_calls但tools_node中没有正确生成 ToolMessageAgent 会一直卡在 agent 和 tools 之间循环直到 recursion_limit 报错。检查每个 tool_call 的 id 是否在 ToolMessage 中回传LangGraph 依赖 tool_call_id 把工具结果和模型请求对齐。9.2 开发环境与生产环境的配置差异开发环境可以关掉大部分安全限制方便调试生产环境需要默认启用防护。下面的表格给出了两者的主要差异。项目开发环境生产环境Prompt 调试直接 print 日志使用 LangSmith 或自定义链路日志API Key环境变量本地配置密钥管理服务禁止进代码库模型调用失败直接抛出重试熔断、降级、告警、错误队列Checkpointer内存版Postgres 或 Redis 持久化工具执行允许任意本地函数白名单工具、参数校验、权限控制日志级别DEBUGINFO 或 WARNING敏感信息脱敏不要在生产环境使用eval这类动态执行代码的工具。ReAct 示例中用了 eval 只是为了演示真实项目应该使用安全的数学表达式库或解析器避免提示注入造成代码执行风险。9.3 提示注入和工具权限控制Agent 一旦接入工具攻击面就扩大了。理论上模型输出完全受用户输入影响如果节点内部用模型输出直接拼接 shell 命令或 SQL就是严重安全漏洞。至少要做到以下几点工具参数必须做类型和取值范围校验。数据库查询建议使用参数化查询或预先定义 SQL 模板。高权限工具必须有人工审批节点不能允许模型直接执行。所有外部工具的调用日志要完整保留便于安全审计。MCP Server 的凭证要独立于模型凭证遵循最小权限原则。9.4 可复用的 Agent 发布前检查清单下面是一份可以打印出来对照的检查清单适合在每次发布前快速走查。检查每个节点是否有异常处理模型调用失败时是否有重试或降级。检查 State 中的字段是否都需要写回敏感字段是否在日志中打印。检查条件边所有可能返回值是否都有目标节点。检查 recursion_limit 是否合理循环保护是否生效。检查工具调用是否记录完整 tool_call_idToolMessage 是否配对。检查 MCP Server 连接方式在目标环境是否可访问、超时设置是否合理。检查 Checkpointer 是否配置thread_id 是否由业务上下文正确传递。检查是否在应用层限流防止模型和工具被高频调用。检查日志是否包含节点名、状态变化、工具输入输出摘要且不含敏感信息。检查是否有人工审批的逃生通道高权限操作不会被模型直接触发。9.5 针对 LangGraph 的学习路径建议LangGraph 的学习不应该一上来就追求复杂。建议按这个顺序推进先跑通 TODO 型最小图两个节点加一条边。加上条件边实现一个简单的循环计数。把循环和 ReAct 结合起来让模型决定是否调用工具。加入 Checkpointer实现多轮对话记忆。接入 MCP Server把真实外部工具嵌入节点。引入子图把可复用的 Agent 能力拆成模块。加入人工审批用 interrupt 或者状态机模拟审批流。最后接入 LangSmith对运行质量做可观测性评估。每一步都要在真实场景里重复三到五次直到你能在不看文档的情况下画出状态流转图才算真正理解。否则只是把代码复制到本地遇到报错还是无法判断问题出在节点、边、状态还是工具上。10. 扩展方向从 LangGraph 走向完整 Agent 工程LangGraph 只是 Agent 运行时的底座一个完整的 Agent 工程还需要处理数据接入、知识库、评估、监控、模型成本和版本迭代。在搜索材料中出现的 Fusion MCP、Playwright MCP、数据库 MCP、蓝湖 MCP、Figma MCP 都是一个方向让 Agent 能触达更多外部系统。真正要关注的问题不是数量而是稳定性和权限可控性。如果团队已经在使用 LangChain 生态LangGraph 是继续深挖 Agent 编排的最自然选择。如果团队从零开始也可以只用 LangGraph 加原生模型 SDK不引入 LangChain保持更轻的依赖。两种路线都成立关键看团队的技术栈、模型供应商和运维能力。对大多数开发者而言最有效的投资是把官方文档中的对话式 ReAct Agent、多节点协作、持久化对话这几个示例吃透再结合自己的业务场景做一版带状态流转图的 Agent。当你能够在纸上画出节点、条件边和状态字段再回看 LangGraph 代码会发现它不过是一套非常朴素的图执行引擎真正复杂的永远是业务规则和模型能力边界。
