跳到主要内容

从零到一:个人 AI 助理记忆系统

七天实战

边学边练请访问 7 天学会 Memory Engineering:首页记忆分层可视化、/learn 七天课表、/lab 五个本地沙盒、社区与资源矩阵。

目标读者

已读 入门介绍快速开始,准备用约一周时间交付可演示的助理记忆后端。对齐子站 Day7 大作业的简化可执行版。

你将交付什么

一个单用户优先、可扩展到多租户的个人助理记忆系统,具备:

  1. 会话 STM + Token 预算拼装
  2. KV 偏好与事实
  3. 向量情景检索(可先 Chroma)
  4. 可选:兴趣图谱多跳
  5. 每日巩固与可信度门槛
  6. 基础可观测与 forget API
  7. 部署检查清单

不在首周范围:多模态大一统、跨公司联邦、亿级向量运维。

总架构

目录建议:

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交付物验收
1Day1需求清单 + File/SQLite KV重启仍记得姓名与语气
2Day2pack() + thread 历史超预算触发摘要
3Day3向量写入/检索 + user 过滤偏好相关问能召回
4Day4兴趣边表或 Neo4j多跳建议非空
5Day5巩固任务 + 置信度低分洞察不进 LTM
6Day6(可选)第二 Agent 只读团队备忘无写权失败
7Day7指标、forget、部署清单检查表全绿

Step 1 — 需求与 KV

写出一页纸:

信息类型保留存储
姓名semantic永久KV
语气偏好affective永久直到修改KV
过敏/禁忌affective永久KV + pinned
昨日任务结论episodic30 天向量
工具失败教训procedural90 天向量/文档

实现 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 消息表。每轮:

  1. 读 KV 锚点进 System 或 Retrieved
  2. 拉最近 N 轮历史
  3. pack(system, history, retrieved, budget)
  4. 调用 LLM
  5. 门禁后写 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 / 夜间任务:

  1. 拉取当日 episodic
  2. 生成 L1 事件摘要、L2 信念
  3. score 低于 0.6 进 quarantine 表
  4. 高分写向量与(可选)更新画像 KV

保留 source_ids 列,方便 FAQ 里说的归因。

Step 6 — 多 Agent 预留

即使只有一个聊天 Agent,也先划分命名空间:

  • ns=user:{id}:private
  • ns=user:{id}:diary
  • ns=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、团队协议与红队。不要并行开六条大前线。

学习反馈闭环

  1. 每天对照 learn 课表 勾选。
  2. /lab 验证直觉。
  3. 卡点查 FAQ
  4. 选型犹豫读 GitHub 项目
  5. 生产加固对照 最佳实践 的上线检查表与 Oncall 提纲。

写周报时用三句话:本周记住了什么能力、测过哪条黄金用例、下周只加一件事——防止范围回潮。

一周日程示例(可照抄)

上午下午晚上验收
D1画五层图、写需求表实现 KV put/get/forget重启仍记得姓名
D2消息表 + pack打印三区 token超长史触发摘要
D3Chroma 接入用户过滤检索过敏问能召回
D4边表或 Neo4j多跳查询返回至少 1 条建议
D5巩固脚本置信度门槛低质不进 LTM
D6第二 Agent 只读权限否定用例无写权报错
D7指标 + compose跑 golden 集检查表勾完

若某天卡住,优先保主路径验收,把图谱或多 Agent 挪到第二周,而不是六线并行都半成品。

风险与缓解

风险早期信号缓解
范围膨胀想同时上 Redis+KG+多模态砍到 KV+向量+pack
模型账单每轮全量巩固夜间批处理、限额
数据损坏直接改生产 db每日快照、先 staging
无法演示只有代码无剧本准备 3 分钟话术+黄金对话
隐私惊吓日志打印全文脱敏中间件先上

演示剧本(3 分钟)

  1. 「我叫演示用户,对花生过敏,请用简短中文。」→ 确认 KV。
  2. 新开 thread:「晚饭吃啥?」→ 应避开花生并偏短句。
  3. 灌水 30 轮后仍能回答当前问题 → 展示摘要/预算。
  4. 「忘掉过敏」→ forget → 再问晚饭,不再提过敏约束。
  5. 打开日志指出 memory_ids → 证明可归因。

完成定义(Done)

当你能对非技术朋友演示:「我说过过敏 → 新会话仍避开;我说忘掉 → 真的忘掉;长聊天不会报错截断;你能指出回答依据哪条记忆」——个人助理记忆系统的从零到一即告成立。之后的工作是评测密度与生产硬度,而不是再换一个时髦框架从头来过。