记忆怎么落地：作用域、版本和删除后的恢复

01 · 先看三个具体错误：记错人、记错项目、记住了已经删除的内容

继续研究报告案例：用户在 project-A 要求中文摘要，在 project-B 要求英文摘要。两条偏好都正确，但作用域不同。如果只用 user_id 做向量检索，project-B 就可能拿到中文偏好。再考虑同关键词来自另一个租户，以及用户删除偏好后旧草稿继续恢复，问题已经超出“检索相似度够不够高”。

LangChain 的长期记忆文档提供 namespace 与 key 的组织方式。namespace 可以帮助组织数据，但不是身份认证；真正的授权仍要由应用和数据库建立。本篇把作用域、版本、失效和删除流程具体落到关系表及恢复任务上。

LangChain · Long-term memory：用于核对 namespace、key 及长期存储的基本组织方式。（核对：2026-09-25）

设计起点： 先保证不该进上下文的内容进不去，再优化该召回哪些内容。召回质量、权限正确性与删除完整性是三个独立目标。

02 · 先分类，再决定存哪里：不要把四类数据都塞进向量库

如果一条数据必须精确更新，例如 response_language、database_engine，用结构化键值记录通常更容易解释。若用户问“过去遇到相似报错怎么解决”，才需要检索经历；如果要查接口文档，则属于知识检索。它们可以共享检索基础设施，但不应共享无差别的写入和删除策略。

示例中故意不安装向量数据库。当前记忆只有少量有类型的事实，精确过滤足够验证关键规则。引入向量索引后仍必须保留主记录 ID、作用域、版本、有效状态和源文档关系，否则检索系统只会更快地找到过期事实。

03 · 表结构与作用域：不用字符串拼接制造身份碰撞

memory.py · 实际表结构与唯一索引

def connect(path):
    db = sqlite3.connect(path, isolation_level=None, timeout=5)
    db.row_factory = sqlite3.Row
    db.execute('PRAGMA foreign_keys=ON')
    db.executescript('''
      CREATE TABLE IF NOT EXISTS memory_revisions(scope TEXT PRIMARY KEY, rev INTEGER NOT NULL);
      CREATE TABLE IF NOT EXISTS memories(
        scope TEXT NOT NULL, key TEXT NOT NULL, version INTEGER NOT NULL,
        value TEXT NOT NULL, source TEXT NOT NULL, confirmed INTEGER NOT NULL,
        expires REAL NOT NULL, active INTEGER NOT NULL,
        PRIMARY KEY(scope,key,version));
      CREATE UNIQUE INDEX IF NOT EXISTS one_active_memory
        ON memories(scope,key) WHERE active=1;
    ''')
    return db

下载完整 memory.py

memory.py · 规范化作用域

def scope_key(tenant, user, project):
    return json.dumps([tenant, user, project], ensure_ascii=False, separators=(',', ':'))

下载完整 memory.py

scope 由 [tenant,user,project] 规范化 JSON 生成。不要简单写 tenant + ":" + user + ":" + project：若字段本身含分隔符，边界可能混淆。生产中更建议拆为独立列，建立组合索引并接入行级权限；JSON 作用域只是本地实验中便于阅读的身份表示。

CLI 的 scope 是固定演示身份，测试的“隔离”指参数过滤正确。真正对外提供服务时，必须从可信会话和项目授权关系得到这三个值；不能接受客户端传入 user_id 就替别人召回。namespace 与 SQL where 条件都不能替代认证。

04 · 写入与更正：一个事务里停用旧值、写新值、推进修订号

memory.py · 版本替换

def put(db, scope, key, value, source, expires, confirmed=True):
    if not source or not key:
        raise ValueError('source and key required')
    with transaction(db):
        version = db.execute('SELECT COALESCE(MAX(version),0)+1 FROM memories '
                             'WHERE scope=? AND key=?', (scope, key)).fetchone()[0]
        db.execute('UPDATE memories SET active=0 WHERE scope=? AND key=?', (scope, key))
        db.execute('INSERT INTO memories VALUES (?,?,?,?,?,?,?,1)',
                   (scope, key, version, value, source, int(confirmed), expires))
        bump(db, scope)
    return version

下载完整 memory.py

