LangGraph 08：时光旅行与 Store

发表于 更新于 分类于
基于 Checkpoint 做错误恢复、replay 与 fork；区分 Store 与检查点，并用 Runtime 注入上下文。

前言

有了 Checkpoint，图就不再是「跑完即弃」：可以从历史快照续跑、改输入再分叉，甚至在并行失败后只重跑出错分支。
Store 则是另一条记忆轴：跨线程的长期键值，不绑在某次对话的检查点链上。
本文压缩讲解错误恢复思路、invoke(None) 回放、update_state 分叉，以及 InMemoryStore 与 Runtime 上下文注入。
Postgres 版 Store/Saver 仅作可选备注；正文演示一律可在无数据库环境下运行。
聊天模型对接 火山方舟 Coding Plan。
下文需要 Python 3.12+，依赖用 uv 管理。

依赖

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

1
2
3
4
uv init langgraph-timetravel-store
cd langgraph-timetravel-store
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。

实现

错误恢复思路

并行图中，同一超步里可能一边成功、一边抛错。
启用 checkpointer 后，成功节点的结果往往已写入检查点；修好出错逻辑后，对同一 Saver 与 thread_id 再 invoke(None, config=...)，可从中断处续跑。

下面用可变开关模拟：第一次 node_joke 失败；关掉开关后空输入续跑。
请保持同一个 InMemorySaver 实例；换进程或新进程时，内存检查点不会自动保留，需要 Postgres 等持久化后端。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
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.3,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)

flags = {"fail_joke": True}


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_change_topic(state: InputState) -> OverAllState:
    return {"topic": f"{state['topic']}:布偶猫"}


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


def node_joke(state: OverAllState) -> OverAllState:
    if flags["fail_joke"]:
        raise RuntimeError("人为抛异常：笑话节点失败")
    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_change_topic", node_change_topic)
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_change_topic")
builder.add_edge("node_change_topic", "node_poem")
builder.add_edge("node_change_topic", "node_joke")
builder.add_edge("node_poem", "node_output")
builder.add_edge("node_joke", "node_output")
builder.add_edge("node_output", END)

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "recover-demo"}}

try:
    graph.invoke({"topic": "猫"}, config=config)
except RuntimeError as exc:
    rprint(f"首次失败: {exc}")
    snap = graph.get_state(config)
    rprint("已写入 keys:", [k for k in snap.values if snap.values.get(k)])
    rprint("next:", snap.next)

flags["fail_joke"] = False
rprint(graph.invoke(None, config=config))

排障流程通常是：get_state_history → 定位失败超步 → 修逻辑或数据 → invoke(None)。
跨进程恢复请改用 Postgres 等持久化 Saver，而不是新建空的 InMemorySaver。

Replay 回放

回放：选中历史里某个检查点的 config，再 invoke(None, config=该检查点)。
输入填 None 表示「不要新用户输入，从该快照的 next 继续」。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
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.3,
    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


topics = ["布偶猫", "狸花猫", "金渐层"]
topic_index = 0


def node_change_topic(state: InputState) -> OverAllState:
    global topic_index
    sub = topics[topic_index % len(topics)]
    topic_index += 1
    return {"topic": f"{state['topic']}:{sub}"}


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


def node_joke(state: OverAllState) -> 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_change_topic", node_change_topic)
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_change_topic")
builder.add_edge("node_change_topic", "node_poem")
builder.add_edge("node_change_topic", "node_joke")
builder.add_edge("node_poem", "node_output")
builder.add_edge("node_joke", "node_output")
builder.add_edge("node_output", END)

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "replay-demo"}}
rprint(graph.invoke({"topic": "猫咪"}, config=config))

history = list(graph.get_state_history(config=config))
target = next(h for h in history if h.next == ("node_poem", "node_joke"))
rprint("回放到:", target.next)
rprint(graph.invoke(None, config=target.config))

回放会再次执行 next 指向的节点，适合复现或对比生成结果。
带副作用的节点（写库、扣费）回放前要自己做幂等。

Fork 分叉

分叉：在某一历史点用 update_state 改状态，再 invoke(None) 走出另一条时间线。
as_node 表示「这些更新视为刚从该节点执行完」，从而影响后续 next。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
import os
from typing import Literal, 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,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)


class OverAllState(TypedDict):
    username: str
    user_input: str
    topic: str
    mode: Literal["poem", "joke"]
    output: str


def router_node(state: OverAllState) -> OverAllState:
    text = state["user_input"]
    if "笑话" in text:
        return {"topic": "荷花", "mode": "joke"}
    return {"topic": "荷花", "mode": "poem"}


def route(state: OverAllState) -> Literal["node_poem", "node_joke"]:
    return "node_joke" if state.get("mode") == "joke" else "node_poem"


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


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


builder = StateGraph(state_schema=OverAllState)
builder.add_node("router_node", router_node)
builder.add_node("node_poem", node_poem)
builder.add_node("node_joke", node_joke)
builder.add_edge(START, "router_node")
builder.add_conditional_edges(
    "router_node", route, path_map=["node_poem", "node_joke"]
)
builder.add_edge("node_poem", END)
builder.add_edge("node_joke", END)

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "fork-demo"}}
rprint(
    graph.invoke(
        {"username": "小王", "user_input": "写一首关于荷花的诗"},
        config=config,
    )
)

