代码仓库ChainReaction

上一篇文章里我们给 agent 挂了 checkpointer，同一个 thread_id 下它能记住用户叫什么。把 thread_id 一换，它就失忆了。这个行为对单次会话来说是对的，做成产品就错了。用户在客服窗口里说过”我不用信用卡”，三天后他打开另一个窗口，这句话不该消失。

LangChain 里管这件事的组件叫 store。它和 checkpointer 是两套独立的东西，一个存 thread 内的消息，一个存跨 thread 的 JSON 文档。这篇把 store 这一层拆开：namespace 怎么设计、写入和召回有哪几条路、语义检索怎么接、什么时候该忘掉、噪声怎么挡在门外。

文里的代码都在 langchain 1.4.3 + langgraph 1.2.12 上跑过，输出是从终端粘出来的。对话模型统一用 DeepSeek，向量用本地跑的 BAAI/bge-small-zh-v1.5，原因在语义检索那一节。

短期记忆和长期记忆的边界

短期记忆就是 agent 的 state。默认的 AgentState 里有一个 messages 键，整段对话历史躺在里面。挂上 checkpointer 之后，这份 state 按 thread_id 落盘，恢复 thread 时读回来。它解决的问题是”这一通对话说到哪了”。

长期记忆是 store。每个条目是一份 JSON 文档，由两层坐标定位：一个 namespace 元组，相当于文件夹；一个 key 字符串，相当于文件名。这两层都不绑定 thread，所以 thread B 能读到 thread A 写的东西，只要 namespace 对得上。

官方文档里把长期记忆按内容分成三类，这个划分对设计 namespace 有实际帮助：

类型 存什么 例子
语义记忆 事实 用户写 Python、用户在上海
情节记忆 经历 上次帮用户调通了 asyncio 超时
程序记忆 规则 agent 自己的系统提示词

三类混在同一个 namespace 里也能跑，但检索的时候会很别扭。后面讲 namespace 设计时会用 (user_id, "preferences") 和 (user_id, "episodes") 把语义记忆和情节记忆分开。

两者的分工用一张图看得更清楚：

图里实线走的是 checkpointer，虚线走的是 store。checkpointer 每个 thread 各存一份，store 是所有 thread 共用的同一份。

store 抽象：从 InMemoryStore 起步

InMemoryStore 是最简单的实现，数据存在一个 Python 字典里，进程一退就没了。开发和测试够用，生产要换成 PostgresStore、MongoDBStore 或 RedisStore，它们都继承同一个 BaseStore。

把 store 单独拿出来用，不接 agent 也能跑。下面这段脚本演示了 put / get / search / delete / list_namespaces 五个方法，以及 namespace 的匹配规则：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

ns_alice = ("alice", "preferences")
ns_alice_ep = ("alice", "episodes")
ns_bob = ("bob", "preferences")

store.put(ns_alice, "pref-language", {"kind": "language", "text": "用户只用中文提问"})
store.put(ns_alice, "pref-editor", {"kind": "tooling", "text": "用户用 VS Code，暗色主题"})
store.put(ns_alice_ep, "ep-2025-11-20", {"kind": "episode", "text": "帮用户调通了 asyncio 超时问题"})
store.put(ns_bob, "pref-language", {"kind": "language", "text": "user prefers English"})

item = store.get(ns_alice, "pref-language")
print(item.key, item.namespace, item.value, item.created_at)
print(store.get(ns_alice, "nope"))          # None

print([i.key for i in store.search(ns_alice, limit=10)])
print([i.key for i in store.search(("alice",), limit=10)])
print([i.key for i in store.search(("alice",), filter={"kind": "episode"})])
print(store.list_namespaces(prefix=("alice",)))

store.delete(ns_alice, "pref-editor")

真实输出（脚本在 LongTermMemory/StoreBasics.py）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
=== get：按 namespace + key 精确取一条 ===
type(item)      = Item
item.key        = pref-language
item.namespace  = ('alice', 'preferences')
item.value      = {'kind': 'language', 'text': '用户只用中文提问'}
item.created_at = 2026-10-04 15:28:09.124471+00:00
item.updated_at = 2026-10-04 15:28:09.124471+00:00
取不存在的 key  -> None

