《零基础 AI 入门》第 7 篇（收官）｜作者：梯度下降的咖啡师｜ 8 年 Java 后端（京东、字节）+ 2 年 AI 应用开发

上一篇我们用 LangGraph 把手写 Agent 重写了一遍，有了图结构、thread_id 隔离和内存级记忆。但当时留了两个尾巴：checkpointer 用的是 InMemorySaver，进程退了还是丢；工具还是全暴露给模型，危险操作没人拦。

今天把这两个尾巴收了。加的东西不多——一处换一行，一处加一个函数调用——但你的 Agent 从"能跑"变成"敢给人用"了。这篇写完，零基础入门系列就收官，下一站是 RAG。

01 从内存到 SQLite：换一行，重启不丢

上一篇最后一行是：

1
checkpointer = InMemorySaver()

InMemorySaver 就是个 Python dict，进程一退全没。换成 SQLite 只需要改两行：

1
2
3
4
from langgraph.checkpoint.sqlite import SqliteSaver

checkpointer = SqliteSaver.from_conn_string("agent.db")
graph = builder.compile(checkpointer=checkpointer)

先装依赖：pip install langgraph-checkpoint-sqlite。

跑起来你会发现目录下多了个 agent.db 文件——就是个 SQLite 数据库，每一步状态快照都存在里面。你 Ctrl+C 退出，重新跑，用同一个 thread_id invoke，它自动从数据库加载上次的 messages，接着聊。

有个细节要说明：实际项目里建议用 with SqliteSaver.from_conn_string("agent.db") as checkpointer: 包一层，退出时自动关连接。教学里为了简洁直接赋值，你自己写的时候加上 with。另外第一次跑之前调一次 checkpointer.setup()，它会自动建好表，后面就不用管了。

就这么简单。你上一篇手写时想了半天的"什么时候存、崩了怎么对齐"，LangGraph 在每个节点执行完自动写一行 SQLite，不用你管。

其他后端也一样，换个 Saver 类就行，拼装方式完全不变：

SQLite（本地开发）：from langgraph.checkpoint.sqlite import SqliteSaver，pip install langgraph-checkpoint-sqlite
PostgreSQL（生产首选）：from langgraph.checkpoint.postgres import PostgresSaver，pip install langgraph-checkpoint-postgres
MongoDB：from langgraph.checkpoint.mongodb import MongoDBSaver，pip install langgraph-checkpoint-mongodb
AWS DynamoDB / Valkey：langgraph-checkpoint-aws 包

Redis 不做 checkpoint（单线程状态快照），它做的是跨线程长期记忆（Store），后面讲到 RAG 和用户画像时再说。MySQL 目前没有官方 checkpointer，生产环境一般直接上 PostgreSQL。

02 为什么需要人工刹车

第五篇体检时说过：你的 tools 列表全暴露给模型了。查天气、查 AQI 是只读的，问题不大。但如果你的工具里有 send_email()、delete_user()——模型被一句话诱导就真的会去调。

Java 后端对这个味道太熟了：写操作接口不能裸奔，必须在关键节点加审批。Agent 也一样——高风险操作不能让模型自己决定，得暂停下来等人点"确认"。

LangGraph 里这个能力叫 Interrupt。你在节点函数里调一个 interrupt()，图就暂停在这一步，把当前状态返回给你。你看完决定继续还是拒绝，再调一次 graph.invoke(Command(resume="approve"), config)，图从暂停的地方接着跑。

03 给 tools_node 加审批

在第六篇代码基础上，加一个需要审批的工具 send_email，然后改 tools_node：

 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
from langgraph.types import interrupt, Command

# 加一个"危险"工具
def send_email(to: str, subject: str, body: str) -> str:
    return f"邮件已发送给 {to}：{subject}"

tool_map = {"get_weather": get_weather, "get_aqi": get_aqi, "send_email": send_email}

# 哪些工具需要人工审批
DANGEROUS_TOOLS = {"send_email", "delete_user", "transfer_money"}

