长期记忆写入、检索与更新

学习目标

把 LongTermMemory 看成有范围、有来源、有版本的事实库，而不是“把聊天存起来”。
设计候选写入、检索过滤、更新替换和删除撤回的完整数据流。
读源码时区分 MemoryStore、检索器、ContextBuilder 和用户授权。

前置知识

已阅读短期记忆与对话历史和 Context 构建、选择与预算。
已理解 Session scope、State ownership、ToolResult、权限和版本冲突。

一条记忆的最小契约

内容与来源

长期记忆至少要有 memory_id、key/value、scope、source_event、created_by、confidence、version、created_at 和 retention。source_event 说明事实从哪条用户消息、工具结果或人工确认产生；没有来源的句子只能是候选，不能冒充可信 Memory。

白话说，Memory 是“带标签和出处的卡片”，不是把一段聊天贴进数据库。未来检索者要知道卡片属于谁、为什么可信、何时过期。

写入

写入流程通常是候选提取 → 意图/敏感性判断 → scope 确定 → provenance 验证 → 去重/冲突检查 → 用户或策略授权 → 版本化存储。短期对话不应默认自动写入跨 Session 的用户或项目 Memory。

检索

检索先按 tenant/user/project scope 和权限过滤，再按查询相关性、新鲜度、confidence 和版本排序。只有通过过滤的结果才能进入 Context；检索器不应绕过 ContextBuilder 直接把 Store 全量交给 Model。

更新与删除

创建只允许在 memory_id 尚不存在时从版本 0 追加；已有记录的更新必须显式携带 expected_version，不能盲写 current。删除用带 expected_version 的新版本或 tombstone，保留必要的审计证据；“把 value 改为空字符串”不是可靠删除语义。

记忆数据流

Memory 的写入和读取是两条不同方向的数据流：写入强调来源、授权和版本，读取强调 scope、相关性和预算。两者都经过 MemoryStore，但责任不能合并成一个无条件的 save/load。

阅读提示：scope 过滤发生在 Rank 之前，避免先对不该读取的数据做相关性处理；更新通过新版本或 tombstone 表达，不能静默覆盖旧事实。

作用域和保留

常见 MemoryScope

session：仅当前会话有效。
user：同一用户的多个 Session 可用。
project：项目成员按权限共享。
tenant：租户内共享，通常要求更严格的写入策略。

scope 越宽，写入和读取门槛越高。用户在一个项目中的临时决定不应自动变成 tenant 级事实。

置信度和新鲜度

confidence 表示来源可信程度，不是模型“感觉很确定”的替代品；freshness 用时间或有效期约束旧事实。检索时可以展示“可能过期”而不是静默当成真值。

记忆进入 Context

ContextBuilder 应携带 memory_id、scope、source 和 version 的安全摘要，方便模型和审计知道事实来自哪里；敏感原文只在授权范围内展开。

本地模拟：写入、检索、更新和删除

用途：下面的标准库示例使用 append-only 版本记录和 current pointer 实现 scope 过滤、严格 create/update、更新和 tombstone 删除；记录按 (memory_id, version) 索引，保留每次 provenance，没有向模型或外部服务发送内容。

python

from __future__ import annotations

from dataclasses import dataclass


@dataclass
class MemoryChange:
    memory_id: str
    key: str
    value: str
    scope: str
    provenance: str
    deleted: bool = False


@dataclass
class MemoryItem:
    memory_id: str
    key: str
    value: str
    scope: str
    version: int
    provenance: str
    deleted: bool = False


class MemoryStore:
    def __init__(self) -> None:
        self.records: dict[tuple[str, int], MemoryItem] = {}
        self.current: dict[str, int] = {}

    def _current(self, memory_id: str) -> MemoryItem | None:
        version = self.current.get(memory_id)
        if version is None:
            return None
        return self.records[(memory_id, version)]

    def _append(
        self,
        change: MemoryChange,
        expected_version: int,
    ) -> MemoryItem:
        actual = self.current.get(change.memory_id, 0)
        if expected_version != actual:
            raise RuntimeError("memory version conflict")
        record = MemoryItem(
            change.memory_id,
            change.key,
            change.value,
            change.scope,
            actual + 1,
            change.provenance,
            change.deleted,
        )
        self.records[(record.memory_id, record.version)] = record
        self.current[record.memory_id] = record.version
        return record

    def create(self, change: MemoryChange) -> MemoryItem:
        if change.memory_id in self.current:
            raise FileExistsError("memory already exists")
        return self._append(change, expected_version=0)

    def update(
        self,
        change: MemoryChange,
        *,
        expected_version: int,
    ) -> MemoryItem:
        if change.memory_id not in self.current:
            raise KeyError("memory does not exist")
        return self._append(change, expected_version=expected_version)

    def history(self, memory_id: str) -> list[MemoryItem]:
        return [
            record
            for (record_id, _), record in sorted(self.records.items())
            if record_id == memory_id
        ]

    def retrieve(self, scope: str, query: str) -> list[MemoryItem]:
        return [
            item
            for memory_id, version in self.current.items()
            for item in [self.records[(memory_id, version)]]
            if not item.deleted
            and item.scope == scope
            and query.lower() in (item.key + " " + item.value).lower()
        ]

    def tombstone(self, memory_id: str, expected_version: int) -> MemoryItem:
        current = self._current(memory_id)
        if current is None:
            raise KeyError("memory does not exist")
        if current.version != expected_version:
            raise RuntimeError("memory version conflict")
        return self.update(
            MemoryChange(
                current.memory_id,
                current.key,
                current.value,
                current.scope,
                "delete-request",
                deleted=True,
            ),
            expected_version=expected_version,
        )


