LangGraph 函数字典从入门到入魂一网打尽所有常用 API如果你写过 Agent十有八九会遇到这类问题LLM 的每次调用都“没记性”多轮对话、多步工具调用、状态回滚、挂起等待人工确认……业务逻辑稍微复杂一点手写的while循环和全局变量就开始失控。LangGraph 就是冲着这个痛点来的它把 Agent 的整个过程变成一张“有状态的计算图”节点是你自定义的函数边是这些函数之间的流转规则状态则是图上大家共享的一块白板。诚然LangGraph 的官方文档质量已经不错但 API 散落在各种 Demo、教程和参考文档里新手最容易懵的是add_edge和add_conditional_edges到底怎么选StateGraph和MessageGraph差在哪中间的interrupt和checkpoint又该怎么配合这篇文章我就把自己平时整理的一份“LangGraph 常用 API 字典”完整放出来。从最基础的概念到生产环境常用的断点续跑、并发分发我会把关键参数、调用时机、容易踩的坑一次性讲清楚。无论你是刚把 LangGraph 装进环境里的新手还是已经能跑通简单 Demo 但想再上个台阶的开发者这篇文章都能帮你省下不少翻源码的时间。1. 先搞懂 LangGraph 的设计核心图、状态、节点边LangGraph 和普通编排框架最大的区别在于它把“流程”抽象成了一张图。这里的图不要求是单向的 DAG它可以有环可以有分支也可以在某条边上“挂起等待”。这种设计从一开始就不是为了简单写个顺序调用而是想让你像画状态机一样去描述复杂的对话和工作流。1.1 “有状态图引擎”到底是什么我用一个生活例子说明。你在公司里处理一份报销单先填表节点A然后等老板审批节点B审批不通过就退回修改回到节点A通过后进入财务打款节点C。整个过程天然是一个“图”而且“退回修改”这一条边让图产生了环。LangGraph 做的事情就是把这张图画出来然后由它帮你管理每一步的执行顺序、状态传递、断点恢复。在这张图里节点Node就是你写的函数比如调用一次 LLM、执行一段搜索、调用一个工具。函数收到当前状态处理后返回状态的一部分更新。边Edge连接节点的路径决定“当前节点跑完下一步去哪个节点”。状态State一个全局共享的数据结构所有节点都往里面读和写。它也是整个图的“记忆”。你不需要手动写“跑完 A 就调用 B”这种硬编码逻辑而是告诉 LangGraph 图的结构由它的运行时引擎去调度。这套设计能解决什么问题最典型的场景是 Agent 的工具调用循环LLM 决定调用工具 → 工具返回结果 → 把结果传回给 LLM → LLM 判断是否继续调用。如果直接用普通 Python 写你很快会发现代码里塞满了循环和异常处理。而用 LangGraph这个逻辑不过是一个“LLM 节点”通过条件边回到自己身上的一条环。1.2 StateGraph、Graph、MessageGraph 到底怎么选很多新手一开始就被这三个类搞晕了。简单说StateGraph最常用的通用图。你需要自定义一个状态数据结构所有节点读写这个状态。这是绝大多数场景的首选。Graph图中的低配版。它也有节点和边但状态只有一个固定的 dict且行为和StateGraph不太一致。官方现在基本主推StateGraph不建议在新项目里用Graph。MessageGraph专门针对“纯聊天消息流”的简化版图。它内部的状态就是一个消息列表任何节点的返回值会被当作新消息追加进去。单个 Agent 多轮对话这种场景非常方便但遇到需要额外字段比如计数字、标记位的场景就不够用了。我的选择原则很简单只要状态里除了消息还有别的字段就老老实实用StateGraph。MessageGraph适合快速验证一个聊天工作流但别贪图省事。还有一个高频概念是StateSchema在早期版本里定义状态用 dataclass 或 TypedDict现在更推荐直接用 TypedDict。比如from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] next_step: str这里messages字段用了add_messages作为 reducer含义是每次节点返回新的消息时自动把新消息追加到已有列表里而不是覆盖整个列表。这点非常重要后面我会专门展开讲。2. 节点与边入门必会的三个核心 API学习 LangGraph 的第一道坎就是理解add_node、add_edge、add_conditional_edges这三个 API 怎么配合。我的建议是先把一个最小闭环跑起来再逐步往上加复杂度。2.1 add_node把普通函数变成图上的节点add_node接受两个参数节点名和节点函数。节点函数有一个固定签名接收一个字典即当前状态返回一个字典即状态更新。from langgraph.graph import StateGraph, START, END def call_model(state: dict) - dict: # 这里假装调一次模型 last_message state[messages][-1] return {messages: [{role: assistant, content: f你说的是{last_message}}]} builder StateGraph(AgentState) builder.add_node(model, call_model)这里最容易忽略的是节点的返回值并不是整个状态而是“状态更新”。LangGraph 会把返回值按照状态字段合并回整体状态中。这就是为什么返回时必须用{messages: ...}这样的字典包一层而不是直接返回一串消息。如果节点写成了return state虽然一般也不会报错但理论上会把整个状态又走一遍 reducer性能上不划算还容易引发意外的字段冲突。我见过有人这么干结果每次循环状态都翻倍增长排查了半天才发现是返回了整个 state。# 错误示范返回整个状态容易造成字段重复合并 def bad_node(state): state[count] state.get(count, 0) 1 return state2.2 add_edge固定路径的连接add_edge用来连接两个节点表示固定调度路径。最典型的用法是定一个起点和终点builder.add_edge(START, model) builder.add_edge(model, END)这里START和END是两个特殊节点。START是图的入口所有流程都必须从它开始END是出口流程走到这里就结束。在只有一个节点的最小 Demo 里两行add_edge就够了。但你一旦有了三四个节点就要考虑“写完顺序逻辑后节点之间的跳转是否永远是固定的”。如果某一步有分支、有判断就走条件边。2.3 add_conditional_edges让流程自己“做决定”这是 LangGraph 里最有价值、也最容易用错的一个 API。我在实际开发里把 Agent 做得像一个会自我判断的流程靠的全是它。条件边的基本用法是def route_after_model(state: dict): if state[next_step] tool: return tool return end builder.add_conditional_edges( model, route_after_model, {tool: tool, end: END} )第一个参数是源头节点第二个参数是一个路由函数第三个参数是一个映射字典路由函数返回的字符串对应到实际要去的节点。这里有两个坑路由函数的返回值必须能被映射字典找到。如果你返回了字典里没写的键运行时会直接报错。我的建议是路由函数里明确返回一个你预先约定的常量而不是凭空造字符串。条件边是从源节点发出去的。别忘了源节点本身已经执行完了它是根据执行结果来决定下一个走向。还有一个容易被忽略的点add_conditional_edges可以不传映射那它默认路由函数返回的字符串就是目标节点名。但为了可维护性我总建议显式写出映射字典尤其当节点名很长、容易打错时映射字典相当于一层编译期检查。2.4 用条件边实现“工具调用循环”Agent 最常见的拓扑——LLM 判断要不要调用工具然后自己循环——用 LangGraph 三行代码就能表达def call_llm(state): # 喂给模型所有历史消息 resp llm.invoke(state[messages]) return {messages: [resp]} def check_tool_needed(state): last state[messages][-1] # 假设模型输出里带了工具调用标记 return tool if getattr(last, tool_calls, None) else end def call_tool(state): # 执行工具调用把结果作为消息追加回去 return {messages: [tool_result]} builder StateGraph(AgentState) builder.add_node(llm, call_llm) builder.add_node(tool, call_tool) builder.add_edge(START, llm) builder.add_conditional_edges(llm, check_tool_needed, {tool: tool, end: END}) builder.add_edge(tool, llm)注意最后一行不管工具执行成功还是失败最终都要回到llm节点让模型看到工具结果后再次决策。这就在llm和tool之间形成了一个环。只要模型不断要求调用工具流程就会一直循环一旦模型说“不需要工具了”走end分支就退出。第一次见到这个写法的人可能会担心死循环。确实如果模型每次都谎称需要工具图就会跑不完。所以我在真实项目里都会给状态里加一个tool_call_count字段在check_tool_needed里判断次数上限超了就强制走end。2.5 并行的多条边怎么理解add_edge和条件边还有一个特性同一个节点可以有多条出边。比如从A出发同时连B和CLangGraph 会并行执行B和C等它们都跑完再汇聚到下一步。这只是图论里的基本行为但在编排里意味着你可以天然实现“分发-聚合”。这种并行模式我用得最多的是一个查询来了同时并行调用多个搜索工具、多个独立分析节点最后汇总结果。用add_edge直接连多条边就能做到不需要额外引入线程池。3. 状态管理与 reducer状态的写法决定图的上限很多人把注意力放在节点函数怎么写却忽略了状态结构的定义。其实 LangGraph 的“灵魂”在状态字段的 reducer 上。状态字段本身的类型、是否用Annotated、用哪个 reducer直接决定了节点间的数据是“覆盖”还是“累加”。3.1 TypedDict 定义状态字段语义要提前想清楚我建议所有新项目都用TypedDict定义状态因为它可读性最好IDE 提示也友好。from typing import TypedDict, Annotated, Literal from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] current_step: str retries: int这里面current_step和retries没有加 reducer意味着任何节点返回这些字段时会直接覆盖旧值。覆盖语义适合放“当前阶段标记”、“最终结果”这类只保留最新值的状态。而加了add_messages的messages字段每次节点返回新消息时LangGraph 会调用add_messages函数把新消息和旧消息合并成一个更长的列表。这是 LangChain 消息模型的标准处理方式能够保证多轮对话消息按顺序累积。3.2 自定义 reducer 并不难add_messages不是唯一的 reducer。你也可以自己写一个。任何函数只要签名是(current_value, new_value) - updated_value就能用Annotated挂到字段上。举个例子我想记录所有节点跑过的路径方便调试def append_path(current: list[str], new: str) - list[str]: return current [new] class AgentState(TypedDict): messages: Annotated[list, add_messages] path: Annotated[list[str], append_path]这样只要某节点返回{path: tool_executed}它就会被自动追加到path列表里而不是覆盖。这种“累积型”字段非常适合日志、痕迹跟踪、中间步骤汇总。要注意的是reducer 函数必须能容忍current_value为初始值比如空列表。因为第一次执行时状态里可能还没有这个字段LangGraph 会传入默认值。所以 reducer 里第一行经常是if current is None: current []。3.3 覆盖 vs 累加最容易翻车的地方我帮别人排查代码时最常见的错误是定义状态时messages字段忘了加add_messages。结果节点第一次返回消息后状态里的消息列表被整体替换第二次节点返回时模型只看到最新一条消息历史全丢了。多轮对话变得像失忆了一样。反过来也常见有些标记位字段如current_step误加了累加 reducer导致值变成一个列表后面节点判断if state[current_step] tool永远不成立。经验法则列表类型、需要追加内容的字段 →Annotated[list, add_messages]或自定义追加 reducer标量类型、只需要最新值的字段 → 不加 reducer直接覆盖需要计数、去重、合并的字段 → 自定义 reducer3.4 MessageGraph 与 StateGraph 的状态差异MessageGraph现在已经不是主推但偶尔还会在老旧代码里见到。它的状态写死在源码里就是一个消息列表。节点返回值会被当作新消息追加你不能在上面存别的字段。所以一旦你需要记录“重试次数”、“用户 ID”这类辅助信息就得立刻切到StateGraph。新手调试时还有个容易蒙的细节在StateGraph里如果你定义的字段是messages: Annotated[list, add_messages]节点返回消息时最好用字典形式也就是{messages: [message]}。有些教程直接用return [message]在老版本里能跑新版本会直接报错因为返回类型不匹配。养成“永远返回字典”的习惯兼容性最好。4. 持久化、中断和人工介入上生产环境的核心 APILangGraph 和其他编排框架拉开差距的地方在于它能“记住执行到哪一步”并能“停下来等人”。这两点正是 Agent 进入生产环境后的刚需。4.1 checkpoint让整张图拥有记忆默认情况下LangGraph 的图执行是一次性的跑完就没了。要让它记住每一步的状态得配置一个checkpointer。它相当于给图的每次执行写一份“存档”记录所有节点的输入输出和状态变化。from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() graph builder.compile(checkpointercheckpointer)MemorySaver是把存档放内存里适合开发和测试。生产环境一般用持久化存储比如SqliteSaver或PostgresSaverfrom langgraph.checkpoint.sqlite import SqliteSaver with SqliteSaver.from_conn_string(checkpoints.sqlite) as checkpointer: graph builder.compile(checkpointercheckpointer) # 执行图...这里有个概念叫thread_id。它表示一条完整的会话。打个比方一张图是一个流程模板thread_id是具体的一笔业务单号。同一个模板不同thread_id有各自独立的执行历史和状态。config {configurable: {thread_id: order-1001}} result graph.invoke({messages: [我想查物流]}, configconfig)之后只要用同一个thread_id再次调用图会从上次中断的地方继续而不是从头开始。这在多轮 Agent 对话里是决定性的能力。4.2 interrupt让图停在半路等人回复interrupt是 LangGraph 里名字最直白的函数之一执行到某个节点时直接把整个图挂起并抛出一个特殊异常把控制权交回给你的代码。from langgraph.types import interrupt def approval_node(state): # 挂起并返回需要人工确认的信息 decision interrupt({question: 是否批准这笔交易, amount: state[amount]}) state[approved] decision return state当图执行到interrupt时你不会拿到最终结果而是拿到一个中断标记。你的程序可以停下来等待人工操作或者把请求发给外部系统。之后你可以带着人工的答案用同一个thread_id继续执行graph.invoke(None, config{configurable: {thread_id: order-1001, input: {decision: True}}})注意中断和普通异常不一样。它不算是“失败”而是设计好的暂停点。人工审批、条件确认、等待用户补充信息都用它来做。4.3 update_state手动改写历史状态有时候你还想做“编辑已发生的事”。比如人工审批时发现某个步骤的输入错了需要修正历史状态再继续走此时update_state就有用了。它允许你直接修改某次执行的指定字段然后从修改后的状态继续往下跑。from langgraph.graph import StateGraph graph.update_state(config, {messages: [修正后的消息]}, as_nodetool)这里的as_node参数特别关键它表示“这次修改相当于某个节点返回的结果”。这样 LangGraph 的 checkpoint 记录里会留有痕迹时间旅行视图依然清晰。4.4 时间旅行与状态回放LangGraph 允许多次从历史断点恢复这就是社区里常说的“时间旅行”。核心 API 是get_state_historystates graph.get_state_history(config) for state in states: print(state.values, state.next)get_state_history会返回该thread_id下所有历史存档。你可以找到某个旧的断点用一个新配置从那里继续跑就相当于回到过去重写某段流程。这在 Agent 调试里非常有用。比如一次执行中工具调用返回了错误数据导致后续结果全错。你不需要重新跑一遍全流程而是定位到出错节点之前的历史断点改写状态后继续执行。整个链路既保留了错误现场又能快速重试。4.5 生产场景人工确认和断点续跑组合我自己的一个生产项目里有一个“邮件自动回复助手”的流程读邮件 → 生成回复草稿 → 等待人工审批 → 修改最终版本 → 发信。过去我用传统代码写手动构建了各种“草稿状态表”和“审批状态表”代码极其脆弱。用 LangGraph 后思路变得很干净读邮件节点从邮箱取邮件。生成草稿节点调用 LLM 生成回复。人工审批节点调用interrupt挂起等待。外部系统收到 interrupt 后把草稿推给人工等待人工反馈。人工确认后用同一个thread_id恢复执行进入发信节点。整个过程每一步的执行记录都被 checkpoint 保存下来出了任何问题都能用get_state_history回溯。这在以前是很难实现的至少不会只靠一个框架就能全做完。5. Send、流式输出与并发进阶但高频的实用 API当你的 Agent 不只是“跑一个流程”而是需要在多个子任务之间做编排时会用上几个并发相关的 API。这里最能快速提升工程效率的是Send和流式输出。5.1 Send动态分发子任务Send在 LangGraph 里是用来做动态并行的。它的使用场景是一个节点执行后发现要把任务分给 N 个子任务并行处理每个子任务有独立的参数和状态。它的设计思路是 Map-Reduce先在一个节点里动态生成任务列表然后用Send把每个子任务发给同一个目标节点或不同目标节点。from langgraph.constants import Send from langgraph.types import Command def plan_task(state): # 假设提取了三个子任务 tasks [summarize, translate, keywords] return Command( goto[Send(worker, {task: t}) for t in tasks] ) builder.add_node(planner, plan_task) builder.add_node(worker, worker_func)这里Send是“发一个任务包”的意思。每次发送都会在目标节点启动一个独立执行分支。所有分支跑完后图会等它们全部完成再继续往后走。这种模式在批量数据处理、多工具并行查询、多角色并行讨论里太实用了。比如用户同时请求“总结文档、翻译段落、提取关键词”你不需要写一堆手动asyncio.gather只需要根据任务数量动态Send。注意Send发的任务越多并发开销越大而且 checkpoint 存档也会变大。单个图内一次分发太多任务可能把执行引擎压垮。我这里一个经验值是 10 个以内轻松50 个以上就要考虑分批或者限流。5.2 stream/astream边跑边看而不是等结局很多时候Agent 的一次调用要好几秒甚至更久。如果让用户盯着一张空白的“加载中”页面体验极差。所以 LangGraph 提供了流式输出stream支持按节点逐步返回结果。for event in graph.stream({messages: [讲个故事]}, configconfig): print(event)stream默认按节点输出你会在事件流里看到“哪个节点在跑输出了什么”。对于聊天效果可以用stream_modemessages拿到 token 粒度的输出for event in graph.stream({...}, configconfig, stream_modemessages): if event[0] and token in event[0]: print(event[0][token], end)stream_mode有几个常用值updates只输出状态更新比默认模式更精简。messages按消息/token 级输出适合聊天流。values每次节点后输出完整状态适合调试。也可以传字符串列表同时开启多种模式。异步场景用astream接口几乎一致。对于 FastAPI 应用astream配合text/event-stream就能快速实现一个 SSE 接口让前端边生成边显示回复。5.3 并发节点与共享状态的安全问题LangGraph 允许多分支并行执行但要注意多分支共享同一份状态时如果两个节点都修改同一个字段而该字段没有合适的 reducer可能后写覆盖先写。所以并行场景下共享字段要么设计成累积型 reducer要么让每个分支只写自己专属的字段。碰到“并行分支都要写同一个字段”的需求我通常的做法是给每个分支一个专属字段最后用一个汇总节点把这些字段合并。比如每个 worker 写result_i最后reduce_node统一汇总到final_result。这样分支之间互相隔离不容易产生状态冲突。另外checkpointer 在并行分支下的存档粒度会变细。如果你开了stream又跑并行分支事件流顺序可能和你直觉不一致因为不同分支交替产生事件。调试时最好用thread_id固定住会话再逐步观察事件序列。6. 常见问题与排查技巧实录最后这部分我把自己几年里踩过的坑和帮别人排查过的代码问题整理成一份速查笔记。遇到问题先翻这里比翻源码快很多。6.1 常用 API 速查表API/概念用途常见误区StateGraph(state_schema)创建有状态图忘记定义 schema直接传 dictadd_node(name, func)注册节点函数返回了完整状态而非状态更新add_edge(start, end)固定流转路径在需要分支的地方误用固定边add_conditional_edges(source, router, mapping)条件路由路由函数返回值不在映射字典里START/END图的入口/出口忘记连到 START图直接不执行Annotated[field, reducer]状态字段合并规则列表字段不加 reducer导致覆盖add_messages消息累加 reducer非消息字段也加了它值变 listcommand里的goto/Send动态跳转/并发分发把 Send 当普通返回值用interrupt挂起等待人工输入误以为它是异常去捕获处理checkpointer持久化执行快照没配thread_id无法续跑get_state_history历史断点/时间旅行试图修改返回值而不是用update_statestream/astream流式输出在 async 环境误用stream6.2 高发问题与排查思路问题一节点函数写了但图不执行。多半是忘了builder.add_edge(START, ...)。没连到START的节点等同于孤岛永远跑不到。检查图只需要做一件事从START出发按边遍历确保所有节点都可达。如果你的图有专门多分支用get_graph().draw_ascii()打印出来看一眼就能定位孤立节点。问题二多轮对话消息一直不累加。老规矩先看messages字段有没有Annotated[list, add_messages]。没有就加上。如果加了还是被覆盖检查节点返回时是不是把整个state[messages]原样返回了一份reducer 接收新旧两份时可能出现拼接重复。我的建议是节点永远只返回“新增的那部分”而不是返回“全部历史”否则虽然结果碰巧正确但每次都在做全量复制性能会逐步劣化。问题三条件边一直走同一个分支。检查路由函数看它依赖的字段在状态里到底存的是最新值还是旧值。很多人在条件边里直接读state[messages][-1]但该节点上一步跑完后消息列表里最后一层可能是工具返回的消息而非模型决策消息于是路由永远判断为“继续调工具”。这时候先打印state结构再调整字段依赖比盲改代码有效率得多。问题四checkpoint 配置了但第二次调用还是从头跑。两个原因最常见一是没传thread_id二是每次调用都重新编译图。记住thread_id相当于“业务单据号”同一个图 同一个thread_id 续跑换了thread_id 新会话。如果你在接口里每次都builder.compile()之前存的 checkpoint 也等于作废。把编译好的图复用进去不要每次调用都重建。问题五interrupt 之后 resume状态不对。interrupt的返回值是“外部恢复时传进来的值”。我的经验是interrupt最好只传一个参数进去一个 dict统一结构避免传到多个参数时漏传导致缺参。恢复时用Command给同一个线程传值from langgraph.types import Command graph.invoke( Command(resume{decision: approve}), config{configurable: {thread_id: order-1001}} )问题六自定义 reducer 序列化 / 类型损坏。如果你自定义了 reducer而字段里出现了自定义对象或复杂嵌套checkpoint 保存时可能失败。遇到“保存时 JSON 序列化失败”之类的报错先把状态里的复杂对象转换为可序列化的基本类型str、dict、list或者自定义序列化逻辑。我们项目里约定状态里不直接放任意 object只放可序列化的数据LLM 相关对象都放在节点内部临时变量里。6.3 几条避坑心得状态结构的设计先画图再写代码。我每次开工前会先想清楚哪些字段是覆盖的哪些字段是累加的。这个决定一下后面节点函数的返回内容就非常明确几乎不会出状态冲突。尽量让节点函数保持“纯函数”风格。不要在里面改全局变量不要偷偷塞外部副作用。节点只依赖入参 state只通过返回值影响后续流程。这样你才能利用 checkpoint 做回放、调试、时间旅行。日志里务必带上 thread_id。LangGraph 的运行时事件里没有 thread_id 会让人抓狂尤其是多用户并发时你根本分不清一条日志属于哪个会话。我在每次 invoke 的 config 里固定带上thread_id和user_id日志查起来直接串成一条线。用get_graph().draw_ascii()快速检查图结构。这个 API 能把图打印成 ASCII 流程图比在文档里脑补结构可靠得多。复杂图里多节点死循环、漏连边一打印就现形。升级版本时留意 API 变化。LangGraph 更新比较快不少 API 的签名和推荐用法会变。网上老代码大概率跑不通升级前先看 changelog别盲抄。最后说两句我自己的使用习惯是新项目不管多简单先用StateGraph写一个最小闭环——一个节点、两条边——然后立刻加上MemorySaver和thread_id。原因很简单LangGraph 真正厉害的不是“跑通一次”而是“跑一半还能接着跑”这个习惯能让你从第一天就体会到它的核心能力而不是停留在 Demo 阶段。如果这篇文章能让你少走点弯路那比什么都值。LangGraph 的学习曲线不算陡但这些 API 的组合方式确实需要大量实战才能摸出感觉。我的建议就一条别只读文档动手搭一个带中断、带并发的完整流程你会比看十篇教程都更懂它。