1. 本章目标 #

第 4 章用手写 history 列表做过多轮对话；第 10 章为 HITL、限流已经用过 checkpointer + thread_id，但那时重点是「暂停与恢复」，不是「怎么设计会话记忆」。

真实客服、办公助理、Copilot 类产品几乎都要回答：

本章要补的能力分三块：

块 解决什么 对应节
持久化 多轮状态存哪、怎么恢复 §3～§5
隔离 谁的历史算谁的 §4
控长 窗口满了怎么裁 / 怎么摘要 §6～§7

本章目标：

用 checkpointer 与 thread 管理 Agent 短期记忆，并学会裁剪 / 摘要策略，做出带多轮记忆的客服 Agent。

学完你应能：

记忆这一层的典型症状是「代码没报错，但产品像失忆」。下面几条是本章实测出来的、和直觉不太一样的结论，先看一眼再往下读：

你可能以为 实际情况 见
每轮重复传整段 history 会导致消息翻倍 传消息对象不会（按 id 去重）；自己重建成 dict 才会真重复 §3.2
SummarizationMiddleware 装上就会摘要 trigger 没达到就完全不工作，示例里常配得太高以致从未触发 §7.3
摘要出来的是中文要点 默认 summary_prompt 是英文的，还带 ## ARTIFACTS 这类写给编程 Agent 的字段 §7.3
系统提示在 messages[0] 里 用 system_prompt= 注入时不进 messages，首条通常是第一轮 Human §7.1
忘传 state_schema 会报错 静默丢弃自定义槽位，get_state 里只剩 messages §7.4.5
摘要只是省点 token 实测连聊 8 轮：不摘要 33 条 vs 摘要 12 条 §8.3

前置依赖： 第 4 章（消息类型）· 第 8 章（工具）· 第 10 章（checkpointer 在 HITL/限流中的用法，可先扫一眼 §5、§7.2）。

参考文档：

2. 短期记忆是什么 #

上一章的客服 Agent 若每次 invoke 都不带 checkpointer，用户第二轮问「刚才那个订单」时，模型看不到第一轮——不是模型笨，是状态没接上。本章要解决的，就是这类「同一场对话内的连续性」。

官方把 Memory 分成两层，不要混：

类型 范围 典型存什么 本文
短期记忆 单个 thread（一条会话线）内 对话 messages、中间件计数、HITL 暂停点 本章
长期记忆 跨 thread / 跨会话 用户偏好、档案、向量知识库 第 16 章（官方 long-term memory）；文档型知识库另见第 12～15 章

用一个客服场景区分两者：

用户说 需要的记忆类型 为什么
「刚才那个 A1002 到哪了？」 短期（本 thread 历史） 上一轮刚查过，在 messages 里
「我上次退货为什么被拒？」（上周另一场聊天） 长期（跨 thread / 档案） 当前 thread 里没有上周对话
「退换货政策是什么？」 外部知识（RAG，第 12 章起） 不在对话里，在文档 / 向量库

短期记忆最常见形态就是 conversation history（对话历史）：
SystemMessage → HumanMessage → AIMessage → ToolMessage → … 随轮次增长（见第 4 章）。

LangChain Agent 把短期记忆放进图状态里的 messages 键，由 checkpointer 按 thread_id 持久化：

用户每发一条消息 → invoke
        │
Agent 从 checkpointer 读出该 thread 已有 state
        │
追加本轮 Human / AI / Tool …
        │
写回 checkpointer
        │
下次同一 thread_id 再 invoke → 自动带上前文

三个容易误解的点：

  1. 模型没有自带「记忆」——是你（或 checkpointer）每次把历史再喂给它。
  2. 短期记忆 ≠ 向量库——RAG 检索到的文档块是外部知识；messages 是「这场对话里说过什么」。
  3. 开了 checkpointer 不等于能无限聊——历史仍会撞上下文窗口，需要 §6～§7 的控长策略。
thread_id = "customer-001"
    │
    ├─ invoke 1 → messages: [H, A]
    ├─ invoke 2 → messages: [H, A, H, A, Tool, A]
    └─ invoke 3 → … 持续增长 → 需要 §7 控长

读本章时，把 checkpointer 想成「每场对话的记事本」：thread_id 是笔记本编号，messages 是写进去的内容。笔记本可以换（新 thread），纸可以撕（trim / 删除），也可以先誊写摘要再撕原件（Summarization）——但这些都只针对同一场对话。

3. checkpointer：会话状态的存取层 #

checkpointer 是 LangGraph 的检查点存储：每次 Agent 跑完一步（或一轮 invoke），会把当前图状态（至少含 messages）序列化存起来；下次同一 thread_id 再进来，先读再追加。

它和数据库 ORM 的直觉类似：你不必在业务层手写 SELECT history / INSERT message，驾驭层（create_agent）会在每次 invoke 边界替你完成读写。

3.1. 为什么需要 checkpointer #

第 4 章手写 history 时，你要自己：

create_agent(..., checkpointer=...) 把这些交给 LangGraph：状态读写 + thread 隔离 成为驾驭层（create_agent）的一等能力。

手写 history checkpointer
多轮追加 业务代码维护 Agent 自动合并
跨请求 自己序列化存储 checkpointer 负责
HITL / thread_limit 难 第 10 章已依赖
适用 单文件脚本 客服、助理、生产 Agent

什么时候仍用手写 history？只有单次脚本、无跨请求、无 HITL，且你完全掌控消息列表时。一旦要 Web 多轮、暂停恢复或限流累计，就应切到 checkpointer。

3.2. 最小多轮示例 #

下面两段 invoke 之间没有传 history 变量——第二轮能答「我是谁」，全靠 checkpointer 在幕后合并状态。读代码时盯住三处：checkpointer=、稳定的 thread_id、每轮只传一条新用户消息。

# 从 langchain.agents 导入创建 Agent 的工厂函数
from langchain.agents import create_agent
# InMemorySaver 是最简单的 checkpointer 实现（存在进程内存里）
from langgraph.checkpoint.memory import InMemorySaver
# 从 dotenv 导入环境变量加载函数
from dotenv import load_dotenv

# 加载 .env 中的 API key
load_dotenv(override=True)

# 创建 Agent，关键是把 checkpointer 传进去
agent = create_agent(
    # 模型标识
    model="deepseek:deepseek-v4-flash",
    # 本示例不需要工具
    tools=[],
    # 系统提示，设定角色
    system_prompt="你是简洁的中文客服助手。",
    # 有了它，多轮状态才会被存下来
    checkpointer=InMemorySaver(),
)

# thread_id 标识一条会话线；两轮用同一个 config 才能接上历史
config = {"configurable": {"thread_id": "customer-001"}}

# 第一轮：用户自报家门
agent.invoke(
    # 只传本轮这一句
    {"messages": [{"role": "user", "content": "你好，我是小王，订单 A1002 查过了吗？"}]},
    # 带上 thread_id
    config=config,
)
# 第二轮：只发新问题，不必手动拼 history
result = agent.invoke(
    # 注意这里没有任何历史消息
    {"messages": [{"role": "user", "content": "我刚才说我是谁？订单号多少？"}]},
    # 同一个 config，历史由 checkpointer 自动带上
    config=config,
)
# 打印最终回答
print(result["messages"][-1].content)

真实运行结果：

您刚才说您是"小王"，订单号是"A1002"。请问您具体想处理什么问题呢？

用 get_state 看这一刻 checkpoint 里存的东西（get_state 的用法见 §5）：

# 读取该 thread 的 checkpoint 快照，看真实存下来的是什么
snapshot = agent.get_state(config)

# snapshot.values 是一个字典，键就是 state 里的字段（至少有 messages）
print("\n----- checkpoint 里的完整轨迹 -----")
# 逐条打印，序号 + 消息类型 + 内容前 80 字
for i, msg in enumerate(snapshot.values["messages"]):
    print(i, type(msg).__name__, getattr(msg, "content", "")[:80])
条数=4
  [0] HumanMessage: '你好，我是小王，订单 A1002 查过了吗？'
  [1] AIMessage: '您好，小王。订单 A1002 已经查到了。请问您是想确认物流进度，还是需要处理售后问题呢？'
  [2] HumanMessage: '我刚才说我是谁？订单号多少？'
  [3] AIMessage: '您刚才说您是"小王"，订单号是"A1002"。'

两轮 4 条，干净地一问一答。顺便记住一个细节，后面 §7.1 调裁剪规则时会用到：

用 create_agent(system_prompt=...) 注入的系统提示，不会出现在 messages 里。 实测这份 messages 里没有任何 SystemMessage，messages[0] 就是第一轮的用户消息。

要点：

3.3 重传整段 #

「每轮重传整段 history」到底会不会重复？

先看把上一轮返回的 messages 原样再传一遍：


# 第一轮：用户自报家门
result = agent.invoke(
    # 只传本轮这一句
    {
        "messages": [
            {"role": "user", "content": "你好，我是小王，订单 A1002 查过了吗？"}
        ]
    },
    # 带上 thread_id
    config=config,
)
# 第二轮：只发新问题，不必手动拼 history
result = agent.invoke(
    # 注意这里没有任何历史消息
    {
        "messages": [
            *result["messages"],
            {"role": "user", "content": "我刚才说我是谁？订单号多少？"},
        ]
    },
    # 同一个 config，历史由 checkpointer 自动带上
    config=config,
)
# 打印最终回答
print(result["messages"][-1].content)

# 读取该 thread 的 checkpoint 快照，看真实存下来的是什么
snapshot = agent.get_state(config)

# snapshot.values 是一个字典，键就是 state 里的字段（至少有 messages）
print("\n----- checkpoint 里的完整轨迹 -----")
# 逐条打印，序号 + 消息类型 + 内容前 80 字
for i, msg in enumerate(snapshot.values["messages"]):
    print(i, msg.id, type(msg).__name__, getattr(msg, "content", "")[:80])

结果没有重复，state 里仍是 4 条：

[0] HumanMessage id=97fb573e-...: '我叫小王。'
[1] AIMessage   id=lc_run--019ffa52-...: '你好，小王！很高兴认识你。'
[2] HumanMessage id=456dfd51-...: '我叫什么？'
[3] AIMessage   id=lc_run--019ffa52-...: '你刚才告诉我你叫小王...'

原因是 messages 用的 add_messages reducer 会按消息 id 去重：已经存在的 id 视为「更新同一条」，而不是新增。上一轮返回的对象都带着 id，所以重传等于无操作。

真正会翻倍的是另一种写法——业务代码自己维护一个 dict 列表当历史。dict 里没有 id，框架只能当成新消息：

# 从 langchain.agents 导入 create Agent 的工厂函数
from langchain.agents import create_agent

# InMemorySaver 是最简单的 checkpointer 实现（存在进程内存里）
from langgraph.checkpoint.memory import InMemorySaver

# 从 dotenv 导入环境变量加载函数
from dotenv import load_dotenv

# 加载 .env 中的 API key
load_dotenv(override=True)

