★ FEATURED ARTICLE 我心里一直想给 LangGraph 维护一份“函数字典”不是官方文档那种干巴巴的 API 列表而是把日常开发里真正高频、真正容易踩坑的函数按使用场景重新组织每个函数配一个能直接跑通的最小示例再说明它背后的设计逻辑。这篇文章就是我整理之后的结果从最基础的StateGraph构建一路到interrupt、Command、Send这些高阶玩法适合两类人看一类是刚接触 LangGraph想找个清晰入口的入门者另一类是已经写过几张图但总感觉某些 API 用法模棱两可、出问题得靠试错的老手。我尽量让每个章节都能独立阅读你完全可以把它当作一本随手翻的手册来用。1. 为什么 LangGraph 值得一份“函数字典”核心执行模型的底层逻辑1.1 状态图不是又一个编排框架而是对控制流的一次“去回调化”你如果写过基于聊天模型的 Agent应该会有这种体验一开始代码很爽就是一个循环把上次的回复塞回模型判断有没有工具调用有就执行、没有就结束。但当需求变成“多轮工具调用”、“人工审核后才能提交订单”、“不同分支召回不同知识库”之后代码就开始失控。到处是if、while、状态标志位甚至要自己维护一套历史消息的合并逻辑。我经常管这种写法叫“面条式编排”因为它把每一步的跳转关系藏在代码深处读代码的人根本无法一眼看出流程长什么样。LangGraph 的做法是把控制流显式地画成一张图。图的节点是处理函数图的状态是全局共享的数据容器图的边决定控制流方向。好处非常直接流程本身成了一个数据结构你可以查看它、持久化它、甚至在运行中间人工干预它。所以 LangGraph 的 API 看起来很多本质上都在回答三个问题状态怎么定义、节点怎么组织和改造状态、边怎么决定下一步去哪。1.2 我建议的入门路径先记住三个概念再去碰 API 细节很多初学者一上来就翻StateGraph的源码结果被Annotated、Reducer、Checkpointer这些概念搞得晕头转向。我的建议是先建立一张心智地图语言模型应用的状态图就是一条“流水线”状态是流水线上流动的料箱节点是工位边是工位之间的传送带。工位可以往料箱里加料更新状态也可以根据料箱里的当前内容决定把料箱送到后面哪个工位条件边。这里有一个很重要的认知LangGraph 的节点并不“返回结果给调用方”而是“返回对状态的部分更新”。更新会按照状态字段上挂载的规约函数Reducer合并到全局状态里。如果你的字段没有挂任何 Reducer默认行为就是整值覆盖这是新手最容易遇到“消息丢了”的根源。理解了这三件套你就可以开始用StateGraph了后面的 API 基本上都是围绕这三件套的配套工具。2. 状态、节点、边构建期三大件 API 全拆解2.1 状态定义TypedDict 不是摆设Reducer 才是关键LangGraph 最常用的状态定义方式是 Python 的TypedDict因为它既是运行时的字典又能让编辑器和类型检查器帮你排查字段名错误。比如一个客服机器人的状态可以长这样from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class State(TypedDict): messages: Annotated[list, add_messages] user_input: str order_id: str | None need_tool: bool这里最值得解释的是Annotated[list, add_messages]。普通的messages字段如果只是list类型每次节点返回{messages: [...]}新列表会直接把旧列表覆盖掉多轮对话就只剩下最后一轮的几条消息。add_messages是一个预置好的 Reducer它知道如何把新消息追加到旧列表里同时还会做去重比如包含相同 ID 的消息会替换而不是重复追加。如果你需要自定义合并逻辑也可以自己写一个函数例如把两个字典深度合并、或者对数字累加from typing import Annotated def total_reducer(prev: int, new: int) - int: return prev new class StatsState(TypedDict): total_count: Annotated[int, total_reducer]我的建议是凡是“多轮累加型”的数据比如消息历史、中间结果列表、累计分数都尽量配上 Reducer凡是“单值快照型”的数据比如当前订单号、用户提问原文直接用普通字段就行不需要过度设计。2.2 add_node 的多种传参方式与节点函数返回规范构建一张图的入口是StateGraph(State)接下来就是往图里塞节点。add_node最常规的用法是传入节点名和处理函数from langgraph.graph import StateGraph, START, END def analyze(state: State) - dict: # 这里可以调用大模型、查数据库、跑计算 return {need_tool: True, messages: [(assistant, 我需要查询一下订单状态)]} builder StateGraph(State) builder.add_node(analyze, analyze)节点函数的签名有一个硬性要求入参一定是一个State类型的完整状态字典返回值必须是“状态字段的部分更新”也就是一个字典里面只能包含状态里定义过的键。如果你试图返回{unknown_key: 1}运行时就会被pydantic校验拦下来。这个设计其实非常合理它保证了图里每个节点都是纯函数式的数据变换方便调试、方便回放。add_node还支持一种更偷懒的写法如果你希望节点名直接等于函数名可以直接用装饰器风格builder.add_node def analyze(state: State) - dict: return {need_tool: True} # 等价于 builder.add_node(analyze, analyze)我个人在实际项目中更推荐显式写名字因为图一旦大了以后节点名往往带有业务语义比如check_human_review、call_search_tool比 Python 函数名更接近流程图上的标签读代码的人不用去猜。2.3 add_edge 与 set_entry_point/set_finish_point别把顺序搞反图只有节点没有边是运行不起来的。边就是控制流本身LangGraph 里最基础的是静态边和入口/出口设置。现代写法推荐直接用START和END哨兵替代早期版本里的set_entry_point和set_finish_pointbuilder.add_edge(START, analyze) builder.add_edge(analyze, answer) builder.add_edge(answer, END)这里有个初学者容易踩的顺序问题如果你既用了add_edge(START, ...)又调用了set_entry_point编译器会提示重复指定入口。我的建议是统一用START/END这套新 API因为它更直观而且入口出口本质上也是两条边概念上是统一的。add_edge本身只负责静态连接它没有条件判断能力。条件判断要交给下一章讲的条件边函数。另外补充一点节点并不需要全部连通没连到END的图会按死路处理编译时也会给出警告运行时如果走到了死路会直接报错。所以设计图的时候最好在纸上先把流程走一遍再把边一条条加进去比在代码里反复改要快得多。3. 条件路由的正确用法add_conditional_edges 与路由函数3.1 真实案例一个带工具判断的多步路由条件边是 LangGraph 最有价值、也最容易犯错的 API。它解决的问题是在执行完某个节点之后下一步不能静态写死而是要根据当前状态决定跳到哪个节点。最常见的场景是判断要不要调用工具。我以客服系统为例流程是先分析用户意图如果消息里包含“订单号查询”“物流跟踪”这类需要外部系统的意图就进工具节点否则直接用模型回答。def analyze(state: State) - dict: # 假设这里得到一个布尔值表示是否需要工具 return {need_tool: True} def route_branch(state: State) - str: if state.get(need_tool): return call_tools return direct_answer builder.add_node(analyze, analyze) builder.add_node(call_tools, call_tools_node) builder.add_node(direct_answer, answer_node) builder.add_conditional_edges( analyze, route_branch, { call_tools: call_tools, direct_answer: direct_answer, }, )路由函数接收当前完整状态返回一个字符串键。这个键必须是映射字典里存在的 key否则运行时直接抛 KeyError 类的错误。映射字典的值才是真实的目标节点名同时也支持指向END。同样的目标节点名也可以从路由函数返回然后省略映射字典但我强烈建议保留映射字典。因为路由函数是纯逻辑映射字典是拓扑关系分开写以后即使未来你想把逻辑上的分支名改成更友好的名字也不用改函数内部。3.2 路由函数的三种写法与返回规范我把路由函数的常见形态归纳为三种你按场景选就行。第一种是最简单的字典取值写法。比如三个分支优先级不同逻辑写在一个函数里返回目标分支名。第二种是封装成闭包提前注入外部依赖。比如某开发者需要在不同环境切换路由策略路由函数可以读取config里的配置后再决定返回哪个分支只依赖状态本身是不够的。第三种是用预置路由函数。LangGraph 的prebuilt模块给了个tools_condition它会检查状态里的最后一条消息是不是带工具调用的 AI 消息是就返回tools否则返回END。这个函数在配合ToolNode时特别省事后面第 6 章会展开。这里有一个我特别想强调的避坑点条件边的映射 dict 里千万不要出现END之外的“不存在节点”。比如你写{call_tools: tools_node}但图里根本不存在名为tools_node的节点编译不报错运行时一走到就崩。排查办法是在画图阶段就对每个条件边做一次静态检查或者干脆维护一个节点名常量集合避免手写字符串出错。4. 编译执行期invoke / stream / config 这组 API 怎么选4.1 invoke 与 ainvoke同步 vs 异步的适用场景图构建完成之后调用builder.compile()得到可执行图对象。最基本的执行方式是invokegraph builder.compile() result graph.invoke({ messages: [(human, 我的订单什么时候到)], user_input: 我的订单什么时候到 }) print(result[messages][-1].content)invoke会从START一路执行到END返回最终状态字典。注意它返回的不是单个节点的结果而是整张图跑完后的完整状态。如果你只想要最后一条消息就得自己从messages里取。异步执行的ainvoke只是把同步调用换成了可等待对象在 FastAPI 接口、爬虫这类 IO 密集场景里非常有用result await graph.ainvoke(initial_input)为什么会有两套因为在 LLM 场景里一个图内部往往要多次调用模型、工具和数据库同步执行会长时间占住线程而异步可以并跑多个会话。但如果你只是本地脚本做实验同步invoke更直接调试也不容易晕。4.2 stream / astreamstream_mode 四种模式对比invoke最大的问题是黑盒你只知道最终状态看不到每步发生了什么。调试复杂图时我几乎都是用stream。它的核心参数是stream_mode我按推荐程度排个表模式输出内容典型用途values每个超步执行后的完整状态快照调试状态变化最接近“逐步展示”updates每个超步中发生变化的节点名和节点返回值的字典只想看“哪一步变了”省 token 和数据量messages逐 token 流式输出产生的消息增量以及消息元数据面向用户的打字机效果custom节点内部通过流写入器主动发出来的自定义事件透出业务中间态比如进度条百分比for chunk in graph.stream(initial_input, stream_modevalues): # chunk 是每个超步之后的完整 state可以直接打印 message if messages in chunk and chunk[messages]: print(--- step ---) print(chunk[messages][-1].content[-100:])四种模式的实际差异在第一眼不容易看出来。我举一个例子在values模式里每次 LLM 输出一条消息你都能看到整个 messages 列表在增长在updates模式里你看到的只是{chat_node: {messages: [新消息]}}要组装完整会话就得自己维护上下文。所以如果你的下游是“展示最终对话”values最省心如果下游是“增量式渲染”updates数据量更小。astream是异步版本配合stream_modemessages可以做真正的流式聊天响应体验比一次性等结果好一个档次。唯一要注意的是这会造成更多网络请求和开销不适合纯离线批处理。4.3 RunnableConfigthread_id、recursion_limit 这些配置项的含义执行时传的config字典新手常常直接忽略但它其实是 LangGraph 的记忆和边界控制工具。最常用的是configurable下的thread_idconfig {configurable: {thread_id: conversation-001}} graph.invoke(initial_input, configconfig)thread_id是状态持久化时的会话标识相当于给每次对话一个独立的“储物柜”。没有它哪怕你挂了检查点所有会话也会混在一起。还有一个翻车率很高的配置项是recursion_limit它限制图最多能执行多少步默认值是 25。如果你写了一个带循环的 Agent工具反复调用超过 25 步就会抛GraphRecursionError。这不是 bug而是保护机制防止死循环把资源吃光。实际使用中我会根据业务复杂度主动调大config { configurable: {thread_id: conversation-002}, recursion_limit: 50, }调用invoke、stream都需要把config传进去尤其是用检查点恢复历史会话时这个参数不能省。5. 持久化与断点Checkpointer 系 API 从入门到生产5.1 MemorySaver单机调试最舒服的兜底方案MemorySaver是 LangGraph 自带的本地内存检查点作用是让一张图在执行中途“记住”自己的状态下次继续同一个thread_id时能接着跑。用法很简单from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: session-1}} graph.invoke({user_input: 你好}, configconfig) # 同一会话继续 graph.invoke({user_input: 帮我查订单}, configconfig)为什么需要这玩意儿因为 LangGraph 的节点本来就是纯函数式的理论上每个节点跑完就结束了。但真实业务需要跨步骤记忆比如多轮对话里第二轮的模型调用必须能看到第一轮的历史消息。MemorySaver提供了一个最简单的“内存数据库”让图的运行状态在同一个 Python 进程内被保存下来。它的优点是零配置、速度极快非常适合开发调试和单机小工具。代价是进程一重启所有会话记录全部消失而且不支持跨进程共享。所以千万别在正式的线上服务里只用MemorySaver。5.2 SqliteSaver / PostgresSaver跨会话恢复的正确打开方式要真正做到持久化项目里最常用的两个选择是SqliteSaver和PostgresSaver。SqliteSaver适合单机部署、不需要高并发的场景一个文件就能存下所有会话状态from langgraph.checkpoint.sqlite import SqliteSaver with SqliteSaver.from_conn_string(checkpoints.sqlite) as checkpointer: graph builder.compile(checkpointercheckpointer) graph.invoke({user_input: 你好}, config{configurable: {thread_id: a}})PostgresSaver则适合多副本、需要跨进程共享状态的生产环境。它可以用连接串初始化配合数据库连接池多个后端实例共享同一份会话历史。在选择上我一般遵循“先本地文件、后集中数据库”的原则。单机爬虫、本地批处理用 SQLite 最稳Web 服务、多实例部署直接上 Postgres。检查点本质上就是把每次超步执行前后的状态差异存下来所以数据库的写入频率不低生产环境要给表做好索引控制好连接池大小。5.3 get_state / update_state / get_state_history手动干预的完整工具链挂上检查点后图形对象会多出三个非常实用的“状态操作” API。get_state可以查看当前会话的最新状态快照snapshot graph.get_state(config) print(snapshot.values) print(snapshot.next) # 下一个将要执行的节点snapshot.next这个字段常被忽略但它实际上是判断“图现在卡在哪一步”的金钥匙。如果一张图执行到某个条件边下一步不知道往哪拐snapshot.next会告诉你候选节点是哪些。get_state_history返回同一线程的所有历史状态版本滚动回溯时特别好用for state in graph.get_state_history(config): print(state.config, state.values)update_state则允许你手工把某些字段改掉常见场景是人工改错数据后再重新触发图执行graph.update_state(config, {order_id: X-2024-01})这三件套结合起来你就拥有了完整的时间旅行能力查当前状态、回溯历史、手动覆盖。对于审核类业务来说这套机制比在模型对话里塞一堆“请忽略之前的指令”靠谱得多。6. 让图具有“人味”interrupt / Command / Send / ToolNode 高阶组合6.1 interrupt 把图停在中间等用户确认很多真实业务里图不能从头到尾闷头跑完。比如客服机器人准备提交退款申请必须等用户点击“确认”才执行。interrupt就是为这个场景设计的 API。from langgraph.types import interrupt, Command def request_confirmation(state: State) - Command: decision interrupt({ type: confirm, message: 是否确认退款, payload: {order_id: state[order_id]}, }) if decision yes: return Command(gotoexecute_refund) return Command(gotoEND)它的执行机制很有意思图跑到interrupt那一行时不会报错而是把请求内容作为中断信号抛出来整个图暂停在“等待恢复”的状态。之后你可以通过Command(resume...)把一个值送回中断点图会从上次中断的地方继续执行。这段逻辑有一个坑下面第 7 章会详细讲那就是恢复时传的数据最终会成为interrupt的返回值。所以中断请求的 payload 和恢复值的类型一定要在结构设计阶段就定好。6.2 Command 才是节点返回的“万能遥控器”上一章里节点返回值只能是一个普通字典。但加上Command之后节点一次性就能干三件事更新状态、指定下一步走向、触发中断或恢复from langgraph.types import Command def smart_node(state: State) - Command: return Command( update{last_step: smart_node}, gotopost_process, )你可以把Command和条件边结合使用在节点内部直接决定路由而不需要单独写一个条件边函数。对比一下就会发现条件边适合“统一从同一个节点分叉”的场景Command适合“每个节点自己决定下一步”的场景。前者的路由逻辑集中可视化时更像流程图后者的路由分散在节点里代码更灵活。另外Command还支持resume参数这是恢复interrupt中断的核心机制。你可以从外部拿到一个值后通过graph.invoke(Command(resume...)直接续跑这套模式实现了“外部人工输入”和“图内节点执行”的闭环。6.3 Send 并行派发子任务现在的图执行默认是单线程串行。如果需要同时处理多个独立子任务比如同时检索多份文档、对多个城市单独计算报价Send就是标准答案。它不是一条普通边而是一次性给多个节点副本发消息。from langgraph.types import Send def dispatch(state: State) - list[Send]: return [ Send(process_city, {city: city, products: state[products]}) for city in state[cities] ] builder.add_node(dispatch, dispatch) builder.add_node(process_city, process_city) builder.add_edge(START, dispatch) builder.add_conditional_edges(dispatch, route_after_dispatch, {done: END})理解Send的关键是你是不是真的需要“并行”我们搭的这套框架本身并不要求一定并发它只是把多个输入一次性分发给同一个节点的多个实例。运行层面是否真正并发取决于执行引擎和资源的调度。所以在设计时你应该把关注点放在“把独立子任务拆开”上而不是直接幻想性能翻倍。拆分合理吞吐量往往会改善但如果你是 CPU 密集的同一份数据强行用Send反而适得其反。6.4 ToolNode工具调用的标准化节点前面讲了这么多手写逻辑其实 LangGraph 里有个专门处理工具调用的现成节点ToolNode。它的作用是读取当前消息列表如果最后一条 AI 消息里包含工具调用就自动执行对应工具并把执行结果按消息格式塞回状态里。from langgraph.prebuilt import ToolNode, tools_condition from langchain_core.tools import tool tool def search_order(order_id: str) - str: 根据订单号查询物流状态 return f订单 {order_id} 的物流状态已发货 tools [search_order] builder.add_node(assistant, call_llm) builder.add_node(tools, ToolNode(tools)) builder.add_conditional_edges(assistant, tools_condition) builder.add_edge(tools, assistant)tools_condition是预置的条件路由函数判断要不要进入工具节点返回tools或END。这套组合的好处是你不需要自己写解析工具调用参数的代码也不用手动接回模型。我实际项目中大概八成工具场景都能直接套用这个模板只有遇到“工具结果要经过人工审核再喂给模型”这种特殊需求时才会换回手动条件边。7. 实测中最容易踩的五个 API 坑7.1 忘了给 TypedDict 字段配 Reducer消息直接被覆盖这是出现频率最高的问题。状态里定义了messages但字段写成了普通list没用Annotated包一层add_messages。结果节点每次返回新消息旧消息就凭空消失。排查方法很简单打印两次invoke之间的消息长度如果第二步 messages 只剩一条基本就是 Reducer 没配对。7.2 add_conditional_edges 的映射字典 key 拼错图直接报错条件边的映射字典 key 是路由函数的字符串返回值不是节点名。很多人会在映射字典里直接写目标节点名却发现路由函数返回call_tool兜不住。我的对策是把所有分支名定义成常量ROUTE_TOOL route_tool ROUTE_ANSWER route_answer字符串常量放在同一个地方写的人不容易拼错读的人更能理解。7.3 interrupt 返回值没处理好恢复执行后状态丢失interrupt返回的值来自Command(resume...)也就是外部传入的数据。如果你在恢复时传的是{action: approve}那么节点里拿到的就是整个字典。常见的错误是把resume和interrupt的请求结构搞混导致节点拿到的是一个嵌套对象后面访问字段时抛AttributeError。我的经验是中断请求里只放展示信息恢复值里只放最小必要数据并给恢复值定义一个明确的类型别用裸字典。7.4 用了 MemorySaver 就以为万事大吉生产挂掉不知所以线上跑了一阵子突然所有会话记录都丢了排查半天发现进程重启了。这不是框架的问题是我自己拿MemorySaver当持久化了。生产环境至少要用PostgresSaver并把thread_id和业务侧的会话 ID 一一对应。记住检查点不等同于业务数据库它只是让图能恢复状态的中间产物业务需要长期保存的数据最终还是要落在你自己的存储里。7.5 stream 模式下想拿到最终答案却拿了一堆中间状态我见过不少人在stream(stream_modeupdates)里反复打印每个节点的输出最后要拼装最终答案时又拼不回来。如果你只想拿最终结果直接用invoke最简单如果你想给用户展示打字机效果用stream_modemessages并只处理与消息有关的增量只有当你确实需要调试“每步状态长什么样”时才用values和updates。这三种诉求别混着选。在把这些坑都踩过一遍之后我反而觉得 LangGraph 的 API 设计整体上是很自洽的。状态、节点、边、检查点、中断恢复每个概念都对应一个明确要解决的问题。你只要带着“状态是怎么流动的”这个视角去查函数大部分用法都能自然推导出来。希望这份函数字典能帮你少走几个我走过的弯路让你从入门直接跳到入魂那一步。
阅读完成 · 觉得有帮助？
