一、编译StateGraph只是一张“蓝图”compile()会把它变成一个可执行的 Runnable。builder StateGraph(MyState) # ... 添加节点和边 ... graph builder.compile() # 编译编译时 LangGraph 会做几件事编译时做的事说明校验图结构检查是否有孤立节点、是否有入口绑定持久化如果传了 checkpointer会绑定到图上生成执行计划把节点和边转换成可调度的执行流程返回 Runnable编译后的图支持 invoke、stream、batch 等方法2. 编译参数graph builder.compile( checkpointermemory, # 持久化 storestore, # 长期记忆 interrupt_before[node_a], # 在某节点前中断 interrupt_after[node_b], # 在某节点后中断 )二、执行四种调用方式编译后的图是一个 Runnable支持标准调用方法。方法用途是否阻塞invoke执行一次返回最终 State阻塞stream流式执行逐个输出非阻塞迭代batch批量执行多个输入阻塞ainvoke/astream异步版本异步1. invoke最简单的执行result graph.invoke({messages: [], count: 0}) print(result)2. stream流式执行推荐stream可以实时看到每个节点的输出而不是等全部跑完。for chunk in graph.stream({messages: []}): print(chunk)stream_mode决定了输出的粒度stream_mode输出内容values每个节点执行后的完整 Stateupdates每个节点返回的更新字典messagesLLM 生成的 token 流custom节点内通过writer自定义的输出# 只看每个节点的更新 for chunk in graph.stream(input, stream_modeupdates): print(chunk) # 输出{node_a: {count: 1}}3. batch批量执行results graph.batch([ {messages: [], count: 0}, {messages: [], count: 100} ])4. 执行时的 configconfig { configurable: { thread_id: user_123 # 会话 ID }, recursion_limit: 25 # 最大迭代次数 } result graph.invoke(input, config)三、持久化Checkpointer没有持久化时每次调用都是独立的图不记得上次说了什么。Checkpointer 让图能记住历史实现多轮对话。使用方式from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() graph builder.compile(checkpointermemory) # 第一次调用 config {configurable: {thread_id: user_1}} graph.invoke({messages: [你好]}, config) # 第二次调用自动恢复上下文 graph.invoke({messages: [继续]}, config)常用 CheckpointerCheckpointer存储位置适用场景MemorySaver内存开发测试SqliteSaverSQLite 文件本地生产PostgresSaverPostgreSQL生产环境RedisSaverRedis高并发查看历史状态# 获取当前状态 state graph.get_state(config) # 获取历史状态列表 history list(graph.get_state_history(config))四、长期记忆Store1. Checkpointer vs Store特性CheckpointerStore作用范围单个会话thread_id跨会话存储内容图的完整状态键值对数据典型用途多轮对话用户偏好、长期知识2. 使用方式from langgraph.store.memory import InMemoryStore store InMemoryStore() graph builder.compile(storestore) # 在节点中访问 store def my_node(state, runtime: Runtime): # 写入 runtime.store.put((users, user_1), preference, {theme: dark}) # 读取 pref runtime.store.get((users, user_1), preference) return {result: pref}五、中断与恢复人机协同1. 中断的两种方式# 方式一编译时指定 graph builder.compile( interrupt_before[approve_node], # 在节点前中断 interrupt_after[review_node] # 在节点后中断 ) # 方式二节点内调用 interrupt from langgraph.types import interrupt def approve_node(state): user_input interrupt(是否批准) return {approved: user_input yes}2. 恢复执行from langgraph.types import Command # 用 Command 恢复 graph.invoke(Command(resumeyes), config)3. 中断流程图六、示例from typing import TypedDict, Annotated from operator import add from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver class ChatState(TypedDict): messages: Annotated[list, add] def chatbot(state: ChatState) - dict: # 模拟 LLM 回复 user_msg state[messages][-1] return {messages: [f回复{user_msg}]} builder StateGraph(ChatState) builder.add_node(chatbot, chatbot) builder.add_edge(START, chatbot) builder.add_edge(chatbot, END) # 编译时绑定 checkpointer memory MemorySaver() graph builder.compile(checkpointermemory) # 多轮对话 config {configurable: {thread_id: user_1}} graph.invoke({messages: [你好]}, config) graph.invoke({messages: [再见]}, config) # 查看完整历史 state graph.get_state(config) print(state.values[messages]) # [你好, 回复你好, 再见, 回复再见]
LangGraph的运行时：编译、执行与持久化（三）
一、编译StateGraph只是一张“蓝图”compile()会把它变成一个可执行的 Runnable。builder StateGraph(MyState) # ... 添加节点和边 ... graph builder.compile() # 编译编译时 LangGraph 会做几件事编译时做的事说明校验图结构检查是否有孤立节点、是否有入口绑定持久化如果传了 checkpointer会绑定到图上生成执行计划把节点和边转换成可调度的执行流程返回 Runnable编译后的图支持 invoke、stream、batch 等方法2. 编译参数graph builder.compile( checkpointermemory, # 持久化 storestore, # 长期记忆 interrupt_before[node_a], # 在某节点前中断 interrupt_after[node_b], # 在某节点后中断 )二、执行四种调用方式编译后的图是一个 Runnable支持标准调用方法。方法用途是否阻塞invoke执行一次返回最终 State阻塞stream流式执行逐个输出非阻塞迭代batch批量执行多个输入阻塞ainvoke/astream异步版本异步1. invoke最简单的执行result graph.invoke({messages: [], count: 0}) print(result)2. stream流式执行推荐stream可以实时看到每个节点的输出而不是等全部跑完。for chunk in graph.stream({messages: []}): print(chunk)stream_mode决定了输出的粒度stream_mode输出内容values每个节点执行后的完整 Stateupdates每个节点返回的更新字典messagesLLM 生成的 token 流custom节点内通过writer自定义的输出# 只看每个节点的更新 for chunk in graph.stream(input, stream_modeupdates): print(chunk) # 输出{node_a: {count: 1}}3. batch批量执行results graph.batch([ {messages: [], count: 0}, {messages: [], count: 100} ])4. 执行时的 configconfig { configurable: { thread_id: user_123 # 会话 ID }, recursion_limit: 25 # 最大迭代次数 } result graph.invoke(input, config)三、持久化Checkpointer没有持久化时每次调用都是独立的图不记得上次说了什么。Checkpointer 让图能记住历史实现多轮对话。使用方式from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() graph builder.compile(checkpointermemory) # 第一次调用 config {configurable: {thread_id: user_1}} graph.invoke({messages: [你好]}, config) # 第二次调用自动恢复上下文 graph.invoke({messages: [继续]}, config)常用 CheckpointerCheckpointer存储位置适用场景MemorySaver内存开发测试SqliteSaverSQLite 文件本地生产PostgresSaverPostgreSQL生产环境RedisSaverRedis高并发查看历史状态# 获取当前状态 state graph.get_state(config) # 获取历史状态列表 history list(graph.get_state_history(config))四、长期记忆Store1. Checkpointer vs Store特性CheckpointerStore作用范围单个会话thread_id跨会话存储内容图的完整状态键值对数据典型用途多轮对话用户偏好、长期知识2. 使用方式from langgraph.store.memory import InMemoryStore store InMemoryStore() graph builder.compile(storestore) # 在节点中访问 store def my_node(state, runtime: Runtime): # 写入 runtime.store.put((users, user_1), preference, {theme: dark}) # 读取 pref runtime.store.get((users, user_1), preference) return {result: pref}五、中断与恢复人机协同1. 中断的两种方式# 方式一编译时指定 graph builder.compile( interrupt_before[approve_node], # 在节点前中断 interrupt_after[review_node] # 在节点后中断 ) # 方式二节点内调用 interrupt from langgraph.types import interrupt def approve_node(state): user_input interrupt(是否批准) return {approved: user_input yes}2. 恢复执行from langgraph.types import Command # 用 Command 恢复 graph.invoke(Command(resumeyes), config)3. 中断流程图六、示例from typing import TypedDict, Annotated from operator import add from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver class ChatState(TypedDict): messages: Annotated[list, add] def chatbot(state: ChatState) - dict: # 模拟 LLM 回复 user_msg state[messages][-1] return {messages: [f回复{user_msg}]} builder StateGraph(ChatState) builder.add_node(chatbot, chatbot) builder.add_edge(START, chatbot) builder.add_edge(chatbot, END) # 编译时绑定 checkpointer memory MemorySaver() graph builder.compile(checkpointermemory) # 多轮对话 config {configurable: {thread_id: user_1}} graph.invoke({messages: [你好]}, config) graph.invoke({messages: [再见]}, config) # 查看完整历史 state graph.get_state(config) print(state.values[messages]) # [你好, 回复你好, 再见, 回复再见]
