LangGraph 07：Checkpoint 持久化

发表于 2026-08-19 更新于 2026-08-20 分类于 ai

用 InMemorySaver 与 thread_id 做会话记忆，查询 get_state/history；PostgresSaver 作可选落库方案。

前言

默认 compile() 不带 checkpointer 时，每次 invoke 都是无状态的一次跑完。
要做多轮对话、断点续跑或事后审计，需要 Checkpoint：按 thread_id 持久化图状态快照。
本文主线用 InMemorySaver（无需数据库即可跑通），再讲 get_state / get_state_history。
生产落库常用 PostgresSaver，文末作可选说明。
聊天模型对接 火山方舟 Coding Plan。
下文需要 Python 3.12+，依赖用 uv 管理。

依赖

建议使用 Python 3.12 及以上。
用 uv 初始化工程并声明基础依赖。

uv init langgraph-checkpoints
cd langgraph-checkpoints
uv venv --python 3.12
uv add "langchain>=1.0,<2.0" langchain-openai langchain-core langgraph python-dotenv rich

在项目根目录创建 .env：

1 2	OPENAI_API_KEY=你的火山方舟 API Key OPENAI_BASE_URL=https://ark.cn-beijing.volces.com/api/coding/v3

不要把 .env 提交进 Git。
可选 Postgres 小节再额外安装 checkpointer 包。

实现

InMemorySaver

InMemorySaver 把检查点存在进程内存里，适合本地演示与单测。
编译时传入 checkpointer=，调用时在 config["configurable"]["thread_id"] 指定线程。

下面用 MessagesState 做两轮对话：同一 thread_id 能记住「我是老王」。

import os

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, MessagesState, StateGraph
from rich import print as rprint

load_dotenv()

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)


class OverAllState(MessagesState):
    output: str


def llm_node(state: OverAllState) -> OverAllState:
    return {"messages": [model.invoke(state["messages"])]}


def output_node(state: OverAllState) -> OverAllState:
    return {"output": state["messages"][-1].content}


builder = StateGraph(state_schema=OverAllState)
builder.add_node("llm_node", llm_node)
builder.add_node("output_node", output_node)
builder.add_edge(START, "llm_node")
builder.add_edge("llm_node", "output_node")
builder.add_edge("output_node", END)

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "demo-memory-01"}}

rprint(graph.invoke({"messages": [HumanMessage("你好,我是老王")]}, config=config))
rprint(graph.invoke({"messages": [HumanMessage("你好,我是谁")]}, config=config))

other = {"configurable": {"thread_id": "demo-memory-01-other"}}

rprint(graph.invoke({"messages": [HumanMessage("你好,我是谁")]}, config=other))

同一 thread_id 的第二轮应能答出「老王」；换线程则像新会话。
进程退出后内存检查点会丢失，这是演示代价。

读最新状态

graph.get_state(config) 返回该线程最新检查点快照（StateSnapshot）。
可查看 values、next、config 等字段，便于调试。

# 承接上一节的 graph / config
snapshot = graph.get_state(config)
rprint(snapshot.values)
rprint(snapshot.next)
rprint(snapshot.config)

next 为空元组通常表示该线程已跑到终点。
人机协同中断时，next 会指向待执行节点（后续 HITL 文展开）。

历史快照

get_state_history(config) 按时间倒序列出该线程的全部检查点。
可用历史里的 checkpoint_id 再 get_state，读取某一超步时的状态。

下面先跑一个并行写诗/笑话图，再打印历史并取倒数第二个检查点。

import os
from typing import TypedDict

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from rich import print as rprint

load_dotenv()

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0.4,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)


class OverAllState(TypedDict):
    topic: str
    poem: str
    joke: str
    final_output: str


class InputState(TypedDict):
    topic: str


class OutputState(TypedDict):
    final_output: str


def node_poem(state: InputState) -> OverAllState:
    poem = model.invoke(
        [HumanMessage(f"写一首关于{state['topic']}的七言绝句，只输出诗句。")]
    ).content
    return {"poem": poem}