# 创建 Agent，关键是把 checkpointer 传进去
agent = create_agent(
    # 模型标识
    model="deepseek:deepseek-v4-flash",
    # 本示例不需要工具
    tools=[],
    # 系统提示，设定角色
    system_prompt="你是简洁的中文客服助手。",
    # 有了它，多轮状态才会被存下来
    checkpointer=InMemorySaver(),
)

# thread_id 标识一条会话线；两轮用同一个 config 才能接上历史
config = {"configurable": {"thread_id": "customer-dup-demo"}}

# 第一轮：先正常跑一轮，拿到助手回复
r = agent.invoke(
    # 只传本轮这一句
    {"messages": [{"role": "user", "content": "我叫小王。"}]},
    # 带上 thread_id
    config=config,
)

# 危险写法：手工把历史重建成 dict（没有 id）
manual_history = [
    # 复述用户上一句
    {"role": "user", "content": "我叫小王。"},
    # 复述助手上一句
    {"role": "assistant", "content": r["messages"][-1].content},
]

# 把重建的历史 + 新问题一起传进去
result = agent.invoke(
    # 前两条没有 id，框架只能当新消息处理
    {"messages": [*manual_history, {"role": "user", "content": "我叫什么？"}]},
    # 同一条会话线
    config=config,
)

# 打印最终回答
print(result["messages"][-1].content)

# 读取该 thread 的 checkpoint 快照，观察消息是否翻倍
snapshot = agent.get_state(config)
print("\n----- checkpoint 里的完整轨迹（可见翻倍） -----")
# 逐条打印，序号 + id + 消息类型 + 内容前 80 字
for i, msg in enumerate(snapshot.values["messages"]):
    print(i, msg.id, type(msg).__name__, getattr(msg, "content", "")[:80])

这次真重复了，4 条变 6 条，第 0/2 条内容一模一样但 id 不同：

您叫小王。请问有什么可以帮您的吗？

----- checkpoint 里的完整轨迹（可见翻倍） -----
0 1207d903-795b-4c7d-8910-5525b49b38e3 HumanMessage 我叫小王。
1 lc_run--01a023df-4bdb-7ac2-98fb-3f1df5ea1c17-0 AIMessage 您好，小王！很高兴为您服务。请问有什么可以帮您的吗？
2 fa181873-a747-45e0-a27c-e45839eb6130 HumanMessage 我叫小王。
3 c480cd33-fbab-4939-8513-8e5e3e2be633 AIMessage 您好，小王！很高兴为您服务。请问有什么可以帮您的吗？
4 fc2e0884-60d7-4807-a770-4f377467dc3b HumanMessage 我叫什么？
5 lc_run--01a023df-503e-70d3-a34b-0cb013198922-0 AIMessage 您叫小王。请问有什么可以帮您的吗？

小结成一句话：

你传的东西 结果
只传本轮新消息（推荐） 正常追加
重传上一轮返回的消息对象 按 id 去重，不会重复
自己重建的 dict 历史 真重复，token 翻倍、模型看到自己说过两次

所以「只传新消息」仍是应当遵守的规范——既为避免第三种情况，也因为传一句话比传整段历史省带宽、更好排查。若你在别处看到「重传必然翻倍」，要知道它取决于消息有没有 id。

一次 invoke 在记忆视角下的数据流：

config.thread_id = "customer-001"
        │
        ▼
checkpointer.get("customer-001")  →  已有 messages（可能为空）
        │
        ▼
追加本轮 HumanMessage → Agent 运行 → 产生 AI / Tool …
        │
        ▼
checkpointer.put("customer-001", 更新后的 state)
        │
        ▼
返回 result（通常含完整 messages，含历史）

3.4. InMemorySaver 是什么 #

测试时首选 InMemorySaver()：零配置、无外部依赖，专注验证「多轮逻辑对不对」。

属性 说明
存储位置 当前 Python 进程内存
依赖 无数据库
进程重启 状态丢失
多实例部署 各进程内存不共享，不能做负载均衡下的统一会话

和「把 history 放 Python 全局变量」的区别： 全局 dict 也能多轮，但和 LangGraph 的 interrupt / resume、middleware 计数、官方 get_state 不对齐；走 checkpointer 才是驾驭层（create_agent）的完整路径。

3.5. 生产 checkpointer 怎么选 #

从开发到上线，通常只换 checkpointer 实现，Agent 代码（tools、middleware、system_prompt）尽量不动；thread_id 生成规则也保持不变。

实现 安装 适用
PostgresSaver pip install -U langgraph-checkpoint-postgres "psycopg[binary]" 生产默认推荐
SQLite（同步/异步包） 见 Checkpointer libraries 单机、边缘、集成测试
MongoDB 等 对应 langgraph-checkpoint-* 包 已有文档库栈时

Postgres 最小形态（需本地已有 PostgreSQL；表结构由 setup() 创建）：

# 导入 os 模块，用于读取环境变量
import os

# 从 langchain 导入创建 Agent 的工厂函数
from langchain.agents import create_agent

# 导入基于 PostgreSQL 的检查点存储实现
from langgraph.checkpoint.postgres import PostgresSaver

# 导入 dotenv，用于从 .env 文件加载 API Key 等配置
from dotenv import load_dotenv

# 加载 .env 中的环境变量，override=True 表示覆盖已存在的同名变量
load_dotenv(override=True)

# 数据库连接串，优先读取 .env 里的 POSTGRES_URI
# sslmode=disable 表示本机连接不启用 SSL
# connect_timeout=5 表示 5 秒连不上就报错，避免无响应
DB_URI = os.getenv(
    "POSTGRES_URI",
    "postgresql://postgres:postgres@localhost:5432/langrag"
    "?sslmode=disable&connect_timeout=5",
)
# PostgresSaver 是上下文管理器，with 块结束时会自动归还数据库连接
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # 首次运行会在数据库中创建 checkpoint 相关的表
    # 这个方法可以重复调用，已存在的表不会被重建
    checkpointer.setup()
    # 创建 Agent，把 checkpointer 传进去即可获得持久化记忆
    agent = create_agent(
        # 使用的模型，与本文其他章节保持一致
        model="deepseek:deepseek-v4-flash",
        # 本示例不需要工具，传空列表
        tools=[],
        # 系统提示，设定助手的角色
        system_prompt="你是简洁的中文客服助手。",
        # 关键：用 PostgresSaver 替代 InMemorySaver
        checkpointer=checkpointer,
    )
    # thread_id 标识一条会话线，同一个 id 的多次调用共享历史
    config = {"configurable": {"thread_id": "customer-001"}}
    # 第一轮：用户提供自己的信息
    agent.invoke(
        {
            "messages": [
                {"role": "user", "content": "你好，我是小王，我的订单号是 A1002。"}
            ]
        },
        config=config,
    )
    # 第二轮：只发新问题，不需要手动拼接历史
    # 历史消息由 checkpointer 从数据库中读出并自动合并
    result = agent.invoke(
        {"messages": [{"role": "user", "content": "我刚才说我是谁？订单号多少？"}]},
        config=config,
    )
    # 打印最后一条消息的内容，应该能正确复述姓名和订单号
    print(result["messages"][-1].content)

psql -U postgres -d postgres
\l
\c langrag
\dt
\d checkpoints
SELECT * from checkpoints;
场景 推荐
本地 / CI InMemorySaver
单机小规模上线 SQLite 类 checkpointer
多实例 + 要 HITL 恢复 Postgres / 云数据库
已有 Mongo 栈 对应 langgraph-checkpoint-mongodb 等

选型口诀：

开发时 用 InMemorySaver；生产用 DB checkpointer；thread_id 由业务层生成（用户 id + 会话 id）。

4. thread_id：会话线程与隔离 #

Thread（线程） 在 LangGraph 里 ≈ 邮箱里的一串往来邮件：同一 thread 内消息有序、状态连续；不同 thread 互不干扰。

Web 产品里可以这样映射：

产品动作 thread_id 策略
用户打开「新对话」 生成新 uuid，旧 thread 只读归档
刷新页面继续聊 前端 localStorage / 服务端 session 带回同一 id
工单系统 一个工单一个 thread，关单后不再写入
匿名访客 guest-{uuid}，登录后可选择「合并」到 user-{id}（进阶，需业务层迁移 state） 
用户 A  session-aaa  ──► thread_id = "userA:2026-08-12"
用户 B  session-bbb  ──► thread_id = "userB:2026-08-12"

4.1. 怎么设计 thread_id #

thread_id 是字符串，框架不解析其内部结构——user-42:chat-9f3a 和 abc 对 LangGraph 等价，只要唯一且稳定。设计时想三问：

做法 示例 说明
每用户一条长期线 f"user-{user_id}" 简单；所有历史进同一线程，需控长（§6）
每「打开客服窗口」一条 f"user-{user_id}:chat-{uuid}" 用户点「新对话」就换新 thread
与业务单绑定 f"ticket-{ticket_id}" 工单闭环内记忆不串
  1. 谁在使用？（用户 id / 访客 id）
  2. 哪一段对话算一场？（窗口 / 工单 / 永久一条线）
  3. 要不要允许「清空记忆」？（新 thread vs 删 messages）

不要把 thread_id 设成常量 "1" 上线——所有用户会共享同一段 messages。本文用固定 id 可以，接口层必须换成业务 id。

4.2. 隔离演示 #

下面用「秘密数字」做对照实验：同一 Agent、同一模型，只改 thread_id，答案应不同。这是验收记忆是否串线最快的方法之一。

# 从 langchain.agents 导入创建 Agent 的工厂函数
from langchain.agents import create_agent
# 内存版 checkpointer
from langgraph.checkpoint.memory import InMemorySaver
# 从 dotenv 导入环境变量加载函数
from dotenv import load_dotenv

# 加载 .env 中的 API key
load_dotenv(override=True)

# 注意：只创建了一个 Agent 实例，两个线程共用它
agent = create_agent(
    # 模型标识
    model="deepseek:deepseek-v4-flash",
    # 不需要工具
    tools=[],
    # 提示模型记住用户说的数字
    system_prompt="你是助手。用户说秘密数字时记住，被问到就回答。",
    # 记忆的开关
    checkpointer=InMemorySaver(),
)

# 线程 1：秘密 42
agent.invoke(
    # 告诉它一个数字
    {"messages": [{"role": "user", "content": "记住，我的秘密数字是 42。"}]},
    # 写入 thread-alice 这条线
    config={"configurable": {"thread_id": "thread-alice"}},
)
# 在同一条线上追问，应该答得出来
r1 = agent.invoke(
    {"messages": [{"role": "user", "content": "我的秘密数字是几？"}]},
    # 仍然是 thread-alice
    config={"configurable": {"thread_id": "thread-alice"}},
)

# 线程 2：从未说过 42
r2 = agent.invoke(
    # 问同样的问题
    {"messages": [{"role": "user", "content": "我的秘密数字是几？"}]},
    # 换成 thread-bob，这条线是全新的
    config={"configurable": {"thread_id": "thread-bob"}},
)

# 对照两个线程的回答
print("Alice:", r1["messages"][-1].content)
print("Bob:", r2["messages"][-1].content)

