很多人在学完 LangChain 之后兴冲冲地去做 Agent 项目结果很快遇到一个尴尬局面模型调用的流程根本不受自己控制。你可能想让 Agent 先做 A 工具调用再根据结果决定走 B 分支还是 C 分支结果模型一口气把所有工具全调了一遍或者在一个循环里出不来。你甚至没办法回答一个最基本的问题Agent 当前到底执行到哪一步了LangGraph 最近成为 Agent 开发领域的主流选择核心原因就是它把执行流程从“链式调用”变成了“状态图”。它真正解决的不是“能不能调大模型”而是“大模型执行流程怎么被可靠控制”这件事。本文会用完整的代码示例带你走一遍 LangGraph 入门到实战的完整路径包括核心概念、环境搭建、条件路由、工具调用、子图拆分和常见坑让你少走弯路。如果你正在做 AI Agent 相关项目或者准备从 LangChain 过渡到 LangGraph这篇文章可以直接作为你的第一篇实战教程来读。1. 这篇文章真正要解决的问题Agent 开发的失控感先讲一个真实的开发场景。假设你要做一个客服机器人用户问“我想退货怎么操作”你希望 Agent 走这样的流程先调用“判断用户意图”的节点再调用“查询订单”的工具最后调用“生成回复”的节点。每一步都需要可控至少你要知道它现在在哪一步。但用传统 LangChain 的LCEL语法写出来的链路是线性的prompt | llm | tool | llm这种线性链适合固定流程一旦遇到“根据模型输出决定下一步走哪里”的需求就会非常别扭。你可以用tool或者RunnableLambda加条件判断但代码会越来越乱尤其是当你需要循环调用工具、多步推理、失败重试、人工审批介入时线性链根本表达不清楚。于是 LangGraph 给出了一个新的心智模型把 Agent 的执行流程画成一张有向图。节点是你要执行的函数边是节点之间的流转关系状态则在整张图中共享和传递。这样你就能精确控制哪个节点先执行。什么条件下走哪条边。什么时候循环回上一个节点。什么时候结束。整张图执行过程中状态如何累积和变化。如果你正在做的项目涉及多轮工具调用、多步决策、循环控制或复杂状态管理LangGraph 比 LangChain 的传统链式写法更合适。这篇文章会一直围绕“控制力”这个核心来展开因为这才是 LangGraph 区别于其他框架的关键。2. LangGraph 是什么从 Chain 到 Graph 的思维转变2.1 LangChain 与 LangGraph 的关系很多初学者会把 LangGraph 和 LangChain 当成两个完全无关的框架这是一个常见的误解。LangChain 是一个大模型应用开发框架负责模型接入、Prompt 管理、输出解析、向量检索、记忆等基础能力。而 LangGraph 是 LangChain 官方团队推出的 Agent 编排框架它依赖于 LangChain 的模型调用和工具抽象但重新设计了执行引擎。简单来说对比维度LangChain传统链式LangGraph图式核心结构线性 Chain有向图 Graph流程表达固定管道节点 边 条件路由状态管理数据在链中流动共享 State 对象可跨节点更新循环控制不擅长原生支持分支控制需要 Hack原生 Conditional Edge适用场景固定流程、RAG复杂 Agent、多步决策、工具循环所谓“从 Chain 到 Graph”本质上是一次思维转变你不再把应用看成“一串连续调用”而是看成一个“节点与边组成的执行图”。这个转变带来两个直接好处一是复杂流程的表达能力大大增强二是你可以观察到任意节点的中间状态调试体验也更好。2.2 核心概念State、Node、Edge、Conditional EdgeLangGraph 有四个核心概念理解它们就能看懂 80% 的代码。State状态整张图共享的数据结构通常用TypedDict定义。每个节点执行完都会返回一个字典这个字典会更新 State 中的部分字段。如果字段需要累积而不是覆盖可以用Annotated[list, operator.add]这样的 reducer 来声明。Node节点一个普通的 Python 函数。函数的入参是当前 State返回值是一个字典表示要更新的 State 字段。Edge边定义节点之间的固定流转方向。比如 A 执行完无条件走向 B。Conditional Edge条件边这也是 LangGraph 最强大的地方。它允许你在某个节点执行完后根据当前 State 动态决定下一步去哪个节点。有点像代码里的if-else但把判断逻辑抽成了一个独立的“路由函数”。我用一个生活化的类比来解释把一个大模型应用想象成一家餐厅。State 是顾客的点菜单Node 是后厨的各个工位Edge 是“切完菜必须传给炒菜工位”这样的固定顺序Conditional Edge 则是“如果顾客是素食主义者就不传肉类菜品改传素菜窗口”这样的动态判断。2.3 LangGraph 的执行过程LangGraph 的执行过程可以简化成三步循环从图入口节点开始读取当前 State。执行当前节点也就是你的 Python 函数拿到返回值并更新 State。根据边或条件边判断下一个节点是谁。如果下一步是 END结束执行。这个“观察 State → 执行节点 → 更新 State → 选择下一个节点”的循环就是 LangGraph 的所有核心机制。3. 环境准备与模型接入配置3.1 安装依赖LangGraph 的 Python 包安装非常简单建议使用pip直接安装以下几个核心包pip install langgraph langchain langchain-openai如果你需要用到 LangGraph 的预构建工具节点、记忆功能或持久化能力可以继续安装pip install langgraph-checkpoint langgraph-prebuilt注意本文不绑定特定版本号因为 LangGraph 迭代很快。你只需要使用pip install --upgrade langgraph保持最新稳定版即可。Python 环境建议使用 3.10 及以上版本避免低版本带来的类型语法兼容问题。3.2 模型服务配置LangGraph 本身不直接调用模型而是通过 LangChain 的模型封装来调用。我这里以 OpenAI 协议兼容的服务为例来说明配置方式。在项目根目录下创建.env文件写入你的 API Key 和模型服务地址OPENAI_API_KEY你的_API_KEY OPENAI_BASE_URL你的模型服务地址这里要特别说明如果你使用的模型服务商提供 OpenAI 协议兼容接口那么langchain-openai这个包可以直接通过base_url参数对接。这样做的好处是模型切换只改配置不动业务代码。如果你希望完全本地化运行也可以考虑使用 Ollama 或其他本地推理服务它们都提供了 OpenAI 兼容接口。LangGraph 的模型接入方式是一样的只需要把模型名改成你本地加载的模型即可。3.3 验证环境安装完成后可以运行下面这段代码验证模型接入是否正常from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0) resp llm.invoke(你好请回复 OK) print(resp.content)能够正常输出内容说明环境基本就绪。接下来可以开始跑第一个 LangGraph 程序了。4. 跑通第一个 LangGraph 程序我用一个最简单的两节点图来说明白 LangGraph 的完整代码结构。4.1 定义 State首先定义一个 State 类型。这里我用TypedDict来表示当前状态中有哪些字段from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): messages: list[str]messages字段是一个字符串列表用于累积节点执行的信息。4.2 定义节点函数每个节点就是一个普通函数。入参是当前 State返回值是要更新的字段字典def node_a(state: State) - dict: return {messages: state[messages] [A 节点已执行]} def node_b(state: State) - dict: return {messages: state[messages] [B 节点已执行]}注意这里我用state[messages] [...]的方式手动拼接列表。这是因为当前的 State 没有声明 reducerLangGraph 默认会用“覆盖”的方式处理节点返回值。如果你希望两个节点分别更新同一字段并且自动合并后面在 Agent 实战中我会用Annotated来演示更好的做法。4.3 构建图并执行接下来就是 LangGraph 的核心代码# 创建状态图 builder StateGraph(State) # 添加节点 builder.add_node(a, node_a) builder.add_node(b, node_b) # 添加边 builder.add_edge(START, a) builder.add_edge(a, b) builder.add_edge(b, END) # 编译并执行 app builder.compile() result app.invoke({messages: []}) print(result)运行结果如下{messages: [A 节点已执行, B 节点已执行]}这完整演示了 LangGraph 的最小执行单元定义 State → 添加节点 → 连接边 → 编译 → 调用。从这个例子能看出LangGraph 并不是一个“黑盒”框架它的每个执行步骤都可以被观察和干预这正是它适合做复杂 Agent 的原因。4.4 如何理解 compile 和 invokecompile()会把状态图编译成一个可调用的应用对象。你可以在编译前后通过app.get_graph().draw_mermaid()需要安装额外依赖来可视化图结构不过本文不展开讲解绘图重点是执行逻辑。invoke()是同步执行入口传入初始 State返回最终 State。如果你有异步需求LangGraph 还提供了ainvoke()、astream()等异步接口适合用于 FastAPI 等异步 Web 服务场景。5. 条件路由与分支控制conditional_edge 深度解析第一个例子展示了固定流程但真实业务里更常见的是“根据内容做分支”。这一节我会给出一个完整的条件路由示例这也是 LangGraph 被使用最多的能力之一。5.1 场景设计假设我们有一个简单的意图路由场景用户输入包含“天气”走天气处理节点。用户输入包含“新闻”走新闻处理节点。其余输入走默认兜底节点。5.2 定义状态与节点from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END class RouterState(TypedDict): query: str route: str def router_node(state: RouterState) - dict: 路由节点根据 query 内容决定 route 字段的值 if 天气 in state[query]: return {route: weather} if 新闻 in state[query]: return {route: news} return {route: default} def weather_node(state: RouterState) - dict: return {query: state[query] 【天气处理完成】} def news_node(state: RouterState) - dict: return {query: state[query] 【新闻处理完成】} def default_node(state: RouterState) - dict: return {query: state[query] 【默认兜底处理完成】}5.3 条件路由函数条件路由的关键在于一个专门的“路由函数”。这个函数读取当前 State返回一个字符串表示下一步要走向哪个节点def decide_route(state: RouterState) - Literal[weather, news, default]: return state[route]这个函数本身没有复杂逻辑它只是返回router_node设置好的route字段。在实际项目中路由函数可以更复杂比如读取数据库配置、调用外部服务来决定路由。5.4 使用 add_conditional_edges 构建分支builder StateGraph(RouterState) # 添加节点 builder.add_node(router, router_node) builder.add_node(weather, weather_node) builder.add_node(news, news_node) builder.add_node(default, default_node) # 起点到路由节点 builder.add_edge(START, router) # 根据路由函数的结果条件分支到不同节点 builder.add_conditional_edges( router, decide_route, { weather: weather, news: news, default: default, }, ) # 每个分支最终都到达 END builder.add_edge(weather, END) builder.add_edge(news, END) builder.add_edge(default, END) app builder.compile() result app.invoke({query: 今天天气怎么样, route: }) print(result)运行结果{query: 今天天气怎么样【天气处理完成】, route: weather}5.5 条件路由的一个关键认知很多初学者会把“条件路由”理解成一个if-else函数但实际上它由两部分组成一部分是节点内部的状态更新这里是router_node另一部分是节点外部的路由选择这里是decide_route。这个分离带来一个好处router_node只负责“判断并写入 State”decide_route只负责“根据 State 决定走向”。两者职责单一后面增加新的分支节点时只需要改路由映射字典和新增节点不用动其他逻辑。另外说一下add_conditional_edges的第一个参数。它指定的是从哪个节点出发。第三个参数是一个字典key 是路由函数可能的返回值value 是目标节点名。如果路由函数返回的 key 不在映射字典里LangGraph 会抛异常所以这个映射必须完整覆盖路由函数的所有返回值。6. 实战构建一个带工具调用的 Agent前面讲完了基础现在进入本文的重头戏构建一个具备工具调用能力的 Agent。这里会用到 LangGraph 预构建的ToolNode和tools_condition它们能帮你省去大量样板代码。6.1 场景设计我们要做一个数学计算 Agent它有两个工具加法和乘法。用户输入“请计算 35 的结果再乘以 2”Agent 需要先调用加法工具得到 8再调用乘法工具得到 16最后给出最终回答。这个过程涉及两次工具调用中间需要循环Agent 节点输出“需要调用工具”的信号 → 工具节点执行工具 → 回到 Agent 节点 → Agent 判断是否还有工具需要调用 → 如果没有输出最终回答并结束。用传统链式结构写这个流程很痛苦用图结构则很自然。6.2 定义工具from langchain_core.tools import tool tool def add(a: int, b: int) - int: 计算两个整数 a 和 b 的和 return a b tool def multiply(a: int, b: int) - int: 计算两个整数 a 和 b 的乘积 return a * btool装饰器来自 LangChain它会把函数包装成 Agent 可调用的工具。注意函数的 docstring 非常重要因为大模型会阅读 docstring 来决定什么时候调用这个工具。6.3 定义 AgentState这里需要用到Annotated和operator.add这是 LangGraph 处理消息列表累积的标准做法import operator from typing import TypedDict, Annotated class AgentState(TypedDict): messages: Annotated[list, operator.add]Annotated[list, operator.add]表示当多个节点都返回messages字段时LangGraph 会用operator.add把新的消息列表追加到原有列表后面而不是覆盖旧消息。这对 Agent 的多轮消息记录至关重要因为模型的历史上下文需要完整保留。6.4 定义节点并构建图from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolNode, tools_condition构建 Agent 节点和状态图def agent_node(state: AgentState) - dict: Agent 节点调用大模型绑定工具决定是否调用工具 llm ChatOpenAI(modelgpt-4o-mini, temperature0) llm_with_tools llm.bind_tools([add, multiply]) return {messages: [llm_with_tools.invoke(state[messages])]} # 工具节点执行 Agent 决定调用的工具 tools_node ToolNode([add, multiply]) builder StateGraph(AgentState) builder.add_node(agent, agent_node) builder.add_node(tools, tools_node) # 起点直接进入 agent 节点 builder.add_edge(START, agent) # 关键从 agent 节点出发根据模型输出判断下一步 builder.add_conditional_edges( agent, tools_condition, {tools: tools, __end__: END} ) # 工具执行完后回到 agent 继续判断 builder.add_edge(tools, agent) app builder.compile()这里的关键是tools_condition。它读取出自模型的消息判断里面是否有工具调用请求。如果有返回toolsLangGraph 就会把执行权交给工具节点如果没有返回__end__LangGraph 就会结束整个执行。6.5 调用并验证result app.invoke({ messages: [ SystemMessage(content你是一个数学计算助手。), HumanMessage(content请计算 35 的结果再乘以 2。), ] }) print(result[messages][-1].content)如果你配置好了模型服务运行后最终输出应该是类似我们先计算 358再计算 8*216所以最终结果是 16。你可以把result[messages]全部打印出来会看到一条完整的执行链用户消息 → AI 调用工具的请求 → 工具返回结果 → AI 最终回答。这意味着整张图确实实现了“Agent 决定调哪个工具 → 工具执行 → 再次回到 Agent”的循环。6.6 这个例子的工程启发这个 Agent 示例虽然只有两个工具但它的架构可以无缝扩展到几十个工具。你只需要把工具列表传给ToolNode和bind_toolsLangGraph 会自动处理“模型返回多个工具调用请求”的情况。在实际项目中我建议把工具按业务域拆分到不同模块再统一汇总给 Agent这样维护成本会低很多。7. 子图与并行分支复杂业务场景怎么拆当业务变复杂之后把所有节点都平铺在一张图里会导致可读性变差。LangGraph 提供了子图Subgraph机制让你可以把一组相关节点封装成一个可复用的子图再作为节点嵌入主图。7.1 子图使用场景一个典型的场景是主图负责全局流程控制比如“先做预处理然后走业务处理最后做结果格式化”。其中“业务处理”内部可能又有多步分支这时候就可以把业务处理的部分抽成一个子图。7.2 子图示例from typing import TypedDict from langgraph.graph import StateGraph, START, END class SubState(TypedDict): text: str processed: int def step1(state: SubState) - dict: return {text: state[text] [已处理], processed: 1} def build_subgraph(): 构建子图内部只有一步处理但可以扩展成多步 sub_builder StateGraph(SubState) sub_builder.add_node(step1, step1) sub_builder.add_edge(START, step1) sub_builder.add_edge(step1, END) return sub_builder.compile() # 主图 class MainState(TypedDict): text: str processed: int subgraph_app build_subgraph() main_builder StateGraph(MainState) # 直接将编译后的子图作为主图的一个节点 main_builder.add_node(sub, subgraph_app) main_builder.add_edge(START, sub) main_builder.add_edge(sub, END) app main_builder.compile() result app.invoke({text: Hello, processed: 0}) print(result)运行结果{text: Hello [已处理], processed: 1}7.3 子图容易出现的问题子图最容易踩的坑是 State 类型不兼容。子图能读取和更新的字段必须在主图的 State 中有定义否则运行时会出现 KeyError。如果你发现子图执行时报错找不到某个字段优先检查主图和子图的 State 字段是否保持一致。另一个问题是子图的 START 和 END。子图必须有明确的 START 和 END 边否则在编译或运行时同样会报错。7.4 并行分支LangGraph 也支持并行执行多个节点。对于没有相互依赖的节点可以用 fan-out 方式让它们同时执行。比如class ParallelState(TypedDict): results: list[str]你可以让 A 节点并行路由到 B 和 C 两个节点两个节点执行完后再汇总到 D 节点def node_b(state: ParallelState) - dict: return {results: state[results] [B]} def node_c(state: ParallelState) - dict: return {results: state[results] [C]} def node_d(state: ParallelState) - dict: return {results: state[results] [D]} builder StateGraph(ParallelState) builder.add_node(b, node_b) builder.add_node(c, node_c) builder.add_node(d, node_d) builder.add_edge(START, b) builder.add_edge(START, c) builder.add_edge(b, d) builder.add_edge(c, d) builder.add_edge(d, END)这里隐含了一种并发模型LangGraph 会看到两个节点同时指向d如果d被合并它在两个上游节点都完成之前不会执行。不过要注意并行节点会同时更新 State所以 State 字段必须使用 reducer比如operator.add来避免覆盖否则后完成的节点会覆盖先完成的节点结果。如果你没有给results字段配置 reducer并行分支的结果就会丢失。正确的做法是使用Annotated[list, operator.add]import operator from typing import TypedDict, Annotated class ParallelState(TypedDict): results: Annotated[list[str], operator.add]并行分支在实际项目中的常见用途包括同时检索多个数据源、同时调用多个模型做投票、同时做多个独立的相似度计算。它能把总耗时从“串行累加”变成“最大单路耗时”是提升 Agent 响应速度的重要手段。8. 常见问题与排查思路学习 LangGraph 的过程中很多问题重复出现。我整理了一张高频问题表方便你直接对照排查。问题现象可能原因排查方式解决方案运行时报错Invalid state key节点返回了 State 中不存在的字段检查节点函数返回的字典 key 是否都在 State 中定义将字段补充到 State 定义中或删除多余返回字段图编译失败提示节点未连接有孤立节点未加边指向 END 或 START用builder.get_graph().print_ascii()检查图结构为孤立节点添加合适的边Agent 不调用工具而是直接回答模型没理解工具用途或 docstring 不清晰打印模型返回的消息检查是否有 tool_calls优化工具 docstring或调整系统提示词tools_condition路由到 END 一直不进入工具模型判断不需要调用工具检查工具列表和模型是否绑定成功确认bind_tools传入了工具列表子图执行时报 KeyError主图和子图的 State 字段不兼容检查主图与子图的 State 定义统一字段名或在子图中补充对应字段映射并行节点结果被覆盖字段未配置 reducer查看最终 State 是否只保留了最后一个节点结果使用Annotated[list, operator.add]消息列表重复累积reducer 被重复叠加观察每个节点的 messages 长度在不需要累积的字段避免使用operator.add无法连接到模型服务环境变量或 base_url 配置错误单独运行ChatOpenAI.invoke()测试检查 API Key、模型名、base_url确认模型服务可用有几个排查思路想重点强调一下第一LangGraph 的报错信息整体上比较清晰但如果你想快速定位是哪个节点出了问题最简单的方式是在节点函数开头加一行print日志。因为节点就是普通函数你可以在里面做任意调试。第二图结构错误可以通过builder.get_graph().print_ascii()打印出 ASCII 图这是快速检查连接关系的好工具。第三模型不调用工具的问题很多时候不是 LangGraph 的问题而是 Prompt 和工具定义的问题。你可以先用llm_with_tools.invoke(请计算 11)单独测试确认模型确实有工具调用能力再放到 LangGraph 里排查。9. 最佳实践与学习路线建议9.1 工程实践建议在真实项目中用 LangGraph有几个实践原则值得遵守。第一State 字段要克制。不要什么都往 State 里塞。State 是整张图共享的内存数据字段越多越容易产生隐式依赖也越难排查问题。建议只保存跨节点需要共享的数据局部变量尽量放在节点函数内部。第二节点职责要单一。一个节点只做一件事。比如“读取用户消息 → 判断意图 → 调用工具 → 生成回复”这四个环节应该拆成四个节点而不是塞进一个节点里。这样每个节点都可以独立测试。第三工具函数要写清楚 docstring。模型是通过 docstring 来理解工具用途的。一个模糊的工具描述会直接导致 Agent 调用错误。工具名和参数名也要符合直觉方便模型理解。第四模型接入要做好统一封装。建议在项目中单独封装一个llm_factory.py统一返回配置好的ChatOpenAI实例。这样切换模型服务商、修改温度参数时只需要动一个文件。第五生产环境优先考虑异步接口。Web 服务场景下不要用invoke用ainvoke或astream避免阻塞事件循环。LangGraph 对异步的支持很成熟这类改造并不复杂。第六做好日志和可观测性。在节点函数里记录关键状态变化在工具调用前后记录耗时。生产环境的 Agent 调试拼的就是日志完整度。LangGraph 支持流式输出你可以用astream_events拿到每个节点的执行事件把它接到监控系统里。9.2 常见认知误区这里再澄清三个常见误区。第一个误区是“LangGraph 只能配合 LangChain 使用”。实际上 LangGraph 是一个独立图执行框架它的节点函数是纯 Python 函数你可以在节点里调用任何你喜欢的库。LangChain 只是它最常见的搭档不是唯一选择。第二个误区是“有了 LangGraph 就不需要 LangChain”。LangChain 的模型接口、工具抽象、Prompt 管理、向量存储能力依然是 LangGraph 的基础设施。两者不是替代关系而是上下层关系。第三个误区是“Agent 一定要用图流程”。如果你的应用只是简单的“接收请求 → 调用一次模型 → 返回结果”用普通 LangChain 就够了引入 LangGraph 反而增加复杂度。框架选型要跟着业务复杂度走。9.3 后续学习路线如果你已经理解了本文的示例接下来可以按这样的顺序深入学习首先阅读 LangGraph 官方文档中的StateGraph和Checkpointer部分。Checkpointer 是 LangGraph 的记忆机制核心它让 Agent 可以保存和恢复历史状态是实现多轮对话持久化的关键。其次学习create_react_agent等预构建 Agent API。它面向简化场景代码更少适合快速原型。但理解底层 StateGraph 原理仍然重要因为复杂需求最终还是要自己控制图。再次研究流式输出和人工中断机制。LangGraph 支持在节点间插入交互式确认比如“调用这个工具前需要人工批准”这在高风险操作场景比如支付、删除数据中非常实用。最后多做一些项目实战。比如从本文的数学 Agent 扩展到带数据库查询、HTTP API 调用、文件处理等真实工具的 Agent。工具越多你越能体会到图编排带来的控制力提升。LangGraph 的核心价值是让 AI Agent 的执行过程变得可理解、可控制、可调试。如果你正在从“Demo 级别的 Agent”迈向“生产级别的 Agent”这个框架值得你花时间深入掌握。建议收藏本文照着示例逐步跑通再用自己的业务场景替换示例内容你会很快体会到从 Chain 到 Graph 带来的变化。
LangGraph实战：从Chain到Graph，构建可控AI Agent流程
很多人在学完 LangChain 之后兴冲冲地去做 Agent 项目结果很快遇到一个尴尬局面模型调用的流程根本不受自己控制。你可能想让 Agent 先做 A 工具调用再根据结果决定走 B 分支还是 C 分支结果模型一口气把所有工具全调了一遍或者在一个循环里出不来。你甚至没办法回答一个最基本的问题Agent 当前到底执行到哪一步了LangGraph 最近成为 Agent 开发领域的主流选择核心原因就是它把执行流程从“链式调用”变成了“状态图”。它真正解决的不是“能不能调大模型”而是“大模型执行流程怎么被可靠控制”这件事。本文会用完整的代码示例带你走一遍 LangGraph 入门到实战的完整路径包括核心概念、环境搭建、条件路由、工具调用、子图拆分和常见坑让你少走弯路。如果你正在做 AI Agent 相关项目或者准备从 LangChain 过渡到 LangGraph这篇文章可以直接作为你的第一篇实战教程来读。1. 这篇文章真正要解决的问题Agent 开发的失控感先讲一个真实的开发场景。假设你要做一个客服机器人用户问“我想退货怎么操作”你希望 Agent 走这样的流程先调用“判断用户意图”的节点再调用“查询订单”的工具最后调用“生成回复”的节点。每一步都需要可控至少你要知道它现在在哪一步。但用传统 LangChain 的LCEL语法写出来的链路是线性的prompt | llm | tool | llm这种线性链适合固定流程一旦遇到“根据模型输出决定下一步走哪里”的需求就会非常别扭。你可以用tool或者RunnableLambda加条件判断但代码会越来越乱尤其是当你需要循环调用工具、多步推理、失败重试、人工审批介入时线性链根本表达不清楚。于是 LangGraph 给出了一个新的心智模型把 Agent 的执行流程画成一张有向图。节点是你要执行的函数边是节点之间的流转关系状态则在整张图中共享和传递。这样你就能精确控制哪个节点先执行。什么条件下走哪条边。什么时候循环回上一个节点。什么时候结束。整张图执行过程中状态如何累积和变化。如果你正在做的项目涉及多轮工具调用、多步决策、循环控制或复杂状态管理LangGraph 比 LangChain 的传统链式写法更合适。这篇文章会一直围绕“控制力”这个核心来展开因为这才是 LangGraph 区别于其他框架的关键。2. LangGraph 是什么从 Chain 到 Graph 的思维转变2.1 LangChain 与 LangGraph 的关系很多初学者会把 LangGraph 和 LangChain 当成两个完全无关的框架这是一个常见的误解。LangChain 是一个大模型应用开发框架负责模型接入、Prompt 管理、输出解析、向量检索、记忆等基础能力。而 LangGraph 是 LangChain 官方团队推出的 Agent 编排框架它依赖于 LangChain 的模型调用和工具抽象但重新设计了执行引擎。简单来说对比维度LangChain传统链式LangGraph图式核心结构线性 Chain有向图 Graph流程表达固定管道节点 边 条件路由状态管理数据在链中流动共享 State 对象可跨节点更新循环控制不擅长原生支持分支控制需要 Hack原生 Conditional Edge适用场景固定流程、RAG复杂 Agent、多步决策、工具循环所谓“从 Chain 到 Graph”本质上是一次思维转变你不再把应用看成“一串连续调用”而是看成一个“节点与边组成的执行图”。这个转变带来两个直接好处一是复杂流程的表达能力大大增强二是你可以观察到任意节点的中间状态调试体验也更好。2.2 核心概念State、Node、Edge、Conditional EdgeLangGraph 有四个核心概念理解它们就能看懂 80% 的代码。State状态整张图共享的数据结构通常用TypedDict定义。每个节点执行完都会返回一个字典这个字典会更新 State 中的部分字段。如果字段需要累积而不是覆盖可以用Annotated[list, operator.add]这样的 reducer 来声明。Node节点一个普通的 Python 函数。函数的入参是当前 State返回值是一个字典表示要更新的 State 字段。Edge边定义节点之间的固定流转方向。比如 A 执行完无条件走向 B。Conditional Edge条件边这也是 LangGraph 最强大的地方。它允许你在某个节点执行完后根据当前 State 动态决定下一步去哪个节点。有点像代码里的if-else但把判断逻辑抽成了一个独立的“路由函数”。我用一个生活化的类比来解释把一个大模型应用想象成一家餐厅。State 是顾客的点菜单Node 是后厨的各个工位Edge 是“切完菜必须传给炒菜工位”这样的固定顺序Conditional Edge 则是“如果顾客是素食主义者就不传肉类菜品改传素菜窗口”这样的动态判断。2.3 LangGraph 的执行过程LangGraph 的执行过程可以简化成三步循环从图入口节点开始读取当前 State。执行当前节点也就是你的 Python 函数拿到返回值并更新 State。根据边或条件边判断下一个节点是谁。如果下一步是 END结束执行。这个“观察 State → 执行节点 → 更新 State → 选择下一个节点”的循环就是 LangGraph 的所有核心机制。3. 环境准备与模型接入配置3.1 安装依赖LangGraph 的 Python 包安装非常简单建议使用pip直接安装以下几个核心包pip install langgraph langchain langchain-openai如果你需要用到 LangGraph 的预构建工具节点、记忆功能或持久化能力可以继续安装pip install langgraph-checkpoint langgraph-prebuilt注意本文不绑定特定版本号因为 LangGraph 迭代很快。你只需要使用pip install --upgrade langgraph保持最新稳定版即可。Python 环境建议使用 3.10 及以上版本避免低版本带来的类型语法兼容问题。3.2 模型服务配置LangGraph 本身不直接调用模型而是通过 LangChain 的模型封装来调用。我这里以 OpenAI 协议兼容的服务为例来说明配置方式。在项目根目录下创建.env文件写入你的 API Key 和模型服务地址OPENAI_API_KEY你的_API_KEY OPENAI_BASE_URL你的模型服务地址这里要特别说明如果你使用的模型服务商提供 OpenAI 协议兼容接口那么langchain-openai这个包可以直接通过base_url参数对接。这样做的好处是模型切换只改配置不动业务代码。如果你希望完全本地化运行也可以考虑使用 Ollama 或其他本地推理服务它们都提供了 OpenAI 兼容接口。LangGraph 的模型接入方式是一样的只需要把模型名改成你本地加载的模型即可。3.3 验证环境安装完成后可以运行下面这段代码验证模型接入是否正常from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0) resp llm.invoke(你好请回复 OK) print(resp.content)能够正常输出内容说明环境基本就绪。接下来可以开始跑第一个 LangGraph 程序了。4. 跑通第一个 LangGraph 程序我用一个最简单的两节点图来说明白 LangGraph 的完整代码结构。4.1 定义 State首先定义一个 State 类型。这里我用TypedDict来表示当前状态中有哪些字段from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): messages: list[str]messages字段是一个字符串列表用于累积节点执行的信息。4.2 定义节点函数每个节点就是一个普通函数。入参是当前 State返回值是要更新的字段字典def node_a(state: State) - dict: return {messages: state[messages] [A 节点已执行]} def node_b(state: State) - dict: return {messages: state[messages] [B 节点已执行]}注意这里我用state[messages] [...]的方式手动拼接列表。这是因为当前的 State 没有声明 reducerLangGraph 默认会用“覆盖”的方式处理节点返回值。如果你希望两个节点分别更新同一字段并且自动合并后面在 Agent 实战中我会用Annotated来演示更好的做法。4.3 构建图并执行接下来就是 LangGraph 的核心代码# 创建状态图 builder StateGraph(State) # 添加节点 builder.add_node(a, node_a) builder.add_node(b, node_b) # 添加边 builder.add_edge(START, a) builder.add_edge(a, b) builder.add_edge(b, END) # 编译并执行 app builder.compile() result app.invoke({messages: []}) print(result)运行结果如下{messages: [A 节点已执行, B 节点已执行]}这完整演示了 LangGraph 的最小执行单元定义 State → 添加节点 → 连接边 → 编译 → 调用。从这个例子能看出LangGraph 并不是一个“黑盒”框架它的每个执行步骤都可以被观察和干预这正是它适合做复杂 Agent 的原因。4.4 如何理解 compile 和 invokecompile()会把状态图编译成一个可调用的应用对象。你可以在编译前后通过app.get_graph().draw_mermaid()需要安装额外依赖来可视化图结构不过本文不展开讲解绘图重点是执行逻辑。invoke()是同步执行入口传入初始 State返回最终 State。如果你有异步需求LangGraph 还提供了ainvoke()、astream()等异步接口适合用于 FastAPI 等异步 Web 服务场景。5. 条件路由与分支控制conditional_edge 深度解析第一个例子展示了固定流程但真实业务里更常见的是“根据内容做分支”。这一节我会给出一个完整的条件路由示例这也是 LangGraph 被使用最多的能力之一。5.1 场景设计假设我们有一个简单的意图路由场景用户输入包含“天气”走天气处理节点。用户输入包含“新闻”走新闻处理节点。其余输入走默认兜底节点。5.2 定义状态与节点from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END class RouterState(TypedDict): query: str route: str def router_node(state: RouterState) - dict: 路由节点根据 query 内容决定 route 字段的值 if 天气 in state[query]: return {route: weather} if 新闻 in state[query]: return {route: news} return {route: default} def weather_node(state: RouterState) - dict: return {query: state[query] 【天气处理完成】} def news_node(state: RouterState) - dict: return {query: state[query] 【新闻处理完成】} def default_node(state: RouterState) - dict: return {query: state[query] 【默认兜底处理完成】}5.3 条件路由函数条件路由的关键在于一个专门的“路由函数”。这个函数读取当前 State返回一个字符串表示下一步要走向哪个节点def decide_route(state: RouterState) - Literal[weather, news, default]: return state[route]这个函数本身没有复杂逻辑它只是返回router_node设置好的route字段。在实际项目中路由函数可以更复杂比如读取数据库配置、调用外部服务来决定路由。5.4 使用 add_conditional_edges 构建分支builder StateGraph(RouterState) # 添加节点 builder.add_node(router, router_node) builder.add_node(weather, weather_node) builder.add_node(news, news_node) builder.add_node(default, default_node) # 起点到路由节点 builder.add_edge(START, router) # 根据路由函数的结果条件分支到不同节点 builder.add_conditional_edges( router, decide_route, { weather: weather, news: news, default: default, }, ) # 每个分支最终都到达 END builder.add_edge(weather, END) builder.add_edge(news, END) builder.add_edge(default, END) app builder.compile() result app.invoke({query: 今天天气怎么样, route: }) print(result)运行结果{query: 今天天气怎么样【天气处理完成】, route: weather}5.5 条件路由的一个关键认知很多初学者会把“条件路由”理解成一个if-else函数但实际上它由两部分组成一部分是节点内部的状态更新这里是router_node另一部分是节点外部的路由选择这里是decide_route。这个分离带来一个好处router_node只负责“判断并写入 State”decide_route只负责“根据 State 决定走向”。两者职责单一后面增加新的分支节点时只需要改路由映射字典和新增节点不用动其他逻辑。另外说一下add_conditional_edges的第一个参数。它指定的是从哪个节点出发。第三个参数是一个字典key 是路由函数可能的返回值value 是目标节点名。如果路由函数返回的 key 不在映射字典里LangGraph 会抛异常所以这个映射必须完整覆盖路由函数的所有返回值。6. 实战构建一个带工具调用的 Agent前面讲完了基础现在进入本文的重头戏构建一个具备工具调用能力的 Agent。这里会用到 LangGraph 预构建的ToolNode和tools_condition它们能帮你省去大量样板代码。6.1 场景设计我们要做一个数学计算 Agent它有两个工具加法和乘法。用户输入“请计算 35 的结果再乘以 2”Agent 需要先调用加法工具得到 8再调用乘法工具得到 16最后给出最终回答。这个过程涉及两次工具调用中间需要循环Agent 节点输出“需要调用工具”的信号 → 工具节点执行工具 → 回到 Agent 节点 → Agent 判断是否还有工具需要调用 → 如果没有输出最终回答并结束。用传统链式结构写这个流程很痛苦用图结构则很自然。6.2 定义工具from langchain_core.tools import tool tool def add(a: int, b: int) - int: 计算两个整数 a 和 b 的和 return a b tool def multiply(a: int, b: int) - int: 计算两个整数 a 和 b 的乘积 return a * btool装饰器来自 LangChain它会把函数包装成 Agent 可调用的工具。注意函数的 docstring 非常重要因为大模型会阅读 docstring 来决定什么时候调用这个工具。6.3 定义 AgentState这里需要用到Annotated和operator.add这是 LangGraph 处理消息列表累积的标准做法import operator from typing import TypedDict, Annotated class AgentState(TypedDict): messages: Annotated[list, operator.add]Annotated[list, operator.add]表示当多个节点都返回messages字段时LangGraph 会用operator.add把新的消息列表追加到原有列表后面而不是覆盖旧消息。这对 Agent 的多轮消息记录至关重要因为模型的历史上下文需要完整保留。6.4 定义节点并构建图from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolNode, tools_condition构建 Agent 节点和状态图def agent_node(state: AgentState) - dict: Agent 节点调用大模型绑定工具决定是否调用工具 llm ChatOpenAI(modelgpt-4o-mini, temperature0) llm_with_tools llm.bind_tools([add, multiply]) return {messages: [llm_with_tools.invoke(state[messages])]} # 工具节点执行 Agent 决定调用的工具 tools_node ToolNode([add, multiply]) builder StateGraph(AgentState) builder.add_node(agent, agent_node) builder.add_node(tools, tools_node) # 起点直接进入 agent 节点 builder.add_edge(START, agent) # 关键从 agent 节点出发根据模型输出判断下一步 builder.add_conditional_edges( agent, tools_condition, {tools: tools, __end__: END} ) # 工具执行完后回到 agent 继续判断 builder.add_edge(tools, agent) app builder.compile()这里的关键是tools_condition。它读取出自模型的消息判断里面是否有工具调用请求。如果有返回toolsLangGraph 就会把执行权交给工具节点如果没有返回__end__LangGraph 就会结束整个执行。6.5 调用并验证result app.invoke({ messages: [ SystemMessage(content你是一个数学计算助手。), HumanMessage(content请计算 35 的结果再乘以 2。), ] }) print(result[messages][-1].content)如果你配置好了模型服务运行后最终输出应该是类似我们先计算 358再计算 8*216所以最终结果是 16。你可以把result[messages]全部打印出来会看到一条完整的执行链用户消息 → AI 调用工具的请求 → 工具返回结果 → AI 最终回答。这意味着整张图确实实现了“Agent 决定调哪个工具 → 工具执行 → 再次回到 Agent”的循环。6.6 这个例子的工程启发这个 Agent 示例虽然只有两个工具但它的架构可以无缝扩展到几十个工具。你只需要把工具列表传给ToolNode和bind_toolsLangGraph 会自动处理“模型返回多个工具调用请求”的情况。在实际项目中我建议把工具按业务域拆分到不同模块再统一汇总给 Agent这样维护成本会低很多。7. 子图与并行分支复杂业务场景怎么拆当业务变复杂之后把所有节点都平铺在一张图里会导致可读性变差。LangGraph 提供了子图Subgraph机制让你可以把一组相关节点封装成一个可复用的子图再作为节点嵌入主图。7.1 子图使用场景一个典型的场景是主图负责全局流程控制比如“先做预处理然后走业务处理最后做结果格式化”。其中“业务处理”内部可能又有多步分支这时候就可以把业务处理的部分抽成一个子图。7.2 子图示例from typing import TypedDict from langgraph.graph import StateGraph, START, END class SubState(TypedDict): text: str processed: int def step1(state: SubState) - dict: return {text: state[text] [已处理], processed: 1} def build_subgraph(): 构建子图内部只有一步处理但可以扩展成多步 sub_builder StateGraph(SubState) sub_builder.add_node(step1, step1) sub_builder.add_edge(START, step1) sub_builder.add_edge(step1, END) return sub_builder.compile() # 主图 class MainState(TypedDict): text: str processed: int subgraph_app build_subgraph() main_builder StateGraph(MainState) # 直接将编译后的子图作为主图的一个节点 main_builder.add_node(sub, subgraph_app) main_builder.add_edge(START, sub) main_builder.add_edge(sub, END) app main_builder.compile() result app.invoke({text: Hello, processed: 0}) print(result)运行结果{text: Hello [已处理], processed: 1}7.3 子图容易出现的问题子图最容易踩的坑是 State 类型不兼容。子图能读取和更新的字段必须在主图的 State 中有定义否则运行时会出现 KeyError。如果你发现子图执行时报错找不到某个字段优先检查主图和子图的 State 字段是否保持一致。另一个问题是子图的 START 和 END。子图必须有明确的 START 和 END 边否则在编译或运行时同样会报错。7.4 并行分支LangGraph 也支持并行执行多个节点。对于没有相互依赖的节点可以用 fan-out 方式让它们同时执行。比如class ParallelState(TypedDict): results: list[str]你可以让 A 节点并行路由到 B 和 C 两个节点两个节点执行完后再汇总到 D 节点def node_b(state: ParallelState) - dict: return {results: state[results] [B]} def node_c(state: ParallelState) - dict: return {results: state[results] [C]} def node_d(state: ParallelState) - dict: return {results: state[results] [D]} builder StateGraph(ParallelState) builder.add_node(b, node_b) builder.add_node(c, node_c) builder.add_node(d, node_d) builder.add_edge(START, b) builder.add_edge(START, c) builder.add_edge(b, d) builder.add_edge(c, d) builder.add_edge(d, END)这里隐含了一种并发模型LangGraph 会看到两个节点同时指向d如果d被合并它在两个上游节点都完成之前不会执行。不过要注意并行节点会同时更新 State所以 State 字段必须使用 reducer比如operator.add来避免覆盖否则后完成的节点会覆盖先完成的节点结果。如果你没有给results字段配置 reducer并行分支的结果就会丢失。正确的做法是使用Annotated[list, operator.add]import operator from typing import TypedDict, Annotated class ParallelState(TypedDict): results: Annotated[list[str], operator.add]并行分支在实际项目中的常见用途包括同时检索多个数据源、同时调用多个模型做投票、同时做多个独立的相似度计算。它能把总耗时从“串行累加”变成“最大单路耗时”是提升 Agent 响应速度的重要手段。8. 常见问题与排查思路学习 LangGraph 的过程中很多问题重复出现。我整理了一张高频问题表方便你直接对照排查。问题现象可能原因排查方式解决方案运行时报错Invalid state key节点返回了 State 中不存在的字段检查节点函数返回的字典 key 是否都在 State 中定义将字段补充到 State 定义中或删除多余返回字段图编译失败提示节点未连接有孤立节点未加边指向 END 或 START用builder.get_graph().print_ascii()检查图结构为孤立节点添加合适的边Agent 不调用工具而是直接回答模型没理解工具用途或 docstring 不清晰打印模型返回的消息检查是否有 tool_calls优化工具 docstring或调整系统提示词tools_condition路由到 END 一直不进入工具模型判断不需要调用工具检查工具列表和模型是否绑定成功确认bind_tools传入了工具列表子图执行时报 KeyError主图和子图的 State 字段不兼容检查主图与子图的 State 定义统一字段名或在子图中补充对应字段映射并行节点结果被覆盖字段未配置 reducer查看最终 State 是否只保留了最后一个节点结果使用Annotated[list, operator.add]消息列表重复累积reducer 被重复叠加观察每个节点的 messages 长度在不需要累积的字段避免使用operator.add无法连接到模型服务环境变量或 base_url 配置错误单独运行ChatOpenAI.invoke()测试检查 API Key、模型名、base_url确认模型服务可用有几个排查思路想重点强调一下第一LangGraph 的报错信息整体上比较清晰但如果你想快速定位是哪个节点出了问题最简单的方式是在节点函数开头加一行print日志。因为节点就是普通函数你可以在里面做任意调试。第二图结构错误可以通过builder.get_graph().print_ascii()打印出 ASCII 图这是快速检查连接关系的好工具。第三模型不调用工具的问题很多时候不是 LangGraph 的问题而是 Prompt 和工具定义的问题。你可以先用llm_with_tools.invoke(请计算 11)单独测试确认模型确实有工具调用能力再放到 LangGraph 里排查。9. 最佳实践与学习路线建议9.1 工程实践建议在真实项目中用 LangGraph有几个实践原则值得遵守。第一State 字段要克制。不要什么都往 State 里塞。State 是整张图共享的内存数据字段越多越容易产生隐式依赖也越难排查问题。建议只保存跨节点需要共享的数据局部变量尽量放在节点函数内部。第二节点职责要单一。一个节点只做一件事。比如“读取用户消息 → 判断意图 → 调用工具 → 生成回复”这四个环节应该拆成四个节点而不是塞进一个节点里。这样每个节点都可以独立测试。第三工具函数要写清楚 docstring。模型是通过 docstring 来理解工具用途的。一个模糊的工具描述会直接导致 Agent 调用错误。工具名和参数名也要符合直觉方便模型理解。第四模型接入要做好统一封装。建议在项目中单独封装一个llm_factory.py统一返回配置好的ChatOpenAI实例。这样切换模型服务商、修改温度参数时只需要动一个文件。第五生产环境优先考虑异步接口。Web 服务场景下不要用invoke用ainvoke或astream避免阻塞事件循环。LangGraph 对异步的支持很成熟这类改造并不复杂。第六做好日志和可观测性。在节点函数里记录关键状态变化在工具调用前后记录耗时。生产环境的 Agent 调试拼的就是日志完整度。LangGraph 支持流式输出你可以用astream_events拿到每个节点的执行事件把它接到监控系统里。9.2 常见认知误区这里再澄清三个常见误区。第一个误区是“LangGraph 只能配合 LangChain 使用”。实际上 LangGraph 是一个独立图执行框架它的节点函数是纯 Python 函数你可以在节点里调用任何你喜欢的库。LangChain 只是它最常见的搭档不是唯一选择。第二个误区是“有了 LangGraph 就不需要 LangChain”。LangChain 的模型接口、工具抽象、Prompt 管理、向量存储能力依然是 LangGraph 的基础设施。两者不是替代关系而是上下层关系。第三个误区是“Agent 一定要用图流程”。如果你的应用只是简单的“接收请求 → 调用一次模型 → 返回结果”用普通 LangChain 就够了引入 LangGraph 反而增加复杂度。框架选型要跟着业务复杂度走。9.3 后续学习路线如果你已经理解了本文的示例接下来可以按这样的顺序深入学习首先阅读 LangGraph 官方文档中的StateGraph和Checkpointer部分。Checkpointer 是 LangGraph 的记忆机制核心它让 Agent 可以保存和恢复历史状态是实现多轮对话持久化的关键。其次学习create_react_agent等预构建 Agent API。它面向简化场景代码更少适合快速原型。但理解底层 StateGraph 原理仍然重要因为复杂需求最终还是要自己控制图。再次研究流式输出和人工中断机制。LangGraph 支持在节点间插入交互式确认比如“调用这个工具前需要人工批准”这在高风险操作场景比如支付、删除数据中非常实用。最后多做一些项目实战。比如从本文的数学 Agent 扩展到带数据库查询、HTTP API 调用、文件处理等真实工具的 Agent。工具越多你越能体会到图编排带来的控制力提升。LangGraph 的核心价值是让 AI Agent 的执行过程变得可理解、可控制、可调试。如果你正在从“Demo 级别的 Agent”迈向“生产级别的 Agent”这个框架值得你花时间深入掌握。建议收藏本文照着示例逐步跑通再用自己的业务场景替换示例内容你会很快体会到从 Chain 到 Graph 带来的变化。