=== search：namespace 是按前缀匹配的 ===
search(('alice','preferences')) 命中 2 条 -> ['pref-language', 'pref-editor']
search(('alice',))              命中 3 条 -> ['pref-language', 'pref-editor', 'ep-2025-11-20']
其中 namespace 列表 = [('alice', 'preferences'), ('alice', 'preferences'), ('alice', 'episodes')]
InMemoryStore 按插入顺序返回，PostgresStore 按 updated_at 倒序，不要依赖顺序

=== filter：按内容等值过滤 ===
filter={'kind':'episode'} -> ['ep-2025-11-20']

=== list_namespaces：先看有哪些 namespace ===
list_namespaces()                     -> [('alice', 'episodes'), ('alice', 'preferences'), ('bob', 'preferences')]
list_namespaces(prefix=('alice',))    -> [('alice', 'episodes'), ('alice', 'preferences')]

=== delete：遗忘一条 ===
delete 之后 search(('alice',)) -> ['pref-language', 'ep-2025-11-20']

有几处行为值得单独拎出来。

search 的第一个参数是 namespace_prefix，不是 namespace 本身。传 ("alice",) 会把 ("alice", "preferences") 和 ("alice", "episodes") 一起捞回来，也就是前缀匹配。想只拿一级，就把完整 namespace 传进去，或者拿回来之后自己按 item.namespace 过滤。

limit 默认 10，超出的部分静默丢弃，没有任何提示。分页靠 offset 手工翻。这个默认值在记忆条目涨到几十条之后很容易踩到，我一般会在检索工具里显式写 limit=20。

返回顺序不能依赖。InMemoryStore 是插入顺序，最新的在最后；PostgresStore 是 updated_at 倒序，最新的在最前。如果代码里写了 items[-1] 来取”最近一条”，换后端的那天就会静默出错。

get 返回的是 Item，字段有 key、value、namespace、created_at、updated_at。带 query 检索时返回的是 SearchItem，多一个 score。

namespace 怎么设计

namespace 是任意长度的字符串元组，官方建议把 user id 或 org id 放进去。实践中我用两级：

1
2
3
4
(user_id, "preferences")   稳定偏好，一个 slot 一个 key
(user_id, "episodes")      做过的事，一个事件一个 key
(user_id, "profile")       结构化画像，key 是字段名
(org_id, "shared_facts")   组织级共享知识

顺序不要反过来写 ("preferences", user_id)。前缀匹配是从左往右的，把 user_id 放在最外层，search((user_id,)) 一次就能把这个用户的所有记忆捞全，跨类别检索也不用改代码。

key 的选择决定了写入是新增还是覆盖。同一 namespace 下用同一个 key 再 put，就是覆盖，条目数不变。用随机 UUID 当 key，就是每次新增一条。偏好类记忆用稳定 key，情节类记忆用 UUID 加时间戳，这是我用下来最省心的分法。

覆盖的时候要留意一个坑。InMemoryStore 的覆盖会把整条记录换掉，created_at 也一起重置。我实测过：

1
2
created_at 是否保留 -> False（InMemoryStore 覆盖时连 created_at 一起重置）
updated_at 是否刷新 -> True

如果业务上需要”首次记住的时间”，得自己往 value 里塞一个字段，别指望 created_at。

还有两件小事。store 的每个方法都有异步版本，aput、aget、asearch、adelete、alist_namespaces，在 FastAPI 这类异步框架里直接用异步版，别把同步调用塞进事件循环。另外 search 的第一个参数是前缀，传空元组 () 就能跨用户检索，做后台审计时用得上，线上接口里别这么干，很容易把别人的记忆喂给当前用户。

写入与检索的三条路径

长期记忆的读写发生在哪里，有三种做法。它们的差别不在 API，而在谁来拍板、什么时候拍板。

路径一：工具里读写，模型自己决定

这是最直接的一条路。把 store 传给 create_agent，工具签名里加一个 ToolRuntime 参数，就能通过 runtime.store 拿到同一个 store。这个参数对模型是不可见的，模型只看到业务参数。

下面这个脚本里，thread A 让模型自己决定写入，thread B 是全新会话，只带着同一个 user_id：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
import os
import zlib
from dataclasses import dataclass