history = list(graph.get_state_history(config=config))
before_router = next(h for h in history if h.next == ("router_node",))

# 改输入后从 START 视角分叉
forked = graph.update_state(
    config=before_router.config,
    values={"user_input": "帮我写一个荷花的笑话"},
    as_node=START,
)
rprint(graph.invoke(None, config=forked))

# 跳过路由：假装 router 已产出 topic/mode
skip_router = graph.update_state(
    config=before_router.config,
    values={"topic": "狸花猫", "mode": "joke"},
    as_node="router_node",
)
rprint(graph.invoke(None, config=skip_router))

as_node=START 适合「改用户输入再整段重跑」。
as_node="router_node" 适合「跳过路由、直接指定下游所需字段」。

Store 对照

Checkpoint Store
作用域 单线程对话时间线 跨线程长期记忆
典型内容 messages、图中间态 用户偏好、资料档案
读取方式 get_state / history store.get / search
注入图 checkpointer= store= + runtime.store

下面用 InMemoryStore 写入用户偏好，并在节点里通过 Runtime 读取。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
import os
from typing import Final, TypedDict

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

load_dotenv()

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

USERS_NS: Final = ("users",)
PREFERENCES_KEY: Final = "preferences"

store = InMemoryStore()
store.put(
    (*USERS_NS, "Alice"),
    PREFERENCES_KEY,
    {"course": "组成原理", "sports": "跑步", "food": "酸奶"},
)


class OverAllState(MessagesState):
    username: str
    user_input: str
    output: str
    preferences: dict[str, str]


def check_preference_node(state: OverAllState, runtime: Runtime) -> OverAllState:
    item = runtime.store.get((*USERS_NS, state["username"]), PREFERENCES_KEY)
    if item is None:
        return {}
    return {"preferences": item.value}


def llm_node(state: OverAllState) -> OverAllState:
    preference = state.get("preferences", {})
    prompt = f"用户偏好:{preference}\n用户需求:{state['user_input']}"
    resp = model.invoke(
        [
            SystemMessage("请根据用户偏好简洁回复。"),
            HumanMessage(prompt),
        ]
    )
    return {"messages": [resp], "output": resp.content}


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

graph = builder.compile(checkpointer=InMemorySaver(), store=store)
config = {"configurable": {"thread_id": "store-demo"}}
rprint(
    graph.invoke(
        {"username": "Alice", "user_input": "推荐一下吃的"},
        config=config,
    )
)

同一用户换 thread_id 仍可通过 Store 读到偏好；Checkpoint 则只服务当前线程。
生产可用 PostgresStore；无数据库时优先 InMemoryStore，API（put / get / search）一致。

可选落库时安装：

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

再用 PostgresStore.from_conn_string(...).setup() 与 PostgresSaver 组合即可。

Runtime 上下文

有些数据既不该进 State（避免进检查点），也不适合放 Store（仅本次请求有效），可用 context_schema + invoke(..., context=...)。
节点通过 runtime.context 读取，例如会员等级、租户 ID。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
import os
from dataclasses import dataclass

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.runtime import Runtime
from rich import print as rprint

load_dotenv()

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


@dataclass
class UserContext:
    username: str
    membership_level: str


class OverAllState(MessagesState):
    user_input: str
    output: str


def llm_node(state: OverAllState, runtime: Runtime[UserContext]) -> OverAllState:
    ctx = runtime.context
    if ctx and ctx.membership_level == "VIP":
        system = (
            f"你是高级助理，当前 VIP 用户是{ctx.username}，"
            "请用『您』称呼，末尾加『VIP服务』。"
        )
    elif ctx:
        system = f"你是助理，当前用户是{ctx.username}，请简洁友好回复。"
    else:
        system = "你是助理，请简洁友好回复。"

    text = model.invoke(
        [SystemMessage(system), HumanMessage(state["user_input"])]
    ).content
    return {"output": text}


builder = StateGraph(state_schema=OverAllState, context_schema=UserContext)
builder.add_node("llm_node", llm_node)
builder.add_edge(START, "llm_node")
builder.add_edge("llm_node", END)
graph = builder.compile()

rprint(
    graph.invoke(
        {"user_input": "最近有什么优惠？"},
        context=UserContext(username="Alice", membership_level="VIP"),
    )
)
rprint(
    graph.invoke(
        {"user_input": "最近有什么优惠？"},
        context=UserContext(username="Bob", membership_level="普通用户"),
    )
)

context 按次注入，默认不进入 Checkpoint。
需要审计时可自行把关键字段拷进 State。

验证

  1. Replay：从 next == ("node_poem", "node_joke") 的检查点空输入续跑，应再次生成诗与笑话。
  2. Fork：改 user_input 含「笑话」后，输出应走笑话分支。
  3. Store：Alice 的回复应提到酸奶等已写入偏好。
  4. Context：VIP 与普通用户的语气/结尾应明显不同。

总结

  1. 恢复 / 回放：同一 checkpointer + invoke(None, config=检查点)。
  2. 分叉：update_state(..., as_node=...) 再续跑，走出新时间线。
  3. Store vs Checkpoint：长期跨线程用 Store；会话时间线用 Checkpoint。
  4. Runtime context：请求级元数据用 context_schema，避免污染状态快照。