1. 本章目标 #

第 11 章我们让 Agent 记住了「这场对话说过什么」，但那种记忆有一条硬边界：它只活在一个 thread_id 里。

1.1 换个 thread_id 就失忆 #

先看一个真实的产品事故。第 11 章的客服 Agent，用户昨天说过「我住杭州、别啰嗦」，今天打开 App 重新进入客服：

用户的感受不是「这个系统设计如此」，而是「这个客服有点傻」。

根因在于 checkpointer 的语义：它保存的是某条会话线的完整状态快照，thread_id 换了就换了一份状态。这不是 bug，正是第 11 章要的隔离性——两个客户的咨询不该互相看见。但这也意味着：

凡是「这个用户永远都成立」的信息，都不该只存在 messages 里。

姓名、常住城市、忌口、语言偏好、公司内部制度……这些东西的生命周期比一场对话长得多。它们需要另一个存储层：长期记忆（long-term memory），在 LangGraph 里就是 store。

1.2 三种「记住」，别混为一谈 #

到本章为止，出现过三种能让 Agent「记住东西」的机制。它们经常被混在一起，导致「该记的没记住，不该记的记了一堆」。先把分工钉死：

一个客服请求同时需要三者：

用户问：「我上次说的那台净化器，滤芯多久换一次？包邮吗？」
   │
   ├── 「上次说的那台」  → checkpointer：本次会话里提到的型号        （第 11 章）
   ├── 「我」的收货地址   → store：用户档案，跨会话有效               （本章）
   └── 「滤芯更换周期」   → 向量知识库：产品手册第 12 页              （第 12～15 章）

三者的判断口诀：

换个会话还要不要？ 要 → store。 是文档里的内容吗？ 是 → 向量库。 只在这轮对话里有意义？ 是 → messages（checkpointer）。

1.3 store 和向量库的界线在哪 #

二者都能「按语义找东西」，甚至 store 本身就支持向量检索（§6），但它们的定位不同。刚学完第 12～15 章，右边那一列你已经很熟了，正好拿来做对照：

一句话：store 存「关于人的事」，向量库存「关于资料的事」。产品手册也能塞进 store，但你会丢掉第 13 章的切分策略、第 15 章的重排与混合检索——这两样刚花两章调出来，不该在这里扔掉；反过来把用户偏好塞进向量库，「精确读某个用户的姓名」就变成一次相似度查询——既不可靠也没必要。

2. store 的数据模型 #

store 是一个带命名空间的 JSON 文档仓库。

2.1 namespace / key / value #

用文件系统类比最直观：

先记住三点：

1. namespace 是元组，不是字符串。("u_zhou", "profile") 是两层，("u_zhou",) 是一层（注意那个逗号，少了它就变成普通字符串了）。层数不限，你可以用 ("org_1", "u_zhou", "memories") 表达「1 号组织下小周的笔记」。
2. value 必须是字典，不能直接放字符串或列表。想存一句话，也要包成 {"text": "..."}。
3. namespace + key 唯一确定一条记录，同名写入就是覆盖（§2.4 会实测这一点）。

2.2 一条记录长什么样：Item #

写进去的是 dict，读出来的却不是 dict，而是一个 Item 对象。这一点每个初学者都会绊一次：store.get(...)["name"] 会报 TypeError，正确写法是 store.get(...).value["name"]。

Item 要包一层，是因为一条记录除了「内容」还有「元信息」——它在哪、叫什么、什么时候写的。这些不该混进 value 污染业务数据，所以单独挂在对象上：

# InMemoryStore：内存版长期记忆，进程退出即丢，适合本地开发
from langgraph.store.memory import InMemoryStore

# 创建一个 store 实例，不传参数就是纯键值存储（不带语义检索）
store = InMemoryStore()

# put(namespace, key, value)：写入一条记录，namespace 是元组
store.put(("u1", "profile"), "basic", {"name": "小周", "city": "杭州"})

# get(namespace, key)：按 key 精确读取，命中返回 Item，未命中返回 None
item = store.get(("u1", "profile"), "basic")

# Item.value 是你写进去的那个字典（原样返回，不做任何包装）
print("value      :", item.value)
# Item.key 回显这条记录的名字，等于 put 时传的第二个参数
print("key        :", item.key)
# Item.namespace 回显它所在的命名空间，仍然是元组
print("namespace  :", item.namespace)
# created_at 由 store 自动维护，是带时区的 datetime（注意下方警告：语义因实现而异）
print("created_at :", item.created_at)
# updated_at 是最后一次写入时间，判断「这条记忆有多旧」只能靠它
print("updated_at :", item.updated_at)

输出（实测，时间戳会不同）：

value      : {'name': '小周', 'city': '杭州'}
key        : basic
namespace  : ('u1', 'profile')
created_at : 2026-08-13 19:00:08.852852+00:00
updated_at : 2026-08-13 19:00:08.852857+00:00

Item 的六个字段：

updated_at 比想象中有用：它是「这条记忆有多旧」的唯一依据。当你要做「三个月没被更新的偏好自动降权」这类策略时，只有它能告诉你答案。

警告：created_at 在两种 store 上语义不同（实测）。 InMemoryStore 覆盖写入时会重新生成整个 Item，created_at 被重置成当前时间；PostgresStore 走 upsert，created_at 保留首次写入时间。

InMemoryStore   覆盖后 created_at 是否被重置: True     ← 每次 put 都刷新
PostgresStore   覆盖后 created_at 是否被重置: False    ← 保留首次时间

所以「这条记忆是什么时候第一次学到的」不能依赖 created_at。要审计首次时间，请自己在 value 里存一个 "first_seen" 字段。顺便一提，InMemoryStore 返回 UTC 时间，PostgresStore 返回数据库本地时区（本机实测是 +08:00），跨实现比较时间时记得统一时区。

想序列化成 JSON（比如落日志、给前端），用 .dict()：

# json 用来把 dict 打成字符串
import json

# .dict() 把 Item 转成普通字典，注意时间戳是 datetime 对象
raw = item.dict()
# default=str 让 json 能处理 datetime；ensure_ascii=False 保留中文
print(json.dumps(raw, ensure_ascii=False, default=str))

输出（实测）：

{"namespace": ["u1", "profile"], "key": "basic", "value": {"name": "小周", "city": "杭州"}, "created_at": "2026-08-13T19:00:08.852852+00:00", "updated_at": "2026-08-13T19:00:08.852857+00:00"}

注意 namespace 变成了 ["u1", "profile"]——元组序列化成 JSON 后就是列表。如果你把这份 JSON 存下来再读回去当 namespace 用，记得 tuple(...) 转回去，否则会报类型错误。

2.3 四个基本操作 #

store 一共五个方法，每个都有异步版（前面加 a）。API 面就这么大，先摆在一起看一眼：

注意 search 的第一个参数叫 namespace_prefix 而不是 namespace——这个命名已经剧透了 §3.1 的行为。先把四个最常用的跑一遍：

# InMemoryStore：内存版长期记忆，进程退出即丢，适合本地开发
from langgraph.store.memory import InMemoryStore

# 新建一个干净的 store 做演示
demo = InMemoryStore()

# ① put：写入。同一个 namespace 下写三条笔记，key 各不相同
demo.put(("u1", "memories"), "m1", {"text": "养了两只猫", "kind": "fact"})
# 第二条：kind 标成 pref，下面演示 filter 时用来区分
demo.put(("u1", "memories"), "m2", {"text": "偏好简短回答", "kind": "pref"})
# 第三条：又是一条 fact，稍后会被 delete 掉
demo.put(("u1", "memories"), "m3", {"text": "对乳制品过敏", "kind": "fact"})

# ② get：按 key 精确取一条
print("get m2   :", demo.get(("u1", "memories"), "m2").value)
# 取不存在的 key 返回 None，不抛异常——所以必须自己判空
print("get 不存在:", demo.get(("u1", "memories"), "m9"))

# ③ search：列举 / 检索。不传 query 就是「列出这个命名空间下的记录」
found = demo.search(("u1", "memories"))
# 三条都在，说明 search 不传条件时等价于「列举」
print("search 条数:", len(found))
# filter 按 value 里的字段做精确匹配（不是模糊匹配）
facts = demo.search(("u1", "memories"), filter={"kind": "fact"})
# 只剩两条 fact，pref 那条被筛掉了
print("filter fact:", [i.value["text"] for i in facts])

# ④ delete：按 key 删除，删不存在的 key 同样不报错
demo.delete(("u1", "memories"), "m3")
# 删完还剩两条，确认删除真的生效了
print("删除后条数:", len(demo.search(("u1", "memories"))))

输出（实测）：

get m2   : {'text': '偏好简短回答', 'kind': 'pref'}
get 不存在: None
search 条数: 3
filter fact: ['养了两只猫', '对乳制品过敏']
删除后条数: 2

三个必须记住的行为：

• get 未命中返回 None，不抛异常。 忘了判空就会 AttributeError: 'NoneType' object has no attribute 'value'——这是本章最常见的报错。
• delete 删不存在的 key 静默成功。 别指望它告诉你「本来就没有」。
• filter 是精确相等匹配，filter={"kind": "fact"} 只匹配 value["kind"] == "fact"，不做包含、不做模糊。想按语义找要用 query（§6）。

异步版行为完全一致，且同一个 store 实例可以混用同步和异步方法（实测：put 写入的数据可以用 aget 读到）。在 async def 节点里请一律用 a* 版，避免阻塞事件循环。

2.4 put 是整体覆盖，不是字段合并 #

这是长期记忆里排名第一的静默数据丢失，一定要亲手验证一遍：

# InMemoryStore：内存版长期记忆，进程退出即丢，适合本地开发
from langgraph.store.memory import InMemoryStore

# 干净的 store
p = InMemoryStore()

# 第一次写入：完整档案，两个字段都在
p.put(("u1", "profile"), "basic", {"name": "小周", "city": "杭州"})
# 读回来确认：name 和 city 都存进去了
print("第一次:", p.get(("u1", "profile"), "basic").value)

# 第二次只写一个字段——很多人以为这是「更新 name」
p.put(("u1", "profile"), "basic", {"name": "小周"})
# 再读一次：city 已经不见了，这就是整体覆盖的后果
print("第二次:", p.get(("u1", "profile"), "basic").value)

输出（实测）：

第一次: {'name': '小周', 'city': '杭州'}
第二次: {'name': '小周'}

city 没了。put 的语义是「把这个 key 的内容整体替换成新 value」，和 dict.update() 完全不同。

正确的更新写法是「读—改—写」：

# InMemoryStore：内存版长期记忆，进程退出即丢，适合本地开发
from langgraph.store.memory import InMemoryStore

# 干净的 store
p = InMemoryStore()

# 第一次写入：完整档案，两个字段都在
p.put(("u1", "profile"), "basic", {"name": "小周", "city": "杭州"})
# 读回来确认：name 和 city 都存进去了
print("第一次:", p.get(("u1", "profile"), "basic").value)

# 读旧值（可能是 None）
old = p.get(("u1", "profile"), "basic")
# 转成普通字典，没有旧值就从空字典开始
merged = dict(old.value) if old else {}
# 在字典上做增量修改：把上一步被冲掉的 city 补回来
merged["city"] = "杭州"
# 再加一个新字段，说明「合并」既能补旧的也能加新的
merged["vip"] = True
# 整体写回：put 永远是整体覆盖，所以要交给它一个完整的字典
p.put(("u1", "profile"), "basic", merged)
# 确认三个字段都在
print("合并后:", p.get(("u1", "profile"), "basic").value)
# 顺手验证上面那条警告：InMemoryStore 里覆盖写入会把 created_at 也刷新
after = p.get(("u1", "profile"), "basic")
# 新旧 created_at 不相等，说明它确实被重置了，不能拿来当「首次写入时间」
print("created_at 被刷新:", after.created_at != old.created_at)

输出（实测）：

合并后: {'name': '小周', 'city': '杭州', 'vip': True}
created_at 被刷新: True

为什么 LangGraph 不提供 merge 或 patch？因为「怎么合并」是业务问题：数组字段该追加还是替换？空字符串算不算「要清空」？框架不猜，交给你写三行代码。 §4.2 会给出一个生产可用的合并函数，§7.5 会讲一个更隐蔽的变体：用抽取模型的输出直接覆盖，会把没提到的字段清空。

3. 命名空间设计 #

namespace 是你唯一的数据组织手段——没有表、没有索引、没有外键。

3.1 search 是前缀匹配，不是精确匹配 #

这是命名空间最反直觉的一点：search(("u1",)) 会把 ("u1", "profile")、("u1", "memories") 里的记录全部捞出来。

from langgraph.store.memory import InMemoryStore
# 准备四个命名空间的数据
s = InMemoryStore()
# u1 的档案
s.put(("u1", "profile"), "basic", {"name": "小周"})
# u1 的笔记
s.put(("u1", "memories"), "m1", {"text": "养了两只猫"})
# u1 的订单：三个不同的第二层，用来观察前缀匹配
s.put(("u1", "orders"), "o1", {"no": "A1002"})
# 另一个用户 u2，key 故意和 u1 的重名，验证 namespace 才是隔离边界
s.put(("u2", "memories"), "m1", {"text": "另一个用户"})

# 传两层：只命中这一层
print("('u1','memories'):", [(i.namespace, i.key) for i in s.search(("u1", "memories"))])
# 传一层：u1 下面所有子空间都会命中
print("('u1',)          :", [(i.namespace, i.key) for i in s.search(("u1",))])
# 传空元组：整个 store
print("()               :", [(i.namespace, i.key) for i in s.search(())])

输出（实测）：

('u1','memories'): [(('u1', 'memories'), 'm1')]
('u1',)          : [(('u1', 'profile'), 'basic'), (('u1', 'memories'), 'm1'), (('u1', 'orders'), 'o1')]
()               : [(('u1', 'profile'), 'basic'), (('u1', 'memories'), 'm1'), (('u1', 'orders'), 'o1'), (('u2', 'memories'), 'm1')]

前缀匹配是特性不是缺陷，它让「导出某个用户的全部长期记忆」变成一行 search((user_id,))。但它也带来两个陷阱：

• 想精确限定一层，必须传完整 namespace，或者拿到结果后自己按 item.namespace 过滤；
• search(()) 会扫全库，条数一多就是慢查询，生产代码里不要写。

3.2 一套够用的命名空间约定 #

命名空间设计没有标准答案，但有一个原则：把「隔离边界」放在前面，把「数据类别」放在后面。因为前缀匹配只支持从左往右，放错顺序会让「按用户导出」变成不可能。

推荐从这套约定起步：

反面例子：("profile", user_id)（类别在前）。这么设计之后，「删除某用户全部数据」就得遍历所有类别，而不是一次 search((user_id,))。

三条硬约定：

1. user_id 必须来自可信来源（登录态），绝不能让模型生成——否则模型可以「越权」读写别人的记忆。§5.4 会展示怎么用 context 保证这一点。
2. 同一个项目里 namespace 层级要统一。有的地方写 (user_id, "memories")、有的地方写 (user_id, "memory")，数据就会分裂成两份，而且不会报错。
3. key 的生成规则要想清楚：profile 用固定 key（"basic"），collection 用 uuid4。§11 会讲为什么不能用「当前条数 + 1」。

3.3 list_namespaces：巡检与「幽灵命名空间」 #

list_namespaces 是你唯一的「目录树」视图，做数据巡检、批量迁移时必用：