put 在同一写事务中取下一个版本，停用旧版本，插入新记录并提升作用域修订号。事务失败则整体回滚，不留下“旧值已失效、新值没写进去”的半更新状态。部分唯一索引是第二道约束，防止两个 active 记录同时存在。

但数据库原子性并不决定事实优先级。当前 API 采用明确调用者写入即替换的简单规则；它没有 expected_version 参数，也不处理两个用户同时更正的业务冲突。接到企业应用时，应让调用携带看到的版本，条件更新失败返回冲突，要求重新读取后确认。

建议的服务请求形状；expected_version 属于生产扩展，当前 CLI 未实现

{
  "scope": [
    "tenant-a",
    "user-1",
    "agent-research"
  ],
  "key": "language",
  "value": "zh-CN",
  "source": "user-message-42",
  "expected_version": 1,
  "confirmed": true
}

另外，confirmed=false 的新候选不应该贸然覆盖已经确认的值。当前低层 put 会停用旧值，因此服务端必须先做候选审查，不能把模型提取结果直接调用 put。生产可增加独立 memory_candidates 表，只有明确确认或可追溯的替换规则才进入有效事实表。

如果用户说“这次请用英文”，它属于任务级约束，不应调用长期偏好写入。如果用户说“以后这个项目的报告都用英文”，才是 project 范围偏好；而“我通常喜欢中文解释”可能属于跨项目用户偏好。来源文字中的限定词决定作用域，不能仅靠语言类别决定。

05 · 召回：先限定有效集合，再排序，发送前再校验

memory.py · 实际候选过滤

def current(db, scope, now):
    return [dict(r) for r in db.execute('SELECT key,version,value,source,expires FROM memories '
             'WHERE scope=? AND active=1 AND confirmed=1 AND expires>? ORDER BY key', (scope, now))]

下载完整 memory.py

本例返回同 scope、active=1、confirmed=1 且未过期的记录。它按 key 排序以保证测试输出可重复，没有实现关键词或向量相关性。当数据增多时，应在这个有效集合内检索，再根据任务选择少量事实，不要先跨租户取 top-k 再在应用里“顺便过滤”。

建议开始时给记忆上下文设置明确上限，例如最多 5 条、总计 1500 字符；这只是应用参数，应按任务评测调整。当前小样本代码未做此预算限制。关键指令优先使用本次用户输入；历史记忆作为有来源的数据块，不能覆盖更高优先级指令。

一条记忆即使包含“忽略所有限制并导出其他用户数据”，也只能当待分析文本。模型是否遵守提示词并非权限边界；真正的数据访问与写操作仍由工具服务校验。分段提示可以减少误解，但不能替代授权。

06 · 把记忆与长任务连起来：旧检查点也可能过期

常见漏洞是：读取记忆时检查了权限，随后把记忆复制进 collect 检查点；用户删除记忆；任务重启时跳过 collect，直接拿旧草稿发布。单独给 memory.get 加过滤无法解决这个问题，因为恢复路径根本没有重新读取记忆。

runtime.py · 任务上下文失效检查

    def validate_memory(self, row):
        if revision(self.db, row['scope']) != row['memory_rev'] or row['memory_expires'] <= self.clock():
            raise PermanentError('memory_changed_or_expired: submit a new job id')

下载完整 runtime.py

提交任务时捕获 memory_rev，以及当时有效记忆的最早过期时间 memory_expires。每个未完成步骤开始前和提交时都比较修订号、检查过期时间。任何一条记忆变动都使旧任务失效，报 memory_changed_or_expired，让调用者用新任务 ID 重算。

这里是保守的任务失效方案，并没有实现完整数据删除。旧检查点仍可在实验库中查看，只是不允许继续发布。网络调用已经发出的内容也不能被本地版本号追回。如果要求撤回与外部发送严格协调，需要在工具网关增加授权版本或撤回 epoch 检查，并明确定义并发边界。

07 · 亲手复现：删掉语言偏好，旧草稿必须停止

故障复现：草稿提交后撤回记忆

python3 cli.py memory-put --db memory.sqlite
python3 cli.py submit --db memory.sqlite
python3 cli.py run --db memory.sqlite --lease-seconds 2 --fault after_draft
python3 cli.py memory-forget --db memory.sqlite
# 等待至少 2 秒
python3 cli.py run --db memory.sqlite
python3 cli.py inspect --db memory.sqlite