预期：Alice 能答 42（或引用前文）；Bob 应表示不知道——证明 checkpoint 按 thread 隔离。

注意这里只有一个 Agent 对象，隔离完全由 thread_id 决定。Web 服务里全局共享一个 Agent 实例是正常做法，用户之间不会串线的前提是每个请求带对的 thread_id。

若 Bob 也答 42，优先查：thread_id 是否写错、是否忘了 checkpointer、是否误把 history 写在 Agent 外全局变量里。

4.3. 与第 9 章 context 的分工 #

初学者常把 context（第 9 章 invoke(..., context=OfficeContext(...))）和 thread_id 混为一谈。记一句：

机制 传参方式 生命周期 典型用途
thread_id + checkpointer config={"configurable": {...}} 持久在 thread 内 对话历史、HITL 状态
context + context_schema invoke(..., context=...) 单次 invoke 当前用户角色、部门（第 9 章）

两者可并存：
context 放「这次请求的角色 / 租户」；thread_id 放「这场对话说过什么」。

反模式： 把整段对话历史塞进 context 每轮重传——等于放弃 checkpointer，且 context_schema 并非为无限 messages 设计。历史应走 messages + checkpointer。

5. 读取状态：get_state #

「Agent 失忆了」是集成时最高频的 bug 之一。排障别猜，用 agent.get_state(config) 看 checkpoint 里真实存了什么——比 print(result["messages"][-1]) 可靠，因为后者只是本次返回值，不一定等于持久化后的全貌（尤其 HITL 暂停时）。

5.1 get_state # 

# 从 langchain.agents 导入创建 Agent 的工厂函数
from langchain.agents import create_agent
# 内存版 checkpointer
from langgraph.checkpoint.memory import InMemorySaver
# 从 dotenv 导入环境变量加载函数
from dotenv import load_dotenv

# 加载 .env 中的 API key
load_dotenv(override=True)

# 创建带记忆的 Agent
agent = create_agent(
    # 模型标识
    model="deepseek:deepseek-v4-flash",
    # 不需要工具
    tools=[],
    # 让模型配合记数字
    system_prompt="你是助手。用户说秘密数字时记住，被问到就回答。",
    # 记忆开关
    checkpointer=InMemorySaver(),
)

# thread_id 标识一条会话线，两轮共用同一个 config 才能接上历史
config = {"configurable": {"thread_id": "demo-001"}}

# 第一轮：把秘密数字告诉它
agent.invoke(
    {"messages": [{"role": "user", "content": "我的秘密数字是 42，记住它。"}]},
    config=config,
)

# 第二轮：只发新问题，历史由 checkpointer 自动合并
result = agent.invoke(
    {"messages": [{"role": "user", "content": "我的秘密数字是多少？"}]},
    config=config,
)
# 这是给用户看的那一句
print("最后一条回复：", result["messages"][-1].content)

# 读取该 thread 的 checkpoint 快照，看真实存下来的是什么
snapshot = agent.get_state(config)

# snapshot.values 是一个字典，键就是 state 里的字段（至少有 messages）
print("\n----- checkpoint 里的完整轨迹 -----")
# 逐条打印，序号 + 消息类型 + 内容前 80 字
for i, msg in enumerate(snapshot.values["messages"]):
    print(i, type(msg).__name__, getattr(msg, "content", "")[:80])

# 条数应随轮次增长；不涨说明 checkpointer 或 thread_id 有问题
print("\nmessages 条数：", len(snapshot.values["messages"]))

# next 非空说明图没跑完（例如 HITL 停在 interrupt），此时不能当最终结果验收
print("snapshot.next：", snapshot.next)

正常跑完时 snapshot.next 是空元组：

messages 条数： 4
snapshot.next： ()

snapshot 上常用的三样东西：

方法 / 属性 作用
agent.get_state(config) 读取该 thread_id 当前快照
snapshot.values 状态字典；["messages"] 是完整轨迹（含 ToolMessage），自定义槽位也在这里（§7.4）
snapshot.next 图停在哪些待执行节点；空元组 = 已跑完

snapshot.next 非空时，说明图还没跑完（例如第 10 章 HITL 在等 Command(resume=...)）。此时不要当作「最终回答」验收记忆。

5.2 get_state_history #

还有两个进阶方法，排查「历史被谁改坏了」时很有用：

方法 作用
agent.get_state_history(config) 倒序遍历这条 thread 的历次 checkpoint，能看出某一轮之后 messages 是怎么变的 
from langgraph.types import StateSnapshot

# 取得该 thread 的历史快照（每步一个 state），通常只有最近几步
history = list[StateSnapshot](
    reversed(list(agent.get_state_history(config)))
)  # 明确将 generator 转为 list
print("\n历史快照数：", len(history))

for idx, snap in enumerate[StateSnapshot](history, start=1):
    print(f"\n=== Step {idx} ===")
    msgs = snap.values["messages"]
    for i, msg in enumerate(msgs):
        print(f"  [{i}] {type(msg).__name__}: {getattr(msg, 'content', '')[:80]}")

5.3 update_state #

方法 作用
agent.update_state(config, values) 手工改写状态（比如客服后台强行修正一个槽位值），慎用 

# 取得当前 state
snapshot = agent.get_state(config)
print("\n--- 手工改写前 ---")
for i, msg in enumerate(snapshot.values["messages"]):
    print(i, type(msg).__name__, getattr(msg, "content", "")[:80])

# 假设要手动修正上一条 AI 回复，改成 "你的秘密数字是 100。"
messages = list(snapshot.values["messages"])
# 找到最后一条 AI 回复，将其内容替换
from langchain_core.messages.ai import AIMessage

for i in range(len(messages) - 1, -1, -1):
    if isinstance(messages[i], AIMessage):
        # 构建一个内容被修正的 AIMessage，保留其它字段（如 id）
        new_msg = AIMessage(
            content="你的秘密数字是 100。", id=getattr(messages[i], "id", None)
        )
        messages[i] = new_msg
        break

# 用 update_state 覆盖写回去（只传有更改的字段即可）
agent.update_state(config, {"messages": messages})

# 再读回 state，确认已生效
snapshot2 = agent.get_state(config)
print("\n--- 手工改写后 ---")
for i, msg in enumerate(snapshot2.values["messages"]):
    print(i, type(msg).__name__, getattr(msg, "content", "")[:80])

5.4 和 result["messages"] 的关系 #

时机 优先看什么
正常跑完一轮 result["messages"][-1] 给用户看；get_state 核对条数
怀疑重复 / 丢失 get_state 逐条打印类型与 content 前缀
HITL 暂停 get_state + version="v2" 的 interrupts（第 10 章 §7.2）

验收多轮记忆时建议固定三步：

  1. get_state 看 messages 条数是否随轮次增长
  2. 新开 thread_id 对照，确认没有串线
  3. 问「上一轮提到的 X」验证模型读到的历史是否足够

若第 1 步条数不涨：checkpointer 或 thread_id 有问题。
若条数涨但第 3 步答错：可能是控长裁掉了关键消息（见 §6～§7），或模型本身未利用历史。

6. 上下文窗口：为什么「能记住」还不够 #

checkpointer 解决存得下、接得上；上下文窗口（context window） 解决模型一次能读多少。两者独立——你可以把 200 条 messages 存进 Postgres，但模型可能只能有效利用最近几万 token。

对话越长，messages 越多，会遇到：

问题 后果
超过模型 context window 报错或截断
未超窗但过长 变慢、变贵、易被旧内容干扰
含大量 ToolMessage token 膨胀（查单返回整段 JSON 等）

Token 从哪来（直觉）： 每条 HumanMessage / AIMessage / ToolMessage 都占 token；工具返回一大段 JSON 时，往往比用户一句「查 A1002」贵一个数量级。长会话控长时，有时先缩短 tool 返回比删用户闲聊更有效（第 8 章工具设计 + 本章 middleware）。

何时该上控长（经验值，按模型调整）：

信号 建议
messages 条数 > 30～50 评估 Summarization 或 trim
单次 invoke 明显变慢 / 变贵 查 Trace 里 input token
模型开始「忘记」早期约定 摘要或自定义 state 槽位
尚未生产上线 先 get_state 看长度，再定 trigger

官方常见策略，本章四种都会动手实现，最后一列是对应小节：

策略 思路 本章
Trim（裁剪） 只保留最近 N 条 / N token §7.1
Delete（删除） 用 RemoveMessage 从 state 永久删掉 §7.2
Summarize（摘要） 旧对话压成摘要消息 §7.3
自定义 state 关键字段单独开槽位，不参与压缩 §7.4

原则：

先保证 thread 记忆正确，再谈裁剪；裁剪要比「整段丢失」更可控。

7. 控长实战 #

控长 middleware 改的是写入 checkpointer 的状态：裁掉或摘要后的 messages 会持久化，下一轮 invoke 读到的就是缩短版历史。因此要在丢信息与省 token 之间做有意识的权衡。

三种策略怎么选：

还要精确记得很久以前的原话？
   ├─ 否 → trim（裁剪，§7.1）或 delete（删除，§7.2）
   └─ 是 → Summarization（摘要，§7.3）压大意
            + 自定义 state 槽位（§7.4）保精确字段
工具返回特别长？
   └─ 优先缩短 tool（工具）输出 + 适当 trim（裁剪）

7.1. @before_model：调模型前裁剪 #

@before_model 在每次模型调用之前执行（第 10 章 middleware 时序里：PII / 限流之后、进模型之前）。它读到的 state["messages"] 是 checkpoint 当前完整列表；返回的 dict 会合并回 state，trim 既影响本次模型输入，也影响之后持久化的历史。

下面示例：保留第一条（通常是 system 或首条锚点消息）+ 最近 4 条，避免窗口爆炸。

# Any 用于标注返回的 dict 值类型
from typing import Any

# create_agent 与 AgentState（state 的类型标注）
from langchain.agents import create_agent, AgentState
# before_model 装饰器：把普通函数变成「调模型前执行」的中间件
from langchain.agents.middleware import before_model
# RemoveMessage 用于从 state 里删消息
from langchain.messages import RemoveMessage
# 内存版 checkpointer
from langgraph.checkpoint.memory import InMemorySaver
# REMOVE_ALL_MESSAGES 是一个特殊常量，表示「清空全部」
from langgraph.graph.message import REMOVE_ALL_MESSAGES
# Runtime 是中间件第二个参数的类型
from langgraph.runtime import Runtime
# 从 dotenv 导入环境变量加载函数
from dotenv import load_dotenv

# 加载 .env 中的 API key
load_dotenv(override=True)


# 装饰器把函数注册成「每次调模型前」执行的中间件
@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """保留首条 + 最近 4 条消息，防止上下文过长。"""
    # 从 state 里取出当前完整消息列表
    messages = state["messages"]
    # 还不够长就什么都不做
    if len(messages) <= 5:
        return None  # 不改动

    # 保留第一条作为锚点（这里是第一轮用户消息）
    first_msg = messages[0]
    # 保留最近 4 条，维持对话连贯
    recent = messages[-4:]
    # 返回的 dict 会合并进 state
    return {
        "messages": [
            # 先清空整个列表
            RemoveMessage(id=REMOVE_ALL_MESSAGES),
            # 再写回想保留的那些
            first_msg,
            # * 展开最近 4 条
            *recent,
        ]
    }