# 列出全部命名空间
print("全部          :", s.list_namespaces())
# prefix 只看某个前缀下的
print("prefix=('u1',):", s.list_namespaces(prefix=("u1",)))
# max_depth 把结果截断到指定层数，用来看「有哪些用户」
print("max_depth=1   :", s.list_namespaces(max_depth=1))
# suffix 按结尾筛选，用来看「所有用户的 memories」
print("suffix=memories:", s.list_namespaces(suffix=("memories",)))

输出（实测）：

全部          : [('u1', 'memories'), ('u1', 'orders'), ('u1', 'profile'), ('u2', 'memories')]
prefix=('u1',): [('u1', 'memories'), ('u1', 'orders'), ('u1', 'profile')]
max_depth=1   : [('u1',), ('u2',)]
suffix=memories: [('u1', 'memories'), ('u2', 'memories')]

max_depth=1 相当于「列出所有用户」，是做全量遍历（比如夜间批处理所有用户的记忆）的入口。

接着是一个只有实测才能发现的坑：

from langgraph.store.memory import InMemoryStore
# 一个全新的 store，什么都没写
ghost = InMemoryStore()
# 目录是空的，符合预期
print("初始:", ghost.list_namespaces())
# 只做一次「读」，读一个完全不存在的命名空间
print("读不存在的:", ghost.get(("ghost", "profile"), "k"))
# 再看一次目录
print("读之后:", ghost.list_namespaces())
# 对比：search 不存在的命名空间不会有这个副作用
print("search 不存在的:", ghost.search(("nobody",)))
# 目录里仍然只有 get 留下的那一个，search 没有再添新的
print("search 之后:", ghost.list_namespaces())

输出（实测）：

初始: []
读不存在的: None
读之后: [('ghost', 'profile')]
search 不存在的: []
search 之后: [('ghost', 'profile')]

get 一个不存在的命名空间，会在 InMemoryStore 里留下一个空目录。 原因是它内部用 defaultdict 存数据，读操作也会建键。后果：

• 巡检脚本会报告一堆「条数 = 0」的命名空间（§10 本章实战跑一遍，巡检结果里就有一个）；
• 如果你用 list_namespaces(max_depth=1) 统计「有多少活跃用户」，数字会虚高。

这是 InMemoryStore 特有的实现细节，PostgresStore 不会（实测：把同一段代码的 store 换成 PostgresStore，巡检结果里没有这个空命名空间）。巡检时要以「条数 > 0」为准，不要以「命名空间存在」为准。

3.4 limit 静默截断、offset 分页、顺序不保证 #

search 有三个默认值会咬人，一次说清：

from langgraph.store.memory import InMemoryStore
# 造 25 条数据
big = InMemoryStore()
# 循环 25 次，key 补零成 k01…k25，方便肉眼看顺序
for n in range(1, 26):
    # 每条只放一个数字，内容不重要，重要的是条数
    big.put(("u1", "logs"), f"k{n:02d}", {"n": n})

# 不传 limit：默认只返回 10 条，而且没有任何「还有更多」的提示
print("默认返回条数:", len(big.search(("u1", "logs"))))
# limit=3 只要前三条
print("limit=3      :", [i.key for i in big.search(("u1", "logs"), limit=3)])
# offset=3 跳过前三条，再取三条——limit / offset 组合就是翻页
print("offset=3     :", [i.key for i in big.search(("u1", "logs"), limit=3, offset=3)])

# 正确的全量读取方式：翻页直到空
keys, offset = [], 0
# 没有「总数」接口，只能循环到取不出东西为止
while True:
    # 每页 10 条，offset 每轮往后挪一页
    page = big.search(("u1", "logs"), limit=10, offset=offset)
    # 返回空列表说明翻完了
    if not page:
        # 唯一的退出条件：没有「总数」可以提前算出来
        break
    # 把这一页的 key 收进结果
    keys.extend(i.key for i in page)
    # 游标前进一页
    offset += 10
# 25 条一条不漏，首尾也对得上
print("翻页取到条数:", len(keys), "| 首尾:", keys[0], keys[-1])

输出（实测）：

默认返回条数: 10
limit=3      : ['k01', 'k02', 'k03']
offset=3     : ['k04', 'k05', 'k06']
翻页取到条数: 25 | 首尾: k01 k25

三条结论：

1. 默认 limit=10，超出部分静默丢弃。 「我明明存了 30 条偏好，Agent 只知道 10 条」——十有八九是这里。要么把 limit 设到业务上限之上，要么翻页。
2. 没有总数接口。 想知道「一共多少条」只能翻完，或者自己维护计数。
3. 顺序不保证跨实现一致。 InMemoryStore 按插入顺序返回（最新的在最后），PostgresStore 按 updated_at 倒序返回（最新的在最前，实测见 §9.3）。

第 3 条尤其阴险：本地用 InMemoryStore 写「取最近 5 条笔记」时你可能写成 search(...)[-5:]，换成 Postgres 上线后语义就反了，而且不报错。要「最近 N 条」，请显式排序：

# 不要依赖 store 的默认顺序，自己按 updated_at 排
recent = sorted(big.search(("u1", "logs"), limit=100), key=lambda i: i.updated_at, reverse=True)[:5]
# 打印排序后的 key，确认是自己控制的顺序
print("最近 5 条:", [i.key for i in recent])

输出（实测，InMemoryStore 里同一批数据时间戳极近，顺序以实际写入为准）：

最近 5 条: ['k25', 'k24', 'k23', 'k22', 'k21']

3.5 非法 namespace 的四种报错 #

命名空间的字符串不是随便取的，四条规则都会立刻抛错（这是好事，属于响亮的失败）：

# InvalidNamespaceError 是 store 层的专用异常
from langgraph.store.base import InvalidNamespaceError
from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
# 逐个尝试四种非法写法：空标签、含点号、保留字前缀、空元组
for ns in [("u1", ""), ("u1", "a.b"), ("langgraph", "x"), ()]:
    # 用 try 包住，让四种都跑完而不是第一次就中断
    try:
        # 校验发生在 put 内部，写入之前
        store.put(ns, "k", {"v": 1})
        # 只有合法的 namespace 才会走到这一行
        print(f"{ns!r:20s} -> 允许")
    # InvalidNamespaceError 是这四种非法情况共用的异常类型
    except InvalidNamespaceError as e:
        # 把异常信息原样打出来，四条报错文案各不相同
        print(f"{ns!r:20s} -> {e}")

输出（实测）：

('u1', '')           -> Namespace labels cannot be empty strings. Got  in ('u1', '')
('u1', 'a.b')        -> Invalid namespace label 'a.b' found in ('u1', 'a.b'). Namespace labels cannot contain periods ('.').
('langgraph', 'x')   -> Root label for namespace cannot be "langgraph". Got: ('langgraph', 'x')
()                   -> Namespace cannot be empty.

对照表：

「邮箱当 user_id」是真实事故高发点：("zhou.wang@x.com", "profile") 会直接抛 InvalidNamespaceError。生产上建议统一用不含 . 的稳定 id（数据库主键、UUID、或邮箱的哈希）。

4. 两种记忆形态：profile 还是 collection #

知道怎么读写之后，真正的设计问题来了：同一个用户的记忆，应该存成「一份不断更新的档案」还是「一堆不断增加的笔记」？

官方把这两种叫做 profile 和 collection。不是品味上的二选一，而是适用边界不同的两种数据结构。

4.1 两种形态的对比 #

选择口诀：

能列成表单的 → profile。只能记成便签的 → collection。

绝大多数生产系统两者都要：profile 装那 5～10 个「不能错」的字段，collection 装源源不断的零散事实。本章实战（§10）就是这种混合形态。

4.2 profile：一个安全的合并函数 #

profile 的难点都在「合并」。§2.4 已经知道 put 是覆盖，这里封装成生产可用的函数：

# BaseStore 是所有 store 实现的公共基类，用作类型标注
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore


def merge_profile(store: BaseStore, user_id: str, patch: dict) -> dict:
    """把 patch 合并进用户档案，只覆盖非空字段，返回合并后的档案。"""
    # profile 形态：命名空间固定两层，key 固定叫 basic
    ns = (user_id, "profile")
    # 先读旧档案，未命中时 get 返回 None
    old = store.get(ns, "basic")
    # 复制成普通字典再改，避免直接修改 Item 内部的对象
    merged = dict(old.value) if old else {}
    # 记下合并前的样子，用来判断这一轮到底有没有新信息
    before = dict(merged)
    # 逐个字段合并
    for field, value in patch.items():
        # 值为 None 表示「本轮没提到」，绝不能拿它覆盖旧值
        if value is not None:
            # 只有非空值才写进合并结果
            merged[field] = value
    # 没有任何变化就不写：省一次 IO，也避免无意义地刷新 updated_at
    if merged != before:
        # 整体写回，key 固定是 basic
        store.put(ns, "basic", merged)
    # 返回结果，方便调用方打印或继续用
    return merged


# 新建 store 演示三次增量更新
prof = InMemoryStore()
# 第一轮：用户说了姓名和城市
print(merge_profile(prof, "u1", {"name": "小周", "city": "杭州", "style": None}))
# 第二轮：只说了风格偏好，姓名城市必须保住
print(merge_profile(prof, "u1", {"name": None, "city": None, "style": "简短，不寒暄"}))
# 第三轮：什么都没提到，档案应当原样不动
print(merge_profile(prof, "u1", {"name": None, "city": None, "style": None}))

输出（实测）：

{'name': '小周', 'city': '杭州'}
{'name': '小周', 'city': '杭州', 'style': '简短，不寒暄'}
{'name': '小周', 'city': '杭州', 'style': '简短，不寒暄'}

这个函数有三处刻意设计，每一处对应一类线上事故：

「空字符串算不算清空」是你必须自己决定的：上面的实现里 "" 会覆盖旧值（因为它不是 None）。如果用户说「不用记我的称呼了」，你希望的语义可能是删除字段，那就要显式处理，例如约定 patch 里传 "__DELETE__" 表示删键。

4.3 collection：一条一个 key #

collection 的写入很简单，难点在 key 怎么取：

# uuid 生成不会撞的 key
import uuid

from langgraph.store.memory import InMemoryStore

# 新建 store
notes = InMemoryStore()
# 命名空间与 profile 平级，方便 search((user_id,)) 一次取全量
NOTE_NS = ("u1", "memories")

# 三条零散事实，一条一个 key
for text in ["家里有两只猫", "对乳制品过敏", "偏好晚间配送"]:
    # key 用随机值：并行写入不会互相覆盖（§11 会讲为什么不能用「条数 + 1」）
    notes.put(NOTE_NS, f"note-{uuid.uuid4().hex[:8]}", {"text": text})

# 读全部：limit 一定要给足，否则默认只回 10 条
allnotes = notes.search(NOTE_NS, limit=100)
# 打印总条数，确认三条都写进去了
print("条数:", len(allnotes))
# 逐条打印，观察随机生成的 key
for i in allnotes:
    # key 是随机的，text 才是业务内容
    print("   ", i.key, i.value["text"])

输出（实测，key 每次都不同）：

条数: 3
    note-3f9c1a2b 家里有两只猫
    note-7d0e4c55 对乳制品过敏
    note-91ab6f30 偏好晚间配送

collection 的三个必答问题，越早想清楚越省事：

1. 怎么去重？ 模型很爱重复记同一件事。最简做法是写入前按原文比对（本章实战用的就是这个），进阶做法是语义查重（§7.6，需要嵌入模型）。
2. 怎么淘汰？ 笔记会无限增长。常见策略：只注入最近 N 条（按 updated_at 排序）、给 value 加 "confidence" 字段做加权、或者用 TTL 自动过期（§9.3）。
3. 怎么修正？ 用户说「我搬到上海了」，旧笔记「住杭州」还在。要么让模型显式调「删除笔记」工具，要么把这类可变字段放进 profile（这就是为什么城市应该在 profile 而不是 collection）。

4.4 混合方案 #

把两者拼起来，加上共享知识，得到一个实用的三层结构：

(user_id, "profile")    basic                 ← 5 个固定字段，合并更新，注入系统提示
(user_id, "memories")   note-xxxx（多条）      ← 零散事实，去重写入，只注入最近 5 条
("org", "policies")     refund / shipping …   ← 全体共享，只读，工具按需查

三层的读取时机也不同，这一点常被忽略：

「档案每轮必读、知识按需查」是一条很实用的经验：档案只有几十字，无脑注入的成本远低于让模型多跑一次工具往返；而制度条款可能上百条，全塞进提示既贵又干扰，交给工具更合适。

5. 在 create_agent 里用 store #

前面把 store 当普通字典用。这一节接上 Agent：多传一个参数，再决定「谁来读、谁来写」。

5.1 一行开关：store= #

# create_agent 是 Agent 工厂
from langchain.agents import create_agent
from langgraph.store.memory import InMemoryStore
# 短期记忆仍然靠 checkpointer
from langgraph.checkpoint.memory import InMemorySaver
# dataclass 用来声明 context 的结构
from dataclasses import dataclass
# dotenv 负责把 .env 里的 API Key 注入环境变量
from dotenv import load_dotenv

# 读取 .env（DEEPSEEK_API_KEY 等）
load_dotenv()


# dataclass 装饰器：几行就能声明一个带字段的上下文类
@dataclass
class Ctx:
    """每次请求都要带的上下文：当前是谁在说话。"""

    # user_id 决定长期记忆的命名空间，必须由业务代码传入
    user_id: str


# 长期记忆容器，整个应用共享一个实例
store = InMemoryStore()
# 预置一条档案，模拟「这个用户以前来过」
store.put(("u_zhou", "profile"), "basic", {"name": "小周", "city": "杭州"})

# 组装 Agent：store 管跨会话，checkpointer 管会话内
agent = create_agent(
    # 对话模型
    model="deepseek:deepseek-v4-flash",
    # 先不给工具，这一节只演示接线
    tools=[],
    # 长期记忆：跨 thread
    store=store,
    # 短期记忆：thread 内
    checkpointer=InMemorySaver(),
    # 声明 context 结构，工具和中间件才能读到 user_id
    context_schema=Ctx,
    # 系统提示暂时写死
    system_prompt="你是客服助手，回答简短。",
)
# 确认参数确实挂上了（agent 是编译好的图，store 会体现在运行时）
print("agent 类型:", type(agent).__name__)

输出（实测）：

agent 类型: CompiledStateGraph

注意：store= 只是「把仓库交给 Agent」，不会让模型自动记住任何东西。 谁读、谁写、读哪个命名空间，都要你显式写出来。这和 checkpointer= 完全不同——后者一传就自动生效。长期记忆最容易在这里被误解：

5.2 工具里读写：ToolRuntime.store #

最常见的接法：把「记住」和「回忆」做成工具，让模型自己决定何时调用：

# create_agent 是 Agent 工厂
from langchain.agents import create_agent
from langgraph.store.memory import InMemoryStore

# 短期记忆仍然靠 checkpointer
from langgraph.checkpoint.memory import InMemorySaver

# dataclass 用来声明 context 的结构
from dataclasses import dataclass

# dotenv 负责把 .env 里的 API Key 注入环境变量
from dotenv import load_dotenv

# tool 装饰器把函数变成工具；ToolRuntime 让工具拿到运行时对象
from langchain.tools import ToolRuntime, tool

# json 用于把档案拼成字符串回灌给模型
import json

# BaseStore 是 store 的抽象基类，所有 store 实现都要继承它
from langgraph.store.base import BaseStore

# 读取 .env（DEEPSEEK_API_KEY 等）
load_dotenv()