from langchain.agents import create_agent
from langchain.tools import ToolRuntime, tool
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore

model = ChatOpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",
    model="deepseek-chat",
    temperature=0.1,
    max_tokens=1000,
)


@dataclass
class Context:
    user_id: str


store = InMemoryStore()
checkpointer = InMemorySaver()


def _stable_key(text: str) -> str:
    """同一句话算同一个 key，重复写入就是覆盖，不会堆垃圾。"""
    return "pref-" + format(zlib.crc32(text.encode("utf-8")), "08x")


@tool
def remember_preference(preference: str, runtime: ToolRuntime[Context]) -> str:
    """把用户的一条长期偏好写进长期记忆。只有用户明确表达、且以后还会用到的偏好才调用。"""
    assert runtime.store is not None
    user_id = runtime.context.user_id
    runtime.store.put(
        (user_id, "preferences"),
        _stable_key(preference),
        {"kind": "preference", "text": preference},
    )
    return f"已写入长期记忆：{preference}"


@tool
def recall_preferences(runtime: ToolRuntime[Context]) -> str:
    """读出当前用户存在长期记忆里的全部偏好。用户问起自己的习惯、环境、要求时调用。"""
    assert runtime.store is not None
    user_id = runtime.context.user_id
    items = runtime.store.search((user_id, "preferences"), limit=20)
    if not items:
        return "长期记忆里没有这个用户的偏好记录。"
    return "\n".join(f"- {it.value['text']}" for it in items)


agent = create_agent(
    model=model,
    tools=[remember_preference, recall_preferences],
    store=store,
    checkpointer=checkpointer,
    context_schema=Context,
    system_prompt=(
        "你是编码助手。用户表达的稳定偏好要调用 remember_preference 记住；"
        "用户问自己的习惯或环境时，先调用 recall_preferences 再回答。"
    ),
)

thread_a = {"configurable": {"thread_id": "thread-A"}}
thread_b = {"configurable": {"thread_id": "thread-B"}}
ctx = Context(user_id="u-42")

agent.invoke(
    {"messages": [{"role": "user", "content": "记一下我的习惯：我写 Python，用 VS Code，回答我先给结论再给代码。"}]},
    thread_a,
    context=ctx,
)
agent.invoke(
    {"messages": [{"role": "user", "content": "我平时用什么编辑器？回答我应该怎么组织？"}]},
    thread_b,
    context=ctx,
)

真实输出（其中一行太长，做了折行）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
=== thread A：模型自己决定写入 ===
  [tool] put namespace=(u-42, 'preferences') key=pref-be6237d9
  [tool] put namespace=(u-42, 'preferences') key=pref-7c9fdab8
  [tool] put namespace=(u-42, 'preferences') key=pref-acb4889a
--- thread A ---
HumanMessage: 记一下我的习惯：我写 Python，用 VS Code，回答我先给结论再给代码。
AIMessage: tool_calls=['remember_preference', 'remember_preference', 'remember_preference']
  args=[{'preference': '用户写 Python'}, {'preference': '用户使用 VS Code 作为编辑器'}, {'preference': '回答时先给结论，再给代码'}]
ToolMessage: 已写入长期记忆：用户写 Python
ToolMessage: 已写入长期记忆：用户使用 VS Code 作为编辑器
ToolMessage: 已写入长期记忆：回答时先给结论，再给代码
thread A 最终回答: 已记住三条偏好：

1. 你写 Python
2. 你用 VS Code
3. 回答先给结论，再给代码

以后我会按这个方式来。

=== store 里现在有什么（和 thread 无关，直接读） ===
key=pref-be6237d9 namespace=('u-42', 'preferences') value={'kind': 'preference', 'text': '用户写 Python'}
key=pref-7c9fdab8 namespace=('u-42', 'preferences') value={'kind': 'preference', 'text': '用户使用 VS Code 作为编辑器'}
key=pref-acb4889a namespace=('u-42', 'preferences') value={'kind': 'preference', 'text': '回答时先给结论，再给代码'}

=== thread B：全新会话，只带着 user_id ===
  [tool] search (u-42, 'preferences') 命中 3 条