# 创建 Agent 并挂上裁剪中间件
agent = create_agent(
    # 模型标识
    model="deepseek:deepseek-v4-flash",
    # 不需要工具，避免裁剪时切断 tool_calls 配对（见 §7.2）
    tools=[],
    # 系统提示
    system_prompt="你是客服助手，回答尽量简短。",
    # 把上面的中间件挂上
    middleware=[trim_messages],
    # 记忆开关
    checkpointer=InMemorySaver(),
)

# 固定一条会话线
config = {"configurable": {"thread_id": "trim-demo"}}
# 连续说四句，把历史撑长
for user in [
    "我叫小陈。",
    "帮我查订单 A1001。",
    "再查 A1002。",
    "我刚才叫什么？",
]:
    # 每轮只传新消息
    agent.invoke({"messages": [{"role": "user", "content": user}]}, config=config)

# 第五轮：此时历史已被裁剪过
result = agent.invoke(
    {"messages": [{"role": "user", "content": "总结一下我们聊过什么。"}]},
    config=config,
)
# 打印最终回答
print(result["messages"][-1].content)

在中间件里加一行 print(len(messages)) 就能看到裁剪的节奏（实测）：

[trim] 进来时 1 条          ← 第 1 轮，不裁
[trim] 进来时 3 条          ← 第 2 轮，不裁
[trim] 进来时 5 条          ← 第 3 轮，刚好等于 5，不裁
[trim] 进来时 7 条 → 裁到 5 条   ← 第 4 轮开始裁
[trim] 进来时 7 条 → 裁到 5 条   ← 第 5 轮

注意最后的稳定状态是 6 条而不是 5 条：裁剪发生在调模型之前，裁到 5 条后模型又追加了一条回答。所以「裁到 N 条」的实际观测值通常是 N+1。

注意：

「保留首条」在这个例子里帮了大忙

裁剪之后再问「我叫什么名字？」，实测模型答对了：

您叫小陈。

这看起来和「裁剪会丢信息」矛盾，其实是因为姓名恰好是第一条消息，被 first_msg 保住了。看一眼裁剪后的 state 就明白：

[0] HumanMessage: '我叫小陈。'              ← 靠「保留首条」活下来
[1] AIMessage:    '好的，小陈。查询 A1002 同样需要...'
[2] HumanMessage: '我刚才叫什么？'
[3] AIMessage:    '您刚才告诉我您叫小陈。'
[4] HumanMessage: '总结一下我们聊过什么。'
[5] AIMessage:    '我们聊过：您自我介绍叫小陈...'

这带来两个实用结论：

  1. 「保留首条」是个廉价又有效的技巧，前提是重要信息真的出现在第一条。用户习惯开场自我介绍时，这个假设常常成立。
  2. 但它非常脆弱。 如果用户先寒暄两句才说姓名，或者中途才给出收货地址，首条就锚不住关键信息了。想稳定保住这类字段，得用 §7.4 的自定义槽位。

关于「首条」到底是什么，前面 §3.2 已经实测过：用 create_agent(system_prompt=...) 注入的系统提示不在 messages 里，所以 messages[0] 就是第一轮的用户消息。如果你是自己往 messages 里塞了 SystemMessage，那首条才是它。调 trim 规则之前，先用 get_state 把列表打印出来看清楚，别凭想象写下标。

7.2. RemoveMessage 与删除策略 #

LangGraph 的 messages 使用 add_messages reducer：普通 append 是追加； RemoveMessage 则是「从 state 里删掉指定 id」或「清空再写」。 这是永久改变 checkpoint 里的历史，不是只影响单次模型输入。

API 作用
RemoveMessage(id=msg.id) 删除指定消息
RemoveMessage(id=REMOVE_ALL_MESSAGES) 清空全部 messages 再追加（常与 trim 联用）

删除时必须保证剩余 history 合法，否则下一跳模型调用会报错：

第二条是实践中最容易踩的：messages[-4:] 这种「按条数切」的写法，完全可能刚好切在 AIMessage(tool_calls=...) 和它的 ToolMessage 之间，保留下来的历史里就有一个「发起了工具调用却没有结果」的悬空调用。这类错误在第 10 章 §8 见过同款报错：

openai.BadRequestError: An assistant message with 'tool_calls' must be followed by
tool messages responding to each 'tool_call_id'.

注意 §7.1 的示例是 tools=[]，压根没有 ToolMessage，所以按条数切很安全。一旦 Agent 带了工具，就必须改成按对切：

# 一个更安全的裁剪：从目标位置往前找，直到落在「安全边界」上
def safe_slice(messages: list, keep: int) -> list:
    """返回最近 keep 条，但保证不切断 tool_calls 与 ToolMessage 的配对。"""
    # 先按条数取一个初始起点
    start = max(len(messages) - keep, 0)
    # 若起点正好是 ToolMessage，说明它的发起者被切掉了，往前挪
    while start > 0 and type(messages[start]).__name__ == "ToolMessage":
        # 往前一条，直到把配对的 AIMessage 一起包含进来
        start -= 1
    # 返回安全的切片
    return messages[start:]

因此生产上更常见的是：trim 保留成对片段，而不是随机删中间几条。

合法 history 小抄（删/裁后自检）：

✓ Human → AI → Human → AI
✓ Human → AI(tool_calls) → Tool → AI
✗ AI(tool_calls) 后面没有 Tool
✗ 只剩 Tool 开头

7.3. SummarizationMiddleware：摘要代替硬删 #

硬删旧消息会丢信息；摘要在触发阈值时调用另一个模型，把远期对话压成一条（或若干条）摘要消息，再与 keep 指定的近期原文拼接。对用户来说，相当于「早期的细节模糊了，但大意还在」——比 trim 更适合长客服会话。

触发后大致流程：

messages 超长 → 达到 trigger
        │
Summarization 模型读旧消息 → 生成摘要 Human/AI 对
        │
state 里旧消息被替换/压缩，保留 keep 条近期原文
        │
主 Agent 继续用缩短后的 history 调模型

先说一个很容易上当的地方

# Any 用于标注返回的 dict 值类型
from typing import Any

# create_agent 与 AgentState（state 的类型标注）
from langchain.agents import create_agent, AgentState

# before_model 装饰器：把普通函数变成「调模型前执行」的中间件
from langchain.agents.middleware import before_model, SummarizationMiddleware

# RemoveMessage 用于从 state 里删消息
from langchain.messages import RemoveMessage

# 内存版 checkpointer
from langgraph.checkpoint.memory import InMemorySaver

# REMOVE_ALL_MESSAGES 是一个特殊常量，表示「清空全部」
from langgraph.graph.message import REMOVE_ALL_MESSAGES

# Runtime 是中间件第二个参数的类型
from langgraph.runtime import Runtime

# 从 dotenv 导入环境变量加载函数
from dotenv import load_dotenv

# 加载 .env 中的 API key
load_dotenv(override=True)


# 装饰器把函数注册成「每次调模型前」执行的中间件
@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """保留首条 + 最近 4 条消息，防止上下文过长。"""
    # 从 state 里取出当前完整消息列表
    messages = state["messages"]
    # 还不够长就什么都不做
    if len(messages) <= 5:
        return None  # 不改动

    # 保留第一条作为锚点（这里是第一轮用户消息）
    first_msg = messages[0]
    # 保留最近 4 条，维持对话连贯
    recent = messages[-4:]
    # 返回的 dict 会合并进 state
    return {
        "messages": [
            # 先清空整个列表
            RemoveMessage(id=REMOVE_ALL_MESSAGES),
            # 再写回想保留的那些
            first_msg,
            # * 展开最近 4 条
            *recent,
        ]
    }


# 反例：装了 SummarizationMiddleware，但根本不会触发
agent = create_agent(
    # 主对话模型
    model="deepseek:deepseek-v4-flash",
    # 不需要工具
    tools=[],
    middleware=[
        # 挂上摘要中间件
        SummarizationMiddleware(
            # 摘要模型
            model="deepseek:deepseek-v4-flash",
            # 4000 token 才触发 —— 问题就出在这一行
            trigger=("tokens", 4000),
            # 保留最近 20 条
            keep=("messages", 20),
        ),
    ],
    # 记忆开关
    checkpointer=InMemorySaver(),
    # 系统提示
    system_prompt="你是客服助手。",
)

# 固定一条会话线
config = {"configurable": {"thread_id": "summary-demo"}}
# 三轮短对话，加起来不过一百来 token
agent.invoke({"messages": [{"role": "user", "content": "我叫小周。"}]}, config=config)
# 第二轮
agent.invoke({"messages": [{"role": "user", "content": "我喜欢猫。"}]}, config=config)
# 第三轮：让它复述前两轮的信息
result = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫什么？喜欢什么？"}]}, config=config
)
# 打印回答 —— 看起来完全正常，所以特别容易误判
print(result["messages"][-1].content)

跑起来回答完全正确：

您叫小周，喜欢猫。对吗？😊

于是很容易得出「摘要生效了」的结论。但看一眼 state 就露馅了：

条数=6
  [0] HumanMessage: '我叫小周。'
  [1] AIMessage:    '您好，小周！很高兴认识您...'
  [2] HumanMessage: '我喜欢猫。'
  [3] AIMessage:    '哈哈，喜欢猫的人通常都很有爱心呢...'
  [4] HumanMessage: '我叫什么？喜欢什么？'
  [5] AIMessage:    '您叫小周，喜欢猫。对吗？'

6 条原文，一条摘要都没有。 三轮短对话离 4000 token 还差得远，中间件从头到尾没干活；回答正确，纯粹是因为普通多轮记忆就够用了。

教训：验收摘要不能看回答对不对，要看 get_state 里有没有出现摘要消息、条数有没有掉下来。 阈值配得比对话总长还高，中间件等于没装。

让它真的触发

想在演示里看到效果，把阈值压到几条消息就行。下面用 trigger=("messages", 6)、keep=("messages", 2)：

# 从 langchain.agents 导入创建 Agent 的工厂函数
from langchain.agents import create_agent
# 内置的摘要中间件
from langchain.agents.middleware import SummarizationMiddleware
# 内存版 checkpointer
from langgraph.checkpoint.memory import InMemorySaver
# 从 dotenv 导入环境变量加载函数
from dotenv import load_dotenv

# 加载 .env 中的 API key
load_dotenv(override=True)

# 创建带摘要能力的 Agent
agent = create_agent(
    # 主对话模型
    model="deepseek:deepseek-v4-flash",
    # 本示例不需要工具
    tools=[],
    middleware=[
        SummarizationMiddleware(
            # 做摘要的模型，可以换成更便宜的
            model="deepseek:deepseek-v4-flash",
            # 演示用的极低阈值：messages 到 6 条就摘要
            trigger=("messages", 6),
            # 摘要后只保留最近 2 条原文
            keep=("messages", 2),
        ),
    ],
    # 摘要结果同样要落到 checkpoint 里
    checkpointer=InMemorySaver(),
    # 系统提示
    system_prompt="你是客服助手，回答简短。",
)