# dataclass 装饰器：几行就能声明一个带字段的上下文类
@dataclass
class Ctx:
    """每次请求都要带的上下文：当前是谁在说话。"""

    # user_id 决定长期记忆的命名空间，必须由业务代码传入
    user_id: str


# 长期记忆容器，整个应用共享一个实例
store = InMemoryStore()


def merge_profile(store: BaseStore, user_id: str, patch: dict) -> dict:
    """把 patch 合并进用户档案，只覆盖非空字段，返回合并后的档案。"""
    # profile 形态：命名空间固定两层，key 固定叫 basic
    ns = (user_id, "profile")
    # 先读旧档案，未命中时 get 返回 None
    old = store.get(ns, "basic")
    # 复制成普通字典再改，避免直接修改 Item 内部的对象
    merged = dict(old.value) if old else {}
    # 记下合并前的样子，用来判断这一轮到底有没有新信息
    before = dict(merged)
    # 逐个字段合并
    for field, value in patch.items():
        # 值为 None 表示「本轮没提到」，绝不能拿它覆盖旧值
        if value is not None:
            # 只有非空值才写进合并结果
            merged[field] = value
    # 没有任何变化就不写：省一次 IO，也避免无意义地刷新 updated_at
    if merged != before:
        # 整体写回，key 固定是 basic
        store.put(ns, "basic", merged)
    # 返回结果，方便调用方打印或继续用
    return merged


@tool
def save_profile(field: str, value: str, runtime: ToolRuntime[Ctx]) -> str:
    """把用户档案里的一个字段写入长期记忆。field 取 name / city / style。"""
    # runtime.store 就是 create_agent 里传进来的那个 store 实例
    st = runtime.store
    # runtime.context 就是 invoke 时传的 Ctx 对象，user_id 决定写到谁名下
    merged = merge_profile(st, runtime.context.user_id, {field: value})
    # 返回值会变成 ToolMessage 回灌给模型
    return f"已记住 {field}={value}，当前档案：{json.dumps(merged, ensure_ascii=False)}"


@tool
def read_profile(runtime: ToolRuntime[Ctx]) -> str:
    """读取当前用户的长期档案。"""
    # 同样通过 runtime 定位到当前用户
    item = runtime.store.get((runtime.context.user_id, "profile"), "basic")
    # 未命中要明确告诉模型「没有」，否则它会自己编
    return json.dumps(item.value, ensure_ascii=False) if item else "暂无档案"


# 带工具重新组装 Agent
agent_tools = create_agent(
    # 对话模型
    model="deepseek:deepseek-v4-flash",
    # 两个工具：一个写一个读
    tools=[save_profile, read_profile],
    # 把长期记忆容器交给 Agent，工具才能通过 runtime.store 拿到它
    store=store,
    # 短期记忆照旧，两者并存
    checkpointer=InMemorySaver(),
    # 声明 context 结构，工具签名里的 ToolRuntime[Ctx] 才有意义
    context_schema=Ctx,
    # 提示里要说清什么时候调工具，否则模型经常不调
    system_prompt=(
        # 写入时机
        "你是客服助手。用户说出个人信息时必须调用 save_profile 记住；"
        # 读取时机
        "被问到个人信息时调用 read_profile 查询。回答简短。"
    ),
)

# 第一轮：让它写
r1 = agent_tools.invoke(
    # 用户主动报出姓名和城市
    {"messages": [{"role": "user", "content": "我叫小周，住杭州，请记住。"}]},
    # thread_id 决定短期记忆落在哪条会话线上
    config={"configurable": {"thread_id": "t1"}},
    # user_id 决定长期记忆写到谁名下
    context=Ctx(user_id="u_zhou"),
)
# 打印最后一条 AI 回复
print("回复:", r1["messages"][-1].content)
# 直接查 store，确认数据真的落盘了（不要只看模型说「记住了」）
print("store:", store.get(("u_zhou", "profile"), "basic").value)

输出（实测，措辞会不同）：

回复: 好的，已记住：小周，杭州。还有什么需要帮忙的吗？
store: {'name': '小周', 'city': '杭州'}

验收长期记忆的第一原则：别看模型嘴上怎么说，看 store 里有什么。 「已记住」只说明它调了工具，不说明数据落到了你以为的位置——命名空间写错、user_id 传错，模型照样会说「记住了」。

5.3 跨会话验证：换 thread_id 还记得 #

这是本章的核心验收：换一个全新的 thread_id，短期记忆清零，看还能不能答对：

# 换 thread_id：这是一条全新的会话线，messages 从空开始
r2 = agent_tools.invoke(
    # 只问不说，答案只能来自长期记忆
    {"messages": [{"role": "user", "content": "你还记得我住哪个城市吗？"}]},
    # thread_id 变了
    config={"configurable": {"thread_id": "t2"}},
    # user_id 没变，所以长期记忆还是这个人的
    context=Ctx(user_id="u_zhou"),
)
# 回复应当包含「杭州」
print("回复:", r2["messages"][-1].content)
# 消息条数很少，说明它不是靠会话历史，而是靠工具查 store
print("本轮消息条数:", len(r2["messages"]))

输出（实测）：

回复: 你住在杭州。
本轮消息条数: 4

4 条消息很说明问题：HumanMessage → 带 tool_calls 的 AIMessage → ToolMessage → 最终 AIMessage。没有历史对话，答案完全来自 store。对比第 11 章：那里靠的是 messages 里的原话，一换 thread_id 就没了。

5.4 隔离靠 context，不是靠 store #

再换一个 user_id，同样的问题应该查不到：

# 第三轮：thread 和 user 都换了
r3 = agent_tools.invoke(
    # 问题和上一轮一字不差，只有身份变了
    {"messages": [{"role": "user", "content": "你还记得我住哪个城市吗？"}]},
    # 又是一条全新会话线
    config={"configurable": {"thread_id": "t3"}},
    # 换成另一个用户
    context=Ctx(user_id="u_other"),
)
# 应当明确表示不知道
print("回复:", r3["messages"][-1].content)
# store 里确实没有这个人的档案
print("u_other 档案:", store.get(("u_other", "profile"), "basic"))

输出（实测）：

回复: 我还没有你的城市信息，能告诉我吗？
u_other 档案: None

这里有一个安全性要点，务必记住：

user_id 必须从 context 传入，绝不能让模型自己决定。

若把工具签名写成 def read_profile(user_id: str) -> str，user_id 就成了模型可填的参数——用户说一句「帮我查一下 u_zhou 的档案」，模型就会照做。这是越权漏洞，不是功能。 正确做法如本节：user_id 只出现在 Ctx 里，由业务代码在 invoke 时注入，模型既看不到也改不了。

5.5 更省的读法：dynamic_prompt 直接注入 #

用工具读档案要多一次「模型 → 工具 → 模型」往返（上面 4 条消息就是证据）。档案这种「每轮都要用、只有几十字」的数据，更好的做法是每次调模型前直接拼进系统提示：

# create_agent 是 Agent 工厂
from langchain.agents import create_agent
from langgraph.store.memory import InMemoryStore

# 短期记忆仍然靠 checkpointer
from langgraph.checkpoint.memory import InMemorySaver

# dataclass 用来声明 context 的结构
from dataclasses import dataclass

# dotenv 负责把 .env 里的 API Key 注入环境变量
from dotenv import load_dotenv

# tool 装饰器把函数变成工具；ToolRuntime 让工具拿到运行时对象
from langchain.tools import ToolRuntime, tool

# dynamic_prompt 装饰器：每次调模型前动态生成系统提示
# ModelRequest 是这次模型请求的完整描述
from langchain.agents.middleware import ModelRequest, dynamic_prompt

# json 用于把档案拼成字符串回灌给模型
import json

# BaseStore 是 store 的抽象基类，所有 store 实现都要继承它
from langgraph.store.base import BaseStore

# 读取 .env（DEEPSEEK_API_KEY 等）
load_dotenv()


# dataclass 装饰器：几行就能声明一个带字段的上下文类
@dataclass
class Ctx:
    """每次请求都要带的上下文：当前是谁在说话。"""

    # user_id 决定长期记忆的命名空间，必须由业务代码传入
    user_id: str


# 长期记忆容器，整个应用共享一个实例
store = InMemoryStore()


def merge_profile(store: BaseStore, user_id: str, patch: dict) -> dict:
    """把 patch 合并进用户档案，只覆盖非空字段，返回合并后的档案。"""
    # profile 形态：命名空间固定两层，key 固定叫 basic
    ns = (user_id, "profile")
    # 先读旧档案，未命中时 get 返回 None
    old = store.get(ns, "basic")
    # 复制成普通字典再改，避免直接修改 Item 内部的对象
    merged = dict(old.value) if old else {}
    # 记下合并前的样子，用来判断这一轮到底有没有新信息
    before = dict(merged)
    # 逐个字段合并
    for field, value in patch.items():
        # 值为 None 表示「本轮没提到」，绝不能拿它覆盖旧值
        if value is not None:
            # 只有非空值才写进合并结果
            merged[field] = value
    # 没有任何变化就不写：省一次 IO，也避免无意义地刷新 updated_at
    if merged != before:
        # 整体写回，key 固定是 basic
        store.put(ns, "basic", merged)
    # 返回结果，方便调用方打印或继续用
    return merged


@tool
def save_profile(field: str, value: str, runtime: ToolRuntime[Ctx]) -> str:
    """把用户档案里的一个字段写入长期记忆。field 取 name / city / style。"""
    # runtime.store 就是 create_agent 里传进来的那个 store 实例
    st = runtime.store
    # runtime.context 就是 invoke 时传的 Ctx 对象，user_id 决定写到谁名下
    merged = merge_profile(st, runtime.context.user_id, {field: value})
    # 返回值会变成 ToolMessage 回灌给模型
    return f"已记住 {field}={value}，当前档案：{json.dumps(merged, ensure_ascii=False)}"


# 带工具重新组装 Agent
agent_tools = create_agent(
    # 对话模型
    model="deepseek:deepseek-v4-flash",
    # 两个工具：一个写一个读
    tools=[save_profile],
    # 把长期记忆容器交给 Agent，工具才能通过 runtime.store 拿到它
    store=store,
    # 短期记忆照旧，两者并存
    checkpointer=InMemorySaver(),
    # 声明 context 结构，工具签名里的 ToolRuntime[Ctx] 才有意义
    context_schema=Ctx,
    # 提示里要说清什么时候调工具，否则模型经常不调
    system_prompt=(
        # 写入时机
        "你是客服助手。用户说出个人信息时必须调用 save_profile 记住；"
        # 读取时机
        "被问到个人信息时调用 read_profile 查询。回答简短。"
    ),
)


@dynamic_prompt
def prompt_with_profile(request: ModelRequest) -> str:
    """调模型前把长期档案拼进系统提示。"""
    # 中间件通过 request.runtime 拿到 store 与 context
    st = request.runtime.store
    # 当前是谁在说话，决定读哪个命名空间
    uid = request.runtime.context.user_id
    # 没接 store 时降级为「无记忆」，不要直接崩
    item = st.get((uid, "profile"), "basic") if st else None
    # 固定的角色说明
    base = "你是客服助手，回答简短。"
    # 没有档案就明确告诉模型「没有」，避免它编造
    if not item:
        # 显式声明「暂无」比什么都不说更能抑制幻觉
        return base + "\n（暂无该用户档案）"
    # 有档案就整段注入，JSON 形式最省 token 也最不容易被误读
    return base + "\n已知用户档案：" + json.dumps(item.value, ensure_ascii=False)


# 这个 Agent 不带任何工具，全靠提示注入
agent_prompt = create_agent(
    # 对话模型
    model="deepseek:deepseek-v4-flash",
    # 空工具列表：证明记忆确实来自提示而不是工具查询
    tools=[],
    # 中间件挂上去
    middleware=[prompt_with_profile],
    # 中间件要从这里读档案
    store=store,
    # 中间件要从 context 拿 user_id
    context_schema=Ctx,
)

# 第一轮：让它写
r1 = agent_tools.invoke(
    # 用户主动报出姓名和城市
    {"messages": [{"role": "user", "content": "我叫小周，住杭州，请记住。"}]},
    # thread_id 决定短期记忆落在哪条会话线上
    config={"configurable": {"thread_id": "t1"}},
    # user_id 决定长期记忆写到谁名下
    context=Ctx(user_id="u_zhou"),
)
# 打印最后一条 AI 回复
print("回复:", r1["messages"][-1].content)
# 直接查 store，确认数据真的落盘了（不要只看模型说「记住了」）
print("store:", store.get(("u_zhou", "profile"), "basic").value)

# 提问：模型应当直接答得出来
r4 = agent_prompt.invoke(
    # 「只回答城市名」是为了让输出好比对
    {"messages": [{"role": "user", "content": "我住哪儿？只回答城市名。"}]},
    # 不传 config 也可以，因为这个 Agent 没有 checkpointer
    context=Ctx(user_id="u_zhou"),
)
# 打印回复
print("回复:", r4["messages"][-1].content)
# 消息条数只有 2，说明零工具往返
print("本轮消息条数:", len(r4["messages"]))

输出（实测）：

回复: 杭州
本轮消息条数: 2

2 条消息——比工具方案少一轮模型往返。两种读法怎么选：

实践配比：档案 + 最近 3～5 条笔记走注入，历史全量与共享知识库走工具。 §10 实战就是这么做的。

5.6 没传 store 会怎样 #

这是个容易踩的坑——它不报错：

# 故意不传 store，看看工具里能拿到什么
probe = {}


@tool
def probe_store(runtime: ToolRuntime[Ctx]) -> str:
    """探测 runtime.store 的取值。"""
    # 把拿到的对象存到外部变量，方便主程序检查
    probe["store"] = runtime.store
    # 直接把它渲染成字符串回灌
    return f"store={runtime.store!r}"


# 这个 Agent 没有 store=
agent_nostore = create_agent(
    # 对话模型
    model="deepseek:deepseek-v4-flash",
    # 只挂探测工具
    tools=[probe_store],
    # context 照常声明，缺的只是 store
    context_schema=Ctx,
    # 强制它调工具，否则可能直接回答就结束了
    system_prompt="必须调用 probe_store 工具，然后简短回答。",
)
# 跑一轮
agent_nostore.invoke(
    # 内容无所谓，目的只是触发工具调用
    {"messages": [{"role": "user", "content": "探测一下"}]},
    # context 传了也没用，因为根本没有 store 可读
    context=Ctx(user_id="u_zhou"),
)
# 关键结论：不传 store 时，runtime.store 是 None，而不是抛异常
print("runtime.store =", probe["store"])

输出（实测）：

runtime.store = None

于是有两种翻车姿势：

6. 语义检索：让记忆「按意思」被找到 #

本节需要嵌入模型（DASHSCOPE_API_KEY），直接复用第 14 章的 embeddings_layer.py。

前面的检索都是「按 key 精确取」或「按字段精确过滤」。笔记类记忆天生模糊：用户问「我有什么忌口」，笔记里写的是「对乳制品过敏」——没有共同关键词。这时需要向量检索。

6.1 配 index：给 store 装上向量索引 #

# sys 用于把项目根目录加进模块搜索路径，好 import embeddings_layer
import sys

# 把项目根目录放到搜索路径最前面，保证 import 到的是本仓库的模块
sys.path.insert(0, r"d:\forever\docs\langrag")
# 第 14 章的统一嵌入层：换供应商只改那个文件
from embeddings_layer import get_embeddings

# 拿到 DashScope 的嵌入模型（text-embedding-v4）
emb = get_embeddings("dashscope")
# 先确认维度：dims 必须和模型实际输出一致，所以现算不硬编码
DIMS = len(emb.embed_query("测试"))
# 打印出来，方便和下面 index 里的配置对照
print("嵌入维度:", DIMS)

