两步：编译时给图一个 cache（缓存后端），add_node 时给节点一个 cache_policy：

from langgraph.graph import StateGraph, START, END
from langgraph.types import CachePolicy
from langgraph.cache.memory import InMemoryCache
from typing import TypedDict
import time

class S(TypedDict):
    x: int
    y: int

def slow(state: S) -> dict:
    time.sleep(2)                       # 假装很慢
    return {"y": state["x"] * 2}

g = StateGraph(S)
g.add_node("slow", slow, cache_policy=CachePolicy(ttl=120))   # ← 缓存 2 分钟
g.add_edge(START, "slow"); g.add_edge("slow", END)
app = g.compile(cache=InMemoryCache())                        # ← 挂缓存后端

app.invoke({"x": 5, "y": 0})            # 慢：真跑 2 秒
app.invoke({"x": 5, "y": 0})            # 快：命中缓存，瞬间返回

结构极简 types.py:518：

# types.py:518
@dataclass(**_DC_KWARGS)
class CachePolicy(Generic[KeyFuncT]):
    """Configuration for caching nodes."""
    key_func: KeyFuncT = default_cache_key   # 怎么把输入变成指纹
    ttl: int | None = None                    # 活多久（秒）；None = 永不过期

默认 key 函数用 pickle 把输入序列化成字节 _internal/_cache.py:26：

# _internal/_cache.py:26
def default_cache_key(*args, **kwargs) -> str | bytes:
    import pickle
    # protocol 5 strikes a good balance between speed and size
    return pickle.dumps((_freeze(args), _freeze(kwargs)), protocol=5, fix_imports=False)

关键在 _freeze——它把输入"规范化"，让"内容相同、顺序不同"的输入产生同一个指纹 _internal/_cache.py:7：

# _internal/_cache.py:7
def _freeze(obj, depth=10):
    if isinstance(obj, Hashable) or depth <= 0:
        return obj                                       # 已可哈希，直接用
    elif isinstance(obj, Mapping):
        # sort keys so {"a":1,"b":2} == {"b":2,"a":1}
        return tuple(sorted((k, _freeze(v, depth-1)) for k, v in obj.items()))
    elif isinstance(obj, Sequence):
        return tuple(_freeze(x, depth-1) for x in obj)
    elif hasattr(obj, "tobytes"):                        # numpy/pandas 等
        return (type(obj).__name__, obj.tobytes(),
                obj.shape if hasattr(obj, "shape") else None)
    return obj

L03 的指纹只是原料。真正存进缓存用的 key 是 CacheKey 三元组 types.py:615：

# types.py:615
class CacheKey(NamedTuple):
    ns: tuple[str, ...]   # 命名空间：区分是哪个节点的缓存
    key: str              # 摘要：输入指纹的哈希
    ttl: int | None       # 存活秒数

它在准备任务时生成 pregel/_algo.py:669：

# pregel/_algo.py:669
if cache_policy:
    args_key = cache_policy.key_func(val)               # ① 算输入指纹（L03）
    cache_key = CacheKey(
        (
            CACHE_NS_WRITES,                             # ② 固定前缀 "__pregel_ns_writes"
            (identifier(proc) or "__dynamic__"),        #    这个节点的身份（模块.函数名）
            name,                                        #    节点名
        ),
        xxh3_128_hexdigest(                             # ③ 把大指纹哈希成短摘要
            args_key.encode() if isinstance(args_key, str) else args_key,
        ),
        cache_policy.ttl,                                # ④ TTL
    )
else:
    cache_key = None

缓存后端的契约 cache/base/__init__.py:15——四对方法（同步+异步）：

# cache/base/__init__.py:15
class BaseCache(ABC, Generic[ValueT]):
    serde: SerializerProtocol = JsonPlusSerializer(pickle_fallback=False)
    @abstractmethod
    def get(self, keys: Sequence[FullKey]) -> dict[FullKey, ValueT]: ...      # 批量取
    @abstractmethod
    def set(self, pairs: Mapping[FullKey, tuple[ValueT, int | None]]) -> None: ... # 批量存(值,TTL)
    @abstractmethod
    def clear(self, namespaces: Sequence[Namespace] | None = None) -> None: ...    # 按 ns 清