# 固定一条会话线
config = {"configurable": {"thread_id": "summary-force"}}
# 连说四句，把条数推过阈值
for t in ["我叫小周。", "我喜欢猫。", "我住在杭州。", "我的订单是 A1002。"]:
    # 每轮只传新消息
    agent.invoke({"messages": [{"role": "user", "content": t}]}, config=config)
    # 每轮打印条数，观察什么时候掉下来
    print(f"说完「{t}」后条数: {len(agent.get_state(config).values['messages'])}")

实测输出，注意第四轮条数从 6 掉回 4——这就是摘要触发的信号：

说完「我叫小周。」后条数: 2
说完「我喜欢猫。」后条数: 4
说完「我住在杭州。」后条数: 6
说完「我的订单是 A1002。」后条数: 4   ← 触发了

摘要长什么样：默认提示词是英文的

触发之后打印 state，会看到一条很长的消息取代了前面的原文：

[0] HumanMessage: 'Here is a summary of the conversation to date:...'   ← 摘要
[1] AIMessage:    '杭州是个好地方呢，环境很美！🌿...'
[2] HumanMessage: '我的订单是 A1002。'
[3] AIMessage:    '好的小周，已记住您的订单号 A1002 😊...'

把那条摘要完整打出来，就是本节最值得知道的一个坑：

Here is a summary of the conversation to date:

## SESSION INTENT

The user is in an initial customer-service greeting phase and has not yet stated a
specific request or problem. The overall task is to identify what the user needs help
with; no concrete goal has been defined yet.

## SUMMARY

- The user's name is 小周 (Xiao Zhou).
- The user likes cats.
- The user lives in Hangzhou (杭州).
- The assistant greeted the user and asked how they could help...

## ARTIFACTS

None.

## NEXT STEPS

- Acknowledge the user's shared details (name, cats, Hangzhou) to build rapport.
- Prompt the user again to state what they need help with...

三个要点：

  1. 摘要是英文的。 对话全程中文，摘要却是英文——因为 SummarizationMiddleware 的默认 summary_prompt 是英文写的。这不会让程序报错，但会让后续每一轮都多带一段英文进上下文，中文模型读英文摘要也可能引入措辞偏差。
  2. 默认提示词是给编程 Agent 设计的。 看 ## ARTIFACTS（产出的文件）和 ## NEXT STEPS（下一步做什么）——这是「让编码助手在超长会话里接力」的模板。客服场景里 ARTIFACTS 永远是 None，纯属浪费 token。
  3. 摘要以 HumanMessage 的形式插入，不是 SystemMessage。也就是说模型会认为「用户告诉了我这段总结」。

换成中文摘要提示词

解决办法是显式传 summary_prompt。提示词里必须包含 {messages} 占位符，中间件会把待压缩的历史填进去：

# 自定义中文摘要提示词，{messages} 是必须保留的占位符
CN_SUMMARY_PROMPT = """请把下面的客服对话历史压缩成中文要点，供后续对话继续使用。

要求：
1. 只输出要点，不要寒暄，不要解释你在做什么。
2. 必须保留：用户姓名、订单号、地址、电话、已确认的金额与承诺。
3. 用简短条目列出用户诉求与已完成的处理。
4. 无法确定的信息不要猜测，直接省略。

对话历史：
{messages}
"""

# 创建 Agent，把中文提示词传给摘要中间件
agent = create_agent(
    # 主对话模型
    model="deepseek:deepseek-v4-flash",
    # 不需要工具
    tools=[],
    middleware=[
        # 挂上摘要中间件
        SummarizationMiddleware(
            # 摘要模型
            model="deepseek:deepseek-v4-flash",
            # 演示用低阈值
            trigger=("messages", 6),
            # 保留最近 2 条原文
            keep=("messages", 2),
            # 关键：换掉默认的英文提示词
            summary_prompt=CN_SUMMARY_PROMPT,
        )
    ],
    # 记忆开关
    checkpointer=InMemorySaver(),
    # 系统提示
    system_prompt="你是客服助手，回答简短。",
)

同样四轮对话，摘要变成了干净的中文要点：

Here is a summary of the conversation to date:

- 用户姓名：小周
- 地址：杭州
- 其他信息（订单号、电话、金额、承诺）：无

注意开头那句 Here is a summary of the conversation to date: 仍然是英文——它是中间件拼接摘要消息时硬编码的前缀，不受 summary_prompt 控制。只有正文受你的提示词影响，这点无伤大雅。

另外，中文提示词里「必须保留订单号」的要求被认真执行了（明确写了「无」而不是遗漏）。摘要质量高度依赖提示词写得细不细——把业务上不能丢的字段一条条列出来，效果远好于笼统的「请总结」。

参数速查

参数 含义
model 执行摘要的模型（必填，可与主 Agent 模型不同）
trigger 何时摘要：("tokens", N) / ("messages", N) / ("fraction", f)；默认 None 表示不主动触发
keep 摘要后保留多少近期原文，默认 ("messages", 20)
summary_prompt 摘要提示词，须含 {messages}；默认是英文
token_counter 自定义 token 计数函数，默认用近似算法
trim_tokens_to_summarize 送去做摘要的历史本身也会被截断，默认上限 4000 token

trigger 还支持列表形式，任一条件满足即触发，适合「既怕条数多又怕 token 大」的场景：

# 消息超过 20 条，或 token 超过 3000，任一满足就摘要
trigger=[("messages", 20), ("tokens", 3000)]

("fraction", f) 是按模型上下文窗口的比例触发，比如 ("fraction", 0.7) 表示用到七成窗口就压缩——换模型时不用重算绝对数字，比较省心。

如果你在旧代码里看到别的参数名，它们仍然可用，只是会给出弃用警告：

max_tokens_before_summary is deprecated. Use trigger=('tokens', value) instead.
messages_to_keep is deprecated. Use keep=('messages', value) instead.

调参建议（客服场景起点）：

参数 保守（少摘要） 积极（长聊）
trigger ("messages", 24) ("tokens", 3000)
keep ("messages", 12) ("messages", 8)
摘要模型 与主模型相同 更小 / 更便宜模型
summary_prompt 中文，列明必保字段 同左，可再压缩长度

一个容易忽略的成本项：摘要本身要调一次模型。trigger 设得太激进，会出现「聊两句就压缩一次」，反而更慢更贵。先用 get_state 观察真实增长速度，再定阈值。

与 §7.1 对比：

@before_model trim SummarizationMiddleware
旧内容 直接丢弃 压进摘要
实现成本 自己写规则 内置，调参数即可
适合 规则简单、允许遗忘 长会话、还要保留要点

长会话客服推荐默认：checkpointer + SummarizationMiddleware，必要时再叠 trim。
两者同时开时，注意 middleware 顺序：一般 Summarization 负责「压缩远期」，trim 负责「硬保最近 N 条」——具体以实测 Trace 中进模型的 messages 为准。

但请记住摘要的本质是有损压缩：上面那条中文摘要保住了姓名和地址，是因为提示词里点名要求了。没被点名的细节（用户的具体措辞、某个被否决的方案）就真的没了。摘要解决「大意还在」，不解决「一个字都不能错」——后者要靠下一节的槽位。

7.4. 自定义 state：给关键信息开专用槽位 #

前面三种策略有个共同软肋：都在压缩 messages，压缩必然有损。trim 直接丢，摘要变成「大意还在、原话没了」。长会话里最容易出这类事故：

用户第 1 轮说了收货地址，第 30 轮下单时，Agent 把地址记错了。

问题不在模型笨，而在你把「必须精确」的信息交给了一个会被压缩的容器。这类字段应单独存：

信息类型 存哪 例子
对话过程、语气、上下文 messages（可裁剪 / 可摘要） 闲聊、追问、工具轨迹
必须精确、不能丢的字段 自定义 state 槽位 姓名、订单号、收货地址、已确认的金额

一句话心智：

messages 是聊天记录，自定义 state 是这场会话的「工单表单」。表单里的字段不会因为聊得久而模糊。

自定义 state 同样存在 checkpointer 里、同样按 thread_id 隔离，所以它是短期记忆的一部分（跨会话的用户档案属于长期记忆，见第 16 章）。

用起来固定三步，缺一步就不生效：

① 定义槽位：继承 AgentState 加字段
        ↓
② 写入槽位：middleware 或工具返回 {"字段": 值}
        ↓
③ 读出槽位：拼进系统提示，模型才看得见

第 ③ 步最常被漏掉——只写不读，槽位就只是数据库里的死值，模型根本不知道它存在。

7.4.1. 第一步：定义槽位 #

继承官方的 AgentState，把自己要的字段加上去。AgentState 本质是 TypedDict，字段用 NotRequired 标注表示「一开始可以没有」（需 Python 3.11+；更早版本从 typing_extensions 导入）。

# NotRequired 表示这个键可以不存在（Python 3.11+ 内置）
from typing import NotRequired

# 官方的 Agent 状态类型
from langchain.agents import AgentState


# 继承 AgentState，原有的 messages 等字段自动保留
class CustomerState(AgentState):
    # 用户姓名：会话开始时还不知道，所以用 NotRequired
    customer_name: NotRequired[str]
    # 当前咨询的订单号
    order_id: NotRequired[str]

再用 state_schema 告诉 create_agent 用你的 state：

# 创建 Agent 时指定自定义 state
agent = create_agent(
    # 模型标识
    model="deepseek:deepseek-v4-flash",
    # 工具列表
    tools=[],
    # 关键：把默认 AgentState 换成扩展后的 CustomerState
    state_schema=CustomerState,
)

注意两点：

补充一点类型细节：AgentState 本质是 TypedDict（typing_extensions._TypedDictMeta），运行时只是个普通字典——state["messages"] 和 state.get("customer_name") 都能用。NotRequired 只影响静态类型检查，运行时不会强制校验；忘写它不会报错，但 IDE 会一直提示「缺少必填键」。

7.4.2. 第二步：写入槽位 #

写入方式和 §7.1 的 trim 完全一样——在 middleware 里返回一个 dict，键就是槽位名。下面演示了本节的核心论点：

故意把 messages 裁到只剩 2 条（姓名早就被丢掉了），但因为姓名和订单号进了槽位，最后一轮仍然答得出来。

# re 用于正则抽取
import re
# Any 标注 dict 值类型；NotRequired 标注可选的 state 字段
from typing import Any, NotRequired

# create_agent 与 AgentState
from langchain.agents import create_agent, AgentState
# before_model：调模型前的钩子；dynamic_prompt：动态系统提示；ModelRequest：请求对象
from langchain.agents.middleware import before_model, dynamic_prompt, ModelRequest
# RemoveMessage 用于删消息
from langchain.messages import RemoveMessage
# 内存版 checkpointer
from langgraph.checkpoint.memory import InMemorySaver
# 清空全部消息的特殊常量
from langgraph.graph.message import REMOVE_ALL_MESSAGES
# 中间件第二个参数的类型
from langgraph.runtime import Runtime
# 从 dotenv 导入环境变量加载函数
from dotenv import load_dotenv

