记忆系统的难点不是「存」，而是「该不该存、存什么形态、什么时候忘」。这个教程用 SQLite + 向量做一个可测试的长期记忆，包含抽取、冲突合并、检索排序和用户删除。
目标
bash
python -m memory remember "我以后都希望用顺丰，别放门口"
python -m memory recall "这个用户对配送有什么偏好？"
python -m memory forget --user u1输出：
text
记住：
- shipping.carrier = 顺丰（置信度 0.92）
- shipping.dropoff = 不放门口（置信度 0.88）
召回：
[0.81] 配送偏好顺丰；不要放门口（2026-10-30，来源 conversation#18）表结构
sql
CREATE TABLE memories (
id TEXT PRIMARY KEY,
user_id TEXT NOT NULL,
kind TEXT NOT NULL, -- preference / fact / episode / procedure
key TEXT NOT NULL,
value TEXT NOT NULL,
confidence REAL NOT NULL,
importance REAL NOT NULL,
source TEXT NOT NULL,
observed_at REAL NOT NULL,
expires_at REAL,
active INTEGER NOT NULL DEFAULT 1,
embedding BLOB NOT NULL
);
CREATE INDEX idx_mem_user ON memories(user_id, active);
CREATE TABLE memory_events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
memory_id TEXT NOT NULL,
action TEXT NOT NULL, -- create / update / deactivate / delete
at REAL NOT NULL,
detail TEXT
);key 用于合并，例如 shipping.carrier；source 用于追溯；active 支持软删除；expires_at 支持 TTL。
Step 1：抽取记忆
不要让模型直接写库。先抽取成结构化、可合并的原子事实：
python
SCHEMA = {
"type": "object",
"properties": {
"items": {
"type": "array",
"items": {
"type": "object",
"properties": {
"kind": {"type": "string", "enum": ["preference", "fact", "episode"]},
"key": {"type": "string"},
"value": {"type": "string"},
"confidence": {"type": "number"},
"importance": {"type": "number"},
"sensitivity": {"type": "string", "enum": ["low", "medium", "high"]},
},
"required": ["kind", "key", "value", "confidence", "importance", "sensitivity"],
},
}
},
"required": ["items"],
}
PROMPT = """从对话中抽取值得长期记住的信息。
规则：
1. 只抽用户明确表达的偏好、事实或经历，不要推测。
2. 每条只包含一个原子事实；key 用点号命名，如 shipping.carrier。
3. 寒暄、一次性上下文、敏感信息（密码、完整证件号）不要抽。
4. 置信度低于 0.6 的不要输出。
5. 最多 5 条。"""
def extract(text: str) -> list[dict]:
data = llm.complete_json(system=PROMPT, user=text, schema=SCHEMA)
return [x for x in data["items"] if x["confidence"] >= 0.6]Step 2：写入与冲突合并
python
import hashlib
import time
import uuid
import numpy as np
from .db import conn, add_event
from .embed import embed
def remember(user_id: str, text: str) -> list[dict]:
items = extract(text)
saved = []
for item in items:
if item["sensitivity"] == "high":
continue # 高敏感信息不进入长期记忆，或单独加密存储
existing = find_by_key(user_id, item["key"])
if existing and existing["value"] == item["value"]:
touch(existing["id"])
saved.append({"key": item["key"], "value": item["value"], "duplicate": True})
continue
if existing:
deactivate(existing["id"], reason="superseded")
mem_id = str(uuid.uuid4())
vec = embed(f"{item['key']}: {item['value']}").astype("float32")
with conn() as c:
c.execute(
"""INSERT INTO memories
(id,user_id,kind,key,value,confidence,importance,source,observed_at,active,embedding)
VALUES (?,?,?,?,?,?,?,?,?,?,?)""",
(
mem_id, user_id, item["kind"], item["key"], item["value"],
item["confidence"], item["importance"], "conversation", time.time(), 1,
vec.tobytes(),
),
)
add_event(mem_id, "create", {"key": item["key"], "value": item["value"]})
saved.append({"key": item["key"], "value": item["value"], "duplicate": False})
return saved冲突处理原则：同一 key 保留最新有效值，旧值标记 inactive，但事件日志保留，便于审计和回滚。
Step 3：检索
python
def recall(user_id: str, query: str, limit: int = 5, min_score: float = 0.2) -> list[dict]:
qv = embed(query).astype("float32")
with conn() as c:
rows = c.execute(
"SELECT * FROM memories WHERE user_id=? AND active=1", (user_id,)
).fetchall()
scored = []
now = time.time()
for row in rows:
vec = np.frombuffer(row["embedding"], dtype="float32")
cosine = float(np.dot(qv, vec) / (np.linalg.norm(qv) * np.linalg.norm(vec) + 1e-8))
age_days = (now - row["observed_at"]) / 86400
recency = 0.5 ** (age_days / 30) # 30 天半衰
score = 0.55 * cosine + 0.20 * recency + 0.15 * row["confidence"] + 0.10 * row["importance"]
if score >= min_score:
scored.append((score, dict(row)))
scored.sort(key=lambda x: x[0], reverse=True)
return [{**row, "score": round(score, 3)} for score, row in scored[:limit]]只按向量相似度会召回「语义相近但过时」的记忆。加上时效、置信度和重要性，能显著改善。
Step 4：遗忘
python
def forget_expired() -> int:
now = time.time()
with conn() as c:
cur = c.execute(
"UPDATE memories SET active=0 WHERE active=1 AND expires_at IS NOT NULL AND expires_at < ?",
(now,),
)
return cur.rowcount
def forget_user(user_id: str) -> int:
with conn() as c:
cur = c.execute("DELETE FROM memories WHERE user_id=?", (user_id,))
return cur.rowcount生产还要：
- 每个用户容量上限，超出按分数淘汰；
- 用户可导出/删除全部记忆；
- 高敏感记忆单独加密、更短 TTL；
- 删除要级联到向量索引与缓存。
Step 5：测试
tests/test_memory.py：
python
def test_remembers_and_recalls():
remember("u1", "我以后都希望用顺丰，别放门口")
hits = recall("u1", "配送偏好")
assert hits
assert any("顺丰" in h["value"] for h in hits)
def test_new_value_supersedes_old():
remember("u1", "我以后都用顺丰")
remember("u1", "算了，以后用京东")
hits = recall("u1", "快递")
active = [h for h in hits if h["key"] == "shipping.carrier"]
assert active[0]["value"] == "京东"
def test_forget_user_removes_everything():
remember("u2", "我喜欢英文回复")
forget_user("u2")
assert recall("u2", "语言偏好") == []再构造一个 20 条记忆的评估集：10 条应召回、10 条噪声。测 Recall@3、错误召回率、跨用户污染（必须为 0）。
Step 6：隐私与作用域
- 检索前必须按
user_id/tenant_id过滤； - 不把其他用户记忆放进上下文；
- 日志与 trace 对记忆内容脱敏；
- 高敏感信息不写入，或单独加密；
- 支持「忘记我」并验证删除后不可检索。
记忆没有作用域是最危险的低级错误
一个 user_id 过滤漏掉，就可能把 A 的地址、偏好、订单信息答给 B。过滤必须在检索之前，而不是生成之后。
常见失败
把所有对话灌进向量库
寒暄、重复、过时信息会淹没真正有用的记忆。先抽取原子事实。
只增不删
记忆越来越慢、越来越吵。TTL、容量上限、用户删除缺一不可。
用模型判断「重要」但不校验
模型会把寒暄记成偏好。写入前用置信度阈值与规则。
旧值覆盖新值但没记录
无法审计。保留事件日志，旧值标记失效而不是物理删除。
验收清单
- 记忆有类型、key、来源、时间、置信度
- 写入前去重、冲突合并、脱敏
- 检索有多维排序与 scope 过滤
- 有 TTL、容量上限、用户删除
- 跨用户污染为 0
- 有召回率与错误召回测试
- 记忆可追溯、可审计
扩展任务
- 把向量检索换成 pgvector/Qdrant；
- 加情景记忆摘要，按事件聚合；
- 加「记忆使用次数」，影响排序；
- 加用户确认：「要记住这个偏好吗？」
- 做一次记忆投毒实验，验证写入审查。
小结
- 记忆不是聊天记录备份，而是提炼后的状态。
- 写入有门槛，检索要多维排序，遗忘要可执行。
- 作用域隔离与删除能力是底线。
- 用测试集验证「该记住的记住，不该用的没用」。
下一篇把整个系统部署到 Cloudflare。