def node_joke(state: InputState) -> OverAllState:
    joke = model.invoke(
        [HumanMessage(f"写一个关于{state['topic']}的笑话，只输出正文。")]
    ).content
    return {"joke": joke}


def node_output(state: OverAllState) -> OutputState:
    return {
        "final_output": (
            f"主题:{state['topic']}\n诗:{state['poem']}\n笑话:{state['joke']}"
        )
    }


builder = StateGraph(
    state_schema=OverAllState,
    input_schema=InputState,
    output_schema=OutputState,
)
builder.add_node("node_poem", node_poem)
builder.add_node("node_joke", node_joke)
builder.add_node("node_output", node_output)
builder.add_edge(START, "node_poem")
builder.add_edge(START, "node_joke")
builder.add_edge("node_poem", "node_output")
builder.add_edge("node_joke", "node_output")
builder.add_edge("node_output", END)

checkpointer = InMemorySaver()
config = {"configurable": {"thread_id": "history-demo"}}
graph = builder.compile(checkpointer=checkpointer)
rprint(graph.invoke({"topic": "莲花"}, config=config))

history = list(graph.get_state_history(config=config))
rprint(f"检查点数量: {len(history)}")
rprint(graph.get_state(config))  # 最新

older_id = history[-2].config["configurable"]["checkpoint_id"]
older_config = {
    "configurable": {
        "thread_id": config["configurable"]["thread_id"],
        "checkpoint_id": older_id,
    }
}
rprint(graph.get_state(config=older_config))

历史里每一项都是完整快照，可据此做时光旅行（见下一篇）。
checkpoint_id 每次运行都会变，请从 history 动态读取，不要写死 UUID。

Postgres 可选

需要跨进程、可重启的持久化时，改用 PostgresSaver。
注意选型时看实际导入的类名：生产常见是 Postgres 检查点，而不是「名字里带 SQL 就当成 SQLite」。

先安装可选依赖，并准备好 Postgres 连接串。

1	uv add "langgraph-checkpoint-postgres" "psycopg[binary,pool]"

下面仅展示骨架；未启动数据库时请跳过，继续用上文的内存示例即可。

import os

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.graph import END, START, MessagesState, StateGraph

load_dotenv()

model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)

DB_URL = "postgresql://langgraph_user:123456@localhost:5432/langgraph_db?sslmode=disable"


class OverAllState(MessagesState):
    output: str


def llm_node(state: OverAllState) -> OverAllState:
    return {"messages": [model.invoke(state["messages"])]}


def output_node(state: OverAllState) -> OverAllState:
    return {"output": state["messages"][-1].content}


builder = StateGraph(state_schema=OverAllState)
builder.add_node("llm_node", llm_node)
builder.add_node("output_node", output_node)
builder.add_edge(START, "llm_node")
builder.add_edge("llm_node", "output_node")
builder.add_edge("output_node", END)

with PostgresSaver.from_conn_string(DB_URL) as checkpointer:
    checkpointer.setup()  # 首次建表，幂等
    graph = builder.compile(checkpointer=checkpointer)
    config = {"configurable": {"thread_id": "demo-pg-01"}}

    print(
        graph.invoke(
            {"messages": [HumanMessage("你好,我是老王")]},
            config=config,
        )
    )

setup() 用于创建检查点表；重复调用通常安全。
连接串、账号与 SSL 请按环境修改，勿把真实密码写进仓库。

验证

同一 thread_id 连续两轮：第二轮应引用第一轮自称。
换 thread_id：模型不应记住「老王」。
get_state_history：并行图跑完后长度应大于 1，且可用旧 checkpoint_id 读中间态。

总结

启用记忆：compile(checkpointer=...) + 每次调用带 thread_id。
查询：get_state 看最新，get_state_history 看时间线。
落库：演示用 InMemorySaver；生产常用 PostgresSaver。