# 加载 .env 中的 API key
load_dotenv(override=True)


# ① 定义槽位
class CustomerState(AgentState):
    # 姓名槽位，一开始可以不存在
    customer_name: NotRequired[str]
    # 订单号槽位
    order_id: NotRequired[str]


# 简易抽取规则；真实项目里更常用工具或小模型抽取（见 §7.4.4）
# 匹配「我叫」后面 2~4 个汉字
NAME_RE = re.compile(r"我叫([\u4e00-\u9fa5]{2,4})")
# 匹配一个大写字母 + 4 位数字，如 A1002
ORDER_RE = re.compile(r"([A-Z]\d{4})")


# ② 写入槽位：从最新一条用户消息里抽取，返回的 dict 会合并进 state
@before_model
def extract_slots(state: CustomerState, runtime: Runtime) -> dict[str, Any] | None:
    """把姓名 / 订单号抽进槽位，保证后续裁剪也丢不掉。"""
    # 取最新一条消息（本轮用户输入）
    last = state["messages"][-1]
    # 多模态消息的 content 可能是 list，这里只处理纯文本
    text = last.content if isinstance(last.content, str) else ""

    # 收集本轮要更新的槽位
    update: dict[str, Any] = {}
    # 姓名只认第一次，避免后面闲聊把它改掉
    if not state.get("customer_name"):
        # 海象运算符：匹配成功就把结果赋给 m
        if m := NAME_RE.search(text):
            # group(1) 是括号里捕获的姓名
            update["customer_name"] = m.group(1)
    # 订单号以最新提到的为准，所以不加 if not 判断
    if m := ORDER_RE.search(text):
        # 写入订单号槽位
        update["order_id"] = m.group(1)

    # 返回 None 表示这一轮没有要更新的槽位
    return update or None


# 故意用很激进的裁剪：只留最近 2 条，用来证明槽位确实独立于 messages
@before_model
def trim_history(state: CustomerState, runtime: Runtime) -> dict[str, Any] | None:
    # 取出当前消息列表
    messages = state["messages"]
    # 本来就很短就不动
    if len(messages) <= 2:
        return None
    # 清空后只写回最近 2 条
    return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES), *messages[-2:]]}


# ③ 读出槽位：把槽位拼进系统提示，模型才看得见
@dynamic_prompt
def prompt_with_slots(request: ModelRequest) -> str:
    # request.state 就是当前状态字典
    state = request.state
    # 提示词的固定部分
    lines = ["你是客服助手，回答简短。"]
    # 槽位里有姓名就拼一行进去
    if name := state.get("customer_name"):
        lines.append(f"当前用户姓名：{name}")
    # 槽位里有订单号就拼一行进去
    if order := state.get("order_id"):
        lines.append(f"当前用户订单号：{order}")
    # 用换行拼成最终的系统提示
    return "\n".join(lines)


# 组装 Agent
agent = create_agent(
    # 模型标识
    model="deepseek:deepseek-v4-flash",
    # 本示例不需要工具
    tools=[],
    # 关键：声明自定义 state，否则槽位会被静默丢弃（见 §7.4.5）
    state_schema=CustomerState,
    # 顺序要紧：先抽槽位，再裁剪。反了就会先把信息删掉再去抽，抽到空
    middleware=[extract_slots, trim_history, prompt_with_slots],
    # 记忆开关
    checkpointer=InMemorySaver(),
)

# 固定一条会话线
config = {"configurable": {"thread_id": "slot-demo"}}

# 前四轮：姓名在第 1 轮说，订单号在第 2 轮说，之后聊别的把它们「挤出」历史
for text in ["我叫小陈。", "帮我看看订单 A1002。", "今天天气不错。", "顺便问下退货政策。"]:
    # 每轮只传新消息
    agent.invoke({"messages": [{"role": "user", "content": text}]}, config=config)

# 最后一轮：历史里已经没有姓名和订单号了，只能靠槽位回答
result = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫什么？我的订单号是多少？"}]},
    config=config,
)
# 打印模型回答
print("回答：", result["messages"][-1].content)

# 验收：messages 很短，但槽位完好
snap = agent.get_state(config)
# 消息条数应该被裁得很短
print("messages 条数：", len(snap.values["messages"]))
# 槽位里的姓名应该还在
print("customer_name：", snap.values.get("customer_name"))
# 槽位里的订单号应该还在
print("order_id：", snap.values.get("order_id"))

实际运行输出（模型措辞会有差异）：

回答： 您好，小陈，您的订单号是A1002。请问还有什么可以帮您？
messages 条数： 3
customer_name： 小陈
order_id： A1002

这三行输出就是本节的全部价值：历史只剩 3 条、姓名的原话早已不在 messages 里，但回答依然精确。

不用只听结论，把 extract_slots 从 middleware 列表里删掉（其余一模一样）再跑一次，模型立刻答不上来：

回答： 抱歉，我无法获取您的姓名和订单号。请提供订单号，我来为您查询。

同一个模型、同样激进的裁剪，差别只在于有没有把关键字段另存一份。这就是槽位存在的理由。

顺便确认一下槽位真的进了 checkpoint——打印 snapshot.values 的全部键：

['messages', 'customer_name', 'order_id']

自定义字段和 messages 并列存在同一个快照里，按同一个 thread_id 持久化。所以换 checkpointer（比如上生产换 Postgres）时，槽位会跟着一起持久化，不需要额外做什么。

7.4.3. 第三步：读出槽位 #

上面示例用 @dynamic_prompt 完成了读出。为什么必须做这一步？自定义槽位不会自动进入模型输入——模型只能看到系统提示 + messages，state 里的其他字段对它是不可见的。

读出的三种常见方式：

方式 做法 适合
@dynamic_prompt（推荐入门） 把槽位拼成一句话加进系统提示 姓名、订单号这类短字段
工具读 state 工具签名里接 runtime: ToolRuntime，读 runtime.state 工具需要知道「当前订单是哪个」
业务代码读 agent.get_state(config).values["order_id"] 落库、埋点、前端展示

第三种值得单独强调：槽位对你的后端代码也可读。比如「用户确认了收货地址」——前端要展示、订单系统要落库，靠解析模型自然语言既不可靠也不优雅，读槽位才是确定性的。

7.4.4. 让工具写槽位 #

正则抽取只适合演示，一到真实场景就跪（「我姓陈」「叫我小陈就行」都匹配不到）。更可靠的做法是让模型自己决定何时记录——给它一个专门写槽位的工具。

工具写 state 的方式是返回 Command 对象，在 update 里同时给出：要更新的槽位，以及本次工具调用对应的 ToolMessage（后者不能省，否则 history 里 tool_calls 没有配对返回，下一跳模型调用会报错，见 §7.2）。

# NotRequired 标注可选的 state 字段
from typing import NotRequired

# create_agent 与 AgentState
from langchain.agents import create_agent, AgentState
# 动态系统提示相关
from langchain.agents.middleware import dynamic_prompt, ModelRequest
# ToolMessage 是工具执行结果的消息类型
from langchain.messages import ToolMessage
# tool 装饰器；ToolRuntime 用于在工具里拿到运行时信息
from langchain.tools import tool, ToolRuntime
# 内存版 checkpointer
from langgraph.checkpoint.memory import InMemorySaver
# Command 用来让工具直接改写 state
from langgraph.types import Command
# 从 dotenv 导入环境变量加载函数
from dotenv import load_dotenv

# 加载 .env 中的 API key
load_dotenv(override=True)


# 定义带订单号槽位的 state
class CustomerState(AgentState):
    # 订单号槽位
    order_id: NotRequired[str]


# 工具返回 Command，即可同时写 state 和补 ToolMessage
# runtime: ToolRuntime 由框架注入，模型不需要（也看不到）这个参数
@tool
def remember_order(order_id: str, runtime: ToolRuntime) -> Command:
    """把用户当前咨询的订单号记入会话槽位。订单号形如 A1001。"""
    # 归一化：去空格并转大写
    key = order_id.strip().upper()
    # 返回 Command 而不是字符串，就能同时更新 state
    return Command(
        update={
            # 写入自定义槽位
            "order_id": key,
            # 必须补上与本次 tool_call 配对的 ToolMessage
            "messages": [
                # tool_call_id 从 runtime 里取，用来和模型的调用请求配对
                ToolMessage(f"已记住订单 {key}。", tool_call_id=runtime.tool_call_id)
            ],
        }
    )


# 读出槽位：让模型知道「当前订单已经记下来了，是哪个」
@dynamic_prompt
def prompt_with_slot(request: ModelRequest) -> str:
    # 从 state 里读订单号槽位
    order = request.state.get("order_id")
    # 提示词基础部分，顺便引导模型去调工具
    base = "你是客服助手。用户第一次提到订单号时，调用 remember_order 记下来。"
    # 槽位有值就附加上去
    if order:
        return f"{base}\n当前用户订单号：{order}"
    # 槽位为空就只返回基础提示
    return base


# 组装 Agent
agent = create_agent(
    # 模型标识
    model="deepseek:deepseek-v4-flash",
    # 把写槽位的工具挂上
    tools=[remember_order],
    # 声明自定义 state
    state_schema=CustomerState,
    # 挂上读槽位的动态提示
    middleware=[prompt_with_slot],
    # 记忆开关
    checkpointer=InMemorySaver(),
)

# 固定一条会话线
config = {"configurable": {"thread_id": "tool-slot-demo"}}

# 第一轮：模型应主动调用 remember_order 把订单号写进槽位
agent.invoke(
    {"messages": [{"role": "user", "content": "帮我记一下，我的订单号是 A1002。"}]},
    config=config,
)

# 直接读 state 验收，不依赖模型怎么措辞
snap = agent.get_state(config)
# 槽位里应该出现 A1002
print("槽位 order_id：", snap.values.get("order_id"))

# 第二轮：槽位已注入系统提示，模型无需重新翻历史
result = agent.invoke(
    {"messages": [{"role": "user", "content": "我的订单号是多少？"}]},
    config=config,
)
# 打印回答
print("回答：", result["messages"][-1].content)

实际运行输出：

槽位 order_id： A1002
回答： 您的订单号是 **A1002**。请问还需要其他帮助吗？😊

第一轮的完整轨迹如下，可以看清 Command 是怎么同时干两件事的：

[0] HumanMessage: '帮我记一下，我的订单号是 A1002。'
[1] AIMessage:    ''  → remember_order({'order_id': 'A1002'})   ← 模型决定调工具
[2] ToolMessage:  '已记住订单 A1002。'                            ← Command 里补的那条
[3] AIMessage:    '好的，我已经记住了您的订单号 **A1002**。'

注意 [1] 的 content 是空字符串——模型这一步只发出工具调用，不说话，这是正常的（第 8 章讲过）。

漏了 ToolMessage会怎样

这是本节唯一容易犯的错。把 update 里的 messages 那一项删掉：