store = MemoryStore()
item = store.create(
    MemoryChange("m-1", "language", "中文", "user:alice", "user-confirmed")
)
print(item.version, [entry.value for entry in store.retrieve("user:alice", "language")], store.current)
updated = store.update(
    MemoryChange("m-1", "language", "English", "user:alice", "user-correction"),
    expected_version=item.version,
)
print([entry.value for entry in store.retrieve("user:alice", "language")])
print([(entry.version, entry.provenance) for entry in store.history("m-1")])
print(store.retrieve("user:bob", "language"))
store.tombstone("m-1", expected_version=2)
print(store.current, store.retrieve("user:alice", "language"))
# 输出：1 ['中文'] {'m-1': 1}
# 输出：['English']
# 输出：[(1, 'user-confirmed'), (2, 'user-correction')]
# 输出：[]
# 输出：{'m-1': 3} []

输出体现五个边界：create 从版本0追加且 current 指针前进；已有记录只能用 expected_version update；纠正保留旧版本和 provenance；Bob 的 scope 查不到 Alice；tombstone 生成版本3并让未来检索为空。真实系统还要保留审计、删除证明和并发存储事务。

源码阅读心智模型

smolagents

轻量 Agent 往往没有内建长期 Memory；若应用在工具或循环外维护向量/键值 Store，要把写入触发、scope 和 Context 注入位置单独画出。不要把 messages 列表误称为跨会话 Memory。

OpenAI Agents SDK

区分 Session history 与应用自己的 Memory/检索服务；RunContextWrapper.context 只提供本地依赖，不能被误读为长期 Memory。阅读时关注记忆候选何时产生、由谁授权、检索结果怎样进入 Runner 的 Context，以及版本删除如何传播。

LangGraph

图 State/checkpoint 适合恢复当前执行，长期 Memory 通常通过 Store namespace 或独立节点能力提供。检查 thread_id/checkpoint_id 与 Store namespace 是否分开，以及节点更新是否把临时 State 错写到跨线程的 Memory。

DeepSeek Harness

将长期 Memory 看作 ctx.sessions/Session 之外的持久化插件或服务；Session 的 append-only Event log 由 ctx.sessionPersistence 的 create/open/flush/close 持久化，ctx.agents/agent-loop 中的 Driver 负责把已授权结果投影给模型，Plugin/Hook 负责候选或事件但不能绕过 scope。源码阅读重点是写入审计、更新版本和删除传播，不能把 SessionEvent 当作长期 Memory API。

易混点

长期 Memory 不是完整聊天记录：它是有筛选、来源和范围的可复用事实。
检索命中不等于可信：仍需 scope、来源、版本、新鲜度和 Context 预算。
更新不是覆盖字符串：版本和 supersedes/tombstone 才能处理纠正与并发。
用户 scope 不等于 tenant scope：扩大可见范围必须有明确授权。
删除不是空值：删除要阻止未来检索，并按策略保留审计证据。

课后小问

为什么 Memory Candidate 不能直接写入 MemoryStore？
答案：候选可能没有明确 scope、来源、授权或冲突处理。
解析：自动提取的事实可能是模型猜测、临时指令或恶意文本。写入前的策略检查决定它能否跨 Session 复用。
检索为什么先过滤 scope 再排序？
答案：不应对无权读取的数据进行相关性处理或暴露统计信号。
解析：排序、缓存和日志都可能泄露内容；先隔离命名空间再做相关性计算是责任边界。
纠正一条偏好时为什么推荐新版本而不是删除旧记录再写？
答案：新版本可以保留来源、顺序和审计关系，避免删除与重写之间出现空窗。
解析：并发读取可以看到一致版本，恢复和审计也能解释何时、谁把中文改成 English。

本节小结

LongTermMemory 是有 memory_id、scope、provenance、confidence、version 和 retention 的持久事实。
写入走候选检查、授权、去重和版本化；读取先 scope/permission，再相关性、新鲜度和预算。
纠正用新版本或 supersedes，删除用 tombstone；空字符串不等于删除。
Session/Run State 的恢复存储和长期 Memory 的可复用存储应分离，进入 Context 还要经过最后一层投影。

快速回顾

能说出一条长期 Memory 的最小字段。
能画出候选写入和查询召回的两条路径。
能解释为何 scope、version、source 不是可选装饰。
下一篇阅读记忆污染、并发冲突与数据隔离，处理错误写入、竞态和跨租户泄露。

长期记忆写入、检索与更新 ​

学习目标 ​

前置知识 ​

一条记忆的最小契约 ​

内容与来源 ​

写入 ​

检索 ​

更新与删除 ​

记忆数据流 ​

作用域和保留 ​

常见 MemoryScope ​

置信度和新鲜度 ​

记忆进入 Context ​

本地模拟：写入、检索、更新和删除 ​

源码阅读心智模型 ​

smolagents ​

OpenAI Agents SDK ​

LangGraph ​

DeepSeek Harness ​

易混点 ​

课后小问 ​

本节小结 ​

快速回顾 ​