# 创建带语义索引的 store
sem = InMemoryStore(
    # index 就是语义检索的全部配置，只有三个键
    index={
        # embed：LangChain 的 Embeddings 对象，也可以传一个 (list[str]) -> list[list[float]] 的函数
        "embed": emb,
        # dims：向量维度
        "dims": DIMS,
        # fields：value 里哪些字段参与嵌入。"$" 表示整条文档
        "fields": ["text"],
    }
)
# 笔记的命名空间，下面反复用到，抽成常量
NS_SEM = ("u1", "memories")
# 写入四条笔记，每次 put 都会触发一次嵌入调用
for key, text in [
    # 宠物类事实
    ("m1", "用户家里养了两只布偶猫"),
    # 风格偏好
    ("m2", "用户希望回答尽量简短，不要寒暄"),
    # 地址信息
    ("m3", "用户的常用收货地址是杭州市西湖区"),
    # 忌口信息
    ("m4", "用户对乳制品过敏"),
]:
    # kind 字段不参与嵌入（fields 只声明了 text），但可以用来做 filter
    sem.put(NS_SEM, key, {"text": text, "kind": "fact"})
# 确认四条都写进去了
print("写入完成，条数:", len(sem.search(NS_SEM, limit=50)))

输出（实测）：

嵌入维度: 1024
写入完成，条数: 4

index 的三个键：

fields 用点号表示嵌套字段——这就是命名空间不能含 . 的原因（§3.5）：点号在这套 API 里是路径分隔符。

6.2 带 query 的 search #

配了 index 之后，search 多了两个关键能力：

• query：传一句自然语言。store 会把它嵌入成向量，再和已存条目比相似度，按相关度排序返回。
• score：命中结果上的相似度分数（常见是余弦相似度，越大越像）。没配 index 时这个字段是 None。

它和关键词检索的差别在于：问法和原文可以完全对不上词，只要意思近就能命中。比如笔记写「他养了猫」，用「宠物」也能搜到。

用的时候记住三点：

• 传了 query 就是语义检索，结果已按 score 降序。
• limit 只管条数；低分结果（比如 < 0.35）多半是噪声，生产上要再加分数阈值。
• 只有 fields 里声明过、且真正被嵌入的字段会参与比对——配错字段或 index=False 写入的条目，搜不着。

# 三个自然语言问题，注意它们和笔记原文几乎没有共同关键词
for q in ["宠物", "他不喜欢啰嗦的回复", "送货到哪里"]:
    # 传 query 就是语义检索，limit 控制返回条数
    hits = sem.search(NS_SEM, query=q, limit=2)
    # 先打印查询词，方便对照下面的命中
    print(f"query={q!r}")
    # 结果已按相似度从高到低排好序
    for i in hits:
        # score 是余弦相似度，越大越像
        print(f"    {i.key}  score={i.score:.4f}  {i.value['text']}")

输出（实测，分数会有细微差异）：

query='宠物'
    m1  score=0.4793  用户家里养了两只布偶猫
    m2  score=0.3168  用户希望回答尽量简短，不要寒暄
query='他不喜欢啰嗦的回复'
    m2  score=0.5254  用户希望回答尽量简短，不要寒暄
    m4  score=0.3326  用户对乳制品过敏
query='送货到哪里'
    m3  score=0.5788  用户的常用收货地址是杭州市西湖区
    m2  score=0.3687  用户希望回答尽量简短，不要寒暄

三个观察：

1. 排第一的都对：「宠物」找到猫、「啰嗦」找到简短偏好、「送货」找到地址。这就是语义检索的价值——用户的问法和记忆的写法可以完全不同。
2. 分数绝对值不高（0.47～0.58）。中文嵌入模型的余弦相似度普遍偏低，不要照搬「0.8 以上才算相关」这种经验值，要在自己的数据上量一遍再定阈值。
3. 第二名基本是噪声（0.31～0.37）。这说明 limit 不能傻给：limit=5 时后面几条纯属凑数。生产上应该同时设 limit 和分数下限：

# 只保留分数达标的结果，阈值要在自己的数据上量过
hits = [i for i in sem.search(NS_SEM, query="宠物", limit=5) if (i.score or 0) >= 0.45]
# 打印过滤后的命中
print("过滤后:", [(i.key, round(i.score, 3)) for i in hits])

输出（实测）：

过滤后: [('m1', 0.479)]

query 和 filter 可以叠加，语义在前、精确条件在后：

# 先按 kind 精确筛，再在其中做语义排序
hits = sem.search(NS_SEM, query="用户的生活习惯", filter={"kind": "fact"}, limit=3)
# 打印 key 与分数
print([(i.key, round(i.score, 4)) for i in hits])

输出（实测）：

[('m2', 0.4721), ('m3', 0.3904), ('m1', 0.3526)]

（四条笔记的 kind 都是 fact，所以 filter 只是没筛掉任何东西；换成 filter={"kind": "pref"} 就只剩符合条件的那些再排序。）

6.3 控制「嵌哪些字段」 #

一条笔记里常常既有正文，也有时间、类型这类结构化字段。结构化字段通常不该进向量索引：白花嵌入费用，还可能掺进噪声。API 给了三档字段级控制：

• put(..., index=False)：完全不嵌入。适合日志、原始事件——存着就行，别被语义检索捞出来。
• put(..., index=["note"])：只嵌 note 字段。字段名不叫 text、或想指定另一段文本参与检索时用。
• index={"fields": ["text"]}：在 store 级统一约定「默认嵌哪些字段」，全库一致。

多字段嵌入时还有一点：每个被选中的字段各自生成一条向量，检索时取各字段得分的最大值（max pooling）。标题和正文主题不一致时，相关的那一侧仍有机会排到前面。

# ① 写入时用 index=False：存得进去，但不参与语义检索
sem.put(NS_SEM, "m6", {"text": "用户讨厌香菜"}, index=False)
# ② 写入时用 index=[...]：覆盖 store 级的 fields 配置
sem.put(NS_SEM, "m7", {"text": "无关内容", "note": "用户是素食主义者"}, index=["note"])
# 验证 ②：搜「素食」应当命中 m7，尽管它的 text 字段毫不相关
hits = sem.search(NS_SEM, query="素食", limit=2)
# m7 排第一，说明参与嵌入的确实是 note 字段而不是 text
print("搜素食:", [(i.key, round(i.score, 3)) for i in hits])
# 验证 ①：m6 能被 get 到，说明数据在
print("m6 能 get 到:", sem.get(NS_SEM, "m6").value)

输出（实测）：

搜素食: [('m7', 0.6625), ('m4', 0.4275)]
m6 能 get 到: {'text': '用户讨厌香菜'}

一条记录可以有多个向量：fields 有两项时，两个字段各嵌一次，检索时按最大值打分（max pooling）。实测很直观：

# 两个字段分别嵌入
multi = InMemoryStore(index={"embed": emb, "dims": DIMS, "fields": ["topic", "detail"]})
# 这条记录的 topic 与 detail 主题完全无关
multi.put(("u1", "m"), "multi", {"topic": "宠物", "detail": "季度财报数字"})
# 这条只在财报上相关
multi.put(("u1", "m"), "single", {"topic": "财报", "detail": "季度财报数字"})
# 搜「猫」：multi 应当靠 topic 字段胜出
print("搜猫  :", [(i.key, round(i.score, 3)) for i in multi.search(("u1", "m"), query="猫", limit=2)])
# 搜「财报」：两条都相关，single 更纯
print("搜财报:", [(i.key, round(i.score, 3)) for i in multi.search(("u1", "m"), query="财报", limit=2)])

输出（实测）：

搜猫  : [('multi', 0.709), ('single', 0.34)]
搜财报: [('single', 1.0), ('multi', 0.495)]

single 拿到 1.0 是因为查询词「财报」正好等于它的 topic 字段原文——满分不代表「非常相关」，只代表「文本几乎相同」。

6.4 三个静默陷阱 #

这三个都不报错，都会让你以为语义检索在工作，其实没有。

陷阱一：没配 index 就传 query。

# 一个没有 index 配置的普通 store
plain = InMemoryStore()
# 第一条：和「宠物」语义相关
plain.put(("u1", "m"), "c1", {"text": "用户养猫"})
# 第二条：和「宠物」毫无关系，用来暴露 query 是否真的生效
plain.put(("u1", "m"), "c2", {"text": "用户住杭州"})
# 传 query 检索「宠物」——注意它不会报错
hits = plain.search(("u1", "m"), query="宠物")
# 打印 key 与 score
print("未配 index 传 query:", [(i.key, i.score) for i in hits])

输出（实测）：

未配 index 传 query: [('c1', None), ('c2', None)]

query 被完全忽略，退化成「列举这个命名空间」，而且没有任何警告。 两条都返回了，包括跟「宠物」毫无关系的「住杭州」。识别办法只有一个：看 score 是不是 None。

陷阱二：score is None 的条目是「补位」，不是命中。

# 带索引的 store，但只索引两条，第三条 index=False
mix = InMemoryStore(index={"embed": emb, "dims": DIMS, "fields": ["text"]})
# 有向量的第一条
mix.put(("u1", "m"), "i1", {"text": "用户养了猫"})
# 有向量的第二条
mix.put(("u1", "m"), "i2", {"text": "用户住杭州"})
# 显式关掉索引：数据存得进去，但不会生成向量
mix.put(("u1", "m"), "skip", {"text": "这条没有被索引"}, index=False)
# 逐个 limit 观察：limit 超过「有向量的条数」时，会拿没向量的条目填满
for lim in [1, 2, 3]:
    # 同一个查询词，只改 limit
    hits = mix.search(("u1", "m"), query="宠物", limit=lim)
    # score 为 None 的就是补位条目，打印时要单独处理避免 round(None) 报错
    print(f"limit={lim}:", [(i.key, None if i.score is None else round(i.score, 3)) for i in hits])

输出（实测）：

limit=1: [('i1', 0.469)]
limit=2: [('i1', 0.469), ('i2', 0.374)]
limit=3: [('i1', 0.469), ('i2', 0.374), ('skip', None)]

limit=3 时那条 index=False 的记录被塞了进来——框架的规则是：「请求条数多于已嵌入条数时，用没打分的条目补满」。判断「这条是不是语义命中」的唯一依据是 score is not None。

同样的道理，fields 里指定的字段不存在，那条记录也没有向量：

# fields 要 text，但这条记录只有 note 字段
mix.put(("u1", "m"), "nofield", {"note": "value 里根本没有 text 字段"})
# 搜索时它只能作为补位出现，score 是 None
hits = mix.search(("u1", "m"), query="字段", limit=5)
# 尽管查询词「字段」和 nofield 的内容高度相关，它依然排在最后且没有分数
print("命中:", [(i.key, None if i.score is None else round(i.score, 3)) for i in hits])

输出（实测）：

命中: [('i2', 0.351), ('i1', 0.328), ('skip', None), ('nofield', None)]

字段名写错 = 这条记忆永远搜不到，且没有任何提示。这是「记忆明明写进去了，Agent 却想不起来」的常见原因之一。

陷阱三：dims 写错，InMemoryStore 不校验。

# 故意把 1024 维模型声明成 8 维
bad = InMemoryStore(index={"embed": emb, "dims": 8, "fields": ["text"]})
# 写入不报错：真实向量是 1024 维，dims=8 完全没被校验
bad.put(("u1", "m"), "b1", {"text": "维度不匹配测试"})
# 检索也不报错，照样算得出相似度
hits = bad.search(("u1", "m"), query="维度", limit=1)
# 分数照样算得出来（因为它直接对真实向量做余弦）
print("dims 写错仍可检索:", [(i.key, round(i.score, 3)) for i in hits])

输出（实测）：

dims 写错仍可检索: [('b1', 0.67)]

InMemoryStore 根本不看 dims，它直接对拿到的向量算余弦。于是本地测试一路绿灯，换成 PostgresStore 上线时才会炸——那边要用 dims 建 vector(N) 列。dims 一定要和模型实际维度一致，并且写成 len(emb.embed_query("x")) 这种自动获取的形式，别手写数字。

6.5 两个响亮报错 #

报错一：同一条记录的两个被索引字段内容完全相同。

# fields 声明了两个字段
dup = InMemoryStore(index={"embed": emb, "dims": DIMS, "fields": ["a", "b"]})
try:
    # 两个字段的内容一模一样
    dup.put(("u1", "m"), "same", {"a": "完全一样的文本", "b": "完全一样的文本"})
# 这是 InMemoryStore 内部去重逻辑的缺陷，抛的是普通 ValueError
except ValueError as e:
    # 打印报错原文，注意它说的是「1 个嵌入对不上 2 个索引位」
    print("报错:", e)
# 内容不同就正常
dup.put(("u1", "m"), "diff", {"a": "第一段文本", "b": "第二段文本"})
# 能读到说明写入成功，反证问题只出在「两个字段文本相同」这一种情况
print("内容不同时正常写入:", dup.get(("u1", "m"), "diff") is not None)

输出（实测）：

报错: Number of embeddings (1) does not match number of indices (2)
内容不同时正常写入: True

原因是 InMemoryStore 内部按「文本」去重待嵌入的内容（同样的文本只嵌一次以省钱），但回填向量时按「字段数」对齐，两者数量不一致就抛错。这是 InMemoryStore 的实现缺陷，规避办法：fields 里的字段语义上就不该重复；如果无法保证，改成 fields: ["$"]。

6.6 成本：一次 put 一次嵌入调用 #

语义索引不是免费的。用一个计数包装可以看得很清楚：

# Embeddings 是嵌入模型的基类，自定义包装必须继承它
from langchain_core.embeddings import Embeddings
from langgraph.store.memory import InMemoryStore
from embeddings_layer import get_embeddings

# 拿到 DashScope 的嵌入模型（text-embedding-v4）
emb = get_embeddings("dashscope")

DIMS = len(emb.embed_query("测试"))


class CountingEmbeddings(Embeddings):
    """包装真实嵌入模型，打印每次调用的批量大小。"""

    # 构造时接收真实的嵌入模型，自己只做转发和计数
    def __init__(self, inner: Embeddings) -> None:
        # 保存真实模型
        self.inner = inner

    # 批量嵌入接口，store 写入时调用它
    def embed_documents(self, texts: list[str]) -> list[list[float]]:
        # 写入路径会走这里，texts 是这一批要嵌入的文本
        print(f"    embed_documents(batch={len(texts)})")
        # 转发给真实模型，自己不改结果
        return self.inner.embed_documents(texts)

    # 单条嵌入接口，检索时调用它
    def embed_query(self, text: str) -> list[float]:
        # 检索路径会走这里
        print(f"    embed_query({text!r})")
        # 同样只做转发
        return self.inner.embed_query(text)


# 用包装后的模型建 store
counted = InMemoryStore(
    index={"embed": CountingEmbeddings(emb), "dims": DIMS, "fields": ["text"]}
)
# 下面三段分别观察写入、带查询检索、不带查询检索各触发几次调用
print("两次 put:")
# 第一次写入：应当看到一次 batch=1
counted.put(("u1", "m"), "d1", {"text": "第一条"})
# 第二次写入：又是一次 batch=1，说明没有攒批
counted.put(("u1", "m"), "d2", {"text": "第二条"})
# 第二段：检索路径
print("带 query 的 search:")
# 传了 query，需要把查询词也嵌入一次
counted.search(("u1", "m"), query="查询")
# 第三段：纯列举路径
print("不带 query 的 search:")
# 不传 query 就是纯列举，下面不会有任何输出，说明零调用
counted.search(("u1", "m"))