# 反例：只写槽位，不补 ToolMessage
@tool
def bad_remember(order_id: str, runtime: ToolRuntime) -> Command:
    """记订单号（故意漏 ToolMessage）。"""
    # update 里只有槽位，少了 "messages" 那一项
    return Command(update={"order_id": order_id.strip().upper()})

好消息是这个错误会立刻响亮地报出来，报错信息还把修法写在里面了：

ValueError: Expected to have a matching ToolMessage in Command.update for tool
'bad_remember', got: []. Every tool call (LLM requesting to call a tool) in the message
history MUST have a corresponding ToolMessage. You can fix it by modifying the tool to
return `Command(update=[ToolMessage("Success", tool_call_id=runtime.tool_call_id)])`

原因是 §7.2 讲的那条规则：历史里每个 tool_calls 都必须有配对的 ToolMessage。工具一旦改成返回 Command，框架就不再帮你自动生成这条消息——责任转移到了你手上。

记法：普通工具返回字符串，框架替你造 ToolMessage；工具返回 Command，你得自己造。

两种写入方式怎么选：

middleware 写（§7.4.2） 工具写（本节）
触发时机 每轮固定执行，确定性 由模型判断，可能不调用
抽取能力 靠你写的规则，语言变化就漏 模型理解语义，鲁棒得多
额外成本 无（不调模型） 多一次工具调用
适合 格式固定的字段（订单号、手机号） 需要理解语义的字段（地址、诉求、意图）

生产上两者常并用：格式化字段用 middleware 兜底，语义字段交工具。

7.4.5. 边界与常见坑 #

自定义 state 很好用，但它不是万能筐。先划清边界：

该放槽位 不该放槽位
少量、精确、要复用的字段 整段对话历史（那是 messages 的活）
本次会话内有效的信息 跨会话的用户档案（属长期记忆 / store）
后端也要读的结构化结果 大块文档、检索结果（放 RAG，第 12 章起）

最坑的一个：忘传 state_schema 不会报错

这是本节唯一的静默失败，值得单独拎出来。把 state_schema=CustomerState 从 create_agent 里删掉，其余代码一字不改：

# 反例：中间件照样返回槽位，但没声明 state_schema
agent = create_agent(
    # 模型标识
    model="deepseek:deepseek-v4-flash",
    # 不需要工具
    tools=[],
    # 这里少了 state_schema=CustomerState
    middleware=[extract_slots, prompt_with_slots],
    # 记忆开关
    checkpointer=InMemorySaver(),
)

跑起来一切正常，没有任何警告。但去看 state：

snapshot.values 键： ['messages']
customer_name： None

extract_slots 明明返回了 {"customer_name": "小陈"}，这个键却被悄悄丢掉了——因为默认的 AgentState 里没有这个字段，框架不知道该往哪存。于是你会看到一个特别费解的现象：抽取逻辑打印日志说抽到了，槽位却永远是空的。

排查口诀：槽位读出来是 None 时，先看 get_state(config).values 有没有这个键。 键都不存在 → 忘了 state_schema；键存在但值是 None → 抽取逻辑没匹配上。

常见坑：

现象 原因 处理
槽位写了，但模型完全不知道 漏了第 ③ 步「读出」 用 @dynamic_prompt 注入系统提示
get_state 里看不到自定义字段 忘了传 state_schema=，静默丢弃 在 create_agent 里补上
槽位被后面的闲聊改乱 每轮无条件覆盖 写入前判断 if not state.get(...)
裁剪后槽位也没了 middleware 顺序反了，先裁再抽 抽取 middleware 放在裁剪之前
工具写完 state 后报 ValueError Command.update 里漏了 ToolMessage 补 tool_call_id=runtime.tool_call_id 的 ToolMessage
换 thread_id 后槽位空了 槽位是thread 内的短期记忆 跨会话信息要走长期记忆

口诀：

会话里「不能记错」的字段，别指望 messages 记住——开个槽位存起来，再拼回提示词。

官方文档：Customizing agent memory

8. 实战：带记忆的客服 Agent #

本节把第 8 章工具与第 11 章记忆与控长串成一条可演示链路。重点验收的不是「又写了两个 tool」，而是同一 thread_id 下三轮对话能否闭环——尤其是第三轮「刚才那个订单」能否指代 A1002。

组装关系：

lookup_order / lookup_policy（第 8 章）
        +
InMemorySaver + thread_id（§3～§4）
        +
SummarizationMiddleware（§7.3，可选）
        → build_customer_agent()

能力清单：

能力 工具 / 机制
查订单 lookup_order
查退换货政策 lookup_policy
多轮记忆 InMemorySaver + thread_id
长聊控长 SummarizationMiddleware（可选）

8.1. memory_customer_service.py #

build_customer_agent(enable_summary=True) 把 Summarization 做成开关，便于 A/B：先关摘要跑通多轮，再打开看 get_state 里 messages 条数变化。

# 从 langchain.agents 导入创建 Agent 的工厂函数
from langchain.agents import create_agent
# 内置摘要中间件，用于长会话控长
from langchain.agents.middleware import SummarizationMiddleware
# tool 装饰器，把普通函数变成工具
from langchain.tools import tool
# 内存版 checkpointer（生产请换 Postgres，见 §3.4）
from langgraph.checkpoint.memory import InMemorySaver
# 从 dotenv 导入环境变量加载函数
from dotenv import load_dotenv

# 加载 .env 中的 API key
load_dotenv(override=True)

# 用字典模拟订单数据库
ORDERS = {
    # 订单号 -> 物流状态
    "A1001": "已发货，预计明天送达",
    "A1002": "运输中，预计后天送达",
}


# 注册为工具：查订单
@tool
def lookup_order(order_id: str) -> str:
    """按订单号查询物流状态。订单号形如 A1001。"""
    # 归一化：去空格并转大写，容忍用户输入 a1002
    key = order_id.strip().upper()
    # 查字典
    status = ORDERS.get(key)
    # 查不到就返回明确的错误文本，让模型知道该追问
    if not status:
        return f"错误：未找到订单 {order_id}。"
    # 查到则返回结构清晰的一句话
    return f"订单 {key}：{status}。"


# 注册为工具：查政策
@tool
def lookup_policy(topic: str) -> str:
    """查询退换货与售后政策摘要。"""
    # 演示用固定文案；真实项目这里应接 RAG（第 12 章起）
    return (
        "退换货：签收 7 日内可无理由退货；"
        "质量问题 15 日内可换货；"
        "跨境订单以页面公示为准。"
    )


# 工厂函数：把摘要做成开关，方便 A/B 对照
def build_customer_agent(*, enable_summary: bool = True):
    # 中间件列表，默认为空
    middleware = []
    # 只有开启摘要时才挂 SummarizationMiddleware
    if enable_summary:
        middleware.append(
            SummarizationMiddleware(
                # 摘要模型，可换更便宜的
                model="deepseek:deepseek-v4-flash",
                # messages 到 12 条就压缩
                trigger=("messages", 12),
                # 压缩后保留最近 8 条原文
                keep=("messages", 8),
            )
        )

    # 返回组装好的 Agent
    return create_agent(
        # 主对话模型
        model="deepseek:deepseek-v4-flash",
        # 两个业务工具
        tools=[lookup_order, lookup_policy],
        # 记忆开关：没有它就没有多轮
        checkpointer=InMemorySaver(),
        # 上面按开关拼出来的中间件
        middleware=middleware,
        system_prompt=(
            # 角色设定
            "你是电商客服助手。"
            # 工具选择规则
            "查订单用 lookup_order；问政策用 lookup_policy。"
            # 防幻觉约束
            "依据工具返回回答，不要编造物流状态。"
            # 显式鼓励利用历史，这是多轮指代能work的关键
            "用户提到过的订单号，后续轮次可主动关联。"
        ),
    )


# 只有直接运行本文件时才执行下面的演示
if __name__ == "__main__":
    # 构建 Agent（默认开启摘要）
    agent = build_customer_agent()
    # 固定一条会话线
    config = {"configurable": {"thread_id": "cs-demo-001"}}

    # 三轮对话，第三轮故意用「刚才那个订单」考验记忆
    turns = [
        "你好，我订单 A1002 到哪了？",
        "退换货几天内可以退？",
        "刚才那个订单，大概哪天到？",
    ]
    # 逐轮对话
    for text in turns:
        # 打印用户输入
        print(f"\n用户: {text}")
        # 每轮只传新消息，历史由 checkpointer 合并
        result = agent.invoke(
            {"messages": [{"role": "user", "content": text}]},
            config=config,
        )
        # 打印助手回答
        print(f"助手: {result['messages'][-1].content}")

    # 调试：查看 thread 内消息条数
    snap = agent.get_state(config)
    # 条数应随轮次增长（未触发摘要时）
    print(f"\n[debug] thread 内 messages 条数: {len(snap.values['messages'])}")

8.2. 真实运行结果 # 

用户: 你好，我订单 A1002 到哪了？
助手: 您好！您的订单 A1002 目前**运输中**，预计**后天送达**。请您留意物流更新哦～

用户: 退换货几天内可以退？
助手: 您好，关于退换货政策：
- **签收 7 日内**可无理由退货；
- **质量问题 15 日内**可换货；
- 跨境订单以页面公示为准。

您是在关注订单 A1002 的退换货吗？如需进一步说明可以告诉我～

用户: 刚才那个订单，大概哪天到？
助手: 您好，根据订单 A1002 的最新物流信息，它目前**运输中**，预计**后天送达**。

[debug] thread 内 messages 条数: 10

第二轮那句「您是在关注订单 A1002 的退换货吗？」很有意思——用户这一轮完全没提订单号，模型是从历史里主动关联出来的。这正是多轮记忆想要的效果。

完整轨迹（get_state 打印）：

[0] HumanMessage: '你好，我订单 A1002 到哪了？'
[1] AIMessage:    ''  → lookup_order({'order_id': 'A1002'})
[2] ToolMessage:  '订单 A1002：运输中，预计后天送达。'
[3] AIMessage:    '您好！您的订单 A1002 目前运输中...'
[4] HumanMessage: '退换货几天内可以退？'
[5] AIMessage:    ''  → lookup_policy({'topic': '退换货'})
[6] ToolMessage:  '退换货：签收 7 日内可无理由退货...'
[7] AIMessage:    '您好，关于退换货政策...'
[8] HumanMessage: '刚才那个订单，大概哪天到？'
[9] AIMessage:    '您好，根据订单 A1002 的最新物流信息...'

注意第三轮（[8]→[9]）没有再调一次 lookup_order：模型直接从 [2] 的 ToolMessage 里读到了答案。这是省钱的好事，但也意味着如果物流状态在两轮之间变了，用户看到的是旧数据。对时效敏感的字段，应在系统提示里明确要求「每次询问状态都重新查询」，或者把结果的时间戳带进工具返回值。

8.3. 摘要到底省了多少 #

enable_summary 这个开关不是摆设。用同一套代码连聊 8 轮（每轮都会触发一次工具调用），只改这个开关：