--- thread B ---
HumanMessage: 我平时用什么编辑器？回答我应该怎么组织？
AIMessage: tool_calls=['recall_preferences'] args=[{}]
ToolMessage: - 用户写 Python
- 用户使用 VS Code 作为编辑器
- 回答时先给结论，再给代码
thread B 最终回答: 你平时用的是 **VS Code**（写 Python）。

至于回答怎么组织，按你之前的要求：**先给结论，再给代码**。

=== checkpointer 的边界：两个 thread 的 state 互不可见 ===
thread A 的消息条数 = 6
thread B 的消息条数 = 4
thread B 里能看到 thread A 的消息吗 -> False

thread B 的 state 里只有 4 条消息，看不到 thread A 说过什么。它答出”VS Code”，靠的是工具去 store 里查。这就是长期记忆该有的样子：消息不共享，知识共享。

模型把一句话拆成了三条记忆，这个行为有利有弊。拆开之后每条更短、更容易被语义检索单独命中，代价是条目数涨得快，也更容易写出互相矛盾的条目。

路径二：中间件或节点里按代码逻辑写

不想让模型拍板的时候，就在 @after_model 中间件或者自定义节点里读 state、按固定规则写 store。适合结构化抽取，比如每轮对话结束后把用户提到的地名抽出来。写入时机完全可控，代价是规则要自己维护，遇到没覆盖的表达就漏了。

路径三：后台任务写

主链路跑完，另起一个任务拿整段对话去提炼记忆。好处是用户感知不到额外延迟，模型也能一次看到完整上下文。官方文档提到的触发方式有定时调度、cron 和用户手动触发。代价是新记忆有延迟，用户这轮说完、下轮问起，可能还没写进去。

按需召回

召回也有两种粒度。上面 recall_preferences 是模型主动调工具，把整个 namespace 拉出来。另一种是代码在调用模型之前先检索，把结果塞进系统提示词。后者不用多花一次工具调用，但每次都要检索，即使这轮对话用不上。

选择依据很简单：记忆条数少、模型知道什么时候需要，就用工具召回；条数多、命中率靠 embedding 撑着，就代码预召回加工具兜底。

三条路径的取舍

路径 谁决定写 发生时机 对主链路的延迟 适合的场景
工具 模型 对话中 多一轮工具调用 用户明确说”记住”的偏好
中间件/节点 代码规则 对话中 同步执行，可控 字段固定的结构化抽取
后台任务 代码调度 对话结束后 无 需要通读全文才能提炼的画像

我默认选第一条。它的好处是用户能看见记忆被写下来了，出问题时也容易从消息里找到是哪一句触发的。

一次对话里，写入和读取分别走哪条路，用一张时序图收一下：

语义检索：给 store 配 embedding

记忆涨到几十上百条之后，把整个 namespace 拉给模型就不合适了。search 支持 query 参数做相似度检索，前提是建 store 的时候配好 embedding。

IndexConfig 有三个字段。embed 是生成向量的东西，可以是 LangChain 的 Embeddings 实例、同步函数、异步函数，或者 "openai:text-embedding-3-small" 这样的 provider 字符串。dims 是向量维度，必须和模型对上。fields 决定把文档的哪些部分拿去算向量，默认 ["$"] 是整个 JSON 对象，也可以写成 ["text"] 只对某一个字段做 embedding。

1
2
3
4
5
6
7
8
9
from langgraph.store.memory import InMemoryStore

store = InMemoryStore(
    index={
        "embed": embed,          # 函数、Embeddings 实例或 provider 字符串
        "dims": 1536,
        "fields": ["text"],      # 只把 text 字段拿去算向量
    }
)

put 的时候还能逐条覆盖索引行为，index=["food_preference"] 指定字段，index=False 表示这条不进向量索引。

这里要如实说一件事：DeepSeek 没有 embedding 接口。我直接打了一下它的端点：

1
2
=== 0. 先确认 DeepSeek 有没有 embedding 端点 ===
HTTPError 404 Not Found -> DeepSeek 没有 embeddings 接口，向量只能另找来源

所以文章里的模型能换成 DeepSeek，向量不行。真实项目里向量来源要么是 OpenAI 的 text-embedding-3-small，要么是本地跑的 sentence-transformers，要么是国内厂商的 embedding API。

