坐标体系:thread_id / checkpoint_ns / checkpoint_id 与超步
前三天的 saver 全都围着一个三元组转:(thread_id, checkpoint_ns, checkpoint_id)。今天专门拆这套寻址坐标——它就像档案馆的"柜号-抽屉号-档案编号"。thread_id 隔离不同对话、checkpoint_ns 用 | 分隔符表达主图/子图的层级、checkpoint_id 用 uuid6(时间有序 UUID)保证单调递增。再往下钻一层:一个 超步(super-step)怎么对应一份 checkpoint,checkpoint 又是怎么在 pregel/_checkpoint.py 里被"造"出来的。搞懂坐标,时间旅行、子图隔离、恢复重放才不再是黑魔法。
· thread_id = 柜号:一个对话/一个用户会话一个柜子,柜子之间互不干扰。
· checkpoint_ns = 抽屉的层级路径:主图的档放最外层抽屉(
"" 空串),子图的档放"抽屉里的小格子",格子路径用 | 串起来(子图A|子子图B)。· checkpoint_id = 档案流水号:每存一份档给一个"越晚越大"的编号(uuid6),排一下就知道谁先谁后、能倒着往回翻(时间旅行)。
而"超步"就是"每开一次组会(所有该跑的节点跑一轮),就拍一张全家福存进抽屉"——一次超步 = 一份 checkpoint。
三元组:一份档案的完整地址
# 每次调用图时,你传的 config 长这样:
config = {"configurable": {
"thread_id": "chat-42", # 哪个对话/会话
"checkpoint_ns": "", # 哪一层图(主图=空串)
"checkpoint_id": "1efab...c3", # 哪一份档(不传则取最新)
}}
# saver 返回的"新坐标"也永远是这三件套(sqlite/__init__.py:437):
return {"configurable": {
"thread_id": thread_id,
"checkpoint_ns": checkpoint_ns,
"checkpoint_id": checkpoint["id"],
}}
三者缺一不可光有 thread_id 定位到"哪个对话"、加 ns 定位到"哪一层图"、再加 id 才定位到"这层图的哪一份档"。少任何一级都会歧义。checkpoint_id 可省略调用时只有 thread_id 是必须的。不给 checkpoint_id 时 saver 取"最新档"(Day 36 的 ORDER BY checkpoint_id DESC LIMIT 1);不给 checkpoint_ns 默认空串(主图)。put 返回带 id 的新坐标存完档,saver 把 config 里的 checkpoint_id 换成刚存的新档 id 返回——这样下一步就知道"当前站在哪份档上",也为下一份档提供父 id。thread_id:对话之间的隔离墙
thread_id 是三级里最外层、也是唯一必填的。基类 docstring 专门解释了它(checkpoint/base/__init__.py:182-198):
# checkpoint/base/__init__.py:186 附近(docstring)
# "The thread_id is the primary key used to store and retrieve checkpoints."
# 用法举例:
config = {"configurable": {"thread_id": "my-thread"}}
# - 对话记忆:同一个 thread_id 跨多次 invoke 复用 → 记得上下文
# - 不同用户/会话:用不同 thread_id → 状态互相隔离,谁也读不到谁
# saver 里 delete_thread 按 thread_id 整柜删除(sqlite/__init__.py:484)
def delete_thread(self, thread_id: str) -> None:
cur.execute("DELETE FROM checkpoints WHERE thread_id = ?", (thread_id,))
cur.execute("DELETE FROM writes WHERE thread_id = ?", (thread_id,))
thread_id = 隔离单位一个 thread_id 就是一条独立的"状态时间线"。同一个复用 → 有记忆;换一个 → 全新开始。这是 LangGraph 实现"多轮对话记忆"的根基。怎么选 thread_id 由你定docstring 说:对话记忆就复用同一个;多用户就一人一个(如 f"user-{uid}");一次性任务就每次随机一个。它是业务语义,框架不替你决定。delete_thread 整柜删删除以 thread_id 为单位——一条 SQL 干掉这个对话的所有档和半成品。对应"用户要求删除我的会话数据"这类需求。checkpoint_ns:用 | 和 : 编码图层级
中间那级 checkpoint_ns(namespace 命名空间)表达"这份档属于哪一层图"。它的分隔符定义在常量里(_internal/_constants.py:87-90):
# _internal/_constants.py:59
CONFIG_KEY_CHECKPOINT_NS = "checkpoint_ns" # "" 表示根图(主图)
# _internal/_constants.py:87
NS_SEP = "|" # 分隔各层级: graph | subgraph | subsubgraph
# _internal/_constants.py:89
NS_END = ":" # 每一层里,分隔"命名空间"和"任务 id"
# 例子:主图里有个子图节点叫 "planner",某次任务 id 是 abc:
# 主图的 checkpoint_ns = ""
# 子图 planner 的 ns = "planner:abc"
# 子图里再套子图 "search" = "planner:abc|search:def"
主图 ns = 空串 ""最外层主图的所有 checkpoint,checkpoint_ns 都是空字符串。这就是为什么你平时不传 ns 也能跑——默认 "" 就是主图。NS_SEP = "|"层级分隔符:子图套子图时,各层的命名空间用 | 串起来,像文件路径的 /。a|b|c 表示三层嵌套。NS_END = ":"每一层内部,用 : 把"子图节点名"和"该次调用的任务 id"分开。为什么要带任务 id?因为同一个子图可能被 Send 并行调用多次(Day 16 map-reduce),每次是独立的一份状态,靠任务 id 区分。checkpoint_ns 字符串,让所有层级的 checkpoint 共享同一 thread_id、存在同一张表里,靠 ns 前缀区分——恢复主图时能顺藤摸瓜找到所有子图的 checkpoint。用一个字符串字段承载"树形层级 + 并行实例 id",是拿"字符串解析的小复杂度"换"统一存储模型"的漂亮取舍。checkpoint_id:为什么是 uuid6 而非普通 UUID
最内层 checkpoint_id 用的不是常见的 uuid4,而是 uuid6——一种按时间排序的 UUID(pregel/_checkpoint.py:28-36、checkpoint/base/id.py:79-109):
# pregel/_checkpoint.py:28
def empty_checkpoint() -> Checkpoint:
return Checkpoint(
v=LATEST_VERSION,
id=str(uuid6(clock_seq=-2)), # ← 用 uuid6 生成 id
ts=datetime.now(timezone.utc).isoformat(),
channel_values={}, channel_versions={}, versions_seen={},
)
# checkpoint/base/id.py:79 uuid6 的核心:时间戳放在 UUID 高位
def uuid6(node=None, clock_seq=None):
nanoseconds = time.time_ns()
timestamp = nanoseconds // 100 + 0x01B21DD213814000
if _last_v6_timestamp is not None and timestamp <= _last_v6_timestamp:
timestamp = _last_v6_timestamp + 1 # ← 同一纳秒内也强制递增,保证单调
...
uuid_int = time_high_and_time_mid << 80 # 时间戳占据最高位
uuid_int |= time_low_and_version << 64
Checkpoint 的 id 字段Day 33 讲的 Checkpoint TypedDict 里,id 的文档就写着(checkpoint/base/__init__.py:97-101)"both unique and monotonically increasing, so can be used for sorting"——既唯一又单调递增,可直接排序。uuid6 时间戳在高位普通 uuid4 是纯随机、无序。uuid6 把时间戳放在 UUID 的最高位——于是"越晚生成的 id 字符串越大",字典序排序 = 时间序排序。这正是 ORDER BY checkpoint_id DESC 能取到"最新档"的根本原因(Day 36)。timestamp <= last → +1关键的单调保证:即使同一纳秒生成两个 id,也强制把后一个的时间戳 +1,确保严格递增、绝不相等。持久执行/时间旅行都依赖"id 能严格排序"。clock_seq=stepcreate_checkpoint 里生成 id 时传 clock_seq=step(pregel/_checkpoint.py:209)——把超步序号也编进去,进一步降低碰撞、隐含步序信息。last+1 补上了单机内的严格递增。超步 → checkpoint:create_checkpoint 造档
坐标讲完,看 checkpoint 本身怎么"诞生"。每个超步(super-step,Day 19-20)结束都会造一份新 checkpoint,主力函数是 create_checkpoint(pregel/_checkpoint.py:149-214):
# pregel/_checkpoint.py:149
def create_checkpoint(checkpoint, channels, step, *, id=None,
updated_channels=None, get_next_version=None,
channels_to_snapshot=None) -> Checkpoint:
ts = datetime.now(timezone.utc).isoformat()
...
values = {}
channel_versions = dict(checkpoint["channel_versions"])
for k in channels:
if k not in channel_versions:
continue
ch = channels[k]
...
v = ch.checkpoint() # ← 逐个通道调 checkpoint() 取当前存档值
if v is not MISSING:
values[k] = v # MISSING(如 UntrackedValue)就不收进来
return Checkpoint(
v=LATEST_VERSION,
ts=ts,
id=id or str(uuid6(clock_seq=step)), # 用超步号造 id
channel_values=values,
channel_versions=channel_versions,
versions_seen=checkpoint["versions_seen"],
updated_channels=None if updated_channels is None else sorted(updated_channels),
)
一超步 → 一次 create_checkpointPregel 每跑完一个超步(一批节点并行执行 + 写入通道),就调一次这个函数给当前全体通道状态拍张快照。所以 checkpoint 链的每一环对应一个超步。for k in channels: ch.checkpoint()核心:遍历每个通道、调它的 checkpoint() 方法(阶段 5 反复讲的那个)收集存档值。这里正是"每个通道自主决定存什么"落地的地方——Topic 交列表、Untracked 交 MISSING、Delta 交哨兵。v is not MISSING 才收呼应 Day 31:返回 MISSING 的通道(UntrackedValue)不会被收进 channel_values——它就这样从存档里"消失"了。这一行代码把 Day 31 的"不持久化"变成现实。id = uuid6(clock_seq=step)用超步序号 step 当 clock_seq 造 id,把"这是第几个超步"编进 id。checkpoint() 方法,就是在这里被超步循环逐个调用、汇总成一份 channel_values 的。通道负责"我这一格存什么",create_checkpoint 负责"把所有格子的存档值收拢成一张全家福",saver(Day 35-37)再负责"把全家福写进数据库"。三层职责一条链,到这里彻底闭环。checkpoint 里还有什么:版本号与 versions_seen
checkpoint 存的不止 channel_values。看 Checkpoint TypedDict 全貌(checkpoint/base/__init__.py:92-123):
# checkpoint/base/__init__.py:92
class Checkpoint(TypedDict):
v: int # 格式版本号(当前 LATEST_VERSION=4)
id: str # uuid6,单调递增(L04)
ts: str # ISO 8601 时间戳
channel_values: dict[str, Any] # 各通道的存档值(L05 收集来的)
channel_versions: ChannelVersions # 各通道当前的版本号("通道→版本")
versions_seen: dict[str, ChannelVersions] # 每个节点"看过"各通道到哪个版本
updated_channels: list[str] | None # 这一步更新了哪些通道
channel_versions"通道名 → 当前版本号"的映射。Day 35/37 里 saver 就是靠它去 blobs 表捞对应版本的通道值。版本号就是 Day 35 讲的"32 位数字.16 位随机"。versions_seen(灵魂字段)"节点 → (它已经看过每个通道到哪个版本)"。引擎靠它决定下一个超步该触发哪些节点:如果某通道的当前版本 > 某节点已看过的版本,说明"有它没消费的新数据"→ 触发它。这是 Pregel 判断"谁该跑"的核心依据。updated_channels记录"这一步哪些通道被更新了",用于优化和调试——不用对比整个 versions 就知道本步动了啥。v: int 格式版本checkpoint 结构本身也会演进(当前 v=4)。存下格式版本号,读回时能判断"这是老格式还是新格式",做兼容转换(Day 39 会看到 v < 4 的迁移处理)。家谱链:parent_checkpoint_id 串起时间线 + 小结
最后一块拼图:checkpoint 之间靠 parent_checkpoint_id 连成一条链,这就是时间旅行的轨道。回看 saver 里怎么串(checkpoint/sqlite/__init__.py:431):
# put 时:新档的父 = 当前 config 里携带的 checkpoint_id
config["configurable"].get("checkpoint_id") # 这就是"上一份档"的 id → 当父
# get_tuple 时:把父 id 组装成 parent_config 返回(sqlite/__init__.py:278-288)
parent_config = ({"configurable": {"thread_id": ..., "checkpoint_ns": ...,
"checkpoint_id": parent_checkpoint_id}}
if parent_checkpoint_id else None) # 第一份档没有父 → None
父 = 写入时的当前 id存新档时,config 里带的 checkpoint_id 正是"刚才那份档",拿它当父。一步步累积,就串成 档1 ← 档2 ← 档3 … 的单向链。parent_config 让你能往回走读档时返回父坐标,get_state_history(Day 44)就靠顺着 parent 一路往回翻,实现"查看/回到任意历史步"。第一份档 parent=None链的起点(empty_checkpoint 之后的首个真档)没有父,标记 None——遍历历史到这就到头了。🧠 今日小结自测
- 唯一定位一份 checkpoint 需要哪三个坐标?(thread_id / checkpoint_ns / checkpoint_id)
- checkpoint_ns 里 | 和 : 分别分隔什么?(| 分图层级、: 分"子图节点名 : 任务 id",主图为空串)
- 为什么 checkpoint_id 用 uuid6 而非 uuid4/自增整数?(时间戳在高位保证单调有序可排序,且纯本地生成无需中心协调)
- 一个超步和 checkpoint 是什么关系?(一超步结束 create_checkpoint 造一份档,遍历通道 checkpoint() 收集值)
- 返回 MISSING 的通道会怎样?(不被收进 channel_values,即 Day 31 的"不持久化"落地)
- versions_seen 干什么用?(记每个节点消费到各通道哪个版本,和 channel_versions 一比决定下步触发谁)
- 时间旅行靠什么串起来?(parent_checkpoint_id 家谱链,第一份档 parent=None)
✋ 10 分钟动手
# 1. Checkpoint 结构 + uuid6 生成
sed -n '92,123p' libs/checkpoint/langgraph/checkpoint/base/__init__.py
sed -n '79,109p' libs/checkpoint/langgraph/checkpoint/base/id.py
# 2. create_checkpoint 怎么遍历通道造档
sed -n '149,214p' libs/langgraph/langgraph/pregel/_checkpoint.py
# 3. 亲手观察坐标与家谱链
python - <<'PY'
from langgraph.graph import StateGraph, START
from langgraph.checkpoint.memory import InMemorySaver
g=StateGraph(dict); g.add_node("a",lambda s:{"n":s.get("n",0)+1}); g.add_edge(START,"a")
app=g.compile(checkpointer=InMemorySaver())
cfg={"configurable":{"thread_id":"t1"}}
app.invoke({"n":0}, cfg); app.invoke({"n":0}, cfg)
for st in app.get_state_history(cfg):
print("id=", st.config["configurable"]["checkpoint_id"][:8],
"parent=", (st.parent_config or {}).get("configurable",{}).get("checkpoint_id","None")[:8])
PY
dumps_typed / loads_typed——把 checkpoint 变成字节、再变回来。明天钻进 JsonPlusSerializer:看它怎么用 ormsgpack 高效编码、遇到 datetime/UUID/Pydantic 模型/numpy 数组等"非 JSON 原生类型"如何用 Ext 扩展编解码、_DeltaSnapshot 这种自定义类型怎么带类型标记往返,以及 LANGGRAPH_STRICT_MSGPACK 的安全考量——为什么"能反序列化任意类型"是把双刃剑。