LangGraph 接入 Redis 持久化：保存多 Agent 状态与恢复人工确认流程
编程Agent 流程里有四个角色：产品经理澄清需求，规划 Agent 拆任务，开发 Agent 写代码，审查 Agent 检查结果。
最开始遇到需求不清楚的情况，就让模型补全后继续执行。后来发现，前面猜错一个要求，后面的任务拆分和代码实现都会跟着偏。比如用户只说“加一个导出功能”，导出格式、哪些角色能用都还没确定，开发 Agent 已经开始写代码了。
所以我把需求确认加进流程：遇到需要用户补充的信息，先保存当前进度，把问题返回给前端。用户回答后，再继续执行后面的节点。
审查退回则通过条件路由回到开发节点，带着审查意见继续修改。只有需要用户选择方案或补充信息时，才暂停等待。
这篇记录 Redis Checkpointer 的接入方式。下面将持久化实现和确认流程整理成两个文件，方便单独运行：redis_saver.py 和 example.py。
代码依赖以下版本，示例状态使用字符串、字典和 LangChain 消息等可序列化数据：
langgraph==1.2.6
langgraph-checkpoint==4.1.0
redis==7.4.0
一、保存哪些内容，恢复时怎么找到
LangGraph 会在执行步骤结束时保存 checkpoint，其中包括当前状态、channel 版本和继续执行所需的信息。并行执行的节点可能属于同一个 super-step。
节点任务产生的写入还会通过 put_writes 单独保存。例如同一步里两个任务并行执行，一个成功、另一个失败，恢复时可以利用已经保存的任务结果，继续处理失败的任务。LangGraph 持久化说明
异步调用需要实现下面四个入口：
| 方法 | 处理内容 |
|---|---|
aput |
保存 checkpoint、metadata 和父 checkpoint ID |
aput_writes |
保存任务写入，供恢复时使用 |
aget_tuple |
根据配置读取指定 checkpoint，或读取当前最新的一份 |
alist |
倒序查询历史，支持 metadata 筛选、before 和 limit |
调用方通过这三个字段定位状态：
| 字段 | 在项目中的作用 |
|---|---|
thread_id |
一次开发会话的标识，发起和恢复请求使用同一个值 |
checkpoint_ns |
区分主图和子图等命名空间，主图通常为空字符串 |
checkpoint_id |
LangGraph 生成的快照 ID，保存时直接使用 checkpoint["id"] |
前端请求恢复时，后端先检查当前用户是否拥有这个会话，再使用对应的 thread_id。会话 ID 和用户权限分别处理。
二、用 Hash 保存数据，用 ZSET 查询历史
这里按 (thread_id, checkpoint_ns) 组织数据，每个 namespace 使用两个 Redis key：
lg-ckpt:{scope}:data Hash
c:<checkpoint_id> 快照、metadata 和父 checkpoint ID
w:<checkpoint_id> 该 checkpoint 下的任务写入
lg-ckpt:{scope}:index ZSET
score = 0
member = checkpoint_id
scope 是两个标识组成的 JSON 数组经过 URL-safe Base64 编码后的结果。这样标识里即使带有冒号，也能区分各个字段。Base64 用于编码，原始标识仍可还原。
ZSET 的 score 统一使用 0，同分成员按 ID 字典序排列。LangGraph Checkpoint 接口约定 ID 单调递增，这里直接按该约定获取最新快照。重试写入旧 checkpoint 时，它仍留在原来的历史位置。Checkpoint 接口源码
读取最新记录走当前 namespace 的索引。管理端查询跨会话历史时使用 SCAN；会话数量增加后，这类查询应限制范围或增加专门的管理索引。KEYS 会扫描整个数据库，请求处理链路里应避免使用。Redis KEYS 说明
过期策略也按 namespace 统一处理：
- 默认保留最近 128 份 checkpoint，超出后同时删除旧快照、任务写入和索引成员。
- 每次写入给数据与索引设置相同的到期时间，默认从最后一次写入起保留 24 小时。
- 查询操作保持原来的到期时间。
这样一个活跃会话的数据量受到历史条数限制。长时间没有继续执行的会话会整体过期。如果业务允许用户隔几天再回来确认，就相应延长 TTL；需要长期保留的需求和交付结果另存到业务数据库。
三、实现异步 Redis Checkpointer
任务写入通过 (task_id, index) 定位。普通 channel 的 index 使用本次 writes 中的位置，同一个标识重复提交时保留第一次结果；WRITES_IDX_MAP 中的保留 channel 使用固定负数索引，后续写入可以更新原值。
这部分沿用当前版本 InMemorySaver 的处理方式。task_path 一起保存，读取后组装成 CheckpointTuple.pending_writes 需要的 (task_id, channel, value) 三元组。InMemorySaver 实现
序列化使用 JsonPlusSerializer，并关闭 pickle 回退。状态里保存需求、消息、任务 ID 和文件路径，路由函数、模型客户端与 Redis 连接留在程序代码中。生成文件较大时，先保存文件，再把路径或对象存储地址写入状态。
下面的 Lua 脚本把数据写入、索引维护、历史裁剪和续期放在一次 Redis 执行中。每份 checkpoint 的任务写入存成一个 JSON 对象，里面保留首次写入顺序；类型和值由 serializer 编码后保存。
保存为 redis_saver.py：
import base64
import json
import re
from langgraph.checkpoint.base import (
BaseCheckpointSaver, CheckpointTuple, WRITES_IDX_MAP,
get_checkpoint_metadata,
)
from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
# 同一 namespace 的数据与索引一起续期；历史超过上限时一起删除。
SAVE = """
local data, index = KEYS[1], KEYS[2]
local mode, cid = ARGV[1], ARGV[2]
if mode == 'checkpoint' then
redis.call('HSET', data, 'c:' .. cid, ARGV[3])
else
local old = redis.call('HGET', data, 'w:' .. cid)
local obj = old and cjson.decode(old) or {items={}, order={}}
for _, row in ipairs(cjson.decode(ARGV[3])) do
local field, overwrite, value = row[1], row[2], row[3]
if obj.items[field] == nil then
table.insert(obj.order, field)
obj.items[field] = value
elseif overwrite then
obj.items[field] = value
end
end
redis.call('HSET', data, 'w:' .. cid, cjson.encode(obj))
end
redis.call('ZADD', index, 0, cid)
local excess = redis.call('ZCARD', index) - tonumber(ARGV[5])
if excess > 0 then
for _, old_id in ipairs(redis.call('ZRANGE', index, 0, excess - 1)) do
redis.call('HDEL', data, 'c:' .. old_id, 'w:' .. old_id)
redis.call('ZREM', index, old_id)
end
end
local now = redis.call('TIME')
local expires = now[1] * 1000 + math.floor(now[2] / 1000) + tonumber(ARGV[4])
redis.call('PEXPIREAT', data, expires)
redis.call('PEXPIREAT', index, expires)
return 1
"""
class RedisCheckpointSaver(BaseCheckpointSaver):
def __init__(self, client, *, prefix="lg-ckpt", ttl_seconds=86400,
max_history=128):
super().__init__(serde=JsonPlusSerializer(pickle_fallback=False))
if not re.fullmatch(r"[A-Za-z0-9_-]+", prefix):
raise ValueError("prefix 只允许字母、数字、下划线和连字符")
if ttl_seconds <= 0 or max_history < 2:
raise ValueError("ttl_seconds 必须大于 0，max_history 至少为 2")
self.client = client
self.prefix = prefix
self.ttl_ms = max(1, int(ttl_seconds * 1000))
self.max_history = max_history
def _scope(self, thread_id, namespace):
raw = json.dumps([thread_id, namespace], ensure_ascii=False).encode()
return base64.urlsafe_b64encode(raw).decode().rstrip("=")
def _keys(self, scope):
root = f"{self.prefix}:{{{scope}}}"
return f"{root}:data", f"{root}:index"
def _parts(self, config):
c = config["configurable"]
tid, ns = str(c["thread_id"]), c.get("checkpoint_ns", "")
return tid, ns, self._scope(tid, ns)
@staticmethod
def _config(tid, ns, cid):
return {"configurable": {
"thread_id": tid, "checkpoint_ns": ns, "checkpoint_id": cid,
}}
def _pack(self, value):
kind, payload = self.serde.dumps_typed(value)
return base64.b64encode(kind.encode() + b"\0" + payload).decode()
def _unpack(self, value):
kind, payload = base64.b64decode(value).split(b"\0", 1)
return self.serde.loads_typed((kind.decode(), payload))
async def _save(self, scope, mode, cid, payload):
await self.client.eval(
SAVE, 2, *self._keys(scope), mode, cid, payload,
self.ttl_ms, self.max_history,
)
async def aput(self, config, checkpoint, metadata, new_versions):
tid, ns, scope = self._parts(config)
cid = checkpoint["id"]
parent_id = config["configurable"].get("checkpoint_id")
# 保存完整快照；new_versions 保留在接口参数中。
record = (tid, ns, checkpoint,
get_checkpoint_metadata(config, metadata), parent_id)
await self._save(scope, "checkpoint", cid, self._pack(record))
return self._config(tid, ns, cid)
async def aput_writes(self, config, writes, task_id, task_path=""):
_, _, scope = self._parts(config)
cid = config["configurable"]["checkpoint_id"]
rows = []
for position, (channel, value) in enumerate(writes):
idx = WRITES_IDX_MAP.get(channel, position)
field = json.dumps([task_id, idx], separators=(",", ":"))
record = self._pack((task_id, channel, value, task_path))
rows.append([field, idx < 0, record])
if rows:
await self._save(scope, "writes", cid, json.dumps(rows))
async def _read(self, scope, cid):
data, _ = self._keys(scope)
raw, raw_writes = await self.client.hmget(data, f"c:{cid}", f"w:{cid}")
if raw is None:
return None
tid, ns, checkpoint, metadata, parent_id = self._unpack(raw)
pending = []
if raw_writes:
obj = json.loads(raw_writes)
for field in obj["order"]:
task_id, channel, value, _ = self._unpack(obj["items"][field])
pending.append((task_id, channel, value))
return CheckpointTuple(
config=self._config(tid, ns, cid), checkpoint=checkpoint,
metadata=metadata,
parent_config=self._config(tid, ns, parent_id) if parent_id else None,
pending_writes=pending,
)
async def aget_tuple(self, config):
_, _, scope = self._parts(config)
cid = config["configurable"].get("checkpoint_id")
if cid:
return await self._read(scope, cid)
_, index = self._keys(scope)
for raw_id in await self.client.zrevrange(index, 0, -1):
item = await self._read(scope, raw_id.decode())
if item is not None:
return item
return None
async def _scopes(self, config):
c = config["configurable"] if config else {}
if "thread_id" in c and "checkpoint_ns" in c:
yield self._parts(config)[2]
return
seen = set()
async for key in self.client.scan_iter(match=f"{self.prefix}:*:data"):
scope = key.decode()[len(self.prefix) + 2:-6]
if scope in seen:
continue
seen.add(scope)
raw = base64.urlsafe_b64decode(scope + "=" * (-len(scope) % 4))
tid, _ = json.loads(raw)
if "thread_id" not in c or tid == str(c["thread_id"]):
yield scope
async def alist(self, config, *, filter=None, before=None, limit=None):
if limit is not None and limit <= 0:
return
before_id = before["configurable"]["checkpoint_id"] if before else None
exact_id = (config or {}).get("configurable", {}).get("checkpoint_id")
candidates = []
async for scope in self._scopes(config):
_, index = self._keys(scope)
for raw_id in await self.client.zrevrange(index, 0, -1):
cid = raw_id.decode()
if before_id and cid >= before_id:
continue
if exact_id and cid != exact_id:
continue
candidates.append((cid, scope))
count = 0
for cid, scope in sorted(candidates, reverse=True):
item = await self._read(scope, cid)
if item is None:
continue
if filter and any(item.metadata.get(k) != v for k, v in filter.items()):
continue
yield item
count += 1
if limit is not None and count >= limit:
return
async def adelete_thread(self, thread_id):
config = {"configurable": {"thread_id": thread_id}}
async for scope in self._scopes(config):
await self.client.delete(*self._keys(scope))
这个实现保存完整 checkpoint，适合状态较小、需要保留近期执行记录的流程。new_versions 作为接口参数保留，channel 数据随快照一起保存。每次写入都会处理当前 checkpoint 的任务记录，使用时应控制单次任务的写入量和状态大小。
客户端使用 decode_responses=False，索引 ID 在读取时显式解码。示例把 serializer 输出转成 Base64 后放入记录，后续改为直接保存二进制时，也可以继续使用当前客户端配置。
所有 Redis 请求通过 redis.asyncio 执行，调用方对应使用 ainvoke、astream、aget_state 和 aget_state_history。如果项目还保留同步调用入口，需要另外实现同步 Saver 方法。redis-py 异步调用说明
四、需求确认后继续执行
为了看清暂停和恢复过程，下面将流程缩小成“确认需求 → 接收开发任务”两个节点。接回项目时，开发节点继续调用原来的开发 Agent。
确认节点通过 interrupt() 返回问题。前端收集用户输入后，后端使用同一个 thread_id，通过 Command(resume=...) 把答案送回图中。
保存为 example.py，与 redis_saver.py 放在同一目录：
import asyncio
import os
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from redis.asyncio import Redis
from redis_saver import RedisCheckpointSaver
class DevState(TypedDict, total=False):
requirement: str
confirmed_requirement: str
status: str
def build_workflow(saver):
def confirm_requirement(state: DevState):
answer = interrupt({
"question": "请补充导出格式和允许导出的角色",
"requirement": state["requirement"],
})
return {"confirmed_requirement": answer}
def code_development(state: DevState):
# 在项目中接入开发 Agent，这里只演示状态传递。
return {"status": f"已接收开发任务：{state['confirmed_requirement']}"}
workflow = StateGraph(DevState)
workflow.add_node("confirm_requirement", confirm_requirement)
workflow.add_node("code_development", code_development)
workflow.add_edge(START, "confirm_requirement")
workflow.add_edge("confirm_requirement", "code_development")
workflow.add_edge("code_development", END)
return workflow.compile(checkpointer=saver)
async def main():
config = {"configurable": {"thread_id": "dev-session-001"}}
client = Redis.from_url(os.environ["REDIS_URL"], decode_responses=False)
try:
saver = RedisCheckpointSaver(client)
app = build_workflow(saver)
first = await app.ainvoke(
{"requirement": "后台增加数据导出功能"}, config, durability="sync",
)
print(first["__interrupt__"][0].value)
finally:
await client.aclose()
# 用新的连接、Saver 和图模拟服务重新创建后的恢复请求。
client = Redis.from_url(os.environ["REDIS_URL"], decode_responses=False)
try:
app = build_workflow(RedisCheckpointSaver(client))
snapshot = await app.aget_state(config)
if not snapshot.values or not any(t.interrupts for t in snapshot.tasks):
raise RuntimeError("会话未处于等待确认状态，请刷新会话信息")
answer = "导出 CSV，仅管理员可用"
if not isinstance(answer, str) or not answer.strip():
raise ValueError("请填写需求确认内容")
final = await app.ainvoke(
Command(resume=answer.strip()), config, durability="sync",
)
print(final["status"])
finally:
await client.aclose()
if __name__ == "__main__":
asyncio.run(main())
先安装依赖，并设置 REDIS_URL。PowerShell 下可以这样运行：
python -m pip install langgraph==1.2.6 langgraph-checkpoint==4.1.0 redis==7.4.0
$env:REDIS_URL = "redis://127.0.0.1:6379/0"
python example.py
dev-session-001 用于单次演示；开始新的演示时换一个会话 ID。实际接口中，由后端创建会话并保存它与用户的关联关系。
第一次调用返回确认问题。第二次调用使用新建的连接、Saver 和图，恢复后可以取得用户确认的需求。示例中 code_development 只返回状态文本，便于检查数据是否传到了后续节点。
这里使用 durability="sync"，等待当前执行步骤的持久化完成后再继续推进图。Redis 写入失败时让异常返回给调用方，前端可以提示保存失败并保留重试入口。
动态中断恢复时，LangGraph 会从中断节点的开头重新执行，interrupt() 返回本次提交的答案。因此确认节点前半段只组织问题；发通知、扣减资源等外部操作放到单独节点，并通过业务请求 ID 处理重复执行。LangGraph 中断与恢复
如果只是调试时希望在开发节点前暂停，可以在编译时设置 interrupt_before=["code_development"]，之后传入 None 继续执行。文章中的需求确认需要接收用户输入，所以使用动态中断。
五、同一会话的请求按顺序处理
前端重复点击确认、接口超时后再次提交，都可能让同一个会话同时收到两个恢复请求。项目接入时，将同一个 thread_id 的运行请求交给同一个串行任务队列处理；有多个服务实例时，也要在共享调度层保证这一点。
处理确认请求的顺序是：
- 验证用户、会话归属和输入内容。
- 按会话顺序领取请求，用请求 ID 检查是否已经处理。
- 读取当前 checkpoint，检查是否仍在等待该次确认。
- 执行
Command(resume=...)，记录处理结果，再返回前端。
前端提交时还可以带上展示问题时对应的 checkpoint ID。后端在领取任务后将它与当前状态比较，旧页面提交的答案可以直接提示刷新。等待多个并行确认任务时，再按 interrupt ID 分别传递答案。
Lua 脚本负责一次存储操作中的数据一致性。同一会话的执行顺序由上面的请求调度负责。example.py 是串行演示，用户鉴权、请求去重和队列接入放在业务接口层实现。
六、恢复后检查状态和任务执行次数
验证持久化时，我主要看两类结果：用户回答是否传到了后面的节点，以及失败重试时已经成功的任务是否被重复执行。
下面保留一个简短的对照脚本。它使用同一张图，分别接入 Redis Saver 和 InMemorySaver，比较暂停问题与最终状态。Redis 部分通过模拟服务运行，便于本地重复验证。
安装 fakeredis[lua]==2.40.0，将下面内容保存为同目录的 test_resume.py：
import asyncio
import fakeredis
import fakeredis.aioredis
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
from example import build_workflow
from redis_saver import RedisCheckpointSaver
async def run(saver, restored_saver):
config = {"configurable": {"thread_id": "resume-test"}}
first = await build_workflow(saver).ainvoke(
{"requirement": "后台增加数据导出功能"}, config, durability="sync",
)
final = await build_workflow(restored_saver).ainvoke(
Command(resume="导出 CSV，仅管理员可用"), config, durability="sync",
)
return first["__interrupt__"][0].value, final
async def main():
server = fakeredis.FakeServer()
writer = fakeredis.aioredis.FakeRedis(server=server)
reader = fakeredis.aioredis.FakeRedis(server=server)
try:
actual = await run(RedisCheckpointSaver(writer), RedisCheckpointSaver(reader))
memory = InMemorySaver()
expected = await run(memory, memory)
assert actual == expected
print("中断问题、确认内容和最终状态一致")
finally:
await writer.aclose()
await reader.aclose()
asyncio.run(main())
应用进程重启后恢复，需要 Redis 中仍保留对应状态，同时使用兼容的图结构和代码。Redis 服务自身重启后的数据保留还取决于 AOF、RDB、持久化卷及淘汰策略，部署时要结合允许丢失的进度配置。会话等待时间在 TTL 的设置里进行设置。
七、 Agent 流程
产品经理节点先整理已经明确的需求，把需要用户回答的问题交给确认节点。收到答案后，再进入规划和开发节点。
审查 Agent 返回修改意见时，把意见写进状态，通过条件路由回到开发节点。再加上明确的修改次数限制，达到上限后把当前结果和待确认问题交给用户，避免同一段代码反复自动修改。
这份 Saver 用普通 Redis 数据结构保存近期状态，保留了按 namespace 续期和按条数裁剪的做法。后续如果需要更完整的 Redis 查询能力，也可以评估 Redis 维护的 langgraph-checkpoint-redis 集成；它使用 RedisJSON、RediSearch 等能力，需要先确认现有 Redis 的版本和模块配置，再按所选包版本完成初始化与迁移。Redis Checkpointer 项目说明
- 本文标签： Langchain Agent
- 本文链接： https://www.tianyajuanke.top/article/105
- 版权声明： 本文由吴沛芙原创发布，转载请遵循《署名-非商业性使用-相同方式共享 4.0 国际 (CC BY-NC-SA 4.0)》许可协议授权