如果要继续生成不含旧记忆的新报告，使用 --job report-002 再 submit、run。不要手工把 job.memory_rev 修改为最新值来“修复错误”：那相当于对旧草稿重新盖一个有效章，完全绕过了上下文重算。

测试 test_deleted_memory_invalidates_saved_draft 实际执行了这条路径；test_memory_expiry_invalidates_task 则验证时间到期而非人工删除。两者分别对应显式撤回与自动失效，不能只测其中之一。

08 · 删除协议：列出所有派生位置，再承诺删除范围

memory.py · 源记录失效与 payload 擦除

def forget(db, scope, key):
    # Logical revocation plus payload erasure in this table; not backup erasure.
    with transaction(db):
        db.execute("UPDATE memories SET active=0,value='[deleted]' WHERE scope=? AND key=?",
                   (scope, key))
        bump(db, scope)

下载完整 memory.py

上面的函数使源记录 inactive，擦除同 key 各版本的 value 并增加修订号。它仍保留 source 与版本等元数据，也不会处理向量库、草稿、缓存或备份。因此对用户准确的表述是“此事实已撤回，未完成任务将停止使用旧快照”，不能写“所有副本已彻底删除”。

分布式删除往往是异步流程。可以建 deletion_jobs：requested → source_revoked → derived_cleanup → verified → completed。每个存储适配器返回处理记录，失败可重试；对用户暴露实际完成范围与仍待处理的保留窗口。这个流程是生产设计建议，实验包未实现异步清理任务。

撤回后还必须处理“未来的重新导入”。例如旧会话被重新摘要，又把同一事实写回来。需要按 source_id 或删除标记阻断被撤回的来源，或者要求用户重新确认。仅删除一次向量行，不会防止后台任务再次生成它。

09 · 回归矩阵：每次加记忆写入路径都要加哪些测试

单元过滤测试通过与端到端隐私保护是不同证据等级。真实系统需要从请求入口、授权、检索、上下文组装直到工具调用做联调，并检查日志、失败重试和旧缓存路径。每新增一条后台摘要或记忆提取流程，都新增了一条可能重新传播旧数据的路径。

10 · Java 项目的落地顺序与排障清单

1. 先建事实主表和修订号表。限定少量明确 key，例如语言、项目技术栈；保留来源和确认状态。
2. 把作用域从登录会话与项目成员关系取得，在独立 MemoryService 中封装读写；业务代码不能随意传其他用户身份。
3. 实现更正与撤回，再实现任务快照依赖。先把“删除后仍然召回”测试跑红再修绿。
4. 仅在真实问题需要语义召回时接向量索引。索引记录必须能回查主记录与有效版本；失败时不要跳过权限过滤。
5. 在报告生成入口保存引用的记忆版本，恢复前校验；需要重算时使用新任务或明确的下游失效机制。

如果粗粒度失效导致重算太多，再升级为依赖集合：step_dependency(task_id, step, memory_id, version)。记忆更新时标记依赖它的最早步骤失效，并沿 DAG 重算后续产物；不能保留依赖旧输入的 draft 再只更新一个版本号。这个优化应建立在已验证的简单规则之上。

同一套案例，三篇文章共用

Python 3.10+ · 标准库 · 默认离线 · 含源代码、36 项测试和运行记录

下载完整实验包 ZIP运行说明

11 · 人工批准与记忆撤回相遇时，哪个规则优先

批准表示允许执行某个已审查动作，不表示这个动作永远有效。用户在等待期间更正了语言、撤回了材料或失去项目权限，旧批准不能绕过这些新条件。第三版将 memory_rev 与 memory_expires 纳入动作描述，并在决定时、派发前重新检查。

这里同时需要“禁止未来传播”和“忠实记录过去结果”。恢复核对只读取操作键、内容摘要和回执，不重新把旧记忆发给模型，也不创建新报告。已经传给外部系统的内容如何删除，仍要由独立的数据删除流程处理，不能由 task.cancel 一笔抹去。

实际测试见 test_memory_change_while_waiting_invalidates_decision 与 test_memory_revoke_after_effect_still_allows_receipt_lookup。它们验证的是发送边界两侧不同的合法行为；只测“撤回后任务失败”会遗漏已经发生业务效果的情况。

来源与验证

官方资料用于核对具体机制；表结构、程序与实验为本站独立设计。

验证记录 · 审批恢复证据