def tools_node(state: AgentState):
    last = state["messages"][-1]
    results = []
    for tc in last.tool_calls:
        # 危险工具，先暂停等人审批
        if tc.name in DANGEROUS_TOOLS:
            decision = interrupt({
                "tool": tc.name,
                "args": tc.args,
                "message": f"Agent 想执行 {tc.name}({tc.args})，批准吗？"
            })
            # 用 Command(resume=...) 恢复时，interrupt() 的返回值就是 resume 传的值
            # 注意：恢复后从这行继续往下走，不是从 tools_node 开头重跑，已经执行过的工具不会再调一遍
            if decision != "approve":
                result = "用户拒绝了这个操作"
            else:
                result = tool_map[tc.name](**tc.args)
        else:
            # 只读工具，直接执行
            result = tool_map[tc.name](**tc.args)

        results.append(ToolMessage(content=result, tool_call_id=tc.id))
    return {"messages": results}

运行方式和之前不一样了——第一次 invoke 遇到 interrupt 会暂停：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
config = {"configurable": {"thread_id": "user-001"}}

# 第一次跑，遇到 send_email 会暂停
result = graph.invoke(
    {"messages": [HumanMessage(content="帮我发个邮件给 boss@company.com，说这周周报写完了")]},
    config
)

# 图停在了 interrupt()，这里能拿到审批请求
print(result["__interrupt__"])
# 输出：Interrupt(value={'tool': 'send_email', 'args': {...}, 'message': 'Agent 想执行...'}, resumable=True)

# 你看完觉得没问题，批准
graph.invoke(Command(resume="approve"), config)

# 如果不批准，传 "reject" 就行，tools_node 里会走到"用户拒绝了这个操作"

你看，审批逻辑全在 tools_node 里——遇到危险工具名就 interrupt，安全工具直接跑。你不用自己写 while 循环等输入，不用自己管理暂停状态，框架帮你存好了。而且这个暂停状态也被 checkpointer 写进了 SQLite——你重启进程，用同一个 thread_id 调 Command(resume="approve")，它还能从上次暂停的地方接着审批。

调试的时候你会看到 interrupt 里的 args 是一坨 JSON——模型吐回来的参数结构，格式不对的时候丢到 ForJSON（forjson.com）里格式化看一眼，比在终端 print 清楚。

04 完整代码长什么样

把前面拼起来，基于第六篇改了三处：

checkpointer 从 InMemorySaver 换成 SqliteSaver
加了一个 send_email 工具和 DANGEROUS_TOOLS 集合
tools_node 里加了 interrupt 审批逻辑

其他部分——State 定义、agent_node、条件边、图的拼装——和第六篇一模一样，不用动。这就是为什么前面让你手写再重写：你现在看每一行改动都知道它在解决什么问题，不是在抄一个陌生框架的 API。

05 零基础入门，到这里就够了

回头看这个系列走了多远：

篇数	你学会了什么
第 1 篇	Java 转 Python 的 6 个坑，大模型就是个字符串函数
第 2 篇	Function Calling + while 循环 = 你的第一个 Agent
第 3 篇	加记忆：滑窗剪枝和摘要压缩，别让它聊两句就失忆
第 4 篇	生产体检：持久化、多用户隔离、工具权限，手写版差什么
第 5 篇	LangGraph 重写：State/Node/Edge/Checkpointer/thread_id
第 6 篇	SQLite 持久化 + Interrupt 人工审批，敢给人用了

到这里，你已经能独立搭一个单 Agent：多轮记忆、多用户隔离、持久化、危险操作审批——这些生产 Agent 该有的东西你都摸过一遍了。再往后就是在这个骨架上长肉：接什么工具、连什么数据、做什么场景。

下一个工具不是查天气了。我们要给 Agent 接一个你自己的知识库——上传几个 PDF，问它里面的内容，它自己去检索相关片段再回答。这就是 RAG，《Agent 特调》栏目的第一篇。

旧炉未冷，新火已燃。关注公众号「梯度下降的咖啡师」，或访问 gradcoffee.com。下一篇开始做 RAG：怎么把你的文档变成 Agent 能查的工具。

聊聊两个问题：

① 你现在最想给 Agent 接的第一个真实工具是什么？

② Interrupt 人工审批这个设计，你觉得在你的场景里是刚需还是过度设计？

咖啡续命，梯度下降。｜作者：梯度下降的咖啡师

01 从内存到 SQLite：换一行，重启不丢#

02 为什么需要人工刹车#

03 给 tools_node 加审批#

04 完整代码长什么样#

05 零基础入门，到这里就够了#

01 从内存到 SQLite：换一行，重启不丢

02 为什么需要人工刹车

03 给 tools_node 加审批

04 完整代码长什么样

05 零基础入门，到这里就够了