Langgraph Persistence

• Checkpointer（检查点保存器）：在每个超级步（super-step）保存/加载图状态
• Thread ID（线程 ID）：标识独立的检查点序列（对话）
• Store（存储）：用于保存用户偏好、事实的跨线程记忆

两种记忆类型：

• 短期记忆（Checkpointer）：线程范围内的对话历史
• 长期记忆（Store）：跨线程的用户偏好、事实

检查点保存器设置

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict, Annotated
import operator

class State(TypedDict):
    messages: Annotated[list, operator.add]

def add_message(state: State) -> dict:
    return {"messages": ["Bot response"]}

checkpointer = InMemorySaver()

graph = (
    StateGraph(State)
    .add_node("respond", add_message)
    .add_edge(START, "respond")
    .add_edge("respond", END)
    .compile(checkpointer=checkpointer)  # 在编译时传入
)

# 务必提供 thread_id
config = {"configurable": {"thread_id": "conversation-1"}}

result1 = graph.invoke({"messages": ["Hello"]}, config)
print(len(result1["messages"]))  # 2

result2 = graph.invoke({"messages": ["How are you?"]}, config)
print(len(result2["messages"]))  # 4 (之前的内容 + 新内容)

import { MemorySaver, StateGraph, StateSchema, MessagesValue, START, END } from "@langchain/langgraph";
import { HumanMessage } from "@langchain/core/messages";

const State = new StateSchema({ messages: MessagesValue });

const addMessage = async (state: typeof State.State) => {
  return { messages: [{ role: "assistant", content: "Bot response" }] };
};

const checkpointer = new MemorySaver();

const graph = new StateGraph(State)
  .addNode("respond", addMessage)
  .addEdge(START, "respond")
  .addEdge("respond", END)
  .compile({ checkpointer });

// 务必提供 thread_id
const config = { configurable: { thread_id: "conversation-1" } };

const result1 = await graph.invoke({ messages: [new HumanMessage("Hello")] }, config);
console.log(result1.messages.length);  // 2

const result2 = await graph.invoke({ messages: [new HumanMessage("How are you?")] }, config);
console.log(result2.messages.length);  // 4 (之前的内容 + 新内容)

import os
from langgraph.checkpoint.postgres import PostgresSaver

# 在部署期间运行一次（不要在应用程序启动时运行）：
#   PostgresSaver.from_conn_string(os.environ["DATABASE_URL"]).setup()

with PostgresSaver.from_conn_string(os.environ["DATABASE_URL"]) as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)

import { PostgresSaver } from "@langchain/langgraph-checkpoint-postgres";

// 在部署期间运行一次（不要在应用程序启动时运行）：
//   await PostgresSaver.fromConnString(process.env.DATABASE_URL!).setup();

const checkpointer = PostgresSaver.fromConnString(process.env.DATABASE_URL!);
const graph = builder.compile({ checkpointer });

线程管理

# 不同的线程维护各自独立的状态
alice_config = {"configurable": {"thread_id": "user-alice"}}
bob_config = {"configurable": {"thread_id": "user-bob"}}

graph.invoke({"messages": ["Hi from Alice"]}, alice_config)
graph.invoke({"messages": ["Hi from Bob"]}, bob_config)

# Alice 的状态与 Bob 的状态互相隔离

// 不同的线程维护各自独立的状态
const aliceConfig = { configurable: { thread_id: "user-alice" } };
const bobConfig = { configurable: { thread_id: "user-bob" } };

await graph.invoke({ messages: [new HumanMessage("Hi from Alice")] }, aliceConfig);
await graph.invoke({ messages: [new HumanMessage("Hi from Bob")] }, bobConfig);

// Alice 的状态与 Bob 的状态互相隔离

状态历史与时光旅行

config = {"configurable": {"thread_id": "session-1"}}

result = graph.invoke({"messages": ["start"]}, config)

# 浏览检查点历史
states = list(graph.get_state_history(config))

# 从过去的检查点重放
past = states[-2]
result = graph.invoke(None, past.config)  # None = 从检查点恢复

# 或者分叉：在过去的检查点更新状态，然后恢复执行
fork_config = graph.update_state(past.config, {"messages": ["edited"]})
result = graph.invoke(None, fork_config)

const config = { configurable: { thread_id: "session-1" } };

const result = await graph.invoke({ messages: ["start"] }, config);

// 浏览检查点历史（异步可迭代对象，收集到数组中）
const states: Awaited<ReturnType<typeof graph.getState>>[] = [];
for await (const state of graph.getStateHistory(config)) {
  states.push(state);
}

// 从过去的检查点重放
const past = states[states.length - 2];
const replayed = await graph.invoke(null, past.config);  // null = 从检查点恢复

