类定义在 memory/__init__.py:33，三个属性是全部家当（memory/__init__.py:69-83）：

# memory/__init__.py:33
class InMemorySaver(BaseCheckpointSaver[str], AbstractContextManager, AbstractAsyncContextManager):
    # thread ID -> checkpoint NS -> checkpoint ID -> (checkpoint, metadata, parent_id)
    storage: defaultdict[str, dict[str, dict[str, tuple[...]]]]
    # (thread ID, checkpoint NS, checkpoint ID) -> {(task_id, write_idx): (task_id, channel, value, path)}
    writes: defaultdict[tuple[str, str, str], dict[tuple[str, int], tuple[...]]]
    # (thread id, checkpoint ns, channel, version) -> (type, bytes)   ← 通道值单独存
    blobs: dict[tuple[str, str, str, str | int | float], tuple[str, bytes]]

构造函数（memory/__init__.py:85）用 factory(lambda: defaultdict(dict)) 建了这套嵌套 defaultdict——好处是 storage[新thread][新ns] 直接可用、不用先判断 key 存不存在。

memory/__init__.py:427，只看核心几行：

# memory/__init__.py:448
c = checkpoint.copy()
thread_id = config["configurable"]["thread_id"]
checkpoint_ns = config["configurable"]["checkpoint_ns"]
values: dict[str, Any] = c.pop("channel_values")     # ① 把通道值从档里抠出来
for k, v in new_versions.items():                    # ② 只遍历“这次变了版本”的通道
    self.blobs[(thread_id, checkpoint_ns, k, v)] = (
        self.serde.dumps_typed(values[k]) if k in values else ("empty", b"")
    )
self.storage[thread_id][checkpoint_ns].update({      # ③ 档本体(已无通道值)存 storage
    checkpoint["id"]: (
        self.serde.dumps_typed(c),                                    # 序列化档
        self.serde.dumps_typed(get_checkpoint_metadata(config, metadata)),  # 序列化元数据
        config["configurable"].get("checkpoint_id"),                  # 父 id
    )
})
return {"configurable": {"thread_id": thread_id, "checkpoint_ns": checkpoint_ns,
                         "checkpoint_id": checkpoint["id"]}}

memory/__init__.py:236，看"没给 checkpoint_id → 取最新档"分支（:281）：

# memory/__init__.py:282
if checkpoints := self.storage[thread_id][checkpoint_ns]:
    checkpoint_id = max(checkpoints.keys())          # ① id 单调递增 → max 就是最新
    checkpoint, metadata, parent_checkpoint_id = checkpoints[checkpoint_id]
    writes = self.writes[(thread_id, checkpoint_ns, checkpoint_id)].values()
    checkpoint_ = self.serde.loads_typed(checkpoint) # 反序列化档(此时无通道值)
    return CheckpointTuple(
        config={"configurable": {"thread_id": thread_id, "checkpoint_ns": checkpoint_ns,
                                 "checkpoint_id": checkpoint_id}},
        checkpoint={
            **checkpoint_,
            "channel_values": self._load_blobs(       # ② 按版本把通道值捞回来拼上
                thread_id, checkpoint_ns, checkpoint_["channel_versions"]),
        },
        metadata=self.serde.loads_typed(metadata),
        pending_writes=[(id, c, self.serde.loads_typed(v)) for id, c, v, _ in writes],
        parent_config=({"configurable": {...parent_checkpoint_id...}} if parent_checkpoint_id else None),
    )

拼装的关键辅助 memory/__init__.py:125：

# memory/__init__.py:125
def _load_blobs(self, thread_id, checkpoint_ns, versions) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for k, ver in versions.items():             # 遍历 档 里的 channel_versions
        kk = (thread_id, checkpoint_ns, k, ver)
        if kk not in self.blobs:                # ← 边界：这个版本没存 blob，跳过
            continue
        vv = self.blobs[kk]
        if vv[0] == "empty":                    # ← 边界：哨兵值，代表“这通道当时是空的”
            continue
        result[k] = self.serde.loads_typed(vv)  # 反序列化，塞进结果
    return result

memory/__init__.py:473，核心循环：

# memory/__init__.py:494
thread_id = config["configurable"]["thread_id"]
checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
checkpoint_id = config["configurable"]["checkpoint_id"]
outer_key = (thread_id, checkpoint_ns, checkpoint_id)
outer_writes_ = self.writes.get(outer_key)
for idx, (c, v) in enumerate(writes):
    inner_key = (task_id, WRITES_IDX_MAP.get(c, idx))    # 普通写用 idx，特殊写用负数
    if inner_key[1] >= 0 and outer_writes_ and inner_key in outer_writes_:
        continue                                          # ← 幂等：普通写已存在就跳过
    self.writes[outer_key][inner_key] = (task_id, c, self.serde.dumps_typed(v), task_path)

D34 说基类默认版本是整数、字符串版本要子类自己实现。InMemorySaver 就重写了它 memory/__init__.py:619：

# memory/__init__.py:619
def get_next_version(self, current: str | None, channel: None) -> str:
    if current is None:
        current_v = 0
    elif isinstance(current, int):
        current_v = current
    else:
        current_v = int(current.split(".")[0])   # 从 "0003.xxxx" 里取出整数部分
    next_v = current_v + 1
    next_h = random.random()                       # 16 位随机小数
    return f"{next_v:032}.{next_h:016}"            # 例：000...0004.0.5123908471

D34 说异步方法要两套。但 InMemorySaver 没有真 IO（就是操作内存字典），所以异步版直接调同步版 memory/__init__.py:529：

# memory/__init__.py:529
async def aget_tuple(self, config): return self.get_tuple(config)
# :543
async def alist(self, config, *, filter=None, before=None, limit=None):
    for item in self.list(config, filter=filter, before=before, limit=limit):
        yield item
# :565
async def aput(self, config, checkpoint, metadata, new_versions):
    return self.put(config, checkpoint, metadata, new_versions)

✋ 10 分钟动手

# 1. put / get_tuple / _load_blobs 三段对照读
sed -n '427,472p' libs/checkpoint/langgraph/checkpoint/memory/__init__.py
sed -n '125,141p' libs/checkpoint/langgraph/checkpoint/memory/__init__.py

# 2. 亲手看 blobs 去重：跑两步只改一个字段，数 blobs 大小
python - <<'PY'
from langgraph.graph import StateGraph, START
from langgraph.checkpoint.memory import InMemorySaver
from typing import Annotated
import operator
class S(dict): pass
g=StateGraph(dict); g.add_node("a",lambda s:{"n":s.get("n",0)+1})
g.add_edge(START,"a"); saver=InMemorySaver(); app=g.compile(checkpointer=saver)
cfg={"configurable":{"thread_id":"t"}}
app.invoke({"n":0,"fixed":"never changes"}, cfg)
print("blobs 键:", list(saver.blobs.keys()))
print("版本号样例:", saver.get_next_version(None, None))
PY