我这条路走本地模型：BAAI/bge-small-zh-v1.5，中文检索用的小模型，512 维，权重一百来兆。装依赖两条命令：

1
pip install langchain-huggingface sentence-transformers

第一次跑会从 HuggingFace 把权重下下来，慢的话等几分钟，之后走本地缓存。HuggingFaceEmbeddings 是 LangChain 对 sentence-transformers 的包装，init_embeddings("huggingface:BAAI/bge-small-zh-v1.5") 是同一个东西的字符串写法，两者可以互换。我用前者，因为要显式传 normalize_embeddings=True：bge 按余弦相似度训练，向量归一化之后点积就是余弦，省得再操心量纲。

dims 我不硬写 512，直接问模型要：

1
DIMS = len(embeddings.embed_query("维度探测"))

换模型的时候这一行跟着变，比手写常量安全。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from langchain_huggingface import HuggingFaceEmbeddings
from langgraph.store.memory import InMemoryStore

embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-small-zh-v1.5",
    encode_kwargs={"normalize_embeddings": True},
)
DIMS = len(embeddings.embed_query("维度探测"))

store = InMemoryStore(index={"embed": embeddings, "dims": DIMS, "fields": ["text"]})

for key, text in {
    "m-food": "用户喜欢吃披萨，尤其是玛格丽特",
    "m-code": "用户主要写 Python，用 pytest 做测试",
    "m-schedule": "用户周三下午三点之后不排会",
    "m-style": "用户要求回答简短，不要用感叹号",
}.items():
    store.put(("u-42", "memories"), key, {"kind": "semantic", "text": text})

store.put(("u-42", "memories"), "m-raw", {"kind": "meta", "text": "这条不建索引"}, index=False)

hits = store.search(("u-42", "memories"), query="用户用什么写代码", limit=3)
for h in hits:
    print(h.score, h.key, h.value["text"])

embed 传函数也完全可以，之前我用一个确定性函数占过位：字符 bigram 经 crc32 哈希到 128 维再归一化。它跑得飞快，也没有任何依赖，缺点是只对齐字面。我把它留成对照组，两套向量跑同一批 query，排序差别正好说明 store 这一层不产生语义。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import zlib
from collections.abc import Sequence


def embed_placeholder(texts: Sequence[str]) -> list[list[float]]:
    """对照组：字符 bigram 哈希到 128 维再归一化。只对齐字面，不是语义。"""
    vectors = []
    for text in texts:
        vec = [0.0] * 128
        for i in range(max(len(text) - 1, 1)):
            gram = text[i : i + 2] if len(text) > 1 else text
            vec[zlib.crc32(gram.encode("utf-8")) % 128] += 1.0
        norm = sum(x * x for x in vec) ** 0.5 or 1.0
        vectors.append([x / norm for x in vec])
    return vectors


store = InMemoryStore(index={"embed": embed_placeholder, "dims": 128, "fields": ["text"]})

真实输出：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
=== -1. 加载 embedding 模型 BAAI/bge-small-zh-v1.5（首次会从 HuggingFace 下载） ===
embed_query('维度探测') 的向量维度 = 512

=== 0. 先确认 DeepSeek 有没有 embedding 端点 ===
HTTPError 404 Not Found -> DeepSeek 没有 embeddings 接口，向量只能另找来源

=== 1. 不配 index 的 store：query 被静默忽略 ===
没有报错，返回 2 条：[('m1', None), ('m2', None)]
也就是说没配 index 时 query 被忽略，只是把 namespace 里的条目按默认顺序给你

=== 2. 占位向量（crc32 bigram，128 维）：query 只对齐字面 ===
query='用户喜欢吃什么'
    score=0.4364 key=m-food text=用户喜欢吃披萨，尤其是玛格丽特
    score=0.2041 key=m-style text=用户要求回答简短，不要用感叹号
    score=0.1179 key=m-schedule text=用户周三下午三点之后不排会
query='用户用什么写代码'
    score=0.3030 key=m-food text=用户喜欢吃披萨，尤其是玛格丽特
    score=0.2182 key=m-schedule text=用户周三下午三点之后不排会
    score=0.0945 key=m-style text=用户要求回答简短，不要用感叹号
