从零到一:个人 AI 助理记忆系统
七天实战
边学边练请访问 7 天学会 Memory Engineering:首页记忆分层可视化、/learn 七天课表、/lab 五个本地沙盒、社区与资源矩阵。
你将交付什么
一个单用户优先、可扩展到多租户的个人助理记忆系统,具备:
- 会话 STM + Token 预算拼装
- KV 偏好与事实
- 向量情景检索(可先 Chroma)
- 可选:兴趣图谱多跳
- 每日巩固与可信度门槛
- 基础可观测与 forget API
- 部署检查清单
不在首周范围:多模态大一统、跨公司联邦、亿级向量运维。
总架构
目录建议:
personal-memory-assistant/
app/
main.py # FastAPI 或 CLI 入口
pack.py # Token 预算
write_path.py
read_path.py
consolidate.py
storage/
kv_sqlite.py
chroma_store.py
graph_stub.py
eval/
golden.yaml
run_eval.py
deploy/
docker-compose.yml
checklist.md
tests/
七步路线(映射七天)
| 步骤 | 对应 Day | 交付物 | 验收 |
|---|---|---|---|
| 1 | Day1 | 需求清单 + File/SQLite KV | 重启仍记得姓名与语气 |
| 2 | Day2 | pack() + thread 历史 | 超预算触发摘要 |
| 3 | Day3 | 向量写入/检索 + user 过滤 | 偏好相关问能召回 |
| 4 | Day4 | 兴趣边表或 Neo4j | 多跳建议非空 |
| 5 | Day5 | 巩固任务 + 置信度 | 低分洞察不进 LTM |
| 6 | Day6 | (可选)第二 Agent 只读团队备忘 | 无写权失败 |
| 7 | Day7 | 指标、forget、部署清单 | 检查表全绿 |
Step 1 — 需求与 KV
写出一页纸:
| 信息 | 类型 | 保留 | 存储 |
|---|---|---|---|
| 姓名 | semantic | 永久 | KV |
| 语气偏好 | affective | 永久直到修改 | KV |
| 过敏/禁忌 | affective | 永久 | KV + pinned |
| 昨日任务结论 | episodic | 30 天 | 向量 |
| 工具失败教训 | procedural | 90 天 | 向量/文档 |
实现 SQLite:
# storage/kv_sqlite.py
import sqlite3
from typing import Optional
class KVMemory:
def __init__(self, path: str = "mem.db") -> None:
self.conn = sqlite3.connect(path)
self.conn.execute(
"""CREATE TABLE IF NOT EXISTS kv (
user_id TEXT, key TEXT, value TEXT, kind TEXT,
updated_at REAL, PRIMARY KEY(user_id, key)
)"""
)
def put(self, user_id: str, key: str, value: str, kind: str) -> None:
self.conn.execute(
"INSERT OR REPLACE INTO kv VALUES (?,?,?,?,strftime('%s','now'))",
(user_id, key, value, kind),
)
self.conn.commit()
def get(self, user_id: str, key: str) -> Optional[str]:
cur = self.conn.execute(
"SELECT value FROM kv WHERE user_id=? AND key=?",
(user_id, key),
)
row = cur.fetchone()
return row[0] if row else None
Step 2 — 会话与 pack()
复用快速开始的预算逻辑;增加 thread_id 消息表。每轮:
- 读 KV 锚点进 System 或 Retrieved
- 拉最近 N 轮历史
pack(system, history, retrieved, budget)- 调用 LLM
- 门禁后写 KV / 向量队列
Step 3 — 向量情景记忆
最小字段:id, user_id, text, kind, created_at, importance。写入会话结束摘要;检索时 where={"user_id": user_id}。
def search_episodes(store, user_id: str, query: str, k: int = 5) -> list[str]:
# 伪代码:chroma collection.query
hits = store.query(query_texts=[query], n_results=k, where={"user_id": user_id})
return hits["documents"][0]
先做纯向量;有余力加 BM25 + RRF(开发指南)。
Step 4 — 结构化兴趣图
若暂无 Neo4j,用 SQLite 边表:
CREATE TABLE edges (
src TEXT, rel TEXT, dst TEXT,
user_id TEXT, since REAL, valid_to REAL, confidence REAL
);
查询「同领域其它兴趣」可用两跳 SQL 或加载 NetworkX。有 Docker 再换 Cypher。
Step 5 — 每日巩固
Cron / 夜间任务:
- 拉取当日 episodic
- 生成 L1 事件摘要、L2 信念
score低于 0.6 进 quarantine 表- 高分写向量与(可选)更新画像 KV
保留 source_ids 列,方便 FAQ 里说的归因。
Step 6 — 多 Agent 预留
即使只有一个聊天 Agent,也先划分命名空间:
ns=user:{id}:privatens=user:{id}:diaryns=team:demo:readonly
第二个「日程 Agent」只读 diary 摘要,验证权限中间件。完整投票协议见最佳实践。
Step 7 — 生产瘦身
| 项 | 做法 |
|---|---|
| 指标 | 写入接受率、检索命中、P95、forget 延迟 |
| 日志 | trace_id + memory_ids |
| 备份 | 每日 SQLite/Chroma 快照 |
| 密钥 | API Key 仅环境变量 |
| 限流 | 每用户写入 QPS |
代码骨架:主循环
def handle_turn(user_id: str, thread_id: str, text: str, budget: int = 8192) -> str:
prefs = load_prefs(user_id) # KV
retrieved = search_episodes(user_id, text) + prefs_as_lines(prefs)
history = load_history(thread_id)
system = build_system(prefs)
messages = pack(system, history, retrieved, budget=budget)
reply = call_llm(messages)
if should_remember(text):
write_path(user_id, text, reply)
append_history(thread_id, text, reply)
return reply
测试计划
| 用例 | 步骤 | 期望 |
|---|---|---|
| P1 持久偏好 | 设定语气 → 重启 → 再聊 | 语气仍遵守 |
| P2 预算 | 灌 50 轮闲聊 | 触发摘要且不超窗 |
| P3 隔离 | 用户 A 记忆不出现在 B | 检索过滤生效 |
| P4 召回 | 存「过敏花生」→ 问晚餐 | 回复规避花生 |
| P5 遗忘 | forget 过敏 → 再问 | 不再提及该约束 |
| P6 巩固 | 造 10 条低质闲聊 | 不进高置信 LTM |
| P7 矛盾 | 先素食后点牛排 | 触发澄清或降权 |
eval/golden.yaml 示例:
- id: allergy_peanut
user_id: u_demo
setup: ["请记住我对花生过敏"]
ask: "今晚吃什么好?"
expect_memory_kinds: [affective]
forbid_substrings: ["花生酱", "宫保鸡丁"]
部署检查清单
-
.env不入库;示例用.env.example - SQLite/Chroma 卷挂载与备份
- 健康检查:
/health与检索探活 - 资源:嵌入与聊天超时、重试上限
- 日志脱敏
- forget API 文档化给用户
- 评测集在 CI 可跑(可先手动)
- 回滚:保留上一版数据目录
docker-compose 最小思路:api + chroma(或内嵌)+ 可选 neo4j;个人演示可单容器。
与子站大作业的差距
子站 Day7 列出多模态、团队记忆、多设备同步等。本教程先拿到可演示主路径;每完成一项主路径验收,再按最佳实践加 L2 Redis、团队协议与红队。不要并行开六条大前线。
学习反馈闭环
写周报时用三句话:本周记住了什么能力、测过哪条黄金用例、下周只加一件事——防止范围回潮。