4.2 状态的序列化与反序列化

4.2 状态的序列化与反序列化

状态怎么存进数据库

本节在全册位置：持久化的前提是可序列化。消息对象、工具结果都要能转成字节再还原。LangGraph 默认用对 LangChain 对象友好的序列化器；自定义状态字段需保证其类型可被序列化，否则检查点写入会失败。我们踩过的坑：把模型实例塞进状态，序列化直接炸。

机制：默认序列化器与自定义

compile(checkpointer=...) 时，状态在每次更新后被序列化进存储。消息用 LangChain 自带序列化；普通 dict/list/基础类型走标准 JSON 类编码。若状态含不可序列化对象（如打开的文件句柄），应在节点内转换为 id 或路径。类比金融：序列化像对账单据的标准报文，双方都能解。序列化器稳定，检查点才能跨进程回放。

from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class S(TypedDict):
    msgs: list

def a(s: S):
    return {"msgs": s["msgs"] + ["hi"]}

b = StateGraph(S)
b.add_node("a", a)
b.add_edge(START, "a")
b.add_edge("a", END)
#    4.2 状态的序列化与反序列化
saver = MemorySaver()
g = b.compile(checkpointer=saver)
print(g.invoke({"msgs": []}, config={"configurable": {"thread_id": "t1"}}))
​

# 含自定义对象时，提供可序列化表示
class Ctx:
    def __init__(self, path): self.path = path
# 保证状态字段都是基本类型或 LangChain 自带对象
​

案例：存不住的检查点

背景：状态里放了模型实例，写检查点报错，本地能跑线上崩。

操作：把模型移到节点闭包/全局，状态只留模型名与参数。

结果：检查点变轻且可写，序列化不再报错。

解读：模型是「逻辑」不是「状态」，不该进状态，这是序列化失败的最高频原因。

变式：需要换模型时，把模型名作为状态字段，运行期按名取实例，兼顾灵活与可序列化。

工程清单

• 编译时传 checkpointer，状态每次更新后被序列化进存储，序列化器对 LangChain 对象友好。

• 消息用 LangChain 自带序列化，自定义对象要保证可序列化，否则检查点写入会失败。

• 不可序列化对象（文件句柄等）在节点内转成 id 或路径，别让状态持有活资源。

常见误区

把模型实例写进状态，生成检查点时把大模型也序列化，既慢又可能泄露；模型是逻辑不是状态，节点闭包里持有即可，状态只留名字。

自定义字段的可序列化

当状态要存非 LangChain 对象，最稳的是存「可序列化的引用」而非对象本身，节点内再重建。

# 状态只存路径，节点内重建重对象
class S(TypedDict):
    doc_path: str        # 而非打开的文件句柄
    rows: list

def load(s):
    with open(s["doc_path"]) as f:   # 用时再开，不持有活资源
        return {"rows": f.readlines()}

# 若必须存复杂对象，转 dict 而非存实例
class Point:
    def __init__(self, x, y): self.x, self.y = x, y
    def to_dict(self): return {"x": self.x, "y": self.y}
# 状态存 to_dict() 结果，读取时 from_dict()
​

⚠️ 常见坑：为了方便把数据库连接塞进状态，检查点写入直接失败或把连接序列化成一堆废字节；连接这类活资源永远用外部持有 + 状态存 id。

💡 关键直觉：状态是「数据快照」不是「运行环境」；能重建的对象就别存对象本身。

自定义对象的序列化实现

状态里确实需要自定义对象时，有两种实现路线：对象提供 to_dict/from_dict 的成对方法，或直接用 dataclass 再转 dict。

# 路线一：成对方法，结构可控
class Point:
    def __init__(self, x, y): self.x, self.y = x, y
    def to_dict(self): return {"x": self.x, "y": self.y}
    @classmethod
    def from_dict(cls, d): return cls(d["x"], d["y"])

# 节点里进出都走成对方法，状态永远存 dict
def node(s):
    raw = {"pt": Point(1, 2).to_dict()}
    return raw

# 读取时再重建
def read(s):
    pt = Point.from_dict(s["pt"])
​

# 路线二：dataclass 一键转 dict，少写样板
from dataclasses import dataclass, asdict
@dataclass
class Point:
    x: int
    y: int
# 状态存 asdict(p)，读取时 Point(**d)
​

两条路线都遵循同一纪律：状态里只存在「能变成基础类型」的数据，重建逻辑收在类自身。测试时给每个自定义类写一个 round-trip 用例（存进去再读出来，断言相等），序列化问题就永远不会悄悄上线。

检查点写入失败的排查顺序

「invoke 正常，一配 checkpointer 就报错」是最典型的序列化事故。按顺序排查：

1. 检查状态 schema 里有没有活资源（连接、句柄、模型实例），有就改成 id。
2. 检查自定义类是否被直接放入状态，是就改成 dict。
3. 检查消息对象是否混用了不兼容的类型（自定义消息子类也要可序列化）。
4. 检查是不是写了超大字段导致超时，而不是真的序列化失败。
5. 最后确认 checkpointer 依赖的数据库连接正常，存储层故障常被误判成序列化问题。

其中 1 和 2 占了九成。把「状态字段类型表」（上一节的表格）贴在代码旁，写节点时对照一眼，比事后排错省事。

字段演进与旧检查点兼容

状态 schema 改版是迟早的事，旧检查点里的历史数据如何兼容，取决于你动了什么：

兼容的通用策略是「读取时兜底」：节点取字段用 .get(默认值)，而不是直接下标访问，这样旧轨迹也能跑。长期不用的旧 thread 允许直接失效，别为兼容拖累新代码——给数据定一个保留期，过期清理，检查点库才不会越滚越大。