# 对照实验：只有 enable_summary 不同
for flag in (False, True):
    # 分别构建开/关摘要的 Agent
    a = build_customer_agent(enable_summary=flag)
    # 用不同 thread_id 避免互相污染
    c = {"configurable": {"thread_id": f"len-{flag}"}}
    # 连聊 8 轮
    for i in range(8):
        a.invoke(
            {"messages": [{"role": "user", "content": f"第{i + 1}个问题：退货要几天？"}]},
            config=c,
        )
    # 打印最终条数
    print(f"enable_summary={flag}: 8 轮后 messages 条数 = {len(a.get_state(c).values['messages'])}")

实测差距相当明显：

enable_summary=False: 8 轮后 messages 条数 = 33
enable_summary=True:  8 轮后 messages 条数 = 12

33 条 vs 12 条。 关掉摘要时是 33 而不是 16——因为每轮都有一次工具调用，一轮实际产生 4 条消息（Human、带 tool_calls 的 AI、Tool、最终 AI）。这解释了为什么带工具的 Agent 比纯聊天更容易撞上下文窗口：ToolMessage 是 token 膨胀的主要来源，往往比用户那句话长得多。

开启摘要后条数稳定在 trigger 附近波动，不再线性增长。这就是控长的意义。

8.4. 验收清单 #

  1. 多轮指代：第三轮「刚才那个订单」能答 A1002 / 运输中（不要求字面复述第一轮原话）。
  2. 工具轨迹：get_state 里可见 tool_calls 与 ToolMessage。
  3. 换 thread 隔离：thread_id 换成 cs-demo-002 后，不应记得 A1002。
  4. 控长（若开启 Summarization）：连续多轮后 messages 条数不会无限线性暴涨（相对 enable_summary=False 对照）。

第 3 条的实测结果，可以作为「隔离生效」的标准答案：

# 换一个全新的 thread_id，注意 agent 对象没变
cfg2 = {"configurable": {"thread_id": "cs-demo-002"}}
# 问和第三轮一模一样的问题
r2 = agent.invoke(
    # 这句话在旧 thread 里能答出 A1002
    {"messages": [{"role": "user", "content": "刚才那个订单，大概哪天到？"}]},
    # 但这里用的是新会话线
    config=cfg2,
)
# 打印回答，应该表示不知道是哪个订单
print(r2["messages"][-1].content)
您好！为了帮您查询准确的物流信息，我需要知道您的**订单号**（形如 A1001）。
麻烦您提供一下订单号，我这就帮您查预计到达时间～

模型反过来找用户要订单号——说明新 thread 里确实一片空白。这里用的是同一个 Agent 对象，隔离完全由 thread_id 决定，与 Agent 实例无关。

与第 10 章护栏 Agent 的关系：本章 Agent 偏记忆与对话体验；把 PIIMiddleware、ModelCallLimitMiddleware 等从第 10 章叠进来就能上线，无需改 checkpointer 用法——记忆层与护栏层是正交的。

9. 实用约定与坑 #

上线前把下面约定当成 Code Review 清单；「常见坑」表对应集成测试里优先写的 5 个用例。

约定 说明
每轮只传新用户消息 省带宽、好排查；自己重建 dict 历史会真重复（§3.2）
thread_id 业务生成 禁止全员共用 "1"
Demo 用 InMemorySaver 生产换 Postgres 等
摘要阈值要实测再定 配得比对话总长还高 = 中间件从未工作（§7.3）
摘要提示词换成中文 默认 summary_prompt 是英文且面向编程 Agent（§7.3）
裁剪 / 摘要会丢细节 姓名、订单号、地址等关键字段开自定义 AgentState 槽位（§7.4）
槽位「写了要读」 写进 state 还需 @dynamic_prompt 注入，模型才看得见（§7.4.3）
自定义槽位必须配 state_schema 忘了会被静默丢弃（§7.4.5）
HITL / thread_limit 与记忆共用 checkpointer（第 10 章）
调试看 get_state 不要只看最后一条 AI 回复

坑分成两类看，静默的那类才危险。

静默类（不报错，但行为不对）：

现象 原因 处理
第二轮「失忆」 未配 checkpointer 或 thread_id 变了 检查创建参数与 config
用户 A 看到用户 B 的历史 thread_id 冲突 按用户 / 会话生成唯一 id
装了摘要中间件但条数一直线性涨 trigger 阈值远高于实际对话长度，从未触发 用 get_state 看真实增长，把阈值调到合理区间（§7.3）
上下文里多出一段英文 摘要用了默认英文 summary_prompt 传中文 summary_prompt（§7.3）
槽位永远是 None，抽取日志却说抽到了 忘传 state_schema，键被静默丢弃 先看 get_state().values 有没有这个键（§7.4.5）
槽位有值但模型不用 只写没读 用 @dynamic_prompt 拼进系统提示（§7.4.3）
自己重建 dict 历史后模型像在自言自语 无 id 的消息被当成新消息，历史出现重复 只传本轮新消息（§3.2）
状态查询返回的是过期数据 模型直接复用了旧 ToolMessage，没重新调工具 提示里要求重新查询，或在返回值带时间戳（§8.2）

响亮类（会抛异常，好定位）：

现象 原因 处理
重启服务后全忘 用了 InMemorySaver 换持久 checkpointer
trim 后模型调用报 tool_calls must be followed by tool messages 按条数切，切断了 tool_calls 与 ToolMessage 的配对 按对保留（§7.2 的 safe_slice）或改用摘要
工具报 Expected to have a matching ToolMessage in Command.update 工具返回 Command 却没补 ToolMessage 补 tool_call_id=runtime.tool_call_id（§7.4.4）
ValueError: Checkpointer requires ... thread_id 配了 checkpointer 却没传 thread_id 在 config 里补上（第 10 章 §5.1）

设计类（不是 bug，是要想清楚的取舍）：

问题 取舍
摘要仍答不对很早以前的信息 摘要是有损压缩；不能错的字段改存槽位（§7.4）
摘要本身也要花钱 trigger 太激进会「聊两句压一次」，反而更慢更贵
保留首条能锚住关键信息吗 只在用户开场就自我介绍时成立，很脆弱（§7.1）

口诀：

thread 管一场对话；checkpointer 管怎么存；控长管能聊多久。

10. 练习 #

每题做完建议打印 get_state 的 messages 条数；涉及隔离的题务必用两个 thread_id 对照。

机制验证类（跑一遍就能纠正直觉，建议全做）：

  1. 新 thread：同一 Agent 用两个 thread_id 各聊一轮，交叉提问，确认隔离。
  2. get_state：每轮后打印 messages 条数与最后两条类型。
  3. 重传是否真重复：照 §3.2 做两组对照——重传 result["messages"] 对象 vs 手工重建 dict 历史，用 get_state 数条数，验证「按 id 去重」。
  4. 摘要有没有触发：先用 trigger=("tokens", 4000) 聊三轮，打印 state 确认一条摘要都没有；再改成 ("messages", 6) 重跑，观察条数从 6 掉回 4。
  5. 摘要是什么语言：把触发后的那条摘要消息完整打印出来，确认默认提示词产出的是英文 + ## ARTIFACTS；然后传中文 summary_prompt 再跑一次对比。
  6. 忘传 state_schema：把 §7.4.2 的 state_schema=CustomerState 删掉，确认程序不报错但 get_state().values 里只有 messages。

能力构建类：

  1. trim 对照：去掉 trim_messages / Summarization，连聊 15 轮，观察 token / 延迟变化（可选 LangSmith）。
  2. 指代理解：先报订单号，第三轮只问「那个单子到哪了」，验收是否调工具或引用前文。
  3. 叠护栏：给 §8 Agent 加第 10 章 PIIMiddleware("email")，输入带邮箱，确认脱敏仍多轮可用（注意第 10 章 §6.4 那个 PII 与工具冲突的坑）。
  4. 自定义 state：照 §7.4.2 跑通示例后，把 extract_slots 从 middleware 里删掉再跑一次，对比最后一轮的回答差异。
  5. 槽位读出：在 §7.4.2 基础上注释掉 prompt_with_slots，验证「只写不读」时模型是否还答得出姓名。
  6. 裁剪切断工具对：给 §7.1 的 Agent 加上 lookup_order 工具，保持按条数裁剪，多聊几轮直到报 tool_calls must be followed by...，再用 §7.2 的 safe_slice 修好。
  7. （扩展）工具写槽位：给 §7.4.4 的 Agent 再加一个 remember_address 工具，写入 address 槽位，并用 get_state 验收；先故意漏掉 ToolMessage 看报错，再补齐。

11. 本章小结 #

记忆章节最容易「代码跑通但产品失忆」——验收务必用 thread 隔离 + 指代问句 + get_state 三件套，不要只看最后一句话顺不顺。

  1. 短期记忆 = 单 thread 内的 messages 状态；长期记忆（跨 thread 的 store）见第 16 章，文档型知识库（RAG）见第 12～15 章。
  2. checkpointer + thread_id 是官方推荐的多轮方案；每轮只传新消息。隔离由 thread_id 决定，全局共享一个 Agent 实例是正常的。
  3. messages 按消息 id 去重：重传带 id 的消息对象不会翻倍，自己重建的 dict 历史才会（§3.2）。
  4. InMemorySaver 适合 Demo；生产用 Postgres 等持久 checkpointer，槽位会一起持久化。
  5. get_state 用于调试完整轨迹与 HITL 中间态；snapshot.next 为空元组才算跑完。
  6. 长会话要 trim / 删除 / 摘要。裁剪要按对切，别切断 tool_calls 与 ToolMessage（§7.2）。
  7. 摘要有两个必须动手改的默认值：trigger 不设或设太高会完全不触发；summary_prompt 默认是英文且面向编程 Agent，中文客服务必替换（§7.3）。
  8. 自定义 state 槽位（§7.4）存「不能记错」的字段，三步齐全才生效：定义 state_schema → middleware/工具写入 → @dynamic_prompt 读出。漏 state_schema 会静默失败。
  9. 工具想写 state 就返回 Command(update=...)，必须自己补 ToolMessage（§7.4.4）。
  10. context（第 9 章） 管单次请求上下文；thread_id 管会话历史——分工不同，可并存。
  11. 本章产出：带记忆的客服 Agent（查单 + 政策 + 多轮指代）；实测摘要把 8 轮对话从 33 条压到 12 条。

一句话记住这章的验收方法：

别问「模型答得对不对」，问「get_state 里存的是什么」。 回答正确可能只是因为对话还太短，机制根本没生效。

下一章：Document Loaders 文档加载——把 PDF、Word、网页里的公司资料读成统一的 Document 对象，开始 RAG 主线（第 12～15 章）。

顺便说清一件事，免得你翻目录时困惑：本章管的是「这场对话说过什么」；「换了 thread_id 也不能忘」的用户档案属于长期记忆，排在第 16 章（RAG 四章之后）。为什么不紧接着讲？因为长期记忆后面要用向量检索做语义查找（store 的 index 参数），嵌入模型和相似度检索是第 14 章的内容。先学完 RAG 再回头看长期记忆，就不必「先跳过、以后再补」了。

← 上一节 10.Middleware与安全护栏 下一节 12.DocumentLoaders文档加载 →