输出（实测）：

两次 put:
    embed_documents(batch=1)
    embed_documents(batch=1)
带 query 的 search:
    embed_query('查询')
不带 query 的 search:

结论：

• 每次 put 都是一次嵌入 API 调用，批量是 1。逐条写 100 条笔记 = 100 次调用。要省钱就攒批（自己收集后一次性写多条，或者用 store.batch(...)）。
• 带 query 的 search 每次一次调用；不带 query 的 search 零调用。所以「列举最近 5 条笔记」这种高频操作应该用不带 query 的 search。
• 写入路径用 embed_documents、查询路径用 embed_query——如果你自定义了嵌入层，两个方法都要实现。

6.7 什么时候才真的需要它 #

大多数长期记忆项目，第一版不需要语义检索。 对照：

不要因为「显得高级」就配 index：它带来嵌入成本、维度配置、pgvector 依赖，还多了本节这三个静默陷阱。§10 的实战默认走「全量列举 + 只注入最近 5 条」，跑起来零嵌入成本，行为完全可预测。

7. 什么时候写记忆 #

读记忆简单，难的是写：写早了进噪声，写晚了用户已走；写多了库膨胀，写少了等于没有。官方把写入时机分成两类——hot path（在线路径上写） 和 background（后台异步写）。再加一个折中，一共三种。

7.1 hot path：做成工具，让模型自己决定 #

这是 ChatGPT 的做法（它有一个 save_memories 工具），也最好上手：§5.2 的 save_profile 就是。

# tool 装饰器把函数变成工具；ToolRuntime 让工具拿到运行时对象
from langchain.tools import ToolRuntime, tool

# dataclass 用来声明 context 的结构
from dataclasses import dataclass
import uuid
from langchain.agents import create_agent
from langgraph.store.memory import InMemoryStore
from dotenv import load_dotenv

# 读取 .env（DEEPSEEK_API_KEY 等）
load_dotenv()

# 长期记忆容器，整个应用共享一个实例
store = InMemoryStore()


# dataclass 装饰器：几行就能声明一个带字段的上下文类
@dataclass
class Ctx:
    """每次请求都要带的上下文：当前是谁在说话。"""

    # user_id 决定长期记忆的命名空间，必须由业务代码传入
    user_id: str


# 复用前面定义好的 store 与 Ctx，做一个「写笔记」的工具
@tool
def remember(text: str, runtime: ToolRuntime[Ctx]) -> str:
    """把一条需要长期记住的用户事实写入长期记忆，text 用一句陈述句。"""
    # 定位到当前用户的笔记命名空间
    ns = (runtime.context.user_id, "memories")
    # 写入前先取全量做朴素查重，避免同一件事记两遍
    existing = runtime.store.search(ns, limit=100)
    # 逐条比对
    for item in existing:
        # 去掉首尾空白再比，避免「多一个空格」被当成新笔记
        if item.value.get("text", "").strip() == text.strip():
            # 命中重复就直接返回，把跳过的原因告诉模型
            return f"已存在相同笔记（{item.key}），跳过"
    # key 用随机值，避免并行调用撞车
    key = f"note-{uuid.uuid4().hex[:8]}"
    # 落盘
    runtime.store.put(ns, key, {"text": text})
    # 回灌给模型的确认信息
    return f"已记住：{text}"


# 组装一个只带这一个工具的 Agent
agent_hot = create_agent(
    # 对话模型
    model="deepseek:deepseek-v4-flash",
    # 只挂写笔记这一个工具
    tools=[remember],
    # 工具要往这里写
    store=store,
    # 工具要从 context 拿 user_id
    context_schema=Ctx,
    # 提示里必须明确「什么时候该记」，否则模型经常忘了调
    system_prompt="你是客服助手。用户提到个人事实（家庭、忌口、偏好等）时必须调用 remember 记录。回答简短。",
)
# 跑一轮，故意在一句话里给两个事实
r = agent_hot.invoke(
    # 一句话两个事实：观察模型会拆成几次工具调用
    {"messages": [{"role": "user", "content": "我家有两只猫，另外我对乳制品过敏。"}]},
    # 写到 u_zhou 名下
    context=Ctx(user_id="u_zhou"),
)
# 打印回复
print("回复:", r["messages"][-1].content)
# 不要只信模型说「记住了」，直接查 store 看写进去几条
notes_now = store.search(("u_zhou", "memories"), limit=100)
# 条数取决于模型怎么拆，每次运行都可能不同
print("笔记条数:", len(notes_now))
# 逐条打印内容，检查措辞是否可用
for i in notes_now:
    # 只看 text 字段，key 是随机的没有信息量
    print("   ", i.value["text"])

输出（实测；拆成几条、每条怎么措辞都由模型决定，每次运行都会不同）：

回复: 好的，已经记住了！有两只猫，且对乳制品过敏。有什么可以帮您的吗？
笔记条数: 2
    用户家里养了两只猫。
    用户对乳制品过敏。

模型把一句话里的两个事实拆成了两次工具调用——这正是 key 必须用随机值的原因（§4.3）：并行调用时用「条数 + 1」算 key，两条笔记会算出同一个 note-1 互相覆盖，最后只剩一条。

hot path 的优缺点：

一个实践细节：工具的返回值要包含写入结果（「已记住 X」），这样模型才能自然地告知用户；如果返回空字符串，模型经常会自己编一句「已保存」，而你无法区分真假。

7.2 回合结束：@after_agent 统一抽取 #

hot path 依赖模型「记得调工具」。更稳的做法是：每个回合结束后，用一次结构化抽取把这轮的信息榨出来，模型完全不参与决策。

# after_agent：一个回合（agent 跑完）之后执行
from langchain.agents.middleware import after_agent

# AgentState 是 Agent 的内置状态类型
from langchain.agents import AgentState
from langchain.agents import create_agent

# Runtime 让中间件拿到 store 与 context
from langgraph.runtime import Runtime

# json 用于把档案拼成字符串回灌给模型
import json

# HumanMessage 用于只挑用户原话
from langchain.messages import HumanMessage

# 短期记忆仍然靠 checkpointer
from langgraph.checkpoint.memory import InMemorySaver  # 短期记忆仍然靠 checkpointer

# init_chat_model 手工构造模型，才能传 extra_body
from langchain.chat_models import init_chat_model

# Pydantic 声明抽取结果的结构
from pydantic import BaseModel, Field
from langgraph.store.memory import InMemoryStore

# dynamic_prompt 装饰器：每次调模型前动态生成系统提示
# ModelRequest 是这次模型请求的完整描述
from langchain.agents.middleware import ModelRequest, dynamic_prompt

# dataclass 用来声明 context 的结构
from dataclasses import dataclass

# dotenv 负责把 .env 里的 API Key 注入环境变量
from dotenv import load_dotenv

# BaseStore 是 store 的抽象基类，所有 store 实现都要继承它
from langgraph.store.base import BaseStore

# 读取 .env（DEEPSEEK_API_KEY 等）
load_dotenv()
# 长期记忆容器，整个应用共享一个实例
store = InMemoryStore()


# dataclass 装饰器：几行就能声明一个带字段的上下文类
@dataclass
class Ctx:
    """每次请求都要带的上下文：当前是谁在说话。"""

    # user_id 决定长期记忆的命名空间，必须由业务代码传入
    user_id: str


class ProfilePatch(BaseModel):
    """一轮对话里能抽到的档案字段，全部可选。"""

    # 每个字段的 description 直接影响抽取质量，务必写清「没提到就留空」
    name: str | None = Field(default=None, description="用户姓名，用户没亲口说就留空")
    # 城市：允许为空，这样「本轮没提到」才有办法表达
    city: str | None = Field(
        default=None, description="用户常住城市，用户没亲口说就留空"
    )
    # 风格偏好：自由文本，练习 1 会把它改成枚举
    style: str | None = Field(
        default=None, description="用户对回答风格的偏好，没提到就留空"
    )


# DeepSeek 做结构化输出必须关掉思考模式，否则 400（第 6 章的坑）
extract_model = init_chat_model(
    # 抽取用的模型，可以和对话模型不同，挑便宜的
    "deepseek:deepseek-v4-flash",
    # 抽取要的是稳定，不是创意，温度拉到 0
    temperature=0,
    # 关闭思考模式，这是 DeepSeek 结构化输出的硬性要求
    extra_body={"thinking": {"type": "disabled"}},
)
# 包装成只输出 ProfilePatch 的抽取器
extractor = extract_model.with_structured_output(ProfilePatch)


def merge_profile(store: BaseStore, user_id: str, patch: dict) -> dict:
    """把 patch 合并进用户档案，只覆盖非空字段，返回合并后的档案。"""
    # profile 形态：命名空间固定两层，key 固定叫 basic
    ns = (user_id, "profile")
    # 先读旧档案，未命中时 get 返回 None
    old = store.get(ns, "basic")
    # 复制成普通字典再改，避免直接修改 Item 内部的对象
    merged = dict(old.value) if old else {}
    # 记下合并前的样子，用来判断这一轮到底有没有新信息
    before = dict(merged)
    # 逐个字段合并
    for field, value in patch.items():
        # 值为 None 表示「本轮没提到」，绝不能拿它覆盖旧值
        if value is not None:
            # 只有非空值才写进合并结果
            merged[field] = value
    # 没有任何变化就不写：省一次 IO，也避免无意义地刷新 updated_at
    if merged != before:
        # 整体写回，key 固定是 basic
        store.put(ns, "basic", merged)
    # 返回结果，方便调用方打印或继续用
    return merged


@after_agent
def auto_extract(state: AgentState, runtime: Runtime[Ctx]) -> None:
    """回合结束后，从用户原话里抽取档案并合并写入。"""
    # 没接 store 就什么都不做，保持可降级
    if runtime.store is None:
        # 直接返回，不抛异常：忘配 store 不应该让对话失败
        return None
    # 只取 HumanMessage 的内容：原因见 §7.5
    said = [
        m.content
        for m in state["messages"]
        if isinstance(m, HumanMessage) and m.content
    ]
    # 用户这轮没说话就跳过，省一次模型调用
    if not said:
        # 没有输入就没有可抽的东西，提前退出
        return None
    # 抽取
    patch = extractor.invoke("从下面的用户原话里抽取用户档案：\n" + "\n".join(said))
    # 结构化输出**偶发返回 None**（模型这次没产出工具调用），不判空会让整个回合崩掉
    if patch is None:
        # 这一轮放弃抽取，下一轮还有机会，绝不能让主流程失败
        return None
    # 用 §4.2 的合并函数写回，只覆盖非空字段
    merge_profile(runtime.store, runtime.context.user_id, patch.model_dump())
    # after_agent 不需要改状态
    return None


@dynamic_prompt
def prompt_with_profile(request: ModelRequest) -> str:
    """调模型前把长期档案拼进系统提示。"""
    # 中间件通过 request.runtime 拿到 store 与 context
    st = request.runtime.store
    # 当前是谁在说话，决定读哪个命名空间
    uid = request.runtime.context.user_id
    # 没接 store 时降级为「无记忆」，不要直接崩
    item = st.get((uid, "profile"), "basic") if st else None
    # 固定的角色说明
    base = "你是客服助手，回答简短。"
    # 没有档案就明确告诉模型「没有」，避免它编造
    if not item:
        # 显式声明「暂无」比什么都不说更能抑制幻觉
        return base + "\n（暂无该用户档案）"
    # 有档案就整段注入，JSON 形式最省 token 也最不容易被误读
    return base + "\n已知用户档案：" + json.dumps(item.value, ensure_ascii=False)


# 组装：读用 dynamic_prompt，写用 after_agent，模型完全不知道有记忆这件事
agent_auto = create_agent(
    # 对话模型
    model="deepseek:deepseek-v4-flash",
    # 一个工具都没有：读写全部由中间件完成
    tools=[],
    # 两个中间件：前者负责读（注入提示），后者负责写（抽取档案）
    middleware=[prompt_with_profile, auto_extract],
    # 两个中间件共用这一个 store
    store=store,
    # user_id 从 context 来
    context_schema=Ctx,
    # 加上短期记忆，方便下面用两个 thread 做对照
    checkpointer=InMemorySaver(),
)
# 新用户，第一轮：说出姓名城市风格
a1 = agent_auto.invoke(
    # 一句话里同时包含姓名、城市、风格三个字段
    {"messages": [{"role": "user", "content": "我叫小李，在成都，说话别绕弯子。"}]},
    # 第一条会话线
    config={"configurable": {"thread_id": "a1"}},
    # 换一个全新用户，避免和前面的 u_zhou 混在一起
    context=Ctx(user_id="u_li"),
)
# 打印回复
print("回复:", a1["messages"][-1].content)
# 直接查 store：三个字段应当都被抽出来了
print("档案:", store.get(("u_li", "profile"), "basic").value)
# 换 thread 再问一次，验证跨会话
a2 = agent_auto.invoke(
    # 只问不说，答案只能来自档案
    {"messages": [{"role": "user", "content": "我在哪个城市？只答城市名。"}]},
    # 全新会话线，短期记忆为空
    config={"configurable": {"thread_id": "a2"}},
    # 同一个用户，长期记忆连续
    context=Ctx(user_id="u_li"),
)
# 只有两条消息（没有工具往返），却答对了
print("跨会话回复:", a2["messages"][-1].content, "| 消息条数:", len(a2["messages"]))

输出（实测）：

回复: 好的小李，明白了。有什么需要直接说，我尽量简短。
档案: {'name': '小李', 'city': '成都', 'style': '说话别绕弯子'}
跨会话回复: 成都 | 消息条数: 2

注意 style 抽出来的是用户原话「说话别绕弯子」，不是规范化的枚举值。若要拿它做程序判断（比如 if style == "concise"），就得在 Field 的 description 里限定取值，或直接用 Literal["concise", "detailed"]——with_structured_output 会把它变成模型必须遵守的约束。

这套写法的好处是行为确定：不管模型配不配合，每轮都会抽一次。代价是每轮多一次模型调用——抽取模型应选便宜的，并可加条件（比如「本轮用户消息少于 10 个字就不抽」）。

上面那句 if patch is None 不是防御形式主义。 本章代码块反复实跑时真踩到过：

AttributeError: 'NoneType' object has no attribute 'model_dump'
During task with name 'auto_extract.after_agent' and id 'c00c1a43-...'

with_structured_output 在模型这一次没有产出工具调用时会返回 None（概率不高，但一定会发生）。而 after_agent 里抛出的异常会让整个回合失败——用户看到的不是「档案没更新」，而是一次彻底的报错。

通用原则：记忆的写入路径永远不能让主流程挂掉。 抽取失败就跳过这一轮，下一轮还有机会；不要为了记住一个可选字段，赔上一次完整的对话。

7.3 后台异步：把抽取搬出请求路径 #

若连「每轮多一次调用」都嫌贵或嫌慢，就把抽取搬出请求路径：

# 这是一个可以放到定时任务 / 队列消费者里的函数
def batch_extract(store_obj, user_id: str, transcript: list[str]) -> dict:
    """离线抽取：给定一段用户原话，抽档案并合并写入。"""
    # 拼成一段文本喂给抽取器
    patch = extractor.invoke("从下面的用户原话里抽取用户档案：\n" + "\n".join(transcript))
    # 同样要判空：离线任务里跳过这一条，比让整批任务失败更合理
    if patch is None:
        # 传空 patch 相当于「只读不改」，直接把现有档案返回去
        return merge_profile(store_obj, user_id, {})
    # 合并写入，返回最新档案
    return merge_profile(store_obj, user_id, patch.model_dump())


