# serde/jsonplus.py:82
class JsonPlusSerializer(SerializerProtocol):
    """Serializer that uses ormsgpack, with optional fallbacks.

    !!! warning
        Security note: ... It should not be used on untrusted python objects.
        If an attacker can write directly to your checkpoint database,
        they may be able to trigger code execution when data is deserialized.
        Set LANGGRAPH_STRICT_MSGPACK=true to restrict deserialization
        to a built-in allowlist of safe types.
    """

注意 saver 里存的从来不是"光秃秃的字节"，而是 (type, bytes) 二元组。看 dumps_typed（serde/jsonplus.py:258-271）：

# serde/jsonplus.py:258
def dumps_typed(self, obj: Any) -> tuple[str, bytes]:
    if obj is None:
        return "null", EMPTY_BYTES          # ① None → 标记 "null" + 空字节
    elif isinstance(obj, bytes):
        return "bytes", obj                  # ② 本来就是字节 → 标记 "bytes"，原样
    elif isinstance(obj, bytearray):
        return "bytearray", obj
    else:
        try:
            return "msgpack", _msgpack_enc(obj)   # ③ 其他一律走 msgpack 编码
        except ormsgpack.MsgpackEncodeError as exc:
            if self.pickle_fallback:
                return "pickle", pickle.dumps(obj)  # ④ 编不了 & 开了兜底 → pickle
            raise exc

loads_typed 就是 dumps 的镜像——按 type 标记选对应解码路径（serde/jsonplus.py:273-290）：

# serde/jsonplus.py:273
def loads_typed(self, data: tuple[str, bytes]) -> Any:
    type_, data_ = data
    if type_ == "null":
        return None
    elif type_ == "bytes":
        return data_
    elif type_ == "bytearray":
        return bytearray(data_)
    elif type_ == "json":                     # ← 老格式：msgpack 迁移前的 JSON
        return json.loads(data_, object_hook=self._reviver)
    elif type_ == "msgpack":
        return ormsgpack.unpackb(
            data_, ext_hook=self._unpack_ext_hook,     # ★ 遇到 Ext 扩展类型时的回调
            option=ormsgpack.OPT_NON_STR_KEYS)          # 允许非字符串 key
    elif self.pickle_fallback and type_ == "pickle":
        return pickle.loads(data_)
    else:
        raise NotImplementedError(f"Unknown serialization type: {type_}")

msgpack 本身也只认基本类型。碰到 datetime、UUID、Pydantic、numpy、以及 LangGraph 自己的 _DeltaSnapshot，靠一个 default 钩子把它们编成带类型码的 Ext（serde/jsonplus.py:295-307,859-860）：

# serde/jsonplus.py:295  —— 每种非原生类型分配一个整数"类型码"
EXT_CONSTRUCTOR_SINGLE_ARG = 0    # 单参数构造（如 UUID(str)）
EXT_CONSTRUCTOR_POS_ARGS   = 1    # 位置参数构造
EXT_CONSTRUCTOR_KW_ARGS    = 2    # 关键字参数构造
EXT_METHOD_SINGLE_ARG      = 3
EXT_PYDANTIC_V1            = 4    # Pydantic v1 模型
EXT_PYDANTIC_V2            = 5    # Pydantic v2 模型
EXT_NUMPY_ARRAY           = 6    # numpy 数组
EXT_DELTA_SNAPSHOT        = 7    # LangGraph 的 DeltaChannel 快照

# serde/jsonplus.py:305  编码钩子：ormsgpack 遇到不认识的对象就调它
def _msgpack_default(obj: Any) -> str | ormsgpack.Ext:
    if isinstance(obj, _DeltaSnapshot):
        return ormsgpack.Ext(EXT_DELTA_SNAPSHOT, _msgpack_enc(obj.value))
    # ... 往下依次判断 Pydantic v2 / v1 / datetime / UUID / numpy / 各种类型 ...

# serde/jsonplus.py:859  编码入口：把 _msgpack_default 作为 default 传给 packb
def _msgpack_enc(data: Any) -> bytes:
    return ormsgpack.packb(data, default=_msgpack_default, option=_option)

解码时，msgpack 每碰到一个 Ext 就调 ext_hook，按类型码重建对象（serde/jsonplus.py:633-666）：

# serde/jsonplus.py:633
def ext_hook(code: int, data: bytes) -> Any:
    if code == EXT_DELTA_SNAPSHOT:
        return _DeltaSnapshot(
            ormsgpack.unpackb(data, ext_hook=ext_hook, option=ormsgpack.OPT_NON_STR_KEYS))
    elif code == EXT_CONSTRUCTOR_SINGLE_ARG:
        try:
            tup = ormsgpack.unpackb(data, ext_hook=ext_hook, ...)  # tup = (模块, 类名, 参数)
            if not _check_allowed(tup[0], tup[1]):     # ① 安全检查：这个类允许 import 吗？
                return tup[2]                          #    不允许 → 只还原原始数据，不构造对象
            return getattr(importlib.import_module(tup[0]), tup[1])(tup[2])   # ② import 并构造
        except Exception:
            return None                                # ③ 构造失败 → 返回 None，不崩
    elif code == EXT_CONSTRUCTOR_POS_ARGS:
        ...
        if tup[0] == "langgraph.types" and tup[1] == "Send":
            return _send_from_args(tup[2])             # 特殊类型走专门重建
        return getattr(importlib.import_module(tup[0]), tup[1])(*tup[2])

回到那条 docstring 警告。默认行为其实是"宽松"的，安全靠一个环境变量收紧（serde/jsonplus.py:97-119）：

# serde/jsonplus.py:107
if allowed_msgpack_modules is _lg_msgpack._SENTINEL:
    if _lg_msgpack.STRICT_MSGPACK_ENABLED:       # 环境变量 LANGGRAPH_STRICT_MSGPACK=true
        allowed_msgpack_modules = None           # 严格：只允许内置的 SAFE_MSGPACK_TYPES
    else:
        # 宽松（默认）：所有类型都允许（带一条 warning）
        allowed_msgpack_modules = True

把今天散落的兼容/降级点收拢——一个生产序列化器要扛住"格式演进"和"数据不完美"：

✋ 10 分钟动手

# 1. dumps_typed / loads_typed 对照读
sed -n '258,290p' libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py

# 2. Ext 类型码 + 编码/解码钩子
sed -n '295,307p' libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py
sed -n '633,666p' libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py

# 3. 亲手看非原生类型的往返
python - <<'PY'
from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
from datetime import datetime, timezone
from uuid import uuid4
s = JsonPlusSerializer()
obj = {"when": datetime.now(timezone.utc), "id": uuid4(), "nums": [1,2,3]}
t, b = s.dumps_typed(obj)
print("格式标记:", t, "| 字节数:", len(b))
back = s.loads_typed((t, b))
print("还原类型:", type(back["when"]).__name__, type(back["id"]).__name__)  # datetime UUID
PY