已完成第1–15天app/graph_agent.py中的基础图可以完成：

decide → tool → decide → finalize

这一阶段不再追求增加更多工具，而是让现有 Agent 能够：

• 保存每一步状态；说起来，
• 进程重启后恢复；
• 高风险操作先人工审批；
• 防止重复执行副作用；话说回来，
• 向客户端流式展示节点进度；
• 正确处理临时故障；
• 连接标准MCP工具；
• 追踪并保护线上请求。怎么说呢，

从第16天来看。Checkpointer、thread_id与断点恢复

今天必须理解

1. State是图运行时数据，Checkpointer负责保存每一步 State。说起来，
2. thread_id是一次可恢复会话的持久游标。不等于user_id。
3. 内存 Checkpointer只能用于测试，进程退出数据会丢失。
4. Checkpoint不是业务数据库，订单、工单等业务数据仍需独立存储。
5. 恢复可能重新执行某些节点，所以副作用必须幂等。

User Pain Point：开发者往往在开发阶段忘记将状态持久化，导致调试失效或生产环境崩溃。User Pain Point：缺乏 thread_id 概念导致跨线程或跨服务的数据冲突。

第一步先这方面，安装 SQLite Checkpointer

pip install langgraph-checkpoint-sqlite
# 在 requirements.txt 追加：
langgraph-checkpoint-sqlite

从接下来来看，让图支持注入 Checkpointer

`app/graph_agent.py` 的构图函数改造如下：

from typing import Any
def build_graph:
builder = StateGraph
builder.add_node
builder.add_node
builder.add_node
builder.add_edge
builder.add_conditional_edges(
"decide",route_after_decide,{
"tool"的观点是。"tool","finalize": "finalize",},)
builder.add_edge
builder.add_edge
# 不传checkpointer时仍可作为普通图运行
return builder.compile
# 保留原来的无持久化图
agent_graph = build_graph

：创建持久化图

`app/persistence.py`：

import sqlite3
from pathlib import Path
from langgraph.checkpoint.sqlite import SqliteSaver
from app.graph_agent import build_graph
CHECKPOINT_PATH = Path
CHECKPOINT_PATH.parent.mkdir
# check_same_thread=False允许 FastAPI 工作线程使用同一连接。checkpoint_connection = sqlite3.connect(
CHECKPOINT_PATH,check_same_thread=False,)
sqlite_checkpointer = SqliteSaver
persistent_graph = build_graph
def thread_config -> dict:
if not thread_id.strip:
raise ValueError
return {"configurable": {"thread_id": thread_id}}

至于第四步，运行并查看 Checkpoint

`checkpoint_demo.py`：

from app.graph_agent import initial_state
from app.persistence import persistent_graph。thread_config
def main -> None:
config = thread_config
result = persistent_graph.invoke(
initial_state,config=config,)
print
print
for snapshot in persistent_graph.get_state_history:
print(
f"step={snapshot.metadata.get}",f"next={snapshot.next}",f"tool_round={snapshot.values.get}",)
if __name__ == "__main__":
main

User Pain Point：没有直观的状态历史查看接口，导致难以定位错误。User Pain Point：单一文件的 Checkpoint 很难在多进程或分布式环境中共享。

第五步的观点是，写 thread_id 测试

`tests/test_persistence.py`：

import pytest
from app.persistence import thread_config
def test_thread_config:
assert thread_config == {"configurable": {"thread_id": "abc"}}
def test_thread_id_cannot_be_empty:
with pytest.raises:
thread_config

故障练习 

• 使用 "a" 运行一次接下来退出 Python 进程。
• 从启动并调用来看，

  config = thread_config snapshot = persistent_graph.get_state print }

  确认状态仍存在。随后改用 "b" 确认不会读取线程

至于第17天，Human-in--Loop 与 幂等写工具

• 查询、计算等只读操作可以自动执行。
• 发邮件、创建工单、删除数据等写操作应该先审批。
• interrupt\t暂停图并返回待审批信息。按理说，
• 恢复时节点会从头运行。而不是从 interrupt\t 下一行继续。
• 幂等键保证同一业务请求重复执行仍只有一个结果。