// 或者分叉：在过去的检查点更新状态，然后恢复执行
const forkConfig = await graph.updateState(past.config, { messages: ["edited"] });
const forked = await graph.invoke(null, forkConfig);

config = {"configurable": {"thread_id": "session-1"}}

# 在恢复之前修改状态
graph.update_state(config, {"data": "manually_updated"})

# 使用更新后的状态恢复执行
result = graph.invoke(None, config)

const config = { configurable: { thread_id: "session-1" } };

// 在恢复之前修改状态
await graph.updateState(config, { data: "manually_updated" });

// 使用更新后的状态恢复执行
const result = await graph.invoke(null, config);

子图检查点保存器作用域

在编译子图时，checkpointer 参数控制着持久化行为。这对于使用中断、需要多轮对话记忆或并行运行的子图至关重要。

各模式的使用场景

• checkpointer=False — 子图不需要中断或持久化。最简单的选项，无检查点开销。
• None（默认 / 省略 checkpointer） — 子图需要 interrupt() 但不需要多轮记忆。每次调用都会全新开始，但可以暂停/恢复。支持并行执行，因为每次调用都会获得一个唯一的命名空间。
• checkpointer=True — 子图需要跨调用记忆状态（多轮对话）。每次调用都会接着上一次的状态继续。

警告：有状态子图（checkpointer=True）不支持在单个节点内多次调用相同的子图实例——这些调用会写入相同的检查点命名空间并发生冲突。

# 不需要中断 — 禁用检查点保存
subgraph = subgraph_builder.compile(checkpointer=False)

# 需要中断但不需要跨调用持久化（默认）
subgraph = subgraph_builder.compile()

# 需要跨调用持久化（有状态）
subgraph = subgraph_builder.compile(checkpointer=True)

// 不需要中断 — 禁用检查点保存
const subgraph = subgraphBuilder.compile({ checkpointer: false });

// 需要中断但不需要跨调用持久化（默认）
const subgraph = subgraphBuilder.compile();

// 需要跨调用持久化（有状态）
const subgraph = subgraphBuilder.compile({ checkpointer: true });

并行子图命名空间隔离

当多个不同的有状态子图并行运行时，将每个子图包装在其自有的 StateGraph 中并使用唯一的节点名称，以实现稳定的命名空间隔离：

from langgraph.graph import MessagesState, StateGraph

def create_sub_agent(model, *, name, **kwargs):
    """使用唯一的节点名称包装智能体以实现命名空间隔离。"""
    agent = create_agent(model=model, name=name, **kwargs)
    return (
        StateGraph(MessagesState)
        .add_node(name, agent)  # 唯一名称 -> 稳定的命名空间
        .add_edge("__start__", name)
        .compile()
    )

fruit_agent = create_sub_agent(
    "gpt-4.1-mini", name="fruit_agent",
    tools=[fruit_info], prompt="...", checkpointer=True,
)
veggie_agent = create_sub_agent(
    "gpt-4.1-mini", name="veggie_agent",
    tools=[veggie_info], prompt="...", checkpointer=True,
)

import { StateGraph, StateSchema, MessagesValue, START } from "@langchain/langgraph";

function createSubAgent(model: string, { name, ...kwargs }: { name: string; [key: string]: any }) {
  const agent = createAgent({ model, name, ...kwargs });
  return new StateGraph(new StateSchema({ messages: MessagesValue }))
    .addNode(name, agent)  // 唯一名称 -> 稳定的命名空间
    .addEdge(START, name)
    .compile();
}

const fruitAgent = createSubAgent("gpt-4.1-mini", {
  name: "fruit_agent", tools: [fruitInfo], prompt: "...", checkpointer: true,
});
const veggieAgent = createSubAgent("gpt-4.1-mini", {
  name: "veggie_agent", tools: [veggieInfo], prompt: "...", checkpointer: true,
});

注意：通过 add_node 作为节点添加的子图会自动获得基于名称的命名空间，不需要此包装器。

长期记忆 (Store)

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

# 保存用户偏好（在所有线程中均可访问）
store.put(("alice", "preferences"), "language", {"preference": "short responses"})

# 带有 store 的节点 — 通过 runtime 访问
from langgraph.runtime import Runtime

def respond(state, runtime: Runtime):
    prefs = runtime.store.get((state["user_id"], "preferences"), "language")
    return {"response": f"Using preference: {prefs.value}"}

# 同时使用 checkpointer 和 store 进行编译
graph = builder.compile(checkpointer=checkpointer, store=store)

# 两个线程访问相同的长期记忆
graph.invoke({"user_id": "alice"}, {"configurable": {"thread_id": "thread-1"}})
graph.invoke({"user_id": "alice"}, {"configurable": {"thread_id": "thread-2"}})  # 偏好设置相同！

import { MemoryStore } from "@langchain/langgraph";

const store = new MemoryStore();

