W6-D1/D2 · 给图装上存档系统：Checkpointer、thread_id 与 SqliteSaver

1
2

graph.invoke({"messages": [("user", "我叫小明，记住我")]})
graph.invoke({"messages": [("user", "我叫什么？")]})   # ❌ 不记得

1
2

无 checkpoint:  invoke → 跑 → 结果 → 【state 丢掉】
有 checkpoint:  invoke → 每个节点后存 checkpoint → 下次从存档续跑

1
2
3
4

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()                                  # ① 注意是实例，不是类名
graph = builder.compile(checkpointer=checkpointer)             # ② 编译时传入

1
2
3

START → agent ⇄ tools（循环）→ END
         │
         └─ 后台：每个节点跑完 → 按 thread_id 存一份 checkpoint

1
2

config = {"configurable": {"thread_id": "demo-1"}}   # 会话身份证
graph.invoke({"messages": [...]}, config)            # 每次 invoke 都要带

 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

"""W6-D1: 给 W5 的 ReAct 图接上 MemorySaver —— 让图记住过去"""
import os
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

# ── 1. 原样复用 W5-D3 的 ReAct 图 ──
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

@tool
def add(a: int, b: int) -> int:
    """计算两个整数之和"""
    return a + b

@tool
def multiply(a: int, b: int) -> int:
    """计算两数之积"""
    return a * b

llm = ChatOpenAI(
    model="deepseek-chat",
    base_url="https://api.deepseek.com",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
).bind_tools([add, multiply])

def agent_node(state: AgentState) -> dict:
    return {"messages": [llm.invoke(state["messages"])]}

# 工厂函数：同一份拓扑，想编译几次编译几次（下面要对照组）
def build():
    b = StateGraph(AgentState)
    b.add_node("agent", agent_node)
    b.add_node("tools", ToolNode([add, multiply]))
    b.add_edge(START, "agent")
    b.add_conditional_edges("agent", tools_condition)
    b.add_edge("tools", "agent")
    return b

# ── 2. 关键差异：同一张图，编译成两个版本 ──
graph_cp   = build().compile(checkpointer=MemorySaver())   # 有存档
graph_nocp = build().compile()                             # 无存档（对照组）

# ── 3. 验证 1：有 checkpoint + 同一 thread_id → 跨 invoke 记忆 ──
config = {"configurable": {"thread_id": "demo-1"}}
r1 = graph_cp.invoke({"messages": [("user", "我叫小明，记住我")]}, config)
print("第一次:", r1["messages"][-1].content)

r2 = graph_cp.invoke({"messages": [("user", "我叫什么名字？")]}, config)
print("第二次:", r2["messages"][-1].content)      # 应该能答出"小明"

# ── 4. 验证 2：读档确认——消息真的在累加（这是铁证，不是看回答猜）──
state = graph_cp.get_state(config)
print("消息总数:", len(state.values["messages"]))   # 期望 ≥ 4 = 2 次提问 + 2 次回答

# ── 5. 验证 3：换一个 thread_id → 开新档 → 失忆 ──
config2 = {"configurable": {"thread_id": "demo-2"}}         # 换槽位
r3 = graph_cp.invoke({"messages": [("user", "我叫什么名字？")]}, config2)
print("换 thread:", r3["messages"][-1].content)    # 答不出小明

# ── 6. 验证 4：无 checkpoint 的图，永远从零开始 ──
graph_nocp.invoke({"messages": [("user", "我叫小明")]})
g2 = graph_nocp.invoke({"messages": [("user", "我叫什么？")]})
print("无cp版消息数:", len(g2["messages"]))        # 期望 = 2（只有这一轮的提问+回答）

1
2

for event in graph_cp.stream({"messages": [("user", "再聊一句")]}, config):
    print("节点跑完:", list(event.keys()))

1
2

graph.invoke(..., {"configurable": {"thread_id": "alice-001"}})  → 读写 alice 的桶
graph.invoke(..., {"configurable": {"thread_id": "bob-001"}})    → 读写 bob 的桶

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