# 模拟：从 checkpointer 里捞出昨天的会话原话，离线处理
yesterday = ["我姓王", "我在广州工作", "回答请详细一些，我需要看到理由"]
# 调一次离线抽取
print("离线抽取结果:", batch_extract(store, "u_wang", yesterday))

输出（实测）：

离线抽取结果: {'name': '王', 'city': '广州', 'style': '详细，需要看到理由'}

生产上常见触发：会话结束（或空闲 N 分钟）后投一条消息到队列 → 消费者读 checkpointer 里的 messages → 抽取 → 写 store。好处是在线路径零额外延迟，代价是记忆不是立刻可用（同一次会话里说的偏好，本轮不会生效）。

7.4 三种时机对照 #

推荐组合（§10 实战用的就是这个）：after_agent 抽档案（确定性）+ remember 工具写笔记（用户显式要求时立刻生效）。两者互补：结构化字段靠抽取保证不漏，自由笔记靠工具保证可控。

7.5 记忆回声污染：抽取器会把 Agent 的话当事实 #

这是长期记忆里最隐蔽也最致命的一类错误，一定要实测一遍。

场景：用户从没说过自己在哪个城市，Agent 却在某轮回答里猜了「您在上海」。若抽取时把整段对话（含 AI 消息）都喂进去：

from pydantic import BaseModel, Field

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model

load_dotenv()

extract_model = init_chat_model(
    # 抽取用的模型，可以和对话模型不同，挑便宜的
    "deepseek:deepseek-v4-flash",
    # 抽取要的是稳定，不是创意，温度拉到 0
    temperature=0,
    # 关闭思考模式，这是 DeepSeek 结构化输出的硬性要求
    extra_body={"thinking": {"type": "disabled"}},
)
# 一段「模型猜错」的对话：用户从未说过城市，助手却断言是上海
full_dialog = (
    # 用户第一句：只问门店，没提城市
    "HumanMessage: 帮我查下最近的门店。\n"
    # 助手凭空猜了一个「上海」——污染源就在这一行
    "AIMessage: 好的，您在上海，附近的门店是南京西路店。\n"
    # 用户只回了「好的」，这不构成对城市的确认
    "HumanMessage: 好的。\n"
)


class CityOnly(BaseModel):
    """只抽一个字段，方便看清污染。"""

    # 注意 description 已经写了「用户没亲口说就留空」
    city: str | None = Field(
        default=None, description="用户常住城市，用户没亲口说就留空"
    )


# 用同一个抽取模型
city_extractor = extract_model.with_structured_output(CityOnly)
# ① 把整段对话（含 AI 消息）喂进去
print(
    "① 整段对话:",
    city_extractor.invoke(f"从下面对话抽取用户档案：\n{full_dialog}").model_dump(),
)
# ② 只喂用户消息：把 AIMessage 那一行整个删掉
human_only = "HumanMessage: 帮我查下最近的门店。\nHumanMessage: 好的。\n"
# 没有污染源，抽取结果应当是空
print(
    "② 只喂用户消息:",
    city_extractor.invoke(f"从下面对话抽取用户档案：\n{human_only}").model_dump(),
)
# ③ 仍喂整段，但在提示里加硬约束
guarded = (
    # 明确规定采信范围
    "从下面对话抽取用户档案。严格规则：只有用户（HumanMessage）亲口说出的信息才可以填写，"
    # 明确规定助手的话不算数
    "助手（AIMessage）说的内容一律视为未确认，必须留空。\n"
    + full_dialog
)
# 有了硬约束，即使看得到「上海」也不会填
print("③ 整段 + 硬约束:", city_extractor.invoke(guarded).model_dump())

输出（实测）：

① 整段对话: {'city': '上海'}
② 只喂用户消息: {'city': None}
③ 整段 + 硬约束: {'city': None}

模型猜的「上海」被当成用户事实写进了长期档案。 此后每轮系统提示都会带上「用户在上海」，模型更确信这件事，下一轮又重复——错误信息自我强化，而且不会自己消失。这就是「回声污染」（echo pollution）。

两个必须做的防御（上面 ② 和 ③ 都实测有效）：

1. 只把 HumanMessage 喂给抽取器（首选，成本为零）；
2. 在抽取提示里明确「只采信用户原话」（当你必须提供上下文时的兜底）。

再加两条工程约束：

3. 给记忆存来源：{"city": "上海", "source": "user_said", "at": "2026-08-14"}。以后排查「这条哪来的」有据可查。
4. 给用户一个「忘掉这条」的入口。长期记忆一旦错了，用户是唯一能纠正它的人；没有删除路径的记忆系统，出错就只能等着挨投诉。

7.6 写入前查重：语义级去重 #

collection 会膨胀。朴素的「原文完全相同才跳过」拦不住换说法的重复：

from langgraph.store.memory import InMemoryStore
from embeddings_layer import get_embeddings
from dotenv import load_dotenv
import uuid

load_dotenv()
emb = get_embeddings("dashscope")
DIMS = len(emb.embed_query("测试"))

# 造一个带语义索引的笔记库
dedup = InMemoryStore(index={"embed": emb, "dims": DIMS, "fields": ["text"]})
# 命名空间抽成常量，下面几处都要用
DNS = ("u1", "memories")
# 先写三条：前两条本身就是同义改写，第三条是无关信息
for i, t in enumerate(["用户养了两只猫", "用户家里有两只猫咪", "用户对乳制品过敏"], 1):
    # key 用 m1/m2/m3，方便在输出里认出是哪条挡住了新笔记
    dedup.put(DNS, f"m{i}", {"text": t})
# 基线条数，后面用它判断到底写进去几条
print("初始条数:", len(dedup.search(DNS, limit=50)))


def put_if_new(text: str, threshold: float = 0.75) -> str:
    """写入前先做一次语义检索，足够相似就跳过。"""
    # 只要 top1
    hits = dedup.search(DNS, query=text, limit=1)
    # score 可能是 None（补位结果），要显式判空
    if hits and hits[0].score is not None and hits[0].score >= threshold:
        # 把「跟哪条重复、有多像」一起返回，方便调阈值
        return f"跳过（与 {hits[0].key} 相似度 {hits[0].score:.3f}）"
    # 不够相似才写入
    dedup.put(DNS, f"note-{uuid.uuid4().hex[:6]}", {"text": text})
    # 告诉调用方这次真的写了
    return "写入"


# 三条候选：两条是换了说法的重复，一条是全新信息
for t in ["用户家养了两只猫", "用户喜欢夜间配送", "用户养了 2 只猫"]:
    # 逐条调用，打印判定结果
    print(f"  {t!r} -> {put_if_new(t)}")
# 最终只应该多出一条
print("最终条数:", len(dedup.search(DNS, limit=50)))

输出（实测）：

初始条数: 3
  '用户家养了两只猫' -> 跳过（与 m2 相似度 0.903）
  '用户喜欢夜间配送' -> 写入
  '用户养了 2 只猫' -> 跳过（与 m1 相似度 0.948）
最终条数: 4

0.903 和 0.948 说明：同义改写的相似度会明显高于普通相关内容（对比 §6.2 里「宠物 vs 养猫」只有 0.479）。去重阈值可以设得比检索阈值高很多。

阈值怎么定？必须在自己的数据上量：

另外三种控制膨胀的手段，按实现成本排序：

1. 只注入最近 N 条（零成本，最有效）：读的时候限量，写的时候不管。
2. TTL 自动过期（见 §9.3，需要 PostgresStore）：给时效性记忆设定寿命。
3. 定期合并（成本最高）：后台任务把同一主题的多条笔记喂给模型，压成一条。

8. 三类记忆：语义、情景、程序 #

官方概念文档借用认知科学分类，把长期记忆分成三种。这不只是理论——它决定你把数据存成什么结构。

注意区分两个容易混的词：语义记忆（semantic memory，认知科学术语，指「存事实」）和语义检索（semantic search，指「按向量找」）。前者是「存什么」，后者是「怎么找」。存语义记忆完全可以不用语义检索。

8.1 语义记忆 #

本章前七节讲的就是它。

姓名、城市、忌口、偏好——这些都是语义记忆，也是绝大多数产品唯一需要的一类。§4 的两种形态（profile / collection）就是它的两种存法，不再重复。

单独说一下三类记忆的优先级：它们不是「都要做」的清单，而是有先后的：

绝大多数项目做完第 1 类就够了。第 2、3 类的共同前提是「你已经积累了数据」——没有历史案例就没有情景记忆，没有反馈渠道就没有程序记忆。先把语义记忆做扎实，再谈另外两类。

8.2 情景记忆：把「做对过的案例」存起来当示例 #

情景记忆落地几乎总是 few-shot：把过去处理成功的案例存下来，下次遇到相似问题时取最像的几条塞进提示。

import uuid
from langgraph.store.memory import InMemoryStore
from dotenv import load_dotenv

load_dotenv()

store = InMemoryStore()
# 情景记忆放在独立命名空间，和事实类记忆分开
EP_NS = ("u_zhou", "episodes")
# 每条情景记忆存「当时的问题 + 当时成功的处理方式」
episodes = [
    # 案例一：ok=True 表示这次处理是成功的，只有成功的才配当示例
    {
        "question": "空气净化器不出风",
        "action": "先确认滤芯保护膜是否撕掉，再判断是否报修",
        "ok": True,
    },
    # 案例二
    {
        "question": "净化器噪音大",
        "action": "确认是否在极速模式，建议切换到睡眠模式",
        "ok": True,
    },
    # 案例三
    {
        "question": "发票没收到",
        "action": "查订单状态，提醒电子发票在订单完成后 24 小时内开具",
        "ok": True,
    },
]
# 逐条写入
for ep in episodes:
    # key 用随机值，理由同 §4.3：避免并行写入撞车
    store.put(EP_NS, f"ep-{uuid.uuid4().hex[:6]}", ep)


def few_shot_block(user_id: str, limit: int = 2) -> str:
    """取出最近的成功案例，拼成 few-shot 文本。"""
    # 这里用「列举 + 取最近」，条数多了可以换成语义检索（§6）
    items = store.search((user_id, "episodes"), limit=100)
    # 只要标记成功的案例
    good = [i for i in items if i.value.get("ok")]
    # 按 updated_at 取最近若干条，避免依赖 store 的默认顺序
    recent = sorted(good, key=lambda i: i.updated_at, reverse=True)[:limit]
    # 拼成「问题 → 处理」的示例块
    return "\n".join(
        f"- 遇到「{i.value['question']}」时：{i.value['action']}" for i in recent
    )

# 打印生成的 few-shot 块
print(few_shot_block("u_zhou"))

输出（实测，顺序取决于写入时间）：

- 遇到「发票没收到」时：查订单状态，提醒电子发票在订单完成后 24 小时内开具
- 遇到「净化器噪音大」时：确认是否在极速模式，建议切换到睡眠模式

把这段文本拼进 dynamic_prompt，就等于让 Agent 从过去的经验里学。选案例的策略决定效果：条数少时取最近，条数多时按语义取最相似（这时才值得给 episodes 配 index）。

情景记忆的两个注意点：

• 必须记「成功」标记。把失败案例当示例是灾难——模型会照着学错。上面用 ok 字段，生产上可以是人工标注或用户点赞。
• 示例要短。few-shot 是按 token 计费的固定开销，每轮都带；两三条足矣，别把二十条案例全塞进去。

8.3 程序记忆：让 Agent 改写自己的系统提示 #

程序记忆最激进的用法：把系统提示本身存进 store，让 Agent 根据反馈改写它。

import uuid
from langgraph.store.memory import InMemoryStore
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model

load_dotenv()

store = InMemoryStore()
# 系统提示存在与用户无关的命名空间里
INS_NS = ("agent", "instructions")
# 初始提示词
store.put(INS_NS, "customer_service", {"text": "你是客服助手，回答用户问题。"})
# 打印初始值
print("改写前:", store.get(INS_NS, "customer_service").value["text"])

# 用户反馈：这就是「让 Agent 学会新规则」的输入
feedback = "用户反馈：你的回答太啰嗦了，而且总是先寒暄。"
# 读出当前提示词
current = store.get(INS_NS, "customer_service").value["text"]
# DeepSeek 做结构化输出必须关掉思考模式，否则 400（第 6 章的坑）
extract_model = init_chat_model(
    # 抽取用的模型，可以和对话模型不同，挑便宜的
    "deepseek:deepseek-v4-flash",
    # 抽取要的是稳定，不是创意，温度拉到 0
    temperature=0,
    # 关闭思考模式，这是 DeepSeek 结构化输出的硬性要求
    extra_body={"thinking": {"type": "disabled"}},
)
# 让模型基于反馈改写提示词本身
rewritten = extract_model.invoke(
    # 说清楚任务：改写的对象是提示词，不是回答用户
    "下面是一个客服 Agent 的系统提示词，以及用户反馈。"
    # 约束输出格式与长度，否则模型会输出一堆解释
    "请输出改进后的系统提示词本身（不要解释，不要加引号），控制在 60 字内。\n\n"
    # 把当前提示词和反馈一起给它
    f"当前提示词：{current}\n反馈：{feedback}"
# 取文本并去掉首尾空白，直接当新提示词用
).content.strip()
# 写回同一个 key（覆盖就是「更新规则」）
store.put(INS_NS, "customer_service", {"text": rewritten})
# 打印改写后的结果
print("改写后:", store.get(INS_NS, "customer_service").value["text"])

输出（实测，措辞会不同）：

改写前: 你是客服助手，回答用户问题。
改写后: 直接回答用户问题，不寒暄，不啰嗦，只给结论和必要步骤。

然后在 dynamic_prompt 里读它，Agent 的行为就被「学到的规则」接管了。下面用同一个问题跑两个 Agent 做对照：

import uuid
from langgraph.store.memory import InMemoryStore
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.agents.middleware import ModelRequest, dynamic_prompt
from langchain.agents import create_agent

# dataclass 用来声明 context 的结构
from dataclasses import dataclass

load_dotenv()


# dataclass 装饰器：几行就能声明一个带字段的上下文类
@dataclass
class Ctx:
    """每次请求都要带的上下文：当前是谁在说话。"""

    # user_id 决定长期记忆的命名空间，必须由业务代码传入
    user_id: str


store = InMemoryStore()
# 系统提示存在与用户无关的命名空间里
INS_NS = ("agent", "instructions")
# 初始提示词
store.put(INS_NS, "customer_service", {"text": "你是客服助手，回答用户问题。"})
# 打印初始值
print("改写前:", store.get(INS_NS, "customer_service").value["text"])

# 用户反馈：这就是「让 Agent 学会新规则」的输入
feedback = "用户反馈：你的回答太啰嗦了，而且总是先寒暄。"
# 读出当前提示词
current = store.get(INS_NS, "customer_service").value["text"]
# DeepSeek 做结构化输出必须关掉思考模式，否则 400（第 6 章的坑）
extract_model = init_chat_model(
    # 抽取用的模型，可以和对话模型不同，挑便宜的
    "deepseek:deepseek-v4-flash",
    # 抽取要的是稳定，不是创意，温度拉到 0
    temperature=0,
    # 关闭思考模式，这是 DeepSeek 结构化输出的硬性要求
    extra_body={"thinking": {"type": "disabled"}},
)
# 让模型基于反馈改写提示词本身
rewritten = extract_model.invoke(
    # 说清楚任务：改写的对象是提示词，不是回答用户
    "下面是一个客服 Agent 的系统提示词，以及用户反馈。"
    # 约束输出格式与长度，否则模型会输出一堆解释
    "请输出改进后的系统提示词本身（不要解释，不要加引号），控制在 60 字内。\n\n"
    # 把当前提示词和反馈一起给它
    f"当前提示词：{current}\n反馈：{feedback}"
    # 取文本并去掉首尾空白，直接当新提示词用
).content.strip()
# 写回同一个 key（覆盖就是「更新规则」）
store.put(INS_NS, "customer_service", {"text": rewritten})
# 打印改写后的结果
print("改写后:", store.get(INS_NS, "customer_service").value["text"])


