企业官网建设流程全解析

线程持久化

许多AI应用程序需要记忆来在单个对话“线程”中的多次交互中共享上下文。在LangGraph中，这种对话级别的记忆可以通过使用检查点（Checkpointers）添加到任何图上。

只需使用兼容的检查点编译图即可。下面是一个使用简单的内存中“MemorySaver”的示例

import{MemorySaver}from"@langchain/langgraph";constcheckpointer=newMemorySaver();constgraph=workflow.compile({checkpointer});

状态 & 状态合并策略

import{Annotation}from"@langchain/langgraph";import{BaseMessage}from"@langchain/core/messages";// 定义整张图的状态结构 GraphStateconstGraphState=Annotation.Root({messages:Annotation<BaseMessage[]>({reducer:(x,y)=>x.concat(y),}),});

Annotation.Root：定义 LangGraph 图的状态 Schema，声明状态里有哪些字段、类型、怎么合并更新
这里状态只有一个字段：messages，类型是消息数组（LLM 对话消息，HumanMessage/AIMessage 都属于BaseMessage）

reducer 归约函数

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 绝对不能接受的

// 当前状态（x）constx=[newHumanMessage("你好")];// 节点返回的更新（y）consty=[newAIMessage("你好，我是智能体")];constnewVal=x.concat(y);// newVal = [HumanMsg, AiMsg]

设置工具

创建一个占位符搜索引擎