query='回复风格有什么要求'
    score=0.2652 key=m-style text=用户要求回答简短，不要用感叹号
    score=0.1387 key=m-code text=用户主要写 Python，用 pytest 做测试
    score=0.0000 key=m-food text=用户喜欢吃披萨，尤其是玛格丽特

=== 3. 真 embedding（BAAI/bge-small-zh-v1.5，512 维） ===
query='用户喜欢吃什么'
    score=0.6418 key=m-food text=用户喜欢吃披萨，尤其是玛格丽特
    score=0.4374 key=m-code text=用户主要写 Python，用 pytest 做测试
    score=0.4276 key=m-style text=用户要求回答简短，不要用感叹号
query='用户用什么写代码'
    score=0.6543 key=m-code text=用户主要写 Python，用 pytest 做测试
    score=0.4837 key=m-style text=用户要求回答简短，不要用感叹号
    score=0.3490 key=m-schedule text=用户周三下午三点之后不排会
query='回复风格有什么要求'
    score=0.5602 key=m-style text=用户要求回答简短，不要用感叹号
    score=0.4097 key=m-code text=用户主要写 Python，用 pytest 做测试
    score=0.3406 key=m-schedule text=用户周三下午三点之后不排会

=== 4. 排序对比：占位向量 vs 真 embedding ===
query='用户喜欢吃什么'    #1 m-food -> m-food
                          #2 m-style -> m-code
                          #3 m-schedule -> m-style
query='用户用什么写代码'  #1 m-food -> m-code        <- 第一名变了
                          #2 m-schedule -> m-style
                          #3 m-style -> m-schedule
query='回复风格有什么要求' #1 m-style -> m-style
                          #2 m-code -> m-code
                          #3 m-food -> m-schedule

=== 6. filter 和 query 可以一起用 ===
命中 ['m-style', 'm-food', 'm-code', 'm-schedule']
index=False 的 m-raw 不会出现在 query 结果里 -> True
但它还能按 key 取到 -> {'kind': 'meta', 'text': '这条不建索引，只做普通记录'}

贴出来的输出做了删减：省掉第 5 节的 query instruction 对照和最后一节，第 4 个 query 也没列。下面正文引用到的数字都在脚本的完整输出里。脚本在 LongTermMemory/SemanticSearch.py，跑一次就能得到完整输出。

输出里有三点值得单独拎出来。

没配 index 的时候，query 不会报错，会被静默忽略。你以为在做语义检索，其实只是把 namespace 里的条目按默认顺序拿回来了。这类静默降级最难查，因为代码看起来完全正常。判断方法很简单，看返回的 SearchItem 有没有 score，没有就是没走向量。

query 和 filter 可以叠加，filter={"kind": "semantic"} 先把范围缩到某一类，再按相似度排序。

第三条是这个对比真正的价值所在。占位向量下，用户用什么写代码 的第一名是”喜欢吃披萨”，两句话共享的字符更多；”Python”和”写代码”没有任何共同字符，m-code 掉出前三。同一批数据、同一个 store，只把 embed 换成 bge，第一名就变成 m-code，分数 0.6543，和它本来的语义对上了。用户喜欢吃什么 和 回复风格有什么要求 这两个 query 的第一名两套向量都一样，所以这次改善挑不出”换什么都一样”的毛病，塌掉的就是占位向量唯一撑不住的那一条：跨字面理解。store 这一层只负责存和排序，语义全在 embedding 模型里。

真模型也不是万能的，有两个地方我自己跑出来觉得别扭，一并说清楚。

一个是分数没有绝对意义，别拿它当阈值。我问了一句 这条不建索引的记录，四条建了索引的记忆里没有一条在讲这件事，按道理应该返回”没有”。真 embedding 下它照样给了一个第一名：m-style，score=0.4910。bge 对任意两个中文句子都能给出 0.3 到 0.5 的底分，0.49 分完全可能是错的。要用分数做过滤就得自己定一条线，比如低于 0.55 当成没命中，或者在检索之后再加一层 rerank。指望”分数低自然就排到后面”不成立。