@dynamic_prompt
def prompt_from_store(request: ModelRequest) -> str:
    """系统提示完全来自 store：改一次记录，全局行为就变了。"""
    # 从共享命名空间读当前生效的提示词
    item = request.runtime.store.get(("agent", "instructions"), "customer_service")
    # 兜底：store 里没有就用一个安全的默认值
    return item.value["text"] if item else "你是客服助手。"


# 对照组：用改写前的原始提示词，写死在代码里
agent_before = create_agent(
    # 对话模型
    model="deepseek:deepseek-v4-flash",
    # 不需要工具
    tools=[],
    # 硬编码的原始提示词，等价于「没有程序记忆」
    system_prompt="你是客服助手，回答用户问题。",
)
# 实验组：提示词完全外置，读 store 里被改写过的版本
agent_after = create_agent(
    # 同一个模型，保证差异只来自提示词
    model="deepseek:deepseek-v4-flash",
    # 同样不带工具
    tools=[],
    # 注意这里没有 system_prompt，提示词全部来自中间件
    middleware=[prompt_from_store],
    # 中间件要从这里读提示词
    store=store,
    # 中间件签名里用到了 Ctx
    context_schema=Ctx,
)
# 同一个带寒暄的问题
question = {
    "messages": [{"role": "user", "content": "你好呀，在吗？我想问下发票的事。"}]
}
# 对照组回复
before_text = agent_before.invoke(question)["messages"][-1].content
# 实验组回复（需要 context，因为 context_schema 声明了 Ctx）
after_text = agent_after.invoke(question, context=Ctx(user_id="u_zhou"))["messages"][
    -1
].content
# 只看开头 40 字，重点观察有没有寒暄；换行替换成空格避免打乱排版
print("改写前 Agent:", before_text[:40].replace("\n", " "))
# 对照组的回复应当明显更干脆
print("改写后 Agent:", after_text[:40].replace("\n", " "))

输出（实测，措辞每次不同，但寒暄与长度的差异是稳定的）：

改写前 Agent: 你好呀！在的～发票方面有什么需要帮助的吗？比如开票、抬头、报销或者发票查询等问题
改写后 Agent: 请问发票具体要问什么？
改写后更短: True

store 里改一条记录，Agent 的说话方式就变了——不用改代码，也不用重新部署。这就是程序记忆的价值：把「怎么做事」的规则从代码挪到数据里。

这个能力很强，但风险同样大，务必配三道护栏：

保守的生产做法：程序记忆只存「风格与偏好」这类软规则，硬规则写死在代码里。像本章实战那样，把「不寒暄」放进 store，把「涉及退款必须查制度」写在代码的系统提示里。

9. 生产化：从 InMemoryStore 换到 PostgresStore #

InMemoryStore 有一个致命限制：进程一退出，所有长期记忆全没了。这跟「长期」直接冲突，所以它只适合开发与测试。

9.1 换一行代码 #

# PostgresStore 需要 langgraph-checkpoint-postgres 与 psycopg
# pip install -U langgraph-checkpoint-postgres "psycopg[binary]"
from langgraph.store.postgres import PostgresStore

# 连接串按你的环境改；生产上从环境变量读，别写死
DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"

# from_conn_string 返回上下文管理器，退出时自动关连接
with PostgresStore.from_conn_string(DB_URI) as pg:
    # setup() 建表并跑迁移，幂等：重复调用会跳过已完成的迁移
    pg.setup()
    # 之后的用法和 InMemoryStore 完全一样
    pg.put(("u_demo", "profile"), "basic", {"name": "小周", "city": "杭州"})
    # 读回来确认真的落库了，注意字段顺序和内存版不同（jsonb 会重排）
    print("Postgres 读回:", pg.get(("u_demo", "profile"), "basic").value)
    # 清理演示数据
    pg.delete(("u_demo", "profile"), "basic")

输出（需要本机 PostgreSQL）：

Postgres 读回: {'city': '杭州', 'name': '小周'}

两条必须记住的操作要求：

1. setup() 必须调一次，否则报表不存在。它是幂等的，放在应用启动时即可。
2. with 块结束后 store 就不可用了。所以 create_agent(store=pg) 必须写在 with 内部，整个服务的生命周期都在这个块里（和第 11 章的 PostgresSaver 同样的结构）。

9.2 setup() 建了什么 #

store 相关表: ['store', 'store_migrations']
store 表结构: [('prefix', 'text'), ('key', 'text'), ('value', 'jsonb'),
              ('created_at', 'timestamp with time zone'),
              ('updated_at', 'timestamp with time zone'),
              ('expires_at', 'timestamp with time zone'),
              ('ttl_minutes', 'integer')]

三点值得注意：

• namespace 被拍平成一个 prefix 文本列（形如 u_zhou.profile）——这正是命名空间标签不能含 . 的底层原因（§3.5）。
• value 是 jsonb，所以你可以直接写 SQL 查记忆：select * from store where prefix like 'u_zhou%'。做数据巡检、导出、批量修正时非常方便，这是 Postgres 版最实用的优势。
• 有 expires_at 和 ttl_minutes 两列，说明 TTL 是数据库层支持的能力（§9.3）。

9.3 四个行为差异 #

同样的代码，换了 store 实现后有四处行为不同。这四条都不会报错，本地测好、上线翻车的概率很高：

前两条的实测对照：

# search 顺序（先写 m1 再写 m2）
InMemoryStore  -> ['m1', 'm2']
PostgresStore  -> ['m2', 'm1']

# 覆盖写入后 created_at 是否被重置
InMemoryStore  -> True
PostgresStore  -> False

TTL 的行为要单独说，因为它有一个反直觉的地方：

# TTLConfig 描述默认过期策略
from langgraph.store.base import TTLConfig
from langgraph.store.postgres import PostgresStore

DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"
# default_ttl 单位是「分钟」；这里设 0.02 分钟 = 1.2 秒，方便演示
ttl_cfg = TTLConfig(
    default_ttl=0.02, refresh_on_read=False, sweep_interval_minutes=None
)
# 把 TTL 配置传给 store，之后所有 put 都默认带这个过期时间
with PostgresStore.from_conn_string(DB_URI, ttl=ttl_cfg) as pg:
    # 建表
    pg.setup()
    # 写一条会过期的记忆
    pg.put(("ttl_demo",), "k1", {"v": 1})
    # 立刻读：当然还在
    print("写入后立刻读:", pg.get(("ttl_demo",), "k1") is not None)
    # 等到早已过期
    import time

    # 睡 2 秒，远超 1.2 秒的 TTL
    time.sleep(2)
    # 关键：过期了，但没人清理，它仍然读得到
    print("过期后直接读:", pg.get(("ttl_demo",), "k1") is not None)
    # 手动触发清理，返回清掉的条数
    print("sweep_ttl 清掉:", pg.sweep_ttl())
    # 清理之后才真的没了
    print("清理后再读  :", pg.get(("ttl_demo",), "k1") is not None)

输出（实测）：

写入后立刻读: True
过期后直接读: True
sweep_ttl 清掉: 1
清理后再读  : False

「过期」不等于「消失」。 expires_at 只是个标记，真正的删除靠清理任务。sweep_interval_minutes 设成正数时 store 会起后台线程定期清理；设成 None 就必须你自己调 sweep_ttl()。

所以：不能把 TTL 当合规删除手段。若法规要求「30 天后必须删除」，你得保证清理任务真的在跑，并且自己验证过。

9.4 选型表 #

从 InMemoryStore 迁到 PostgresStore 的检查清单：

1. 所有「取最近 N 条」的地方是否显式排序？（默认顺序会反）
2. 有没有依赖 created_at 判断「首次学到的时间」？（语义不同）
3. 命名空间标签有没有可能含 .？（会抛错）
4. 语义检索用到了吗？（要装 pgvector）
5. create_agent(store=...) 是否在 with 块内？（出块即失效）

10. 实战：带长期记忆的客服 Agent #

10.1 它做了什么 #

三条设计决策，都是前面各节踩过坑之后的结论：

1. 档案用抽取（确定性），笔记用工具（可控性）——§7.4 的推荐组合。
2. 抽取只喂 HumanMessage——避免 §7.5 的回声污染。
3. 档案注入全量，笔记只注入最近 5 条——控制系统提示的长度，不配 index（§6.7）。

10.2 一个文件跑完整条链路 #

"""带长期记忆的客服 Agent

三个命名空间分别承担三件事：
    (user_id, "profile")   用户档案：姓名 / 城市 / 风格偏好，profile 模式（单 key 合并覆盖）
    (user_id, "memories")  自由笔记：用户随口提到的事实，collection 模式（一条一个 key）
    ("org", "policies")    内部知识库：全体用户共享的制度条款，只读

用法：
    python memory_agent.py    # 四幕演示：建档 → 换会话回忆 → 换用户隔离 → 查共享制度，最后巡检
"""

# 让 str | None 这类新式类型标注在旧版本 Python 上也能用
from __future__ import annotations

# json 用于把档案字典拼进系统提示、以及巡检时打印
import json

# uuid 给每条笔记生成唯一 key：并行工具调用时用「计数」会互相覆盖
import uuid

# dataclass 用来声明 context_schema（每次请求要带的上下文）
from dataclasses import dataclass

# dotenv 把 .env 里的 DEEPSEEK_API_KEY 等注入环境变量
from dotenv import load_dotenv

# create_agent 是 Agent 工厂；AgentState 是内置状态类型
from langchain.agents import AgentState, create_agent

# after_agent：一个回合结束后执行；dynamic_prompt：每次调模型前动态拼系统提示
from langchain.agents.middleware import ModelRequest, after_agent, dynamic_prompt

# init_chat_model 用来手工构造模型对象（结构化输出需要传 extra_body）
from langchain.chat_models import init_chat_model

# HumanMessage 用于「只采信用户原话」的过滤
from langchain.messages import HumanMessage

# tool 装饰器把普通函数变成工具；ToolRuntime 让工具拿到 store 和 context
from langchain.tools import ToolRuntime, tool

# 短期记忆（同一会话的多轮上下文）仍然由 checkpointer 负责
from langgraph.checkpoint.memory import InMemorySaver

# BaseStore 是所有 store 的公共类型，用于类型标注
from langgraph.store.base import BaseStore

# Runtime 让中间件拿到 store 与 context
from langgraph.runtime import Runtime

# InMemoryStore 是内存版长期记忆，进程退出即丢
from langgraph.store.memory import InMemoryStore

# Pydantic 声明「档案抽取」的输出结构
from pydantic import BaseModel, Field

# 读取 .env
load_dotenv()

# 对话主模型：普通聊天不需要关思考模式
CHAT_MODEL = "deepseek:deepseek-v4-flash"

# 演示用的用户标识，换一个就是一片全新的记忆空间
DEMO_USER = "u_zhou"

# 抽取档案要走结构化输出，DeepSeek 必须显式关掉思考模式，否则 400
EXTRACT_MODEL = init_chat_model(
    # 模型标识
    "deepseek:deepseek-v4-flash",
    # 抽取任务用 0 温度，减少字段填写的随机性
    temperature=0,
    # extra_body 原样透传给 DeepSeek 接口，关掉 thinking
    extra_body={"thinking": {"type": "disabled"}},
)

# 内部制度条款：演示用的「长期知识库」，全体用户共享
POLICIES = {
    # 退货政策，第 4 幕演示时会被查到
    "refund": "7 天无理由退货，需保持包装完好；生鲜类不支持无理由退货。",
    # 运费政策
    "shipping": "满 99 元包邮；杭州、上海、南京支持次日达。",
    # 发票政策
    "invoice": "电子发票在订单完成后 24 小时内开具，可在订单详情页下载。",
}


# dataclass 让这个类不用手写 __init__，create_agent 的 context_schema 认它
@dataclass
class Ctx:
    """每次请求都要带的上下文：当前是谁在说话。

    user_id 决定命名空间，也就决定了「读到谁的记忆」。
    它不能放进 store，也不该由模型决定——必须由业务代码在 invoke 时传入。
    """

    # 用户标识，来自你的登录态 / 会话系统
    user_id: str


class Profile(BaseModel):
    """从对话里抽取的用户档案。

    三个字段全部可选：本轮没提到就必须留空，不能瞎猜。
    """

    # 姓名字段，没提到留空
    name: str | None = Field(default=None, description="用户姓名，用户没亲口说就留空")
    # 常住城市，没提到留空
    city: str | None = Field(
        default=None, description="用户常住城市，用户没亲口说就留空"
    )
    # 回答风格偏好，没提到留空
    style: str | None = Field(
        default=None, description="用户对回答风格的偏好，没提到就留空"
    )


# 把模型包装成「只输出 Profile 结构」的抽取器
EXTRACTOR = EXTRACT_MODEL.with_structured_output(Profile)


def profile_ns(user_id: str) -> tuple[str, ...]:
    """用户档案的命名空间。"""
    # 约定：第一层放用户标识，第二层放数据类别
    return (user_id, "profile")


def memory_ns(user_id: str) -> tuple[str, ...]:
    """用户自由笔记的命名空间。"""
    # 与 profile 平级，便于 search((user_id,)) 一次取全量
    return (user_id, "memories")


# 内部知识库的命名空间：不带 user_id，表示全体共享
POLICY_NS = ("org", "policies")


def seed_policies(store: BaseStore) -> None:
    """把内部制度写进共享命名空间（幂等：同 key 覆盖）。"""
    # 逐条写入；key 用制度编号，重复执行只是覆盖，不会产生重复记录
    for key, text in POLICIES.items():
        # value 必须是字典，所以把正文包成 {"text": ...}
        store.put(POLICY_NS, key, {"text": text})


def load_profile(store: BaseStore, user_id: str) -> dict[str, str]:
    """读取用户档案，没有就返回空字典。"""
    # profile 模式：整个档案就是一个 key
    item = store.get(profile_ns(user_id), "basic")
    # get 命中返回 Item，未命中返回 None
    return dict(item.value) if item else {}


def merge_profile(
    store: BaseStore, user_id: str, patch: dict[str, str | None]
) -> dict[str, str]:
    """把抽取结果合并进档案：只覆盖非空字段。

    这是本脚本最关键的一个约定。store.put 是**整体覆盖**，
    如果直接把抽取结果写回去，本轮没提到的字段会被 None 冲掉，
    出现「越聊越空」的失忆事故。
    """
    # 先读旧档案
    merged = load_profile(store, user_id)
    # 记下合并前的样子，用于判断这一轮到底有没有新信息
    before = dict(merged)
    # 只把「本轮真的抽到了」的字段覆盖上去
    for field, value in patch.items():
        # None 表示「这轮没提到」，拿它覆盖等于把老信息删掉
        if value is not None:
            # 非空才写进合并结果
            merged[field] = value
    # 没有任何变化就不写：避免刷新 updated_at，也避免写出空档案
    if merged != before:
        # put 是整体覆盖，所以交给它的必须是完整档案
        store.put(profile_ns(user_id), "basic", merged)
    # 返回合并后的结果，方便调用方打印
    return merged