import{tool}from"@langchain/core/tools";import{z}from"zod";constsearchTool=tool(async({}:{query:string})=>{// This is a placeholder for the actual implementationreturn"Cold, 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:"What's the weather like?"});consttools=[searchTool];

现在可以将这些工具包装在一个简单的 ToolNode 中。每当 LLM 调用这些工具（函数）时，此对象将实际运行它们

import{ToolNode}from"@langchain/langgraph/prebuilt";consttoolNode=newToolNode(tools);

设置模型

加载聊天模型

它应该能与消息配合使用。将所有代理状态都以消息的形式表示，因此它需要能够很好地与消息配合使用

什么是消息BaseMessage？

LangChain 里的消息就是对话单元：

• HumanMessage：用户说的话
• AIMessage：大模型输出（可以是文字，也可以带工具调用）
• ToolMessage：工具执行完返回的结果

Agent 的全部状态，就存成一个消息数组messages[]
整个智能体的记忆、历史对话、工具返回结果，全部塞进这个消息列表，不用单独定义一堆零散字段
所以聊天模型必须支持接收消息数组作为输入，而不是只接收纯字符串

普通文本模型：model.invoke("一句话字符串")
聊天模型 (ChatModel)：model.invoke([msg1, msg2, msg3])，这就是「和消息配合」

它应该能与工具调用配合使用，这意味着它可以在响应中返回函数参数

支持工具调用的 ChatModel（gpt-4o 这类）可以输出结构化指令：我要调用某个工具，并且带上这个工具需要的参数
不是自然语言，是机器能识别的tool_calls结构，类似：

{ "tool_calls": [ { "name": "search_weather", "args": {"city": "北京"} } ] }

意思：别把这段当成文字回答，去执行 search_weather 函数，入参 city = 北京
这个就是在响应中返回函数参数

代码加载

注意
这些模型要求并非 LangGraph 的通用要求——它们仅是此示例的要求

import{ChatOpenAI}from"@langchain/openai";// 基础对话大模型，只会正常输出自然语言文本constmodel=newChatOpenAI({model:"gpt-4o"});

完成此操作后，应确保模型知道它可以使用这些工具。可以通过调用bindTools来完成此操作

constboundModel=model.bindTools(tools);

定义图

现在可以把它们放在一起。将首先在没有检查点的情况下运行它

这是 LangGraph 最经典ReAct 工具调用智能体流程图代码（反复思考 + 调用工具，就是ReAct）
流程：START → agent（调用大模型） → 路由判断：

• 如果模型输出没有工具调用→ 走到 END，结束
• 如果模型输出有 tool_calls→ 走到 tools 节点执行工具，执行完自动回到 agent，循环
  

// - `StateGraph`：用来构建状态图import{END,START,StateGraph}from"@langchain/langgraph";import{AIMessage}from"@langchain/core/messages";// - `RunnableConfig`：运行配置（里面包含 checkpoint 持久化、线程 id 等）import{RunnableConfig}from"@langchain/core/runnables";// 1. 拿到当前状态里的`messages`消息数组，取最后一条消息（就是刚大模型返回的 AIMessage）// 2. 判断这条消息有没有 `tool_calls`：// - `tool_calls`为空 / 不存在 → 返回`END`，整张图结束，回答用户// - 存在 tool_calls → 返回字符串`"tools"`，流转到 tools 节点执行工具// `addConditionalEdges`：条件边，根据函数返回值动态选择下一个节点constrouteMessage=(state:typeofGraphState.State)=>{const{messages}=state;constlastMessage=messages[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 toolsreturn"tools";};// 1. 读取状态里全部`messages`消息// 2. 调用绑定了工具的 boundModel，把消息丢给 LLM// 3. LLM 返回 AIMessage（可能是普通文字，也可能带 tool_calls）// 4. 返回 `{ messages: [response] }`，利用你前面写的`reducer: x.concat(y)`，追加这条新消息到全局 messages 数组，不会覆盖历史对话constcallModel=async(state:typeofGraphState.State,config?:RunnableConfig,)=>{const{messages}=state;constresponse=awaitboundModel.invoke(messages,config);return{messages:[response]};};constworkflow=newStateGraph(GraphState).addNode("agent",callModel).addNode("tools",toolNode).addEdge(START,"agent")// `addConditionalEdges(起点, 路由函数)`：动态分支.addConditionalEdges("agent",routeMessage).addEdge("tools","agent");constgraph=workflow.compile();

假设用户问：帮我算 3+5，add 工具可用

1. START→agent(callModel)，messages=[HumanMessage (“帮我算 3+5”)]，boundModel 调用 LLM
2. LLM 判断需要调用 add 工具，返回 AIMessage，携带tool_calls
3. 状态 messages 追加这条 AIMessage
4. 执行routeMessage：发现 lastMessage 有 tool_calls → 返回tools
5. 进入tools节点：ToolNode 解析 tool_call，执行 add (3,5)，生成 ToolMessage，追加进 messages
6. tools执行完成，走固定边回到agent
7. 再次调用 LLM，此时 messages 里面带上了工具返回结果8
8. LLM 这次不需要调用工具，直接输出自然语言回答
9. routeMessage检查：无 tool_calls，返回END，流程结束

// 传给 Agent 的初始状态：一条用户消息，打招呼。底层会自动转成 LangChain 内部的`HumanMessage`letinputs={messages:[{role:"user",content:"Hi I'm Yu, nice to meet you."}]};// 使用`streamMode: "values"`流式输出，每一步拿到完整的 messages 状态，打印最新一条消息forawait(const{messages}ofawaitgraph.stream(inputs,{// 每次 chunk 返回完整最新 state（不是增量 delta），这是调试 Agent 最常用的模式streamMode:"values",})){// 取消息数组里【最后一条】最新消息letmsg=messages[messages?.length-1];if(msg?.content){// 如果消息有文本内容，打印文本console.log(msg.content);}elseif(msg?.tool_calls?.length>0){// 如果是工具调用消息，打印tool_calls结构化信息console.log(msg.tool_calls);}else{// 兜底打印整条消息对象console.log(msg);}console.log("-----\n");}

Hi I'm Yu, nice to meet you. ----- Hi Yu! Nice to meet you too. How can I assist you today? -----

用户输入：Hi I'm Yu, nice to meet you.

1. 图启动，进入agent节点调用大模型
2. 模型判断：不需要调用任何工具，直接返回文本回复
3. agent 节点执行完毕，stream 吐出第一个完整 state：messages = [用户消息，AIMessage (“Hi Yu! Nice to meet you too.”)]
4. 取最后一条 msg，有 content，打印：Hi Yu! Nice to meet you too.
5. 路由routeMessage检测：没有 tool_calls → 走到 END，图结束，循环终止

如果是需要调用工具的提问（例如：帮我算 1+2），输出会多一轮：
① 模型返回 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",})){letmsg=messages[messages?.length-1];if(msg?.content){console.log(msg.content);}elseif(msg?.tool_calls?.length>0){console.log(msg.tool_calls);}else{console.log(msg);}console.log("-----\n");}

Remember my name?-----You haven't shared your name with me yet. What's your name?-----

添加记忆

再次尝试使用检查点。将使用MemorySaver，它会将检查点“保存”到内存中

MemorySaver是内存级别的检查点存储器，用来保存每一轮图执行后的状态快照；依靠thread_id区分不同对话会话，下次用同一个 thread_id，就能恢复上次对话的全部 messages 历史

import{MemorySaver}from"@langchain/langgraph";// Here we only save in-memory// 内存存储，只保存在当前Node进程内存里// 每当图跑完一个节点，LangGraph 自动把当前完整 state（`messages`数组）保存一个快照// 只在当前 Node.js 进程内存，进程退出，数据消失，适合本地调试constmemory=newMemorySaver();// 编译图的时候，传入checkpointer，开启状态持久化// 开启整张图的状态断点 / 持久化能力：// 不加`checkpointer`：图执行完就丢弃状态，无记忆，无法恢复// 加了 checkpointer：自动保存每一步状态快照，支持中断恢复、记住历史对话constpersistentGraph=workflow.compile({checkpointer:memory});// 会话配置：thread_id 就是会话ID，同一个对话固定一个thread_id// `config`（里面包含`thread_id`）必须传给`stream`/`invoke`，LangGraph 靠这个找到对应的会话存档letconfig={configurable:{thread_id:"conversation-num-1"}};// 用户初始输入inputs={messages:[{role:"user",content:"Hi I'm Jo, nice to meet you."}]};// 流式执行，把config传入stream，绑定thread_idforawait(const{messages}ofawaitpersistentGraph.stream(inputs,{...config,streamMode:"values",})){letmsg=messages[messages?.length-1];if(msg?.content){console.log(msg.content);}elseif(msg?.tool_calls?.length>0){console.log(msg.tool_calls);}else{console.log(msg);}console.log("-----\n");}

第一轮执行（当前代码）
用户：Hi I'm Jo, nice to meet you.
Agent 回复：Hello Jo! Nice to meet you too.
执行结束，MemorySaver 自动保存这个 thread 的 messages 完整数组

第二轮：继续同一会话，继续提问（复用同一个 config，thread_id不变）
此时 LangGraph 会自动读取 thread_id=conversation-num-1 的历史 messages，模型记得你叫 Jo
不需要手动拼接历史消息，checkpoint 会自动加载上一轮保存的 state
重要：第二轮输入的inputs只需要传本轮用户新消息，不用带上全部历史，持久化会自动加载历史

1. 图每执行完一个节点，自动把当前完整GraphState存入 MemorySaver，以thread_id作为索引
2. 再次调用persistentGraph.stream，传入相同 thread_id
3. LangGraph 先从内存读取该 thread 最新的检查点（messages 历史）
4. 把新输入合并进状态，继续跑图

保存的是每完成一个节点执行之后，整张图的完整状态快照（state 状态，也就是 messages 数组），也就是节点跑完、状态更新完成之后才存检查点

Hi I'm 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",})){letmsg=messages[messages?.length-1];if(msg?.content){console.log(msg.content);}elseif(msg?.tool_calls?.length>0){console.log(msg.tool_calls);}else{console.log(msg);}console.log("-----\n");}

Remember my name? ----- Yes, I'll remember that your name is Jo. How can I assist you today? -----

什么叫「整张图的状态快照」？
当我们说保存整张图的状态，意思是：不只是刚刚新增的那一条消息，把当前这张图里全部全局数据一次性打包存下来。

对本文这个 Agent，整张图状态就只有一个字段：

{ messages: [ HumanMessage("帮我算3+5"), AIMessage(带tool_calls), ToolMessage("结果=8") ] }

这个对象，就是整张图此刻完整状态
哪怕这次节点只新增了最后一条 ToolMessage，checkpoint 保存的依然是完整的 messages 数组，不是只存新增那一条

区分：节点局部数据 VS 整张图全局状态

• 节点（agent /tools）里面的临时变量：函数内部的临时变量，执行完就销毁，不属于整张图状态，不会被 checkpoint 保存
• GraphState：所有节点共享的全局容器。agent能读、tools也能读；任何节点修改它，整张图所有节点都能看到最新值。这就是会被 checkpoint 快照保存的东西。

START → agent节点 → 路由判断 → tools节点 → agent节点 → END

agent和tools是两个独立节点，但是它们共用同一份 GraphState（messages）

1. agent 节点往 state 追加 AI 消息
2. 流转到 tools 节点，tools 读取同一份 state 里的 messages，拿到 tool_call
3. tools 追加 ToolMessage，更新同一份 state
4. 再回到 agent 节点，agent 读取更新后的 state
   GraphState 就是整张图所有节点共享的唯一全局内存。

回到 checkpoint 保存时机
每跑完任意一个节点，LangGraph 就把当前这个全局 GraphState 完整打包快照存入 MemorySaver。

• agent 跑完：存一份整张图状态
• tools 跑完：再存一份整张图状态

快照里面包含：

1. 完整 state（你的 messages）
2. 图的执行指针：当前执行到哪个节点了（中断恢复时，知道下一步从哪里继续跑）

新建对话线程

如果要开始新的对话，可以传入不同的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",})){letmsg=messages[messages?.length-1];if(msg?.content){console.log(msg.content);}elseif(msg?.tool_calls?.length>0){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）

1. 父图流转进入子图节点
2. 子图内部执行SubNodeA，子节点跑完
   • 触发一次 checkpoint
   • 保存顶层父图完整 state 快照（子图内对子图 state / 消息的修改，已经合并进顶层 state）
3. 子图流转到SubNodeB，子节点跑完
   • 再触发一次 checkpoint
   • 再次保存顶层完整 state
4. 子图全部执行完毕，退出子图，回到父图

子图里每一个节点跑完，都会保存快照；保存的内容永远是父图的全局状态，不是单独存一份子图状态。
子图只是逻辑上的封装，它的状态是合并到父图顶层 state 里的。

重要区分：子图的状态规则

• 子图可以定义自己独立的 StateSchema（子图局部 state）
• 当子图执行结束退出时，子图会把子图 state 的更新，合并回父图顶层 state（依靠 reducer）
• 子图运行过程中，子图内部节点产生的 checkpoint 快照，依然保存的是父图全局 state；子图内部临时状态，一旦子图退出，局部子图状态就销毁，不会单独存档

并行扇出（fan-out，多个子图并行）场景

父图同时启动 SubGraph1、SubGraph2 两个并行子图：

• SubGraph1 内部每跑完一个子节点 → 写一次顶层 checkpoint
• SubGraph2 内部每跑完一个子节点 → 写一次顶层 checkpoint
  并行分支各自节点完成，都会触发快照，都是同一个 thread_id 下的全局状态快照

其他

1. 子图不会单独开 thread_id
   子图属于同一个对话线程，共用父图的thread_id，所有快照都存在这条会话的检查点链里
2. 子图内部中断（interrupt）是全局中断
   如果子图里面某个节点 interrupt 暂停，整个父图流程全部暂停；恢复的时候，从子图内部中断的那个节点继续执行

因为 checkpoint 同时记录了整张图执行位置，包括 “当前正在子图内部哪一步”