"""W6-D2：thread_id 多会话隔离 —— 一张图，两个用户，互不干扰"""
# （构图部分与 D1 完全相同：State / 工具 / agent_node / builder）

graph = builder.compile(checkpointer=MemorySaver())   # 编译时只传一次 checkpointer

alice = {"configurable": {"thread_id": "alice-001"}}
bob   = {"configurable": {"thread_id": "bob-001"}}

graph.invoke({"messages": [("user", "我叫爱丽丝，我喜欢猫")]}, alice)
graph.invoke({"messages": [("user", "我是鲍勃")]}, bob)

# 各自问自己的信息
r_a = graph.invoke({"messages": [("user", "我叫什么？我喜欢什么？")]}, alice)
r_b = graph.invoke({"messages": [("user", "我叫什么？我喜欢什么？")]}, bob)
print("Alice 记得:", r_a["messages"][-1].content)   # 应答出：爱丽丝 / 猫
print("Bob   记得:", r_b["messages"][-1].content)   # 应答出：鲍勃（不知道猫）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

"""W6-D2：SqliteSaver 落盘 + 重启恢复（save 与 load 分两次运行）"""
# （构图部分与前面完全相同，到 builder 为止，先不要 compile）

from langgraph.checkpoint.sqlite import SqliteSaver

def save():
    """第一次运行：写入会话"""
    with SqliteSaver.from_conn_string("checkpoints.db") as saver:
        graph = builder.compile(checkpointer=saver)
        cfg = {"configurable": {"thread_id": "persist-1"}}
        graph.invoke({"messages": [("user", "我叫小明，住在北京")]}, cfg)
        print("已写入会话")

def load():
    """关掉进程后再运行：读回存档"""
    with SqliteSaver.from_conn_string("checkpoints.db") as saver:
        graph = builder.compile(checkpointer=saver)
        cfg = {"configurable": {"thread_id": "persist-1"}}
        r = graph.invoke({"messages": [("user", "我叫什么？住哪？")]}, cfg)
        print("重启后回复:", r["messages"][-1].content)   # 应答出 小明 / 北京

if __name__ == "__main__":
    save()      # 第一次跑这个
    # load()    # 然后注释掉 save、打开这行，重启进程再跑

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

def inspect():
    with SqliteSaver.from_conn_string("checkpoints.db") as saver:
        graph = builder.compile(checkpointer=saver)
        cfg = {"configurable": {"thread_id": "persist-1"}}

        # ① get_state：读当前存档
        state = graph.get_state(cfg)
        print("消息总数:", len(state.values["messages"]))   # 应 > 2
        print("下一步节点:", state.next)                      # 正常跑完应为空

        # ② get_state_history：看存档时间线（新 → 旧）
        print("历史 checkpoint 时间线:")
        for i, snap in enumerate(graph.get_state_history(cfg)):
            cid = snap.config["configurable"]["checkpoint_id"]
            n = len(snap.values["messages"])
            print(f"  #{i}  checkpoint_id={cid[:8]}  消息数={n}")

        # ③ update_state：从图外面改档（人注入，不是节点返回）
        graph.update_state(cfg, {"messages": [("user", "顺便记住：我喜欢喝美式咖啡")]})

        # ④ 验证注入真的被 agent 感知
        r = graph.invoke({"messages": [("user", "我喜欢喝什么？")]}, cfg)
        print("注入后回复:", r["messages"][-1].content)   # 应答出 美式咖啡

1
2

节点 return {"messages": [...]}   →  内部改档：流水线工人自己填工单
graph.update_state(cfg, {...})    →  外部改档：车间主任从外面改工单

1
2
3

from langgraph.checkpoint.memory import MemorySaver      # 或 .sqlite 的 SqliteSaver
graph = builder.compile(checkpointer=MemorySaver())      # 编译期接入
graph.invoke(payload, {"configurable": {"thread_id": "会话ID"}})   # 运行期带槽位