如果你正在学习大语言模型应用开发接触过 LangChain最近大概率会被一个词刷屏LangGraph。各大技术社区、B 站教程、GitHub 开源项目里都开始出现它的身影。但你打开官方文档时很快会被一堆概念搞懵State、Node、Edge、Conditional Edge、Subgraph、RecursionLimit……英文文档读着费劲中文资料又很多是表面翻译真正能落地跑通的教程少之又少。这篇文章的目标很直接用一条清晰的实战主线把 LangGraph 最核心的四个能力讲透。包括如何定义状态、如何用节点组装流程、如何用条件边做分支和循环、如何用子图和并行分支应对复杂业务场景。同时我会专门回答一个搜索量很高的问题在节点函数里到底怎么改 State 状态值。先说一个核心判断LangGraph 真正降低的不是“调用大模型”的门槛而是“编排多条工具调用、多步推理流程”的工程复杂度。如果你只是写一个“输入一句话、返回一段回答”的简单 Demo不需要 LangGraph但如果你要做带工具调用、多步判断、人工介入、可恢复状态的 AgentLangGraph 是目前少数能优雅解决这些问题的框架。1. 为什么 LangGraph 值得学它真正解决的问题1.1 LangChain 的痛点LangChain 早期最常用的抽象是 Chain也就是把 Prompt、LLM、工具按顺序串起来。这个设计在处理“固定步骤”时非常好用比如先查数据库再写总结最后翻译成英文。但真实的 Agent 场景不是固定的线性流程而是有判断、有循环、有分叉的图结构。举个例子用户问“帮我查一下上海明天天气再对比一下北京”。传统 Chain 很难优雅地表达“先判断用户提到了几个城市再分别查询最后合并结果”这样的流程。就算强行写代码也会变成大段 if-else维护成本非常高。1.2 LangGraph 与 LangChain 的关系LangGraph 是 LangChain 团队推出的“有状态、可编排”的 Agent 框架。它不是 LangChain 的替代品而是底层流程引擎。你可以继续用 LangChain 的 Prompt 模板、大模型封装、工具集成然后把这些组件放进 LangGraph 的图里由 LangGraph 来控制执行顺序、状态流转、分支跳转和循环退出。两者最直白的区别是维度LangChainLangGraph核心抽象Chain、AgentExecutorStateGraph流程结构线性链条为主有向图支持分支、循环、并行状态管理靠 Chain 内部传递不直观统一 State节点间显式传递适用场景固定流程任务复杂 Agent、多步推理、人工审核流程很多朋友问“langchain 和 langgraph 的区别”答案不是谁取代谁而是分工不同。你把 LangChain 当工具箱把 LangGraph 当流水线控制系统配合使用效率最高。1.3 读完这篇文章你能得到什么读完并动手跑完整篇文章的示例后你应该能完成三件事第一独立理解 LangGraph 官方文档中 State、Node、Conditional Edge 这些基本概念第二用 LangGraph 写一个带条件路由和循环的 Agent 流程第三知道在图里哪些地方容易踩坑以及生产环境里推荐怎么组织代码。2. 核心概念State、Node、Edge 与执行原理2.1 State 状态Agent 的“记忆黑板”LangGraph 的核心是 State。你可以把它理解成一张在所有节点之间流动的“黑板”。每个节点都能读取黑板上的内容也能往黑板上写新内容下一个节点读到的就是更新后的黑板。State 通常通过 TypedDict 定义例如from typing import TypedDict class ChatState(TypedDict): messages: list user_id: str在简单的顺序流程里State 就是一个字典在复杂场景下可以通过Annotated和 reducer 控制某个字段的更新方式这个后面详细讲。2.2 Node 节点每一步只做一件事Node 是图里的最小执行单元本质上就是一个函数。函数输入是当前 State输出是一个字典字典里的键会被更新到 State 中。def my_node(state: ChatState) - dict: return {user_id: u_123}设计原则是“一个节点只做一件事”。不要在一个节点里既调模型又查数据库又做判断拆开之后流程可观测性和复用性都会好很多。2.3 Edge 与 Conditional Edge流程的“红绿灯”Edge 是节点之间的连线表达“当前节点执行完下一步去哪个节点”。普通 Edge 是固定连接Conditional Edge 则是“根据当前 State 动态决定下一步走向”。Conditional Edge 是 LangGraph 最吸引人的点也是很多教程没讲透的地方。它由一个路由函数和一张路由表组成路由函数读 State返回一个字符串路由表把字符串映射到对应节点。这就是 Agent 能“判断”的关键机制。2.4 子图、并行、循环复杂流程的三种组合方式学习了 Node 和 Edge 之后图的基本单位已经足够但真实业务还要用到三种组合方式循环Conditional Edge 指回之前的节点实现“做得不对就再来一次”。子图把一个完整流程封装成一个图再作为一个节点嵌入父图。并行从一个节点同时发散到多个节点再合并结果。这三种能力组合起来基本能覆盖 90% 的 Agent 编排需求。3. 环境准备与安装3.1 运行环境要求LangGraph 是 Python 库建议使用 Python 3.9 及以上版本本文示例在 Python 3.10 环境下验证。安装前建议先创建虚拟环境避免污染系统 Python。python -m venv .venv source .venv/bin/activate # Windows 使用 .venv\Scripts\activate3.2 安装 LangGraph安装核心库即可pip install langgraph本文示例涉及读取大模型但为了降低联网和 API Key 的干扰我会尽量让示例不依赖真实大模型。如果后面你希望接入 OpenAI 兼容接口需要额外安装pip install langchain-openai python-dotenv版本提示LangGraph 迭代很快API 可能有细微调整本文示例以较新的稳定版为准。如果你安装的是老版本建议先升级pip install --upgrade langgraph3.3 验证安装在命令行执行python -c import langgraph; print(langgraph.__version__)如果正常输出版本号说明安装成功。如果报ModuleNotFoundError检查是否激活了正确的虚拟环境。4. 第一个 LangGraph 应用跑通一条完整流程4.1 定义 State先来一个最小示例两个节点按顺序执行分别向 State 中追加一条消息并更新计数。# graph_basic.py from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, START, END class BasicState(TypedDict): messages: Annotated[list, operator.add] count: int这里messages使用了Annotated[list, operator.add]含义是每次节点返回messages字段时不覆盖旧值而是用operator.add把新旧列表拼接起来。这是 LangGraph 中 reducer 的典型用法后面会细讲。4.2 创建节点并连线定义节点函数并创建图结构def node_a(state: BasicState) - dict: return { messages: [fA 收到 count{state[count]}], count: state[count] 1, } def node_b(state: BasicState) - dict: return { messages: [B 处理完成], count: state[count] 1, } builder StateGraph(BasicState) builder.add_node(node_a, node_a) builder.add_node(node_b, node_b) builder.add_edge(START, node_a) builder.add_edge(node_a, node_b) builder.add_edge(node_b, END)START和END是 LangGraph 内置的特殊节点。START表示流程入口END表示流程结束。4.3 编译并运行图构建好后需要调用compile()得到可执行对象graph builder.compile() result graph.invoke({ messages: [], count: 0, }) print(result)运行结果中messages会是两条消息拼接后的列表count变成 2。这证明两个节点按顺序执行且 State 在整个流程中保持共享。这个最小示例虽然简单但已经包含了 LangGraph 最关键的机制State 定义、节点函数、普通边、编译执行。后面所有复杂功能都是在这个基础上扩展的。5. 条件路由conditional_edge 深度解析与分支控制5.1 为什么需要条件路由顺序执行无法应对真实业务。用户输入“今天天气如何”Agent 需要走天气查询工具输入“3.5 乘以 8”Agent 需要走计算工具。传统 if-else 写在单个函数里会导致一个函数越来越长。LangGraph 的做法是把“走向判断”抽成一条边让路由函数只负责决策业务节点只负责执行。5.2 路由函数与路由表条件路由由两部分组成路由函数接收 State返回一个字符串 key。路由表字典key 对应目标节点名。写法如下builder.add_conditional_edges( analyze, # 路由函数的源节点 route_function, # 路由函数 { weather: weather_node, math: math_node, default: END, }, )当analyze节点执行完后LangGraph 会调用route_function(state)拿到返回的字符串 key再根据路由表跳到对应节点。5.3 完整示例按问题类型分发下面的示例用规则区分“天气”和“计算”两类问题完全不需要大模型方便本地直接跑通# graph_conditional.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class RouteState(TypedDict): query: str route: str result: str def analyze(state: RouteState) - dict: query state[query] if 天气 in query: return {route: weather} if 计算 in query: return {route: math} return {route: default} def weather_node(state: RouteState) - dict: return {result: 今天晴转多云适合出行。} def math_node(state: RouteState) - dict: # 这里只是演示生产环境不要直接计算任意表达式 expression state[query].replace(计算, ).strip() return {result: f收到计算需求{expression}} def route_function(state: RouteState) - str: return state[route] builder StateGraph(RouteState) builder.add_node(analyze, analyze) builder.add_node(weather_node, weather_node) builder.add_node(math_node, math_node) builder.add_edge(START, analyze) builder.add_conditional_edges( analyze, route_function, { weather: weather_node, math: math_node, default: END, }, ) builder.add_edge(weather_node, END) builder.add_edge(math_node, END) graph builder.compile() print(graph.invoke({query: 上海明天天气})) print(graph.invoke({query: 计算 12345 6789}))运行后两次 invoke 会走向不同节点最终result字段内容不同。这就是最核心的分支控制。5.4 循环检测与条件路由的配合条件路由不仅能做分支还能做循环。只要路由函数返回一个“之前的节点”流程就会形成循环。循环时最容易出现死循环因此 LangGraph 提供了一个安全机制recursion_limit。如果一次运行中节点执行次数超过该限制会抛出异常。try: graph.invoke({query: 循环测试}, config{recursion_limit: 10}) except Exception as e: print(触发限制, e)实际开发中这个配置要保留防止模型判断失误导致 Agent 无限循环。6. 循环与循环检测让 Agent 学会“再来一次”6.1 Agent 中的循环长什么样Agent 的典型循环模式是“思考并行动”。大模型根据当前信息决定调用某个工具拿到工具结果后再次思考直到认为已经得到最终答案。每一步都可以看作图中的一个节点而“是否继续”就是一条从“判断节点”指回“思考节点”的边。6.2 用条件边实现循环下面示例模拟一个最多执行 5 次迭代的流程# graph_loop.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class LoopState(TypedDict): step: int result: str def process(state: LoopState) - dict: step state[step] 1 if step 5: return {step: step, result: f第 {step} 次迭代} return {step: step, result: 迭代完成} def controller(state: LoopState) - str: return process if state[step] 5 else end_node def end_node(state: LoopState) - dict: return {result: 最终结束 state[result]} builder StateGraph(LoopState) builder.add_node(process, process) builder.add_node(end_node, end_node) builder.add_edge(START, process) builder.add_conditional_edges( process, controller, { process: process, end_node: end_node, }, ) builder.add_edge(end_node, END) graph builder.compile() result graph.invoke({step: 0, result: }) print(result)关键在于controller返回process时流程会回到process节点形成循环。这是 LangGraph 中实现“不断重试、不断执行工具”的通用方式。6.3 防止死循环recursion_limit 与步骤计数在需要调用外部 API 的 Agent 场景中模型可能反复选择同一个工具导致流程无法结束。推荐双重保护第一在invoke时传入recursion_limitgraph.invoke(input_state, config{recursion_limit: 20})第二在 State 中加入step字段由节点函数负责增长计数并在路由函数中判断上限。只要进入容错分支就跳到结束节点避免一直空转。从 LangGraph 的行为来看recursion_limit更像安全网不要依赖它做业务结束判断。业务上的“是否结束”应该由路由函数显式决定。7. 子图与并行分支复杂流程的两种组合方式7.1 什么是子图什么时候用子图就是把一个流程封装成图再当成节点嵌入到另一个图里。当某个子流程足够独立时就应该拆成子图。典型例子一个“文章自动发布”系统主图负责“选题、写稿、审核、发布”其中“写稿”内部又可以拆成“搜索资料、生成大纲、逐段生成、合并润色”四个步骤。这四个步骤内聚度很高对外只暴露“输入主题、输出文章”非常适合做成子图。7.2 子图作为节点接入父图子图编译后的对象可以直接作为节点函数使用。假设已经有一个编译好的writing_subgraphparent_builder StateGraph(ParentState) parent_builder.add_node(topic_select, topic_select_node) parent_builder.add_node(writing, writing_subgraph) parent_builder.add_node(publish, publish_node) parent_builder.add_edge(START, topic_select) parent_builder.add_edge(topic_select, writing) parent_builder.add_edge(writing, publish) parent_builder.add_edge(publish, END)子图内部怎么跑父图不关心父图只关心子图的输入 State 和输出 State。这能显著降低复杂系统的认知负担。7.3 并行分支并发执行与结果合并并行分支适合“多个独立任务同时执行”的场景。比如用户要同时查三个城市的天气没必要串行查询可以在一个节点分裂出三个分支都指向合并节点。LangGraph 支持从一个节点发出多条边这些目标节点会在同一个 super-step 中并行执行。示例# graph_parallel.py from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, START, END class ParallelState(TypedDict): cities: list results: Annotated[list, operator.add] done: bool def prepare(state: ParallelState) - dict: return {cities: [北京, 上海, 广州]} def query_beijing(state: ParallelState) - dict: return {results: [北京晴]} def query_shanghai(state: ParallelState) - dict: return {results: [上海小雨]} def query_guangzhou(state: ParallelState) - dict: return {results: [广州多云]} def merge(state: ParallelState) - dict: return {done: True, results: state[results]} builder StateGraph(ParallelState) builder.add_node(prepare, prepare) builder.add_node(query_beijing, query_beijing) builder.add_node(query_shanghai, query_shanghai) builder.add_node(query_guangzhou, query_guangzhou) builder.add_node(merge, merge) builder.add_edge(START, prepare) builder.add_conditional_edges( prepare, lambda state: [query_beijing, query_shanghai, query_guangzhou], )注意这里add_conditional_edges的路由函数可以直接返回一组节点名让流程同时走向多个节点。LangGraph 会尝试并行执行它们。最后连上合并节点builder.add_edge(query_beijing, merge) builder.add_edge(query_shanghai, merge) builder.add_edge(query_guangzhou, merge) builder.add_edge(merge, END) graph builder.compile() result graph.invoke({cities: [], results: [], done: False}) print(result)合并的关键是results字段使用了Annotated[list, operator.add]。如果没有这个 reducer多个并行节点返回results时后写的会覆盖先写的最终只能拿到一个城市的结果。这是并行分支最容易踩的坑。8. 在节点函数中改变 State 状态值正确姿势与常见误区搜索“langgraph如何在节点函数改变state状态值”的人很多这里专门展开讲。LangGraph 官方推荐的更新方式是节点函数返回一个字典字典中键与 State 中的字段对应。def update_node(state: MyState) - dict: return {name: 张三, age: 18}执行这个节点后State 中name变为“张三”age变为 18。这个方式最简单、最可控也是官方文档示例的主路径。8.1 返回 dict 更新状态适合覆盖式更新单个字段。注意如果没有在 State 字段上指定 reducer返回的新值会直接覆盖旧值。class ProfileState(TypedDict): name: str age: int def change_name(state: ProfileState) - dict: # 读取 current state[name] # 返回更新 return {name: current _new}8.2 使用 reducer 合并状态如果多个节点同时更新同一个字段或者希望“追加”而不是“覆盖”就需要 reducer。from typing import Annotated import operator class ListState(TypedDict): history: Annotated[list, operator.add] def node_1(state: ListState) - dict: return {history: [第一步]} def node_2(state: ListState) - dict: return {history: [第二步]}两个节点都执行后history是[第一步, 第二步]而不是只剩[第二步]。这个机制在并行分支和子图状态回传中尤其重要。8.3 避免直接修改 State 的三个原因很多人为了省事会在节点函数里直接写state[name] 张三这样做在部分场景下能工作但我不推荐。原因有三个第一LangGraph 的状态更新机制依赖返回值直接修改不等于显式更新在开启持久化、时间旅行、断点功能时可能无法正确记录状态变化。第二函数变得难以测试因为你既要传入 State又需要关心函数是否修改了入参不符合纯函数习惯。第三遇到 reducer 时直接修改很容易绕过合并逻辑导致数据出现意想不到的覆盖。更稳妥的写法是每个节点函数只通过返回 dict 声明要更新的字段。读取操作直接访问state修改操作全部交给返回值。这样图和状态流都非常清晰。9. 常见问题与排查思路问题现象可能原因排查方式解决方案ImportError: cannot import name STARTlanggraph 版本过旧查看版本号打印langgraph.__version__升级pip install --upgrade langgraph启动后抛 GraphRecursionError循环条件一直为真看异常堆栈中循环节点名修正路由函数或调大recursion_limit业务上用 step 字段设上限并行分支最终只保留一个结果多节点写同一 key 但没用 reducer打印中间 state 观察覆盖将字段类型改为Annotated[list, operator.add]子图返回的数据没有更新父图子图 State 与父图 State 字段不对应检查子图输入输出 State 定义统一子图与父图 State或做显式字段映射调大模型时报 API Key 错误环境变量未加载检查终端环境变量或.env文件使用python-dotenv加载配置确认 Key 有效graph.invoke卡住不返回内部工具调用超时或死循环打开日志追踪节点执行顺序给外部调用设置超时增加步骤计数兜底排查 LangGraph 问题时最有效的工具不是反复看文档而是把每个节点的返回值和中间 State 打印出来。你可以把graph.invoke改成逐步调试也可以给节点函数加日志import logging logging.basicConfig(levellogging.INFO) def my_node(state: MyState) - dict: logging.info(my_node 收到 state: %s, state) return {result: ok}看日志比猜逻辑快得多。10. 最佳实践与后续学习方向10.1 工程建议第一State 定义要克制。很多项目一开始就把所有临时变量都放进 State导致图很难调试。State 只保存需要在节点之间传递的业务数据不保存无关的中间计算结果。第二节点函数保持“读 State返回新 State”的纯函数风格。不要做文件操作后不返回任何状态也不要在一个节点里塞太多逻辑。一个节点做太多事会导致断点续跑和错误定位都很痛苦。第三所有外部调用都要设置超时和异常兜底。大模型接口、数据库、第三方 API 都可能失败失败后路由到重试节点还是结束节点需要在画图前就想清楚。第四生产环境考虑使用 LangGraph 的持久化扩展。它能记录每一步状态支持断点恢复和人工介入。对需要“用户确认后继续”的 Agent 流程这是很有价值的能力。第五测试要覆盖路由函数。分支是最容易出 bug 的地方建议单独测试路由函数在不同输入下返回的目标节点。10.2 后续学习方向这篇文章只覆盖了 LangGraph 的基础图结构。如果再深入建议按这个顺序学习State 的进阶用法Pydantic State、多个 reducer 组合、状态校验。Checkpoint 与持久化把图运行中间状态保存到数据库。断点与人工审核在关键节点暂停等人工确认后再继续。多 Agent 协作多个图或子图之间传递结果。LangGraph Platform把调试好的图部署成服务。这些能力并不是并列的而是层层递进。先把本文中的最小示例和条件路由跑通再往持久化和部署方向走就不会被底层概念劝退。10.3 最后提醒LangGraph 的学习曲线不算陡但它和普通链式调用完全不同。不要期望一次性写出完美的大图我的建议是从“一个节点 一条边”开始逐步加条件分支再加循环再加子图和并行。每加一个能力都先打印一次 State确认状态流向符合预期再继续下一层。另外有朋友问 LangGraph 是否有 Rust 版本。目前官方主推是 Python 和 TypeScript 版本并没有看到官方维护的 Rust 版本。如果你所在团队是 Rust 技术栈更务实的路径是通过 HTTP API 或 LangGraph Platform 调用服务端而不是强行把图逻辑落到 Rust 代码里。通篇读到这里你已经有能力把 LangGraph 放进自己的 Agent 项目里了。推荐的做法是找一个真实小需求比如“用户输入城市名查天气并生成出行建议”用 LangGraph 重写一遍。写的过程中遇到异常就对照常见问题表排查很快能积累属于自己的实战经验。