内存实现里 TTL 就是"存的时候算出过期时间戳，取的时候比一下现在" cache/memory/__init__.py:38：

# cache/memory/__init__.py:38 (set)
now = datetime.datetime.now(datetime.timezone.utc)
for (ns, key), (value, ttl) in keys.items():
    if ttl is not None:
        expiry = (now + datetime.timedelta(seconds=ttl)).timestamp()   # 现在+TTL
    else:
        expiry = None
    self._cache[ns][key] = (*self.serde.dumps_typed(value), expiry)    # 存: 序列化值 + 过期戳

# cache/memory/__init__.py:17 (get)
now = datetime.datetime.now(datetime.timezone.utc).timestamp()
...
enc, val, expiry = self._cache[ns][key]
if expiry is None or now < expiry:              # 没过期
    values[(ns, key)] = self.serde.loads_typed((enc, val))
else:
    del self._cache[ns][key]                    # 过期 → 顺手删掉（惰性清理）

命中：执行前，引擎拿所有"有 cache_key 且还没写"的任务去 cache 批量查，命中的直接把 writes 灌回去 pregel/_loop.py:1549：

# pregel/_loop.py:1549
def match_cached_writes(self) -> Sequence[PregelExecutableTask]:
    if self.cache is None:
        return ()
    matched = []
    if cached := {(t.cache_key.ns, t.cache_key.key): t
                  for t in self.tasks.values()
                  if t.cache_key and not t.writes}:        # 有 key 且尚未有写
        for key, values in self.cache.get(tuple(cached)).items():
            task = cached[key]
            task.writes.extend(values)                     # 命中 → 把缓存的 writes 填进任务
            matched.append(task)
    return matched

写入：任务跑完 put_writes 时，若它可缓存，就把 writes 存进 cache pregel/_loop.py:1609：

# pregel/_loop.py:1609
def put_writes(self, task_id, writes):
    super().put_writes(task_id, writes)
    if not writes or self.cache is None or not hasattr(self, "tasks"):
        return
    task = self.tasks.get(task_id)
    if task is None or task.cache_key is None:
        return
    self.submit(self.cache.set, {
        (task.cache_key.ns, task.cache_key.key): (task.writes, task.cache_key.ttl)
    })

✋ 10 分钟动手

# 1. 读四段核心
sed -n '7,31p'     libs/langgraph/langgraph/_internal/_cache.py   # _freeze + default_cache_key
sed -n '668,687p'  libs/langgraph/langgraph/pregel/_algo.py       # CacheKey 生成
sed -n '17,57p'    libs/checkpoint/langgraph/cache/memory/__init__.py  # 内存 TTL
sed -n '1549,1625p' libs/langgraph/langgraph/pregel/_loop.py      # 命中 + 写入

# 2. 亲手看命中：数节点执行次数 + 计时
python - <<'PY'
import time
from langgraph.graph import StateGraph, START, END
from langgraph.types import CachePolicy
from langgraph.cache.memory import InMemoryCache
from typing import TypedDict
class S(TypedDict):
    x: int; y: int
runs = {"n": 0}
def slow(s):
    runs["n"] += 1; time.sleep(1); return {"y": s["x"] * 2}
g = StateGraph(S); g.add_node("slow", slow, cache_policy=CachePolicy(ttl=60))
g.add_edge(START,"slow"); g.add_edge("slow",END)
app = g.compile(cache=InMemoryCache())
t=time.time(); print(app.invoke({"x":5,"y":0}), round(time.time()-t,2))  # ~1s
t=time.time(); print(app.invoke({"x":5,"y":0}), round(time.time()-t,2))  # ~0s 命中
print("slow 执行次数:", runs["n"])                                        # 1
PY