@tool
def remember_note(text: str, runtime: ToolRuntime[Ctx]) -> str:
    """把一条需要长期记住的用户事实写入长期记忆。text 用一句陈述句。"""
    # 工具里通过 runtime 拿 store，和 create_agent 传进去的是同一个对象
    store = runtime.store
    # 命名空间由 context 里的 user_id 决定，模型无法越权写到别人名下
    ns = memory_ns(runtime.context.user_id)
    # 先做一次朴素查重：完全同文的笔记不再重复写入
    existing = store.search(ns, limit=100)
    # 逐条比对已有笔记
    for item in existing:
        # 去掉首尾空白再比，避免「多打一个空格」被当成新笔记
        if item.value.get("text", "").strip() == text.strip():
            # 命中重复就返回，把跳过的原因明确告诉模型
            return f"已存在相同笔记（{item.key}），本次跳过"
    # collection 模式：一条笔记一个 key。key 必须用随机值，
    # 因为模型可能在同一步里并行调用两次本工具，用「条数+1」会算出同一个 key 互相覆盖
    key = f"note-{uuid.uuid4().hex[:8]}"
    # 写入，value 是一个普通字典
    store.put(ns, key, {"text": text})
    # 工具的返回值会作为 ToolMessage 回灌给模型
    return f"已记住：{text}（{key}）"


@tool
def recall_notes(keyword: str, runtime: ToolRuntime[Ctx]) -> str:
    """按关键词回忆用户的长期笔记；keyword 传空字符串表示列出全部。"""
    # 同样通过 runtime 定位到当前用户
    store = runtime.store
    # 命名空间由 context 决定，模型改不了
    ns = memory_ns(runtime.context.user_id)
    # 无 query 的 search 就是「列举这个命名空间」，limit 要给足
    items = store.search(ns, limit=100)
    # 关键词为空则全量返回，否则做一次朴素包含匹配
    hits = [
        i for i in items if not keyword.strip() or keyword in i.value.get("text", "")
    ]
    # 没有命中时明确告诉模型「没有」，避免它自己编
    if not hits:
        # 返回一句明确的否定，比返回空字符串更能抑制幻觉
        return "没有相关笔记"
    # 拼成短文本回灌，行首加序号方便模型引用
    return "\n".join(f"{n}. {i.value['text']}" for n, i in enumerate(hits, 1))


@tool
def search_policy(keyword: str, runtime: ToolRuntime[Ctx]) -> str:
    """查询公司内部制度（退货 / 运费 / 发票等）。"""
    # 制度放在共享命名空间，与用户无关
    store = runtime.store
    # 依然是列举 + 关键词过滤；换成语义检索见教程 §6
    items = store.search(POLICY_NS, limit=100)
    # 制度条数很少，朴素的子串匹配就够用
    hits = [
        i for i in items if not keyword.strip() or keyword in i.value.get("text", "")
    ]
    # 关键词没命中就把全部条款给模型，让它自己判断
    chosen = hits or items
    # 带上 key 前缀，模型引用时能说清依据的是哪一条
    return "\n".join(f"[{i.key}] {i.value['text']}" for i in chosen)


@dynamic_prompt
def prompt_with_memory(request: ModelRequest) -> str:
    """每次调模型前，把长期记忆拼进系统提示。

    这一步不可省：store 里的数据不会自动进入模型输入，
    模型只能看到系统提示 + messages。
    """
    # 中间件通过 request.runtime 拿 store 与 context
    store = request.runtime.store
    # 当前是谁在说话，决定读哪个命名空间
    user_id = request.runtime.context.user_id
    # 固定的角色说明，三句分别对应「语气 / 何时写 / 何时查」
    base = [
        # 语气要求
        "你是电商客服助手，回答简短、不寒暄。",
        # 写入时机：说清楚才不会漏调工具
        "用户提到新的个人事实（住址、家庭、忌口、偏好等）时，调用 remember_note 记下来。",
        # 查证要求：制度类问题不许凭记忆答
        "涉及退货 / 运费 / 发票等制度问题时，必须调用 search_policy 查证后再回答。",
    ]
    # 没配 store 时降级为「无记忆模式」，而不是直接崩
    if store is None:
        # 明确写出「未接入」，方便从回复里看出配置问题
        return "\n".join(base) + "\n（当前未接入长期记忆）"
    # 档案是短字段，直接整段注入最省事
    profile = load_profile(store, user_id)
    # 空档案就不加这一段，免得给模型一个空对象徒增困惑
    if profile:
        # JSON 形式最省 token，也最不容易被模型误读
        base.append("已知用户档案：" + json.dumps(profile, ensure_ascii=False))
    # 笔记可能很多，只注入最近若干条，避免把系统提示撑爆
    notes = store.search(memory_ns(user_id), limit=5)
    # 同样，没有笔记就不加这一段
    if notes:
        # 用分号连成一行，比逐条换行更省 token
        base.append(
            "已知用户笔记：" + "；".join(i.value.get("text", "") for i in notes)
        )
    # 拼成一段系统提示返回
    return "\n".join(base)


@after_agent
def extract_profile(state: AgentState, runtime: Runtime[Ctx]) -> None:
    """一个回合结束后，从**用户原话**里抽取档案并合并写入。

    只喂 HumanMessage 是刻意的：如果把 AI 的回复也喂进去，
    模型自己猜错的信息会被当成用户事实写进档案（记忆回声污染）。
    """
    # 没接 store 就什么都不做
    if runtime.store is None:
        # 直接返回而不是抛错：漏配 store 不该让整个回合失败
        return None
    # 只取用户消息的文本，AI 与工具消息一律丢弃
    said = [
        m.content
        for m in state["messages"]
        if isinstance(m, HumanMessage) and m.content
    ]
    # 用户这一回合没说话（例如只是恢复中断）就跳过，省一次模型调用
    if not said:
        # 没有输入就没有可抽的东西
        return None
    # 调用结构化抽取器；抽取失败不能影响主流程，所以整段包在 try 里
    try:
        # 把这一轮所有用户原话拼成一段喂给抽取器
        got = EXTRACTOR.invoke("从下面的用户原话里抽取用户档案：\n" + "\n".join(said))
    # 这里刻意捕获所有异常：记忆写入永远不该拖垮对话
    except Exception:  # noqa: BLE001
        # 网络抖动、限流、模型报错都归到这里：这一轮不更新档案，下一轮还有机会
        return None
    # with_structured_output 偶发返回 None（模型没产出工具调用），必须判空
    if got is None:
        # 同样跳过这一轮，不能让 .model_dump() 在 None 上炸掉
        return None
    # 合并写回，只覆盖非空字段
    merge_profile(runtime.store, runtime.context.user_id, got.model_dump())
    # after_agent 不需要改状态，返回 None
    return None


def build_agent(store: BaseStore):
    """组装 Agent：短期记忆用 checkpointer，长期记忆用 store。"""
    # store 由调用方传入，本函数不关心它是哪种实现
    return create_agent(
        # 主对话模型
        model=CHAT_MODEL,
        # 三个工具分别对应「写笔记 / 读笔记 / 查制度」
        tools=[remember_note, recall_notes, search_policy],
        # 读：注入档案与笔记；写：回合结束抽档案
        middleware=[prompt_with_memory, extract_profile],
        # 长期记忆：换 thread 也还在（内存版仅限本进程）
        store=store,
        # 短期记忆：同一 thread 内的多轮上下文
        checkpointer=InMemorySaver(),
        # 声明 context 的结构，工具与中间件才能读到 user_id
        context_schema=Ctx,
    )


def say(agent, store: BaseStore, user_id: str, thread_id: str, text: str) -> None:
    """跑一轮对话并打印关键结果。"""
    # 打印本轮输入，方便对照
    print(f"\n[{user_id} @ {thread_id}] 用户：{text}")
    # thread_id 决定短期记忆，context 决定长期记忆的归属
    result = agent.invoke(
        # 本轮用户输入
        {"messages": [{"role": "user", "content": text}]},
        # thread_id 决定这轮接在哪条会话线后面
        config={"configurable": {"thread_id": thread_id}},
        # user_id 决定读写谁的长期记忆
        context=Ctx(user_id=user_id),
    )
    # 只看最后一条 AI 回复
    print(f"    助手：{result['messages'][-1].content}")
    # 顺手打印工具调用轨迹，证明记忆是「查出来的」不是「猜出来的」
    calls = [
        # 取工具名
        c["name"]
        # 遍历本轮所有消息
        for m in result["messages"]
        # 只有 AIMessage 才有 tool_calls，用 getattr 兜底成空列表
        for c in getattr(m, "tool_calls", []) or []
    ]
    # 消息条数能直观反映走了几次工具往返：2 条表示零往返
    print(f"    工具调用：{calls or '无'}｜消息条数：{len(result['messages'])}")
    # 打印当前档案，观察它随对话增长
    print(f"    档案：{load_profile(store, user_id) or '（空）'}")


def inspect_store(store: BaseStore) -> None:
    """巡检：把 store 里所有命名空间与条目打印出来。"""
    # list_namespaces 先看有哪些「目录」
    namespaces = store.list_namespaces()
    # 注意这个数字可能虚高：InMemoryStore 的 get 会留下空的幽灵命名空间
    print(f"命名空间共 {len(namespaces)} 个：")
    # 逐个命名空间列举条目，limit 给足以免被静默截断
    for ns in namespaces:
        # 不传 query 的 search 就是纯列举，零嵌入成本
        items = store.search(ns, limit=100)
        # 用 / 把元组拼成路径形式，比原始元组好读
        print(f"  {'/'.join(ns)}  条数={len(items)}")
        # 再逐条打印内容
        for item in items:
            # ensure_ascii=False 保证中文原样显示
            print(f"      {item.key}: {json.dumps(item.value, ensure_ascii=False)}")


# 内存版 store：零依赖，进程退出即丢，演示够用
store = InMemoryStore()
# 先把共享制度灌进去，第 4 幕要查它
seed_policies(store)
# 组装 Agent
agent = build_agent(store)
# 第 1 幕：同一会话内建档 + 写笔记
print("=" * 68)

print("第 1 幕：thread=t1，建立档案与笔记")
# 下分隔线
print("=" * 68)
# 第一轮：一句话给出姓名、城市、风格三个档案字段
say(agent, store, DEMO_USER, "t1", "我叫小周，住杭州，回答别啰嗦。")
# 第二轮：给出两个零散事实，观察模型会不会并行调两次 remember_note
say(agent, store, DEMO_USER, "t1", "顺便记一下，我家有两只猫，对乳制品过敏。")
# 第 2 幕：换 thread，短期记忆清零，长期记忆仍在
print("\n" + "=" * 68)

print("第 2 幕：thread=t2（全新会话），验证跨会话记忆")
# 下分隔线
print("=" * 68)
# 换了 thread_id，答案只可能来自 store
say(agent, store, DEMO_USER, "t2", "我住哪个城市？家里有什么宠物？")
# 第 3 幕：换用户，命名空间隔离，查不到别人的档案
print("\n" + "=" * 68)

print("第 3 幕：换 user_id，验证隔离")
# 下分隔线
print("=" * 68)
# 同样的问题换个人问，应当答不出来
say(agent, store, "u_other", "t3", "我住哪个城市？")
# 第 4 幕：共享知识库照常可查，与用户无关
print("\n" + "=" * 68)

print("第 4 幕：共享知识库（与 user 无关）")
# 下分隔线
print("=" * 68)
# 制度不属于任何用户，所以 u_other 也能查到
say(agent, store, "u_other", "t3", "生鲜能七天无理由退货吗？")
# 最后巡检一次，看清数据落在哪
print("\n" + "=" * 68)

print("巡检")
# 下分隔线
print("=" * 68)
# 把所有命名空间和条目打出来
inspect_store(store)

11. 约定与坑 #

11.1 数据模型与命名空间 #

11.2 Agent 集成 #

11.3 语义检索 #

11.4 写入时机与内容 #

11.5 两条通用原则 #

第一条：长期记忆的错误不会自愈。 短期记忆错了，换个 thread_id 就重置；长期记忆错了，它会跟着用户走很久，还会每轮加固自己。所以：写入要保守（宁可少记，不要记错），并且永远给用户和运维留一条删除路径。

第二条：能不存就不存。 每条长期记忆都是持续的 token 成本（每轮注入）+ 合规负担（PII）+ 出错风险。上线前对每个字段问一句：「不记它会怎样？」答不上来的，就不要记。

12. 练习 #

1. 档案字段规范化（§7.2）。把 Profile.style 从自由文本改成 Literal["concise", "detailed", "friendly"]，重跑 §7.2 的例子，观察抽取结果从「说话别绕弯子」变成 concise。再想一下：如果用户说的偏好不属于这三类，模型会怎么填？
2. 加一个「忘掉」工具（§10.2）。给本章实战加 forget_note(keyword)：先 search 找到匹配的笔记，再逐条 delete，返回删了几条。跑一轮「忘掉我养猫这件事」验证。
3. 把注入换成语义检索（§6.1）。给本章实战的 memories 命名空间配 index，把 prompt_with_memory 里「最近 5 条」改成「与最后一条用户消息最相关的 3 条」。先写 20 条笔记，再对比两种读法注入的内容差异。
4. 实测回声污染（§7.5）。故意把 extract_profile 里的 HumanMessage 过滤去掉，然后设计一段对话：用户不说城市，想办法让 Agent 猜一个。观察错的城市进入档案后，后面几轮是怎么被反复强化的。
5. TTL 清理任务（§9.3）。给一类时效性记忆（比如「用户正在咨询订单 A1002」）设 ttl=60（一小时），并写一个每 10 分钟调 sweep_ttl() 的循环。验证过期条目在 sweep 前后的可读性差异。
6. 迁移演练（§9.4）。把本章实战的 store 从内存换成 Postgres，然后逐条走 §9.4 的五项检查清单，看你的代码里有几条会踩雷。

13. 小结 #

这一章把「跨会话记忆」拆成三个独立问题：存在哪（store 数据模型）、怎么找（命名空间与语义检索）、什么时候写（时机与污染防御）。

七条最该带走的结论：

1. checkpointer 和 store 解决的不是同一个问题。前者保存一条会话的过程，后者保存跨会话的事实；缺前者是「同一会话内失忆」，缺后者是「换会话就失忆」，两者不可互相替代（§1.1）。
2. namespace 是唯一的隔离边界，而它必须由业务代码决定。user_id 只能从 context 来，绝不能做成工具入参——否则模型可以读任何人的记忆（§5.4）。
3. put 是整体覆盖。任何「更新档案」的地方都必须先读再合并，且只覆盖非空字段，否则档案会越聊越空（§4.2）。
4. store 里的数据不会自动进入模型输入。要么显式注入系统提示（省钱，适合短字段），要么做成工具（灵活，适合按需查）（§5.5）。
5. 语义检索是可选项，不是标配。几十条记忆用「全量列举 + 只注入最近 N 条」又快又准；配 index 会引入嵌入成本和三个静默陷阱（§6.4、§6.7）。
6. 回声污染是长期记忆最危险的失效模式。抽取时只喂用户原话，否则模型会把自己的猜测存成用户事实，并且一路自我强化（§7.5）。
7. InMemoryStore 只能用于开发。上生产换 PostgresStore，并且要预期到四处行为差异：默认排序、created_at 语义、幽灵命名空间、TTL 支持（§9.3）。