很多人在学习 LangGraph 时最大的困惑不是“不知道它是什么”而是看完一堆 Demo 之后依然不知道如何组织一个真正能用的 Agent 流程。网上资料要么只讲一个 Hello World要么直接甩出一大段源码中间缺失的状态管理、条件路由、工具接入这些关键环节反而要靠自己去踩坑。这篇文章按 LangGraph 从入门到实战的完整路径展开把核心概念、可运行代码和真实工程中的坑放在一起讲。你会看到一个 Agent 是如何从“固定链路”变成“可路由、可循环、可接工具”的完整过程而不是散装知识点罗列。全文示例代码都考虑到了可复制性建议在本地跑一遍再继续往下读很多疑问会在运行结果中自动解开。关于 LangGraph 和 LangChain 的关系我的判断是LangChain 依然是重要的基础组件库但 Agent 编排层正在明显向 LangGraph 集中。如果你只想学一个框架来支撑未来一两年的 Agent 开发LangGraph 是优先级很高的一环。1. 为什么 Agent 开发绕不开 LangGraph先聊一个经常被问的问题LangChain 过时了吗准确地说过时的不是 LangChain而是“只靠 Chain 串流程”的 Agent 开发方式。早期用 LangChain 写 Agent最常见的形态是 AgentExecutor它内部把“模型思考 - 调用工具 - 再把结果交给模型”这一步包装成了一个循环。这个循环用起来简单但它是一个黑盒你想在中间插入一步人工确认想在特定条件下跳出循环想回退到之前的某一步重试都会非常别扭。LangGraph 把这件事从根本上改变了。它把 Agent 流程建模成一张显式有向图每个状态转换都看得见、可以打断、可以恢复、可以精细控制。你不再被“框架内置的 Agent 循环”限制而是自己决定模型什么时候该思考、什么时候该调工具、什么时候该结束。所以我的观点是LangGraph 的核心价值不是“画图很酷”而是把 Agent 流程的控制权还给了开发者。LangGraph 与 LangChain 的分工可以用这个表格快速理解对比维度LangChainLangGraph核心定位组件库与集成层图编排引擎流程模型Chain 和可组合组件显式有向图状态处理外置 Memory链式传递State 贯穿全部节点可叠加 Reducer循环能力弱依赖内置 Agent 循环原生支持环结构适用场景模型调用、工具封装、文档处理复杂 Agent、多轮任务、人工介入、流程回放你完全可以在 LangGraph 的节点里使用 LangChain 的 prompt 模板、输出解析器、文档加载器。两者不是替代关系而是不同层级的配合关系。2. LangGraph 核心概念State、Node、Edge、GraphLangGraph 的模型非常简洁一共四个核心物件State、Node、Edge、Graph。State 是贯穿整个图的全局数据。你可以把它理解为所有节点共享的一张“工作台”每个节点都能读它也可以往里面写数据。State 的类型一般是 TypedDict也就是带字段类型声明的字典。Node 是图中的一个处理单元本质上就是一个普通 Python 函数。函数的输入是当前 State返回值是需要更新到 State 中的字段。返回值不需要返回全部字段只返回要修改的部分即可。Edge 是节点之间的连接线。普通 Edge 表示固定的执行顺序一个节点跑完后无条件走到下一个节点。Conditional Edge 则允许根据当前 State 的内容动态决定下一步去哪个节点。这是 LangGraph 支持分支和循环的基础。Graph 是承载 State、Node、Edge 的容器。你把所有元素注册进去调用 compile() 编译后就得到一个可执行对象。这个对象的使用方式很像 Web 框架中的 app通过 invoke() 传入初始 State然后按图驱动执行。这里要特别说明一点LangGraph 并不是一个普通的 DAG 工作流框架。DAG 只能单向无环但 Agent 的核心能力恰恰在于“循环”。LangGraph 的图允许出现环这就为 Agent 反复迭代提供了最底层的支持。打个比方如果把开发 Agent 比作搭建一条生产线State 就是传送带上的物料Node 是每个工位的操作员Edge 是传送带的方向控制。LangGraph 允许你自由设计传送带走向甚至让物料回到上一个工位重新加工。3. 环境准备与第一个 LangGraph 程序LangGraph 是一个纯 Python 库环境准备非常简单。推荐使用 Python 3.9 及以上版本建立一个虚拟环境后直接安装即可。python -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate pip install langgraph如果你希望在节点里调用真实的大模型还需要安装对应的模型提供商 SDK例如pip install langchain-openai但为了先理解 LangGraph 本身我们可以暂时不接任何模型。先写一个最小可运行的程序两个节点一条固定边把字符串依次处理。# minimal_graph.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): message: str def node_a(state: State): print(节点 A 收到:, state[message]) return {message: state[message] - A} def node_b(state: State): print(节点 B 收到:, state[message]) return {message: state[message] - B} graph StateGraph(State) graph.add_node(a, node_a) graph.add_node(b, node_b) graph.add_edge(START, a) graph.add_edge(a, b) graph.add_edge(b, END) app graph.compile() result app.invoke({message: hello}) print(最终 State:, result)这段代码演示了 LangGraph 最基本的运行机制通过StateGraph(State)创建图声明使用哪个 State 类型。用add_node注册两个节点节点名分别是a和b。用add_edge连接 START 到节点 a、节点 a 到节点 b、节点 b 到 END。compile()返回可执行对象invoke()传入初始 State 并执行整张图。运行命令python minimal_graph.py预期输出节点 A 收到: hello 节点 B 收到: hello - A 最终 State: {message: hello - A - B}注意一个细节node_b接收到的 State 中message已经是hello - A。这说明 LangGraph 会把前一个节点的更新结果自动合并到 State 中再传给下一个节点。这个更新机制是整个框架的基石下一节会重点展开。4. 状态管理如何在不同节点之间共享和修改 State节点函数的返回值如何合并到 State 中是 LangGraph 新手最容易踩坑的地方。默认情况下返回值中的字段是直接覆盖式写入。比如 State 中有一个count字段节点返回{count: 10}那么 State 里的count就变成 10。但如果多个节点都要往同一个 list 字段里追加内容覆盖式写入就会把前一个节点的结果冲掉。LangGraph 的解决方案是 Reducer归约器。Reducer 不是 LangGraph 独有的概念在很多状态管理库中都有当多个更新需要作用于同一个字段时用 Reducer 定义“如何合并这些更新”。类型注解的写法如下# state_reducer.py from typing import TypedDict, Annotated from operator import add from langgraph.graph import StateGraph, START, END class State(TypedDict): count: int logs: Annotated[list[str], add] def node_a(state: State): return {count: state[count] 1, logs: [A 执行完成]} def node_b(state: State): return {count: state[count] 1, logs: [B 执行完成]} graph StateGraph(State) graph.add_node(node_a, node_a) graph.add_node(node_b, node_b) graph.add_edge(START, node_a) graph.add_edge(node_a, node_b) graph.add_edge(node_b, END) app graph.compile() result app.invoke({count: 0, logs: []}) print(result)这里的关键点logs字段用Annotated[list[str], add]声明含义是“追加而不是覆盖”。两个节点分别返回一个只有一个元素的列表最终结果是[A 执行完成, B 执行完成]。count没有加 Reducer所以每个节点返回的都是基于当前值的计算结果最终值是 2。运行结果{count: 2, logs: [A 执行完成, B 执行完成]}如果你把Annotated[list[str], add]改成普通的list[str]第二个节点返回的日志会直接覆盖第一个节点最终 logs 只剩[B 执行完成]。这就是“看起来没问题结果却不对”的常见元凶。在实际的对话类 Agent 中还有一个更常用的内置 Reduceradd_messages。它用于处理消息列表并且会根据消息的 id 去重避免多轮对话中消息被重复追加。from langgraph.graph.message import add_messages class ChatState(TypedDict): messages: Annotated[list, add_messages]在多轮 Agent 场景中所有模型输入、工具返回、系统消息都统一存在messages字段里。这样每个节点只需要往列表里追加消息不需要手动维护“历史上下文”这个全局变量。5. 条件路由与分支控制Conditional Edge 深度解析如果 LangGraph 只有固定边那它和普通工作流框架没有区别。真正体现 Agent 编排能力的是条件路由。条件路由的 API 是add_conditional_edge。它做的事情是当某个节点执行完成后调用一个路由函数这个函数读取当前 State返回下一个要进入的节点名。路由函数本身也是一个普通 Python 函数输入是 State输出是一个字符串代表目标节点名称。你可以把判断逻辑做得非常简单也可以做得相当复杂。下面用一个关键词分级路由的示例来说明不依赖任何大模型直接跑通# routing.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): question: str answer: str def route_by_keyword(state: State): if 天气 in state[question]: return weather if 时间 in state[question]: return time return default def weather_node(state: State): return {answer: 当前城市晴天气温 26 ℃东南风 3 级} def time_node(state: State): return {answer: 当前时间为 2026-04-20 14:30:00} def default_node(state: State): return {answer: 我暂时无法回答这个问题请换个说法试试} graph StateGraph(State) graph.add_node(weather, weather_node) graph.add_node(time, time_node) graph.add_node(default, default_node) graph.add_conditional_edge( START, route_by_keyword, {weather: weather, time: time, default: default} ) graph.add_edge(weather, END) graph.add_edge(time, END) graph.add_edge(default, END) app graph.compile() print(app.invoke({question: 北京天气怎么样})) print(app.invoke({question: 现在几点了})) print(app.invoke({question: 你会写代码吗}))这里需要重点理解的是add_conditional_edge的三个参数第一个参数是起始节点。上例中直接从 START 路由意味着图一进来就要做判断。第二个参数是路由函数。它负责返回目标节点的名称字符串。第三个参数是路径映射表。它把路由函数的返回值映射到图中真实存在的节点名。这个映射表在概念上非常重要路由函数返回的是一个“逻辑分支名”映射表负责把它翻译成“实际执行节点”。运行结果{question: 北京天气怎么样, answer: 当前城市晴天气温 26 ℃东南风 3 级} {question: 现在几点了, answer: 当前时间为 2026-04-20 14:30:00} {question: 你会写代码吗, answer: 我暂时无法回答这个问题请换个说法试试}条件路由也可以放在图的中间节点而不是只放在入口。比如一个“意图识别”节点跑完后根据识别结果分发到不同的业务处理节点。这种模式在真实项目里是最常见的结构之一。实际工程中有一个容易踩的坑路由函数和路径映射表中返回的节点名如果不匹配运行时会直接报错提示找不到对应的节点。建议所有分支名统一用字符串常量维护而不是随手写字符串字面量。6. 循环、子图与并行分支从线性流程到 Agent 循环条件路由解决了“下一步走哪条路”的问题但 Agent 的另一个核心需求是“反复执行同一段逻辑”也就是循环。经典的 Agent 工作流是这样的模型决定调用工具工具返回结果模型根据结果继续思考如果还没完成任务再次调用工具。这个过程本质上就是一个循环。下面是一个最简单的循环骨架。这里用一个假消息列表模拟多轮对话节点每次往消息列表里追加一条 assistant 消息当消息条数达到阈值时跳转到结束节点。# agent_loop.py from typing import TypedDict, Annotated, Literal from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] finished: bool def tool_node(state: AgentState): # 这里模拟调用工具向消息列表追加一条新的内容 # 真实场景中会去调用搜索引擎、数据库或外部 API if len(state[messages]) 4: return {finished: True} return { messages: [{role: assistant, content: 工具返回结果}], finished: False, } def should_continue(state: AgentState) - Literal[tool_node, finish]: if state[finished]: return finish return tool_node def finish_node(state: AgentState): return {messages: [{role: assistant, content: 流程结束}]} graph StateGraph(AgentState) graph.add_node(tool_node, tool_node) graph.add_node(finish, finish_node) graph.add_edge(START, tool_node) graph.add_conditional_edge( tool_node, should_continue, {tool_node: tool_node, finish: finish} ) graph.add_edge(finish, END) app graph.compile() result app.invoke({messages: [{role: user, content: 你好}], finished: False}) print(result[messages])这里最关键的是add_conditional_edge中出现了自环路由结果中的tool_node: tool_node表示“继续调用自己”。这是 LangGraph 与普通 DAG 工作流最大的不同。只要路由函数决定继续图就会一直循环下去因此必须在循环体内设置明确的退出条件否则会形成死循环。上面这个例子的退出条件比较粗暴消息条数达到 4 就结束。真实项目中退出条件一般是模型判断“我已经给出最终答案”或者代码判断“工具调用次数已达上限”。工程上还有一种兜底做法在 State 里维护一个step_count每循环一次加一超过 N 次就强制性结束防止异常场景把资源耗尽。除了循环LangGraph 还支持子图和并行分支。子图的概念很简单一个编译后的图可以作为另一个图的节点。如果你有一段流程在很多地方都要复用就可以把这段流程封装成子图在主图里用一条边指向它。这和函数拆分的思路一致只不过拆分的粒度从“函数”变成了“图”。并行分支常用在“多个独立任务同时处理”的场景。比如用户提交一份文档需要同时做摘要、关键词提取、敏感信息检测。这三个任务互不依赖可以并行执行最后再把结果汇总写入 State。并行分支要注意的是多个并行节点如果同时写入 State 中的同一个字段必须给该字段配置 Reducer否则后面的写入会覆盖前面的结果。7. MCP 协议与 Agent SkillLangGraph 如何接入真实工具LangGraph 把流程控制得很好但一个 Agent 要解决实际问题必须接入外部工具。这就绕不开当前越来越热的 MCPModel Context Protocol模型上下文协议。MCP 出现之前给 Agent 接一个工具的流程大概是写一个工具函数把函数的名称、描述、参数结构整理给模型再在 Agent 执行循环中处理模型发起的工具调用。每接一个新工具都要重复这个过程工具数量一多代码会变得非常琐碎。MCP 的思路是标准化把工具放在 MCP Server 中通过协议暴露给 MCP Client。客户端负责发现工具、获取工具描述、调用工具并返回结果。这样一套协议可以同时对接很多不同的工具提供方。在 LangGraph 中加入 MCP本质上是在某个节点内完成“与 MCP Server 通信”的动作。图本身不关心你的工具是普通 Python 函数还是 MCP 工具它只关心节点是否把结果写回了 State。一个典型的 MCP 接入思路是在节点中创建或连接 MCP Client。向 MCP Server 发起list_tools请求拿到工具列表。将工具列表交给大模型让模型决定调用哪个工具。将模型选择的工具调用请求发送给 MCP Server。把工具结果写回 State交给后续节点继续判断。一个 MCP Server 的配置文件通常长这样{ mcpServers: { weather-server: { command: python, args: [weather_server.py], env: {} } } }在 LangGraph 节点中与 MCP Server 通信的示意代码大致如下。注意不同 SDK 的 API 略有差异实际使用时以官方文档为准# 示意代码在 LangGraph 节点中使用 MCP 获取工具 # 具体客户端实现请参考所用 SDK 的官方文档 async def mcp_tool_node(state): from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client params StdioServerParameters(commandpython, args[weather_server.py]) async with stdio_client(params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() # 将 tools 转换为模型可调用的格式 # 在本节点中完成工具调用并把结果写入 state return {tool_results: tools}这里更推荐的方式是不要在图的每个节点上都直接创建 MCP Client而是把 MCP Client 的生命周期管理放在流程外围节点内只负责调用已经就绪的客户端。这样可以避免反复建立连接的开销。关于 MCP 和 Agent Skill 的区别也是很多初学者容易混淆的点。我建议这样理解对比项Agent SkillMCP解决的核心问题把“怎么做某类事”封装成可复用技能把“外部能力”标准化接入典型粒度一个较完整的工作流或领域能力一组工具或资源关注点流程、Prompt、约束、产出协议、传输、鉴权、工具发现二者关系一个 Skill 内部可以调用多个 MCP Server 的工具MCP 是 Skill 获取工具的一种接入方式MCP 解决的是“如何标准化接入工具”Skill 解决的是“如何把一套做事方法沉淀下来”。一个 Skill 可能会用到多个 MCP Server 提供的能力而一个 MCP Server 也可以被多个 Skill 复用。两者不是同一层的东西。8. 综合实战构建一个带路由和工具调用的客服 Agent把前面几节的内容串起来做一个不依赖真实大模型也能跑通的综合示例。这个示例模拟了一个智能客服能根据用户输入路由到订单查询、人工客服、寒暄问候三个分支。# customer_service_agent.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class ServiceState(TypedDict): user_input: str reply: str need_human: bool def route_intent(state: ServiceState): text state[user_input] if 订单 in text or 快递 in text: return order if 人工 in text or 投诉 in text: return human return greeting def order_node(state: ServiceState): # 模拟调用订单查询工具 # 真实场景中可以在这里调用 MCP 工具或普通 Python 工具 return { reply: 正在为您查询订单状态请稍等。您的订单预计明天送达。, need_human: False, } def human_node(state: ServiceState): return { reply: 已为您转接人工客服工号 1024 正在处理。, need_human: True, } def greeting_node(state: ServiceState): return { reply: 您好我是智能客服可以查询订单或转人工。, need_human: False, } graph StateGraph(ServiceState) graph.add_node(order, order_node) graph.add_node(human, human_node) graph.add_node(greeting, greeting_node) graph.add_conditional_edge( START, route_intent, {order: order, human: human, greeting: greeting} ) graph.add_edge(order, END) graph.add_edge(human, END) graph.add_edge(greeting, END) app graph.compile() cases [ 我的订单什么时候到, 我要投诉请转人工, 你好, ] for case in cases: result app.invoke({user_input: case, need_human: False}) print(f用户: {case}) print(f客服: {result[reply]}) print(f是否需要人工: {result[need_human]}) print(- * 40)运行结果预期如下用户: 我的订单什么时候到 客服: 正在为您查询订单状态请稍等。您的订单预计明天送达。 是否需要人工: False ---------------------------------------- 用户: 我要投诉请转人工 客服: 已为您转接人工客服工号 1024 正在处理。 是否需要人工: True ---------------------------------------- 用户: 你好 客服: 您好我是智能客服可以查询订单或转人工。 是否需要人工: False这个示例已经把“路由”这一步完整跑通了。如果希望接上真实的大模型只需把greeting_node中的返回内容替换为一次 LLM 调用如果希望调用真实订单接口把order_node中的模拟返回替换为调用 MCP 工具或 HTTP API 的代码即可。在这个骨架基础上扩展时最需要保留的设计是状态中始终维护好reply和need_human这类跨节点共享的字段。后续无论是加人机协作、加日志审计还是加多轮对话都会比直接在节点内部保存全局变量要清晰得多。9. 常见问题与排查思路LangGraph 本身是一个比较新的框架版本迭代快遇到问题先看报错信息再看依赖版本通常能解决大部分问题。下面整理了一些高频问题。问题现象可能原因排查方式解决方案安装 langgraph 时依赖冲突本机已存在旧版 LangChain 或 pydantic查看 pip 依赖树和错误日志在虚拟环境中安装统一升级相关依赖节点返回值没有更新到 State返回的字典键名与 State 字段不一致或字段被覆盖打印节点输出和 State 内容检查返回键名为合并类字段配置 Reducerlist 字段被后一个节点覆盖没给 list 字段添加Annotated[list, add]查看 State 中该字段的值使用Annotated[list, add]或add_messages条件路由总跳到同一个分支路由函数内部逻辑判断顺序有问题或 path_map 键名不一致在路由函数中打印 State 的 Input 值修正判断逻辑统一节点名常量循环不退出进程卡住循环体内没有满足退出条件或退出条件写错在循环节点打印消息数量和退出标志增加最大步数限制设置明确的结束判断MCP 工具注册不上Server 地址错误、认证失败、工具名称冲突先用独立 MCP Client 单独测试 Server检查配置文件确认服务端日志最小化接入多个并行节点写入同一字段时值丢失字段没有配置 Reducer检查运行结果中字段内容给该字段添加合并型 Reducer编译时报“no edges between nodes”节点没有连接到 START 或 END存在孤立节点检查 add_edge 和 add_conditional_edge 是否完整补齐所有节点到 START/END 的边如果问题日志不够直观建议先用一个小图复现问题。最小化复现是排查 LangGraph 问题最高效的方式因为图越复杂定位成本越高。10. 最佳实践与工程建议把 LangGraph 从 Demo 带到生产环境以下几个实践建议值得认真对待。State 字段设计要克制。State 是贯穿全图的全局数据尽量不要把大对象、临时变量、敏感信息全部塞进去。只保留跨节点需要的字段。如果一个数据只在一个节点内部使用就应该放在节点内部而不是污染全局状态。节点函数保持单一职责。一个节点最好只做一件事调用模型、调用工具、做路由判断、写结果。如果把很多逻辑堆在同一个节点里图结构会退化成一个大函数调试和测试都会变得困难。路由函数保持纯逻辑。不要在路由函数里调用外部服务也不要写复杂的 IO 操作。路由函数应该是“给定 State返回字符串”的纯函数这样最容易测试和覆盖。工具调用前必须校验输入。如果某个节点会调用数据库、执行命令、访问外部 API一定要注意参数校验。真实项目中很多人直接把模型输出拼进 SQL 或 shell 命令风险极大。涉及数据库变更、系统命令或生产环境操作时先确认合法授权在测试环境验证保留回滚方案并遵循最小权限原则。日志要记录关键状态变化。在节点入口和出口打印 State 的关键字段对定位问题非常有帮助。生产环境建议接入结构化日志把图名、节点名、会话 ID 都带上排查问题时能直接串起一整条链路。大模型相关代码要允许 Mock。写单元测试时不能每次都真实调用大模型。建议在节点设计中预留模型接口测试时注入一个返回固定结果的假模型这样才能快速验证图结构和路由逻辑。版本管理要做锁定。LangGraph 迭代很快如果项目已经跑通就把核心依赖版本锁定到 requirements.txt 或 pyproject.toml 中。不要让生产环境每次部署都拉取最新版本否则很容易出现“昨天还能跑今天升级后报错”的情况。会话级状态用 Checkpointer 管理。如果 Agent 需要支持长期对话或断点恢复建议了解 LangGraph 的 Checkpointer 机制把每个步骤的状态保存下来这样在回复异常时可以直接从某个节点重新执行。11. 总结与后续学习方向这篇文章走完了 LangGraph 的最小学习闭环理解了 State、Node、Edge 四个核心概念亲手写了最小图掌握了状态管理中的 Reducer 机制深度理解了条件路由的用法看到了循环在 Agent 场景中的价值也理清了 MCP 工具接入和 LangGraph 的结合方式。最后的客服示例虽然简单但它已经具备了真实 Agent 的骨架入口路由、业务节点、状态返回、结束判定。如果让我给一个学习路径上的建议我会说先不要急着接大模型把 State 和路由这两块吃透因为它们决定了你后续能写出多复杂的 Agent。等你能不查文档就画出任意流程的 LangGraph 结构时再开始接入真实模型和 MCP 工具会顺手很多。接下来可以继续深入的几个方向是Checkpointer 与持久化状态、Human-in-the-loop 人工介入、流式输出与实时进度、子图复用与团队内部共享。LangGraph 官方文档也值得反复翻阅它现在的内容更新速度比大多数二手教程快得多。建议把文中的示例代码在本地亲手跑一遍然后改造成你自己的场景比如订单助手、知识库问答、运维排障机器人。只有把示例代码变成自己的代码这个框架才算真正开始为你工作。
很多人在学习 LangGraph 时最大的困惑不是“不知道它是什么”而是看完一堆 Demo 之后依然不知道如何组织一个真正能用的 Agent 流程。网上资料要么只讲一个 Hello World要么直接甩出一大段源码中间缺失的状态管理、条件路由、工具接入这些关键环节反而要靠自己去踩坑。这篇文章按 LangGraph 从入门到实战的完整路径展开把核心概念、可运行代码和真实工程中的坑放在一起讲。你会看到一个 Agent 是如何从“固定链路”变成“可路由、可循环、可接工具”的完整过程而不是散装知识点罗列。全文示例代码都考虑到了可复制性建议在本地跑一遍再继续往下读很多疑问会在运行结果中自动解开。关于 LangGraph 和 LangChain 的关系我的判断是LangChain 依然是重要的基础组件库但 Agent 编排层正在明显向 LangGraph 集中。如果你只想学一个框架来支撑未来一两年的 Agent 开发LangGraph 是优先级很高的一环。1. 为什么 Agent 开发绕不开 LangGraph先聊一个经常被问的问题LangChain 过时了吗准确地说过时的不是 LangChain而是“只靠 Chain 串流程”的 Agent 开发方式。早期用 LangChain 写 Agent最常见的形态是 AgentExecutor它内部把“模型思考 - 调用工具 - 再把结果交给模型”这一步包装成了一个循环。这个循环用起来简单但它是一个黑盒你想在中间插入一步人工确认想在特定条件下跳出循环想回退到之前的某一步重试都会非常别扭。LangGraph 把这件事从根本上改变了。它把 Agent 流程建模成一张显式有向图每个状态转换都看得见、可以打断、可以恢复、可以精细控制。你不再被“框架内置的 Agent 循环”限制而是自己决定模型什么时候该思考、什么时候该调工具、什么时候该结束。所以我的观点是LangGraph 的核心价值不是“画图很酷”而是把 Agent 流程的控制权还给了开发者。LangGraph 与 LangChain 的分工可以用这个表格快速理解对比维度LangChainLangGraph核心定位组件库与集成层图编排引擎流程模型Chain 和可组合组件显式有向图状态处理外置 Memory链式传递State 贯穿全部节点可叠加 Reducer循环能力弱依赖内置 Agent 循环原生支持环结构适用场景模型调用、工具封装、文档处理复杂 Agent、多轮任务、人工介入、流程回放你完全可以在 LangGraph 的节点里使用 LangChain 的 prompt 模板、输出解析器、文档加载器。两者不是替代关系而是不同层级的配合关系。2. LangGraph 核心概念State、Node、Edge、GraphLangGraph 的模型非常简洁一共四个核心物件State、Node、Edge、Graph。State 是贯穿整个图的全局数据。你可以把它理解为所有节点共享的一张“工作台”每个节点都能读它也可以往里面写数据。State 的类型一般是 TypedDict也就是带字段类型声明的字典。Node 是图中的一个处理单元本质上就是一个普通 Python 函数。函数的输入是当前 State返回值是需要更新到 State 中的字段。返回值不需要返回全部字段只返回要修改的部分即可。Edge 是节点之间的连接线。普通 Edge 表示固定的执行顺序一个节点跑完后无条件走到下一个节点。Conditional Edge 则允许根据当前 State 的内容动态决定下一步去哪个节点。这是 LangGraph 支持分支和循环的基础。Graph 是承载 State、Node、Edge 的容器。你把所有元素注册进去调用 compile() 编译后就得到一个可执行对象。这个对象的使用方式很像 Web 框架中的 app通过 invoke() 传入初始 State然后按图驱动执行。这里要特别说明一点LangGraph 并不是一个普通的 DAG 工作流框架。DAG 只能单向无环但 Agent 的核心能力恰恰在于“循环”。LangGraph 的图允许出现环这就为 Agent 反复迭代提供了最底层的支持。打个比方如果把开发 Agent 比作搭建一条生产线State 就是传送带上的物料Node 是每个工位的操作员Edge 是传送带的方向控制。LangGraph 允许你自由设计传送带走向甚至让物料回到上一个工位重新加工。3. 环境准备与第一个 LangGraph 程序LangGraph 是一个纯 Python 库环境准备非常简单。推荐使用 Python 3.9 及以上版本建立一个虚拟环境后直接安装即可。python -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate pip install langgraph如果你希望在节点里调用真实的大模型还需要安装对应的模型提供商 SDK例如pip install langchain-openai但为了先理解 LangGraph 本身我们可以暂时不接任何模型。先写一个最小可运行的程序两个节点一条固定边把字符串依次处理。# minimal_graph.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): message: str def node_a(state: State): print(节点 A 收到:, state[message]) return {message: state[message] - A} def node_b(state: State): print(节点 B 收到:, state[message]) return {message: state[message] - B} graph StateGraph(State) graph.add_node(a, node_a) graph.add_node(b, node_b) graph.add_edge(START, a) graph.add_edge(a, b) graph.add_edge(b, END) app graph.compile() result app.invoke({message: hello}) print(最终 State:, result)这段代码演示了 LangGraph 最基本的运行机制通过StateGraph(State)创建图声明使用哪个 State 类型。用add_node注册两个节点节点名分别是a和b。用add_edge连接 START 到节点 a、节点 a 到节点 b、节点 b 到 END。compile()返回可执行对象invoke()传入初始 State 并执行整张图。运行命令python minimal_graph.py预期输出节点 A 收到: hello 节点 B 收到: hello - A 最终 State: {message: hello - A - B}注意一个细节node_b接收到的 State 中message已经是hello - A。这说明 LangGraph 会把前一个节点的更新结果自动合并到 State 中再传给下一个节点。这个更新机制是整个框架的基石下一节会重点展开。4. 状态管理如何在不同节点之间共享和修改 State节点函数的返回值如何合并到 State 中是 LangGraph 新手最容易踩坑的地方。默认情况下返回值中的字段是直接覆盖式写入。比如 State 中有一个count字段节点返回{count: 10}那么 State 里的count就变成 10。但如果多个节点都要往同一个 list 字段里追加内容覆盖式写入就会把前一个节点的结果冲掉。LangGraph 的解决方案是 Reducer归约器。Reducer 不是 LangGraph 独有的概念在很多状态管理库中都有当多个更新需要作用于同一个字段时用 Reducer 定义“如何合并这些更新”。类型注解的写法如下# state_reducer.py from typing import TypedDict, Annotated from operator import add from langgraph.graph import StateGraph, START, END class State(TypedDict): count: int logs: Annotated[list[str], add] def node_a(state: State): return {count: state[count] 1, logs: [A 执行完成]} def node_b(state: State): return {count: state[count] 1, logs: [B 执行完成]} graph StateGraph(State) graph.add_node(node_a, node_a) graph.add_node(node_b, node_b) graph.add_edge(START, node_a) graph.add_edge(node_a, node_b) graph.add_edge(node_b, END) app graph.compile() result app.invoke({count: 0, logs: []}) print(result)这里的关键点logs字段用Annotated[list[str], add]声明含义是“追加而不是覆盖”。两个节点分别返回一个只有一个元素的列表最终结果是[A 执行完成, B 执行完成]。count没有加 Reducer所以每个节点返回的都是基于当前值的计算结果最终值是 2。运行结果{count: 2, logs: [A 执行完成, B 执行完成]}如果你把Annotated[list[str], add]改成普通的list[str]第二个节点返回的日志会直接覆盖第一个节点最终 logs 只剩[B 执行完成]。这就是“看起来没问题结果却不对”的常见元凶。在实际的对话类 Agent 中还有一个更常用的内置 Reduceradd_messages。它用于处理消息列表并且会根据消息的 id 去重避免多轮对话中消息被重复追加。from langgraph.graph.message import add_messages class ChatState(TypedDict): messages: Annotated[list, add_messages]在多轮 Agent 场景中所有模型输入、工具返回、系统消息都统一存在messages字段里。这样每个节点只需要往列表里追加消息不需要手动维护“历史上下文”这个全局变量。5. 条件路由与分支控制Conditional Edge 深度解析如果 LangGraph 只有固定边那它和普通工作流框架没有区别。真正体现 Agent 编排能力的是条件路由。条件路由的 API 是add_conditional_edge。它做的事情是当某个节点执行完成后调用一个路由函数这个函数读取当前 State返回下一个要进入的节点名。路由函数本身也是一个普通 Python 函数输入是 State输出是一个字符串代表目标节点名称。你可以把判断逻辑做得非常简单也可以做得相当复杂。下面用一个关键词分级路由的示例来说明不依赖任何大模型直接跑通# routing.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): question: str answer: str def route_by_keyword(state: State): if 天气 in state[question]: return weather if 时间 in state[question]: return time return default def weather_node(state: State): return {answer: 当前城市晴天气温 26 ℃东南风 3 级} def time_node(state: State): return {answer: 当前时间为 2026-04-20 14:30:00} def default_node(state: State): return {answer: 我暂时无法回答这个问题请换个说法试试} graph StateGraph(State) graph.add_node(weather, weather_node) graph.add_node(time, time_node) graph.add_node(default, default_node) graph.add_conditional_edge( START, route_by_keyword, {weather: weather, time: time, default: default} ) graph.add_edge(weather, END) graph.add_edge(time, END) graph.add_edge(default, END) app graph.compile() print(app.invoke({question: 北京天气怎么样})) print(app.invoke({question: 现在几点了})) print(app.invoke({question: 你会写代码吗}))这里需要重点理解的是add_conditional_edge的三个参数第一个参数是起始节点。上例中直接从 START 路由意味着图一进来就要做判断。第二个参数是路由函数。它负责返回目标节点的名称字符串。第三个参数是路径映射表。它把路由函数的返回值映射到图中真实存在的节点名。这个映射表在概念上非常重要路由函数返回的是一个“逻辑分支名”映射表负责把它翻译成“实际执行节点”。运行结果{question: 北京天气怎么样, answer: 当前城市晴天气温 26 ℃东南风 3 级} {question: 现在几点了, answer: 当前时间为 2026-04-20 14:30:00} {question: 你会写代码吗, answer: 我暂时无法回答这个问题请换个说法试试}条件路由也可以放在图的中间节点而不是只放在入口。比如一个“意图识别”节点跑完后根据识别结果分发到不同的业务处理节点。这种模式在真实项目里是最常见的结构之一。实际工程中有一个容易踩的坑路由函数和路径映射表中返回的节点名如果不匹配运行时会直接报错提示找不到对应的节点。建议所有分支名统一用字符串常量维护而不是随手写字符串字面量。6. 循环、子图与并行分支从线性流程到 Agent 循环条件路由解决了“下一步走哪条路”的问题但 Agent 的另一个核心需求是“反复执行同一段逻辑”也就是循环。经典的 Agent 工作流是这样的模型决定调用工具工具返回结果模型根据结果继续思考如果还没完成任务再次调用工具。这个过程本质上就是一个循环。下面是一个最简单的循环骨架。这里用一个假消息列表模拟多轮对话节点每次往消息列表里追加一条 assistant 消息当消息条数达到阈值时跳转到结束节点。# agent_loop.py from typing import TypedDict, Annotated, Literal from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] finished: bool def tool_node(state: AgentState): # 这里模拟调用工具向消息列表追加一条新的内容 # 真实场景中会去调用搜索引擎、数据库或外部 API if len(state[messages]) 4: return {finished: True} return { messages: [{role: assistant, content: 工具返回结果}], finished: False, } def should_continue(state: AgentState) - Literal[tool_node, finish]: if state[finished]: return finish return tool_node def finish_node(state: AgentState): return {messages: [{role: assistant, content: 流程结束}]} graph StateGraph(AgentState) graph.add_node(tool_node, tool_node) graph.add_node(finish, finish_node) graph.add_edge(START, tool_node) graph.add_conditional_edge( tool_node, should_continue, {tool_node: tool_node, finish: finish} ) graph.add_edge(finish, END) app graph.compile() result app.invoke({messages: [{role: user, content: 你好}], finished: False}) print(result[messages])这里最关键的是add_conditional_edge中出现了自环路由结果中的tool_node: tool_node表示“继续调用自己”。这是 LangGraph 与普通 DAG 工作流最大的不同。只要路由函数决定继续图就会一直循环下去因此必须在循环体内设置明确的退出条件否则会形成死循环。上面这个例子的退出条件比较粗暴消息条数达到 4 就结束。真实项目中退出条件一般是模型判断“我已经给出最终答案”或者代码判断“工具调用次数已达上限”。工程上还有一种兜底做法在 State 里维护一个step_count每循环一次加一超过 N 次就强制性结束防止异常场景把资源耗尽。除了循环LangGraph 还支持子图和并行分支。子图的概念很简单一个编译后的图可以作为另一个图的节点。如果你有一段流程在很多地方都要复用就可以把这段流程封装成子图在主图里用一条边指向它。这和函数拆分的思路一致只不过拆分的粒度从“函数”变成了“图”。并行分支常用在“多个独立任务同时处理”的场景。比如用户提交一份文档需要同时做摘要、关键词提取、敏感信息检测。这三个任务互不依赖可以并行执行最后再把结果汇总写入 State。并行分支要注意的是多个并行节点如果同时写入 State 中的同一个字段必须给该字段配置 Reducer否则后面的写入会覆盖前面的结果。7. MCP 协议与 Agent SkillLangGraph 如何接入真实工具LangGraph 把流程控制得很好但一个 Agent 要解决实际问题必须接入外部工具。这就绕不开当前越来越热的 MCPModel Context Protocol模型上下文协议。MCP 出现之前给 Agent 接一个工具的流程大概是写一个工具函数把函数的名称、描述、参数结构整理给模型再在 Agent 执行循环中处理模型发起的工具调用。每接一个新工具都要重复这个过程工具数量一多代码会变得非常琐碎。MCP 的思路是标准化把工具放在 MCP Server 中通过协议暴露给 MCP Client。客户端负责发现工具、获取工具描述、调用工具并返回结果。这样一套协议可以同时对接很多不同的工具提供方。在 LangGraph 中加入 MCP本质上是在某个节点内完成“与 MCP Server 通信”的动作。图本身不关心你的工具是普通 Python 函数还是 MCP 工具它只关心节点是否把结果写回了 State。一个典型的 MCP 接入思路是在节点中创建或连接 MCP Client。向 MCP Server 发起list_tools请求拿到工具列表。将工具列表交给大模型让模型决定调用哪个工具。将模型选择的工具调用请求发送给 MCP Server。把工具结果写回 State交给后续节点继续判断。一个 MCP Server 的配置文件通常长这样{ mcpServers: { weather-server: { command: python, args: [weather_server.py], env: {} } } }在 LangGraph 节点中与 MCP Server 通信的示意代码大致如下。注意不同 SDK 的 API 略有差异实际使用时以官方文档为准# 示意代码在 LangGraph 节点中使用 MCP 获取工具 # 具体客户端实现请参考所用 SDK 的官方文档 async def mcp_tool_node(state): from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client params StdioServerParameters(commandpython, args[weather_server.py]) async with stdio_client(params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() # 将 tools 转换为模型可调用的格式 # 在本节点中完成工具调用并把结果写入 state return {tool_results: tools}这里更推荐的方式是不要在图的每个节点上都直接创建 MCP Client而是把 MCP Client 的生命周期管理放在流程外围节点内只负责调用已经就绪的客户端。这样可以避免反复建立连接的开销。关于 MCP 和 Agent Skill 的区别也是很多初学者容易混淆的点。我建议这样理解对比项Agent SkillMCP解决的核心问题把“怎么做某类事”封装成可复用技能把“外部能力”标准化接入典型粒度一个较完整的工作流或领域能力一组工具或资源关注点流程、Prompt、约束、产出协议、传输、鉴权、工具发现二者关系一个 Skill 内部可以调用多个 MCP Server 的工具MCP 是 Skill 获取工具的一种接入方式MCP 解决的是“如何标准化接入工具”Skill 解决的是“如何把一套做事方法沉淀下来”。一个 Skill 可能会用到多个 MCP Server 提供的能力而一个 MCP Server 也可以被多个 Skill 复用。两者不是同一层的东西。8. 综合实战构建一个带路由和工具调用的客服 Agent把前面几节的内容串起来做一个不依赖真实大模型也能跑通的综合示例。这个示例模拟了一个智能客服能根据用户输入路由到订单查询、人工客服、寒暄问候三个分支。# customer_service_agent.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class ServiceState(TypedDict): user_input: str reply: str need_human: bool def route_intent(state: ServiceState): text state[user_input] if 订单 in text or 快递 in text: return order if 人工 in text or 投诉 in text: return human return greeting def order_node(state: ServiceState): # 模拟调用订单查询工具 # 真实场景中可以在这里调用 MCP 工具或普通 Python 工具 return { reply: 正在为您查询订单状态请稍等。您的订单预计明天送达。, need_human: False, } def human_node(state: ServiceState): return { reply: 已为您转接人工客服工号 1024 正在处理。, need_human: True, } def greeting_node(state: ServiceState): return { reply: 您好我是智能客服可以查询订单或转人工。, need_human: False, } graph StateGraph(ServiceState) graph.add_node(order, order_node) graph.add_node(human, human_node) graph.add_node(greeting, greeting_node) graph.add_conditional_edge( START, route_intent, {order: order, human: human, greeting: greeting} ) graph.add_edge(order, END) graph.add_edge(human, END) graph.add_edge(greeting, END) app graph.compile() cases [ 我的订单什么时候到, 我要投诉请转人工, 你好, ] for case in cases: result app.invoke({user_input: case, need_human: False}) print(f用户: {case}) print(f客服: {result[reply]}) print(f是否需要人工: {result[need_human]}) print(- * 40)运行结果预期如下用户: 我的订单什么时候到 客服: 正在为您查询订单状态请稍等。您的订单预计明天送达。 是否需要人工: False ---------------------------------------- 用户: 我要投诉请转人工 客服: 已为您转接人工客服工号 1024 正在处理。 是否需要人工: True ---------------------------------------- 用户: 你好 客服: 您好我是智能客服可以查询订单或转人工。 是否需要人工: False这个示例已经把“路由”这一步完整跑通了。如果希望接上真实的大模型只需把greeting_node中的返回内容替换为一次 LLM 调用如果希望调用真实订单接口把order_node中的模拟返回替换为调用 MCP 工具或 HTTP API 的代码即可。在这个骨架基础上扩展时最需要保留的设计是状态中始终维护好reply和need_human这类跨节点共享的字段。后续无论是加人机协作、加日志审计还是加多轮对话都会比直接在节点内部保存全局变量要清晰得多。9. 常见问题与排查思路LangGraph 本身是一个比较新的框架版本迭代快遇到问题先看报错信息再看依赖版本通常能解决大部分问题。下面整理了一些高频问题。问题现象可能原因排查方式解决方案安装 langgraph 时依赖冲突本机已存在旧版 LangChain 或 pydantic查看 pip 依赖树和错误日志在虚拟环境中安装统一升级相关依赖节点返回值没有更新到 State返回的字典键名与 State 字段不一致或字段被覆盖打印节点输出和 State 内容检查返回键名为合并类字段配置 Reducerlist 字段被后一个节点覆盖没给 list 字段添加Annotated[list, add]查看 State 中该字段的值使用Annotated[list, add]或add_messages条件路由总跳到同一个分支路由函数内部逻辑判断顺序有问题或 path_map 键名不一致在路由函数中打印 State 的 Input 值修正判断逻辑统一节点名常量循环不退出进程卡住循环体内没有满足退出条件或退出条件写错在循环节点打印消息数量和退出标志增加最大步数限制设置明确的结束判断MCP 工具注册不上Server 地址错误、认证失败、工具名称冲突先用独立 MCP Client 单独测试 Server检查配置文件确认服务端日志最小化接入多个并行节点写入同一字段时值丢失字段没有配置 Reducer检查运行结果中字段内容给该字段添加合并型 Reducer编译时报“no edges between nodes”节点没有连接到 START 或 END存在孤立节点检查 add_edge 和 add_conditional_edge 是否完整补齐所有节点到 START/END 的边如果问题日志不够直观建议先用一个小图复现问题。最小化复现是排查 LangGraph 问题最高效的方式因为图越复杂定位成本越高。10. 最佳实践与工程建议把 LangGraph 从 Demo 带到生产环境以下几个实践建议值得认真对待。State 字段设计要克制。State 是贯穿全图的全局数据尽量不要把大对象、临时变量、敏感信息全部塞进去。只保留跨节点需要的字段。如果一个数据只在一个节点内部使用就应该放在节点内部而不是污染全局状态。节点函数保持单一职责。一个节点最好只做一件事调用模型、调用工具、做路由判断、写结果。如果把很多逻辑堆在同一个节点里图结构会退化成一个大函数调试和测试都会变得困难。路由函数保持纯逻辑。不要在路由函数里调用外部服务也不要写复杂的 IO 操作。路由函数应该是“给定 State返回字符串”的纯函数这样最容易测试和覆盖。工具调用前必须校验输入。如果某个节点会调用数据库、执行命令、访问外部 API一定要注意参数校验。真实项目中很多人直接把模型输出拼进 SQL 或 shell 命令风险极大。涉及数据库变更、系统命令或生产环境操作时先确认合法授权在测试环境验证保留回滚方案并遵循最小权限原则。日志要记录关键状态变化。在节点入口和出口打印 State 的关键字段对定位问题非常有帮助。生产环境建议接入结构化日志把图名、节点名、会话 ID 都带上排查问题时能直接串起一整条链路。大模型相关代码要允许 Mock。写单元测试时不能每次都真实调用大模型。建议在节点设计中预留模型接口测试时注入一个返回固定结果的假模型这样才能快速验证图结构和路由逻辑。版本管理要做锁定。LangGraph 迭代很快如果项目已经跑通就把核心依赖版本锁定到 requirements.txt 或 pyproject.toml 中。不要让生产环境每次部署都拉取最新版本否则很容易出现“昨天还能跑今天升级后报错”的情况。会话级状态用 Checkpointer 管理。如果 Agent 需要支持长期对话或断点恢复建议了解 LangGraph 的 Checkpointer 机制把每个步骤的状态保存下来这样在回复异常时可以直接从某个节点重新执行。11. 总结与后续学习方向这篇文章走完了 LangGraph 的最小学习闭环理解了 State、Node、Edge 四个核心概念亲手写了最小图掌握了状态管理中的 Reducer 机制深度理解了条件路由的用法看到了循环在 Agent 场景中的价值也理清了 MCP 工具接入和 LangGraph 的结合方式。最后的客服示例虽然简单但它已经具备了真实 Agent 的骨架入口路由、业务节点、状态返回、结束判定。如果让我给一个学习路径上的建议我会说先不要急着接大模型把 State 和路由这两块吃透因为它们决定了你后续能写出多复杂的 Agent。等你能不查文档就画出任意流程的 LangGraph 结构时再开始接入真实模型和 MCP 工具会顺手很多。接下来可以继续深入的几个方向是Checkpointer 与持久化状态、Human-in-the-loop 人工介入、流式输出与实时进度、子图复用与团队内部共享。LangGraph 官方文档也值得反复翻阅它现在的内容更新速度比大多数二手教程快得多。建议把文中的示例代码在本地亲手跑一遍然后改造成你自己的场景比如订单助手、知识库问答、运维排障机器人。只有把示例代码变成自己的代码这个框架才算真正开始为你工作。
