LangGraph 的线程持久性 - CheckPointer 检查点

线程持久化许多AI应用程序需要记忆来在单个对话“线程”中的多次交互中共享上下文。在LangGraph中这种对话级别的记忆可以通过使用检查点Checkpointers添加到任何图上。只需使用兼容的检查点编译图即可。下面是一个使用简单的内存中“MemorySaver”的示例import{MemorySaver}fromlangchain/langgraph;constcheckpointernewMemorySaver();constgraphworkflow.compile({checkpointer});状态 状态合并策略import{Annotation}fromlangchain/langgraph;import{BaseMessage}fromlangchain/core/messages;// 定义整张图的状态结构 GraphStateconstGraphStateAnnotation.Root({messages:AnnotationBaseMessage[]({reducer:(x,y)x.concat(y),}),});Annotation.Root定义 LangGraph 图的状态 Schema声明状态里有哪些字段、类型、怎么合并更新这里状态只有一个字段messages类型是消息数组LLM 对话消息HumanMessage/AIMessage 都属于BaseMessagereducer 归约函数reducer:(x,y)x.concat(y)LangGraph 规则x当前状态里该字段的旧值已有值y节点本次返回的增量更新值reducer 返回值 这个字段的新状态x.concat(y)旧数组后面追加新数组返回合并后的新数组等价于[...x, ...y]业务含义节点执行完成不要直接覆盖 messages而是追加消息旧messages: [用户提问]节点返回增量{ messages: [AI回答] }reducer 执行[用户提问].concat([AI回答])新状态messages: [用户提问, AI回答]如果不写 reducer默认行为是直接覆盖旧数组直接被 y 替换历史对话消息直接丢失这是对话 Agent 绝对不能接受的// 当前状态xconstx[newHumanMessage(你好)];// 节点返回的更新yconsty[newAIMessage(你好我是智能体)];constnewValx.concat(y);// newVal [HumanMsg, AiMsg]设置工具创建一个占位符搜索引擎import{tool}fromlangchain/core/tools;import{z}fromzod;constsearchTooltool(async({}:{query:string}){// This is a placeholder for the actual implementationreturnCold, with a low of 13 ℃;},{name:search,description:Use to surf the web, fetch current information, check the weather, and retrieve other information.,schema:z.object({query:z.string().describe(The query to use in your search.),}),});awaitsearchTool.invoke({query:Whats the weather like?});consttools[searchTool];现在可以将这些工具包装在一个简单的 ToolNode 中。每当 LLM 调用这些工具函数时此对象将实际运行它们import{ToolNode}fromlangchain/langgraph/prebuilt;consttoolNodenewToolNode(tools);设置模型加载聊天模型它应该能与消息配合使用。将所有代理状态都以消息的形式表示因此它需要能够很好地与消息配合使用什么是消息BaseMessageLangChain 里的消息就是对话单元HumanMessage用户说的话AIMessage大模型输出可以是文字也可以带工具调用ToolMessage工具执行完返回的结果Agent 的全部状态就存成一个消息数组messages[]整个智能体的记忆、历史对话、工具返回结果全部塞进这个消息列表不用单独定义一堆零散字段所以聊天模型必须支持接收消息数组作为输入而不是只接收纯字符串普通文本模型model.invoke(一句话字符串)聊天模型 (ChatModel)model.invoke([msg1, msg2, msg3])这就是「和消息配合」它应该能与工具调用配合使用这意味着它可以在响应中返回函数参数支持工具调用的 ChatModelgpt-4o 这类可以输出结构化指令我要调用某个工具并且带上这个工具需要的参数不是自然语言是机器能识别的tool_calls结构类似{ tool_calls: [ { name: search_weather, args: {city: 北京} } ] }意思别把这段当成文字回答去执行 search_weather 函数入参 city 北京这个就是在响应中返回函数参数代码加载注意这些模型要求并非 LangGraph 的通用要求——它们仅是此示例的要求import{ChatOpenAI}fromlangchain/openai;// 基础对话大模型只会正常输出自然语言文本constmodelnewChatOpenAI({model:gpt-4o});完成此操作后应确保模型知道它可以使用这些工具。可以通过调用bindTools来完成此操作constboundModelmodel.bindTools(tools);定义图现在可以把它们放在一起。将首先在没有检查点的情况下运行它这是 LangGraph 最经典ReAct 工具调用智能体流程图代码反复思考 调用工具就是ReAct流程START → agent调用大模型 → 路由判断如果模型输出没有工具调用→ 走到 END结束如果模型输出有 tool_calls→ 走到 tools 节点执行工具执行完自动回到 agent循环// - StateGraph用来构建状态图import{END,START,StateGraph}fromlangchain/langgraph;import{AIMessage}fromlangchain/core/messages;// - RunnableConfig运行配置里面包含 checkpoint 持久化、线程 id 等import{RunnableConfig}fromlangchain/core/runnables;// 1. 拿到当前状态里的messages消息数组取最后一条消息就是刚大模型返回的 AIMessage// 2. 判断这条消息有没有 tool_calls// - tool_calls为空 / 不存在 → 返回END整张图结束回答用户// - 存在 tool_calls → 返回字符串tools流转到 tools 节点执行工具// addConditionalEdges条件边根据函数返回值动态选择下一个节点constrouteMessage(state:typeofGraphState.State){const{messages}state;constlastMessagemessages[messages.length-1]asAIMessage;// If no tools are called, we can finish (respond to the user)if(!lastMessage.tool_calls?.length){returnEND;}// Otherwise if there is, we continue and call the toolsreturntools;};// 1. 读取状态里全部messages消息// 2. 调用绑定了工具的 boundModel把消息丢给 LLM// 3. LLM 返回 AIMessage可能是普通文字也可能带 tool_calls// 4. 返回 { messages: [response] }利用你前面写的reducer: x.concat(y)追加这条新消息到全局 messages 数组不会覆盖历史对话constcallModelasync(state:typeofGraphState.State,config?:RunnableConfig,){const{messages}state;constresponseawaitboundModel.invoke(messages,config);return{messages:[response]};};constworkflownewStateGraph(GraphState).addNode(agent,callModel).addNode(tools,toolNode).addEdge(START,agent)// addConditionalEdges(起点, 路由函数)动态分支.addConditionalEdges(agent,routeMessage).addEdge(tools,agent);constgraphworkflow.compile();假设用户问帮我算 35add 工具可用START→agent(callModel)messages[HumanMessage (“帮我算 35”)]boundModel 调用 LLMLLM 判断需要调用 add 工具返回 AIMessage携带tool_calls状态 messages 追加这条 AIMessage执行routeMessage发现 lastMessage 有 tool_calls → 返回tools进入tools节点ToolNode 解析 tool_call执行 add (3,5)生成 ToolMessage追加进 messagestools执行完成走固定边回到agent再次调用 LLM此时 messages 里面带上了工具返回结果8LLM 这次不需要调用工具直接输出自然语言回答routeMessage检查无 tool_calls返回END流程结束// 传给 Agent 的初始状态一条用户消息打招呼。底层会自动转成 LangChain 内部的HumanMessageletinputs{messages:[{role:user,content:Hi Im Yu, nice to meet you.}]};// 使用streamMode: values流式输出每一步拿到完整的 messages 状态打印最新一条消息forawait(const{messages}ofawaitgraph.stream(inputs,{// 每次 chunk 返回完整最新 state不是增量 delta这是调试 Agent 最常用的模式streamMode:values,})){// 取消息数组里【最后一条】最新消息letmsgmessages[messages?.length-1];if(msg?.content){// 如果消息有文本内容打印文本console.log(msg.content);}elseif(msg?.tool_calls?.length0){// 如果是工具调用消息打印tool_calls结构化信息console.log(msg.tool_calls);}else{// 兜底打印整条消息对象console.log(msg);}console.log(-----\n);}Hi Im Yu, nice to meet you. ----- Hi Yu! Nice to meet you too. How can I assist you today? -----用户输入Hi Im Yu, nice to meet you.图启动进入agent节点调用大模型模型判断不需要调用任何工具直接返回文本回复agent 节点执行完毕stream 吐出第一个完整 statemessages [用户消息AIMessage (“Hi Yu! Nice to meet you too.”)]取最后一条 msg有 content打印Hi Yu! Nice to meet you too.路由routeMessage检测没有 tool_calls → 走到 END图结束循环终止如果是需要调用工具的提问例如帮我算 12输出会多一轮① 模型返回 tool_calls打印 tool_call 数组② 进入 tools 节点执行工具追加 ToolMessage③ 回到 agent模型拿到工具结果输出最终文本回答tools 节点ToolNode和 agent 节点都跑在同一个宿主 Agent 进程 / 同一个 JS 服务里面同运行时inputs{messages:[{role:user,content:Remember my name?}]};forawait(const{messages}ofawaitgraph.stream(inputs,{streamMode:values,})){letmsgmessages[messages?.length-1];if(msg?.content){console.log(msg.content);}elseif(msg?.tool_calls?.length0){console.log(msg.tool_calls);}else{console.log(msg);}console.log(-----\n);}Remember my name?-----You havent shared your name with me yet. Whats your name?-----添加记忆再次尝试使用检查点。将使用MemorySaver它会将检查点“保存”到内存中MemorySaver是内存级别的检查点存储器用来保存每一轮图执行后的状态快照依靠thread_id区分不同对话会话下次用同一个 thread_id就能恢复上次对话的全部 messages 历史import{MemorySaver}fromlangchain/langgraph;// Here we only save in-memory// 内存存储只保存在当前Node进程内存里// 每当图跑完一个节点LangGraph 自动把当前完整 statemessages数组保存一个快照// 只在当前 Node.js 进程内存进程退出数据消失适合本地调试constmemorynewMemorySaver();// 编译图的时候传入checkpointer开启状态持久化// 开启整张图的状态断点 / 持久化能力// 不加checkpointer图执行完就丢弃状态无记忆无法恢复// 加了 checkpointer自动保存每一步状态快照支持中断恢复、记住历史对话constpersistentGraphworkflow.compile({checkpointer:memory});// 会话配置thread_id 就是会话ID同一个对话固定一个thread_id// config里面包含thread_id必须传给stream/invokeLangGraph 靠这个找到对应的会话存档letconfig{configurable:{thread_id:conversation-num-1}};// 用户初始输入inputs{messages:[{role:user,content:Hi Im Jo, nice to meet you.}]};// 流式执行把config传入stream绑定thread_idforawait(const{messages}ofawaitpersistentGraph.stream(inputs,{...config,streamMode:values,})){letmsgmessages[messages?.length-1];if(msg?.content){console.log(msg.content);}elseif(msg?.tool_calls?.length0){console.log(msg.tool_calls);}else{console.log(msg);}console.log(-----\n);}第一轮执行当前代码用户Hi Im Jo, nice to meet you.Agent 回复Hello Jo! Nice to meet you too.执行结束MemorySaver 自动保存这个 thread 的 messages 完整数组第二轮继续同一会话继续提问复用同一个 configthread_id不变此时 LangGraph 会自动读取 thread_idconversation-num-1 的历史 messages模型记得你叫 Jo不需要手动拼接历史消息checkpoint 会自动加载上一轮保存的 state重要第二轮输入的inputs只需要传本轮用户新消息不用带上全部历史持久化会自动加载历史图每执行完一个节点自动把当前完整GraphState存入 MemorySaver以thread_id作为索引再次调用persistentGraph.stream传入相同 thread_idLangGraph 先从内存读取该 thread 最新的检查点messages 历史把新输入合并进状态继续跑图保存的是每完成一个节点执行之后整张图的完整状态快照state 状态也就是 messages 数组也就是节点跑完、状态更新完成之后才存检查点Hi Im Jo, nice to meet you. ----- Hello Jo, nice to meet you too! How can I assist you today? -----inputs{messages:[{role:user,content:Remember my name?}]};forawait(const{messages}ofawaitpersistentGraph.stream(inputs,{...config,streamMode:values,})){letmsgmessages[messages?.length-1];if(msg?.content){console.log(msg.content);}elseif(msg?.tool_calls?.length0){console.log(msg.tool_calls);}else{console.log(msg);}console.log(-----\n);}Remember my name? ----- Yes, Ill remember that your name is Jo. How can I assist you today? -----什么叫「整张图的状态快照」当我们说保存整张图的状态意思是不只是刚刚新增的那一条消息把当前这张图里全部全局数据一次性打包存下来。对本文这个 Agent整张图状态就只有一个字段{ messages: [ HumanMessage(帮我算35), AIMessage(带tool_calls), ToolMessage(结果8) ] }这个对象就是整张图此刻完整状态哪怕这次节点只新增了最后一条 ToolMessagecheckpoint 保存的依然是完整的 messages 数组不是只存新增那一条区分节点局部数据 VS 整张图全局状态节点agent /tools里面的临时变量函数内部的临时变量执行完就销毁不属于整张图状态不会被 checkpoint 保存GraphState所有节点共享的全局容器。agent能读、tools也能读任何节点修改它整张图所有节点都能看到最新值。这就是会被 checkpoint 快照保存的东西。START → agent节点 → 路由判断 → tools节点 → agent节点 → ENDagent和tools是两个独立节点但是它们共用同一份 GraphStatemessagesagent 节点往 state 追加 AI 消息流转到 tools 节点tools 读取同一份 state 里的 messages拿到 tool_calltools 追加 ToolMessage更新同一份 state再回到 agent 节点agent 读取更新后的 stateGraphState 就是整张图所有节点共享的唯一全局内存。回到 checkpoint 保存时机每跑完任意一个节点LangGraph 就把当前这个全局 GraphState 完整打包快照存入 MemorySaver。agent 跑完存一份整张图状态tools 跑完再存一份整张图状态快照里面包含完整 state你的 messages图的执行指针当前执行到哪个节点了中断恢复时知道下一步从哪里继续跑新建对话线程如果要开始新的对话可以传入不同的thread_id。咻所有记忆都消失了开玩笑它们会永远存在于那个线程中config{configurable:{thread_id:conversation-2}};{configurable:{thread_id:conversation-2}}inputs{messages:[{role:user,content:you forgot?}]};forawait(const{messages}ofawaitpersistentGraph.stream(inputs,{...config,streamMode:values,})){letmsgmessages[messages?.length-1];if(msg?.content){console.log(msg.content);}elseif(msg?.tool_calls?.length0){console.log(msg.tool_calls);}else{console.log(msg);}console.log(-----\n);}you forgot? ----- output Could you please provide more context or details about what you are referring to? This will help me assist you better. -----如果涉及到子图子图中每个子节点执行完也会持久化吗会。子图内部每一个子节点执行完成同样会触发 checkpoint 持久化保存【包含子图修改之后的整张图全局状态】关键点子图没有独立的、单独的 checkpoint 存储。子图复用父图同一个 checkpointer 同一个 thread_id持久化的永远是父图定义的顶层 GraphState不是子图自己的局部状态。子图执行 快照保存时序父图挂载了一个子图子图内部有 SubNodeA → SubNodeB父图流转进入子图节点子图内部执行SubNodeA子节点跑完触发一次 checkpoint保存顶层父图完整 state 快照子图内对子图 state / 消息的修改已经合并进顶层 state子图流转到SubNodeB子节点跑完再触发一次 checkpoint再次保存顶层完整 state子图全部执行完毕退出子图回到父图子图里每一个节点跑完都会保存快照保存的内容永远是父图的全局状态不是单独存一份子图状态。子图只是逻辑上的封装它的状态是合并到父图顶层 state 里的。重要区分子图的状态规则子图可以定义自己独立的 StateSchema子图局部 state当子图执行结束退出时子图会把子图 state 的更新合并回父图顶层 state依靠 reducer子图运行过程中子图内部节点产生的 checkpoint 快照依然保存的是父图全局 state子图内部临时状态一旦子图退出局部子图状态就销毁不会单独存档并行扇出fan-out多个子图并行场景父图同时启动 SubGraph1、SubGraph2 两个并行子图SubGraph1 内部每跑完一个子节点 → 写一次顶层 checkpointSubGraph2 内部每跑完一个子节点 → 写一次顶层 checkpoint并行分支各自节点完成都会触发快照都是同一个 thread_id 下的全局状态快照其他子图不会单独开 thread_id子图属于同一个对话线程共用父图的thread_id所有快照都存在这条会话的检查点链里子图内部中断interrupt是全局中断如果子图里面某个节点 interrupt 暂停整个父图流程全部暂停恢复的时候从子图内部中断的那个节点继续执行因为 checkpoint 同时记录了整张图执行位置包括 “当前正在子图内部哪一步”