// 保存用户偏好（在所有线程中均可访问）
await store.put(["alice", "preferences"], "language", { preference: "short responses" });

// 带有 store 的节点 — 通过 runtime 访问
const respond = async (state: typeof State.State, runtime: any) => {
  const item = await runtime.store?.get(["alice", "preferences"], "language");
  return { response: `Using preference: ${item?.value?.preference}` };
};

// 同时使用 checkpointer 和 store 进行编译
const graph = builder.compile({ checkpointer, store });

// 两个线程访问相同的长期记忆
await graph.invoke({ userId: "alice" }, { configurable: { thread_id: "thread-1" } });
await graph.invoke({ userId: "alice" }, { configurable: { thread_id: "thread-2" } });  // 偏好设置相同！

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

store.put(("user-123", "facts"), "location", {"city": "San Francisco"})  # 写入 (Put)
item = store.get(("user-123", "facts"), "location")  # 获取 (Get)
results = store.search(("user-123", "facts"), filter={"city": "San Francisco"})  # 搜索 (Search)
store.delete(("user-123", "facts"), "location")  # 删除 (Delete)

修复方案

# 错误：没有 thread_id - 状态不会被持久化！
graph.invoke({"messages": ["Hello"]})
graph.invoke({"messages": ["What did I say?"]})  # 无法记住！

# 正确：务必提供 thread_id
config = {"configurable": {"thread_id": "session-1"}}
graph.invoke({"messages": ["Hello"]}, config)
graph.invoke({"messages": ["What did I say?"]}, config)  # 成功记住！

// 错误：没有 thread_id - 状态不会被持久化！
await graph.invoke({ messages: [new HumanMessage("Hello")] });
await graph.invoke({ messages: [new HumanMessage("What did I say?")] });  // 无法记住！

// 正确：务必提供 thread_id
const config = { configurable: { thread_id: "session-1" } };
await graph.invoke({ messages: [new HumanMessage("Hello")] }, config);
await graph.invoke({ messages: [new HumanMessage("What did I say?")] }, config);  // 成功记住！

# 错误：进程重启后数据丢失
checkpointer = InMemorySaver()  # 仅限内存！

# 正确：在生产环境中使用持久化存储
from langgraph.checkpoint.postgres import PostgresSaver
with PostgresSaver.from_conn_string("postgresql://...") as checkpointer:
    checkpointer.setup()  # 仅首次使用创建数据表时需要
    graph = builder.compile(checkpointer=checkpointer)

// 错误：进程重启后数据丢失
const checkpointer = new MemorySaver();  // 仅限内存！

// 正确：在生产环境中使用持久化存储
import { PostgresSaver } from "@langchain/langgraph-checkpoint-postgres";
const checkpointer = PostgresSaver.fromConnString("postgresql://...");
await checkpointer.setup(); // 仅首次使用创建数据表时需要

from langgraph.types import Overwrite

# 带有 reducer 的状态：items: Annotated[list, operator.add]
# 当前状态：{"items": ["A", "B"]}

# update_state 会经过 reducers 处理
graph.update_state(config, {"items": ["C"]})  # 结果：["A", "B", "C"] - 已追加！

# 若要进行替换，请使用 Overwrite
graph.update_state(config, {"items": Overwrite(["C"])})  # 结果：["C"] - 已替换

import { Overwrite } from "@langchain/langgraph";

// 带有 reducer 的状态：items 使用 concat reducer
// 当前状态：{ items: ["A", "B"] }

// updateState 会经过 reducers 处理
await graph.updateState(config, { items: ["C"] });  // 结果：["A", "B", "C"] - 已追加！

// 若要进行替换，请使用 Overwrite
await graph.updateState(config, { items: new Overwrite(["C"]) });  // 结果：["C"] - 已替换

# 错误：store 在节点中不可用
def my_node(state):
    store.put(...)  # NameError! store 未定义

# 正确：通过 runtime 访问 store
from langgraph.runtime import Runtime

def my_node(state, runtime: Runtime):
    runtime.store.put(...)  # 正确的 store 实例

// 错误：store 在节点中不可用
const myNode = async (state) => {
  store.put(...);  // ReferenceError!
};

// 正确：通过 runtime 访问 store
const myNode = async (state, runtime) => {
  await runtime.store?.put(...);  // 正确的 store 实例
};

• 不要在生产环境中使用 InMemorySaver — 重启后数据会丢失；请使用 PostgresSaver
• 不要遗漏 thread_id — 没有它状态将无法持久化
• 不要期望 update_state 能绕过 reducer — 它会经过 reducer 处理；如需直接替换请使用 Overwrite
• 不要在单个节点内并行运行相同的有状态子图（checkpointer=True）— 会导致命名空间冲突
• 不要直接在节点内部访问 store — 应该通过 Runtime 参数使用 runtime.store 访问