另一个是低分区的名次基本是噪声。还是 用户喜欢吃什么 这个 query，第二名 m-code 0.4374、第三名 m-style 0.4276，差 0.0098。这个差距换个 embedding 模型、甚至换个 encode_kwargs 就可能翻过来。真正稳的是第一名和后面拉开的那一截：0.6418 对 0.4374，差 0.2 以上。做 top-k 召回时 k 取 1 到 3 是有意义的，把 k 开到 10 再让模型自己挑，等于把噪声一起递过去了。

顺带一个 bge 特有的坑。官方建议检索时给 query 加一句前缀「为这个句子生成表示以用于检索相关文章：」，文档侧不加。我试了一下，加了前缀分数整体往下掉（m-code 从 0.6543 掉到 0.5838），名次没变。真正的问题是加不了：IndexConfig 只有一个 embed 函数，文档和 query 走同一条路，框架不区分调用方。要在 store 里用上这个前缀，得自己包一层函数，靠调用时的上下文去判断是不是 query，代价是这条路不再通用。我的选择是不加，省下的这点召回率不值得破坏 embed 的纯粹性。

还有一件容易漏的事。部署到 LangGraph Server 的时候，索引配置从 Python 代码里搬到了 langgraph.json 的 store.index 字段。本地跑得好好的语义检索，上线之后突然不生效，八成是漏了这一步。

记忆的治理

写进去容易，管起来难。记忆一旦有了噪声和矛盾，模型会一直照着错的做。

什么时候更新

偏好像是”用户喜欢中文回答”，应该覆盖而不是追加。我用的办法是给工具加一个受约束的 slot 参数，让 key 等于 slot 名，同一个 slot 再写就是覆盖。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from typing import Literal

from langchain.tools import ToolRuntime, tool


@tool
def remember(
    slot: Literal["language", "tooling", "style", "project"],
    fact: str,
    runtime: ToolRuntime[Context],
) -> str:
    """把一条以后还会用到的稳定信息写进长期记忆。

    只有用户明确表达长期要求（"以后"、"记住"、"我的习惯"），或者透露了稳定的事实时才调用。
    一次性的任务请求、临时提问、寒暄都不要调用。
    slot 是这条记忆的归属，同一个 slot 再写会覆盖旧值，所以 slot 要选准。
    """
    assert runtime.store is not None
    user_id = runtime.context.user_id
    runtime.store.put((user_id, "profile"), slot, {"slot": slot, "fact": fact})
    return f"已写入 slot={slot}"

Literal 会让模型看到的 JSON Schema 里带一个 enum，它能选的 slot 就那几个，不会随手编一个新 key 出来。

什么时候遗忘

delete(namespace, key) 是硬删除。用户注销、记忆被判定为错误、条目过期，都用它。

BaseStore.put 的签名里还有一个 ttl 参数，看起来很适合做自动过期。实测下来 InMemoryStore 没实现：

1
2
3
=== ttl：写入时给过期时间（治理用） ===
put(..., ttl=60) -> NotImplementedError: TTL is not supported by InMemoryStore. Use a store implementation that supports TTL or set ttl=None.
TTL 在 BaseStore.put 的签名里，但 InMemoryStore 没实现，要靠 PostgresStore 之类

本地开发想验证 TTL 逻辑，得换成 PostgresStore。用 InMemoryStore 时 TTL 只能靠自己的定时任务扫 updated_at 来做。

怎么把噪声挡在门外

热路径写入最大的风险是模型什么都记。我把判断写进工具描述里，然后拿六条消息试了一遍，其中两条是一次性任务和临时提问：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
=== 逐条消息，看模型调不调 remember ===
[0] 用户: 以后回答我都用中文，术语也尽量别夹英文。
      调用 remember: 是 [{'slot': 'language', 'fact': '用户要求始终用中文回答，术语也尽量不夹英文。'}]
[1] 用户: 帮我把这段 JSON 转成 CSV。
      调用 remember: 否
[2] 用户: 今天几号？
      调用 remember: 否
[3] 用户: 我最近在做一个 Hexo 博客项目。
      调用 remember: 是 [{'slot': 'project', 'fact': '用户正在做一个 Hexo 博客项目'}]
[4] 用户: 算了，以后回答可以保留英文术语，我反而看得更顺。
      调用 remember: 是 [{'slot': 'style', 'fact': '回答中保留英文术语（如 API、commit、deploy 等），不要翻译成中文'}]
