在 AI 对话系统中实现自主记忆管理
目录
1. 引言
你有没有遇到过这样的场景——用户跟 AI 助手聊了三天，第四天打开对话，AI 完全不记得之前说过什么。用户说”我上次提到的那个项目”，AI 只能回一句”请问您指的是哪个项目？”。
这就是无状态对话的核心痛点：每次对话都是一张白纸。
市面上有不少记忆框架（比如 mem0）试图解决这个问题，但在实际落地时我们发现几个绕不开的限制：
- 黑盒提取逻辑：无法精确控制”记住什么、忘记什么”
- 存储耦合：通常绑定特定向量数据库，难以与已有基础设施整合
- 生命周期缺失：记忆只增不减，随着对话积累会变成噪声源
所以我们选择自己实现一套完整的记忆管理系统，核心闭环只有五步：
提取 → 向量化存储 → 召回 → 注入 → 清理
本文将以一个实际的 FastAPI + PostgreSQL 项目为背景，完整拆解这套方案的架构与实现，记忆提取/更新/删除，向量化存储，召回，记忆应用，过期记忆清理全流程都可以在本文中进行参考。
提示一下，本文中虽然是使用 DIFY 工作流提取记忆，使用 RAGFLOW 作为对话服务，实现方案实际上是通用的，对这些工具做了弱化处理。
2. 整体架构
全景流程图
用户发送消息
│
▼
┌──────────────────────────────────────────────────────────┐
│ 对话引擎 │
│ │
│ ① 向量检索记忆 ──→ ② 格式化记忆文本 ──→ ③ 注入提示词 │
│ │
└──────────────────────────┬───────────────────────────────┘
│
▼
LLM 生成回复
│
▼
用户收到回答
│
┌──────────────────┴──────────────────┐
│ 后台定时任务（每5分钟） │
│ │
│ ④ 扫描新对话 │
│ │ │
│ ▼ │
│ ⑤ LLM 提取记忆（add/update/delete） │
│ │ │
│ ▼ │
│ ⑥ 写入 PostgreSQL + pgvector │
└─────────────────────────────────────┘
│
┌──────────────────┴──────────────────┐
│ 后台定时任务（每天凌晨） │
│ │
│ ⑦ 清理过期的知识类记忆 │
└─────────────────────────────────────┘
核心组件
| 组件 | 职责 |
|---|---|
| PostgreSQL + pgvector | 存储记忆文本和向量，支持余弦距离检索 |
| Embedding 模型 | 将文本转为向量（bge-m3，1024 维） |
| LLM 工作流 | 分析对话历史，决定 add/update/delete 操作 |
| 定时任务 | 增量提取记忆、清理过期记忆 |
两类记忆
我们将用户记忆分为两类，它们在召回策略和生命周期上完全不同：
| 类别 | 说明 | 召回策略 | 生命周期 |
|---|---|---|---|
| persona（用户画像） | 用户的身份、偏好、习惯等 | 全量召回 | 永久保留 |
| knowledge（专业知识） | 对话中涉及的领域知识、事实 | 向量相似度检索 | 定时过期清理 |
为什么这样分？用户画像数量少且始终相关（”用户是化工行业的技术总监”），全量加载不会产生噪声。知识类记忆数量多且时效性强，必须按相关性检索，并定期淘汰过时信息。
3. 数据模型
Memory 表结构
CREATE TABLE memory (
id VARCHAR(36) PRIMARY KEY, -- UUID
user_id INTEGER NOT NULL, -- 用户ID
fact VARCHAR(2000) NOT NULL, -- 记忆内容
embedding VECTOR(1024), -- pgvector 向量字段
category VARCHAR(20) NOT NULL, -- persona | knowledge
created_at TIMESTAMP,
updated_at TIMESTAMP
);
CREATE INDEX ix_memory_user_id ON memory (user_id);
SQLModel + pgvector 的字段定义技巧
在 SQLModel（SQLAlchemy 2.0+ 的上层封装）中定义 pgvector 向量字段时，不能直接使用 Python 类型注解，需要通过 sa_column 绕过类型推断：
from pgvector.sqlalchemy import Vector
from sqlalchemy import Column
from sqlmodel import Field
class Memory(BaseModel, TimestampMixin, table=True):
id: str = Field(
default_factory=lambda: str(uuid.uuid4()),
max_length=36,
primary_key=True,
)
user_id: int = Field(index=True, nullable=False)
fact: str = Field(max_length=2000, nullable=False)
# 关键：用 sa_column 绕过 SQLModel 的类型推断
embedding: Optional[Any] = Field(
default=None,
sa_column=Column(Vector(1024), nullable=True),
)
category: str = Field(max_length=20, nullable=False)
这里有两个细节：
- 类型声明为
Optional[Any]：SQLModel 会尝试将 Python 类型映射到 SQL 类型，但它不认识Vector。声明为Any可以跳过这个推断，实际的列类型由sa_column决定。 -
sa_column=Column(Vector(1024))：直接使用 SQLAlchemy 的Column来定义，pgvector 提供的Vector(1024)指定了向量维度。
会话表扩展
在会话表中添加两个字段来支持增量记忆提取：
class Conversation(table=True):
# ... 其他字段 ...
# 上次记忆提取时最后一个用户消息的 ID
memory_extracted_message_id: str | None = Field(
None, max_length=64, nullable=True,
)
# 上次记忆提取的时间戳
memory_extracted_at: datetime | None = Field(
None, nullable=True,
)
memory_extracted_message_id：标记上次提取到哪条消息，下次只处理它之后的新消息memory_extracted_at：与last_conversation_at对比，判断是否有新对话需要提取
4. 记忆提取
记忆提取是整个系统最复杂的环节。我们通过定时任务异步完成，不影响对话的实时响应。
9. 总结
回顾整个方案，核心设计决策有三个：
- 两类记忆、两种召回：persona 全量 + knowledge 向量检索，在全面性和精确性之间找到平衡
- LLM 驱动的增量提取：将”记什么、改什么、删什么”的决策交给 LLM，同时通过输入现有记忆实现去重和冲突消解
- 写入即向量化 + 定时清理：保证向量与文本的一致性，同时通过生命周期管理控制记忆噪声
整套方案的依赖非常轻量——PostgreSQL（pgvector 扩展）、一个 Embedding 模型、一个 LLM 工作流，再加上已有的 Celery 定时任务基础设施。不需要额外的向量数据库、不需要第三方记忆框架，所有数据都在自己的 PostgreSQL 里，可控、可查、可调。
对于中小规模的 AI 对话系统（用户量在万级以内），这套方案完全够用。如果规模再大，可以考虑在 pgvector 上加 HNSW 索引加速检索，或者将 Embedding 调用改为批量异步以提高吞吐。
最后，记忆管理的本质不是”存更多”，而是”存对的、忘该忘的”。希望本文对你有所启发。