[5] 用户: 再说一次：回答必须用中文，这条不要忘。
      调用 remember: 是 [{'slot': 'language', 'fact': '回答必须用中文'}]

=== 长期记忆最终内容（同一 slot 只剩最后一次写入） ===
key=language   value={'slot': 'language', 'fact': '回答必须用中文'}
key=project    value={'slot': 'project', 'fact': '用户正在做一个 Hexo 博客项目'}
key=style      value={'slot': 'style', 'fact': '回答中保留英文术语（如 API、commit、deploy 等），不要翻译成中文'}
一共 3 条

=== 对照：全量记录会变成什么样 ===
不做闸门、每条都存 -> 6 条，其中 3 条有长期价值，2 条是任务和临时提问

六条消息，模型只写了四次，任务和临时提问都放过去了。第 5 条把第 0 条的 language 覆盖掉了，条目数没涨。闸门是有效的。

但它也暴露了一个真问题。第 4 条和第 0 条是互相矛盾的，模型把第 4 条归到了 style，第 0 条留在 language。结果 store 里同时躺着”回答必须用中文”和”保留英文术语”。两条都不算错，凑在一起就自相矛盾。

靠固定 slot 只能收敛同一类里的更新，跨 slot 的矛盾它管不了。我的处理办法是写入之前先把同 namespace 下的现有记忆读出来，一起给模型看，让它决定是覆盖、新增还是删除。这一步可以在热路径的同一个工具里做，也可以丢给后台任务。官方文档里提到的 profile 与 collection 之争说的就是这个：单一 profile 文档好更新但容易丢信息，文档集合好追加但冲突要自己处理。

图里最后一步是定期审计。我一般会看两个信号：一条记忆从写入到现在有没有被检索命中过，以及同 namespace 里有没有语义高度重复的条目。前者说明它是不是噪声，后者说明去重没做好。

和 checkpointer 的分工

这两个组件经常被搞混，因为它们都要在 create_agent 里传。分清楚之后就不会用错了。

checkpointer store
存什么 agent state，主要是 messages 任意 JSON 文档
怎么定位 thread_id namespace + key
生命周期 一个 thread 结束可以留着，也可以清 跨 thread 长期存在
访问方式 自动读写，框架管 代码或工具里显式读写
换 thread 后 读不到 读得到
典型用途 多轮对话、中断恢复、时间旅行 用户画像、偏好、跨会话知识

两个可以同时挂上，互不干扰：

1
2
3
4
5
6
7
agent = create_agent(
    model=model,
    tools=[remember_preference, recall_preferences],
    checkpointer=InMemorySaver(),   # thread 内的消息
    store=store,                    # 跨 thread 的知识
    context_schema=Context,
)

我的判断标准是：这条信息在用户下次开新窗口时还有用吗？有用就进 store，没用就让它留在 thread 的 state 里。

小结

  • 短期记忆是 thread 内的 state，长期记忆是跨 thread 的 store。判断标准是换一个 thread_id 之后这条信息还要不要。
  • namespace 用 (user_id, 类别) 两级，search 的第一个参数是前缀，传 (user_id,) 就能跨类别捞。limit 默认 10 且静默截断，返回顺序不保证，这两点要在代码里显式处理。
  • 写入有工具、中间件、后台任务三条路。默认选工具那条，模型拍板的记忆用户看得见，出问题也查得到。写入前把现有记忆读出来一起判断，能挡掉大部分矛盾。
  • 语义检索靠 IndexConfig 的 embed / dims / fields。DeepSeek 没有 embedding 端点（实测 404），向量得另找来源，本地跑 BAAI/bge-small-zh-v1.5 就够用，换掉 embed 之后排序立刻从字面变成语义。没配 index 时 query 会被静默忽略，用返回项有没有 score 来判断是否真的走了向量。
  • 相似度分数只有相对意义。第一名和后面拉开 0.2 才可信，0.4 上下的名次基本是噪声，完全不相关的 query 也能拿到 0.49 分。要做阈值过滤就自己定一条线，或者在检索后面再加一层 rerank。
  • 治理三件事：同 slot 覆盖、用 delete 和 TTL 遗忘、按”以后还会不会用到”过滤写入。InMemoryStore 不支持 TTL，生产上换 PostgresStore 再说。