1. 从“AI 失忆”说起:为什么记忆是智能的最短木板
做过 NLP、跑过对话系统、搭过智能客服的朋友,大概率都遇到过同一个尴尬场景:模型上一轮还能准确回答“我叫小明,今年 28 岁”,下一轮换个句式问“我多大了”,它就一脸茫然。这不是模型笨,而是它压根没有“记忆”这个概念。传统的无状态接口设计,每次对话都是“初见”,模型的所有知识都来自训练时的权重,无法感知上下文,更别提跨会话的长期信息沉淀。
“ai-memory”这个项目标题,短短一个连字符,其实指向的是当下 AI 应用落地时最痛的那一环——记忆系统。我在实际项目里测过很多方案,从最简单的拼接历史消息到向量检索 + 知识库,再到专门的记忆管理层,踩坑无数。这个标题背后,本质上是在问一个问题:怎么让大模型像一个靠谱的同事,而不是一个每次都要重新自我介绍的新人?
这个项目解决的,正是“会话记忆、用户画像沉淀、跨会话知识复用”这三层问题。适合谁看?如果你正在搭聊天机器人、私人助理、Agent 工作流,或者任何需要“记住用户”的 AI 产品,这篇内容能帮你省掉几周的试错时间。我会从设计思路、存储选型、核心实现、坑点排查四个维度,把 ai-memory 这类项目拆开揉碎讲清楚。
2. 整体设计思路:记忆不是数据库,而是分层的信息管道
2.1 三种记忆的边界:短期、长期、事实型
很多人一上来就想把所有对话历史塞进 context,这是最典型的坑。Context 窗口再大,也扛不住无限增长,而且塞进去的噪声会让模型注意力分散。我自己的经验是,记忆系统必须先分层,最实用的划分方式如下:
- 短期记忆(会话内):当前轮次对话的上下文,用于保持话题连贯,通常就是最近 N 轮原始消息。
- 长期记忆(跨会话):用户在不同时间点表达过的偏好、习惯、身份信息,需要结构化存储并定时召回。
- 事实型记忆(知识库):产品自身的领域知识、FAQ、业务规则,属于静态或半静态数据,和用户无关,但需要在对话中被检索引用。
这个分层不是学术空想,而是直接对应存储和查询策略。短期记忆放 Redis 或内存队列,长期记忆放向量库 + 结构化表,事实型记忆走独立的检索引擎。三者各司其职,才能避免“一锅炖”导致性能雪崩。
2.2 为什么“直接塞 context” 是饮鸩止渴
我先说一个很反直觉的结论:把记忆统统塞进 prompt,短期看效果不错,长期看是灾难。原因有三。
第一,Token 成本线性上涨。假设用户每轮对话 500 token,聊 100 轮就是 50000 token,按商用模型价格算,单次请求成本高到离谱,而且延迟随输入长度显著增加。第二,模型对超长上下文的注意力会稀释,中间部分的信息召回质量明显下降,我实测过 32k 上下文下,超过 8k 后关键信息命中率掉的非常明显。第三,无法结构化更新,用户说“我不吃香菜”,系统只能把原文堆进去,换个说法“我忌口香菜”就成了两条噪声。
所以 ai-memory 这类项目普遍采用“提取 + 存储 + 召回”的管道式架构。对话进来后,先做意图和实体抽取,把值得记忆的信息提炼成结构化条目,再写入记忆库;需要时,根据当前 query 做相关性召回,拼接进 prompt。这才是可持续的方案。
2.3 记忆的生命周期:写入、读取、遗忘
记忆系统不是存了就完事,它要像人脑一样有生命周期。
写入阶段要做信息筛选。不是每句话都值得记,比如“嗯嗯”“好的”这类寒暄完全无用;但“我下周去上海出差”就值得记,因为它有时间、地点、事件三要素。筛选可以通过规则引擎做初筛,再用模型判断是否值得入库。读取阶段要做相关性排序。我见过不少项目直接把记忆全部倒进 prompt,结果模型被老旧的偏好带偏,比如用户三年前喜欢喝奶茶,现在早改喝美式了。所以召回一定要结合时间衰减和相关性打分。遗忘阶段最容易被忽略,但恰恰是记忆系统的灵魂。用户改变想法是常态,记忆需要支持覆盖和过期,否则就是刻舟求剑。
3. 核心细节解析:存储选型与结构化设计
3.1 向量库、关系库还是键值对?我选混合双打
存储选型是 ai-memory 里争议最大的地方。我用过纯向量库(如 Milvus、Qdrant),也试过只用 PostgreSQL + pgvector,还试过纯 JSON 文件硬存。结论是:单一存储永远不够,混合双打才是正解。
向量库负责语义召回,适合“模糊但表达不一致”的记忆。比如用户今天说“我家猫叫咪咪”,明天问“我的宠物叫啥”,字面完全不重叠,但向量空间里距离很近。关系型存储负责精确结构化查询,比如“用户的城市”“用户的会员等级”“用户上次购买时间”,这些字段适合放 PostgreSQL 或 MySQL,方便条件过滤。键值存储适合短期会话状态,Redis TTL 天然适配对话过期清理。
我的参考实现是:长期记忆落 PostgreSQL(结构化字段)+ pgvector(语义向量),短期会话状态放 Redis。这套组合便宜、好运维,单机就能跑,不需要额外引入重型中间件。如果你的并发量极大,或者记忆条目过亿,再考虑拆出独立向量库。
3.2 记忆条目的最小单元设计
记忆不能整段原样存储,否则召回时噪声太多。我会把每条记忆拆成如下最小单元:
| 字段 | 类型 | 说明 | 示例 |
|---|---|---|---|
| memory_id | string | 唯一 ID | mem_8f3a2b |
| user_id | string | 归属用户 | user_1001 |
| content | text | 记忆摘要 | 用户在减肥,忌高糖 |
| entities | jsonb | 结构化实体 | {"diet": "low_sugar"} |
| source_turn_id | string | 来源对话轮次 | turn_72 |
| created_at | datetime | 创建时间 | 2024-05-01 10:00 |
| last_accessed_at | datetime | 最近召回时间 | 2024-05-02 15:30 |
| access_count | int | 召回次数 | 3 |
| embedding | vector | 内容向量 | [0.01, -0.02, ...] |
为什么要加 last_accessed_at 和 access_count?因为这两个字段是实现时间衰减和记忆巩固的关键。被高频召回的记忆说明对用户重要,可以提升权重;长期未被召回的则逐渐降权,最终可清理。这个机制模拟的就是人类记忆的“提取强化效应”。
3.3 写入前的信息抽取:规则先行,模型兜底
信息抽取是记忆质量的分水岭。纯规则抽取快但覆盖差,纯模型抽取效果好但慢且贵。我建议分两级。第一级用正则和词表匹配抓确定性信息,比如邮箱、电话、日期、地名。第二级用 LLM 做开放域抽取,Prompt 设计让模型输出 JSON,包括记忆类型、实体、重要性评分。
这里有个实操技巧:不要用主模型做抽取。我一般用一个小模型或者同一个模型但 temperature 调到 0.1,并且把抽取 prompt 设计成“只输出 JSON,不解释”,这样既能保证速度,又避免污染主对话流。抽取之后还有一道过滤:置信度低于阈值,或者内容长度超过限制,直接丢弃,宁缺毋滥。
4. 实操过程与核心实现:手把手搭一个可用记忆模块
4.1 环境准备与依赖清单
我以最通用的技术栈为例:Python 3.10+、FastAPI、PostgreSQL(含 pgvector 扩展)、Redis、OpenAI 兼容接口。如果你用的是别的模型服务,逻辑完全一致,换个 client 即可。
# 安装依赖 pip install fastapi uvicorn sqlalchemy redis openai pgvector psycopg2-binary数据库初始化需要先启用 pgvector 扩展:
CREATE EXTENSION IF NOT EXISTS vector;然后建记忆表。注意 embedding 字段的类型是 vector(1536),维度要和你的 Embedding 模型对齐。OpenAI 的 text-embedding-3-small 是 1536 维,换成其他模型记得改。
CREATE TABLE IF NOT EXISTS memories ( id SERIAL PRIMARY KEY, user_id VARCHAR(64) NOT NULL, content TEXT NOT NULL, entities JSONB, importance FLOAT DEFAULT 0.5, embedding VECTOR(1536), created_at TIMESTAMP DEFAULT NOW(), last_accessed_at TIMESTAMP, access_count INT DEFAULT 0 );4.2 写入流程:从对话到记忆条目的完整链路
对话文本进来之后,先触发抽取。我封装了一个函数,输入原始用户语句,输出结构化的记忆候选。
from openai import OpenAI client = OpenAI() def extract_memory(user_id: str, user_text: str) -> dict | None: prompt = f""" 从用户话语中抽取值得长期记忆的信息,仅输出 JSON。 字段:memory_content, entities, importance_score (0-1)。 如果无值得记忆信息,输出 {{"memory_content": null}}。 用户说:{user_text} """ resp = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], temperature=0.1, ) result = eval(resp.choices[0].message.content) # 生产环境请用 json.loads if not result.get("memory_content"): return None return { "user_id": user_id, "content": result["memory_content"], "entities": result.get("entities", {}), "importance": result.get("importance_score", 0.5), }这里特别提醒:eval 只适合演示,生产必须用 json.loads 并做异常捕获,我一开始图省事直接 eval,遇到模型输出带注释或者 Markdown 代码块的时候直接崩,后来改成先提取 JSON 子串再解析,稳定很多。
拿到候选记忆后,写入前先做去重判断。去重不是简单字符串相等,而是计算新候选和已有记忆的向量余弦相似度,超过阈值(我常用 0.85)就视为重复。重复的更新 last_accessed_at 和 content,而不是新增一条,否则记忆库会迅速膨胀。
def add_memory(conn, mem: dict, embedding: list[float]): # 先查重 cur = conn.cursor() cur.execute(""" SELECT id, content FROM memories WHERE user_id = %s ORDER BY id DESC LIMIT 50 """, (mem["user_id"],)) rows = cur.fetchall() dup_sim = 0.0 for row_id, old_content in rows: sim = cosine_similarity(embedding, get_embedding(old_content)) if sim > dup_sim: dup_sim = sim dup_id = row_id if dup_sim > 0.85: cur.execute(""" UPDATE memories SET content = %s, entities = %s, last_accessed_at = NOW(), access_count = access_count + 1 WHERE id = %s """, (mem["content"], json.dumps(mem["entities"]), dup_id)) else: cur.execute(""" INSERT INTO memories (user_id, content, entities, importance, embedding) VALUES (%s, %s, %s, %s, %s) """, (mem["user_id"], mem["content"], json.dumps(mem["entities"]), mem["importance"], embedding)) conn.commit()4.3 召回流程:加权排序,组合 prompt
召回不是只看相似度一个维度。我采取线性加权打分,让重要记忆排在前面:
score = 0.5 * 语义相似度 + 0.3 * 重要性分 + 0.2 * 时间衰减分
时间衰减分用 last_accessed_at 距今的指数衰减函数计算,越久远分越低。这个权重组合实测下来比较稳,你可以根据业务调整。召回数量我一般限制在 5~10 条,太多会稀释模型注意力。
def recall_memories(user_id: str, query: str, top_k: int = 7) -> list[str]: query_emb = get_embedding(query) cur = conn.cursor() cur.execute(""" SELECT content, embedding, importance, 1 - (embedding <=> %s::vector) AS sim, CASE WHEN last_accessed_at IS NULL THEN 0.5 ELSE exp(-EXTRACT(EPOCH FROM (NOW() - last_accessed_at)) / 86400.0) END AS time_decay FROM memories WHERE user_id = %s ORDER BY 0.6 * sim + 0.25 * importance + 0.15 * time_decay DESC LIMIT %s """, (query_emb, user_id, top_k)) rows = cur.fetchall() return [r[0] for r in rows]pgvector 的距离运算符<=>是余弦距离,1 减去它就是余弦相似度,需要在 SQL 里计算,避免把全量向量拉出来在 Python 里算,否则数据量一上来性能必崩。
召回结果最后拼进 system prompt:
def build_prompt(user_message: str, memories: list[str]) -> list[dict]: memory_block = "\n".join(f"- {m}" for m in memories) system_prompt = f""" 你是用户的私人助手。请基于已知的用户信息回答问题。 已知信息: {memory_block} """ return [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_message}, ]这个方案跑起来后,我再回归测试之前“记不住名字”的场景,第二轮问“我多大了”,只要记忆里有过出生年份相关的表述,就能准确回答。但是注意,接线时不能把记忆和当前轮对话混淆,记忆块放 system,当前轮放 user,这样模型能明确区分先验知识和即时上下文。
5. 常见问题与排查技巧实录
5.1 记忆写入过猛:用户随口一提,系统当真了
我最开始调的抽取 prompt 阈值太低,用户说“今天有点累”都被记成永久记忆,导致后续对话里模型频繁提起“您最近比较累”,体验很差。
排查下来,问题出在 importance_score 的判定标准太模糊。我的修复方式是:在抽取 prompt 里明确写下“只记忆事实型、偏好型、长期稳定型信息;情绪、瞬时状态、闲聊不记”。同时在代码里加硬性后置过滤:importance_score 低于 0.6 直接丢弃,short_text 少于 4 个字符也丢弃。这轮调整后,记忆库的写入量下降了约 60%,但回答准确度反而提升,因为噪声少了。
5.2 记忆冲突:用户上个月说 A,这个月说 B
冲突是必然的,不处理就会精神分裂。实务上我分成两个场景处理。如果是直接否定型,比如“我不喝奶茶了”,新记忆写入时把旧的“喜欢喝奶茶”降权标记为 replaced,并将新记忆置顶。如果是补充型,比如“我平时喝美式,但偶尔也喝拿铁”,两条可以共存,靠 importance 和 access_count 共同排序,让模型在回答时自行权衡。
还有个细节:用户说“我喜欢”,可能后面马上就要说“但我现在不喜欢了”。我建议写入记忆时加一个 cr_status 字段判断语句的时态色彩,如果是转折句的中间部分,暂不写入,等完整句识别后再做覆盖。
5.3 向量召回偏离:语义相近但事实相反
向量相似度高不代表逻辑一致。我踩过一个典型案例:用户说“我住在北京朝阳区”,后来搬家到“上海浦东”,两条记忆的 embedding 相似度有 0.82,没达到 0.85 的覆盖阈值,结果模型被问“你在哪”时回答“用户可能在北京或上海”,直接血压拉满。
对策是在结构化字段里做闭环校验。entities 中如果存在“居住地”“工作城市”这类单值字段,新写入时直接查旧值并执行覆盖,不走向量阈值。语义相似度只用来判断“是否是同一条内容”,用户画像里的枚举型字段必须走精确覆盖逻辑。
5.4 性能瓶颈:召回越来越慢,怎么办
记忆表数据量超过百万后,全表 TOP-N 排序会明显变慢。排查后发现两个点:一是没建 HNSW 索引,二是查询同时做了向量距离和复杂字段计算,触发了全表扫描。修复方式是建好索引后把不必要的 CASE 表达式简化,提前在应用层算好时间衰减权重。
CREATE INDEX ON memories USING hnsw (embedding vector_cosine_ops);另外,user_id 过滤条件命中率极低时,查询仍可能绕开索引。所以我在查询条件里显式锁死 user_id 等值匹配,并在 explain analyze 里确认走的 Index Scan。实测百万级数据下,单次召回从 2 秒降到 60 毫秒以内,完全可以接受。
6. 进阶扩展:从单用户记忆到协作记忆网络
如果 ai-memory 只停留在“记住一个人”,那它的上限也就是私人助理。我后来在项目里加了群体记忆和共享记忆两层,玩法完全不同。
群体记忆是指按用户标签聚合出共性偏好。比如“健身人群普遍关注蛋白质摄入”,这类信息可以在没有个人记忆时作为冷启动模板,给新用户推荐初始话术。共享记忆则是让多个 Agent 共同维护一个知识池,比如客服场景中,A 客服发现用户投诉了支付问题,B 客服再次服务同一位用户时自动携带这一条,避免用户重复描述问题。实现上就是在 memories 表增加 scope 字段和 visibility 字段,查询时多带一个 scope 条件即可。
还有一点感受很深:记忆系统永远需要人工介入的开关。不是所有用户都希望被记住,产品层面必须提供“清除记忆”“关闭记忆”的选项,并在技术上快速执行 DELETE 的连锁更新。这既是合规要求,也是产品温度。我在调研很多开源记忆项目时发现,绝大多数都忽视了这一点,只顾着如何记得更好,忘了如何忘得更彻底。
7. 最后分享一点实操体会
我做完这个记忆模块又回头重构了两次,最大的领悟是:记忆系统的核心指标不是“记住了多少”,而是“何时该忘”。很多开发者包括我自己早期都执着于把更多信息塞进记忆库,后来才发现,一个干净、规范、敢于遗忘的记忆库,配合合理的召回排序,才是让模型从“人工智障”变成“贴心助理”的关键分水岭。
如果你准备动手实现,我建议先跑通最小闭环:短期会话记忆用 Redis TTL 顶住,长期记忆就用 PostgreSQL + pgvector,抽取模型用最便宜的 mini 级模型,把流程跑通后再逐步加复杂排序、加共享记忆、加 HNSW 索引优化。不要一开始就上重型分布式向量库,那是等并发量真正起来之后才需要考虑的事。
记忆是智能的底层能力,但底层的底层,是需要被谨慎设计的边界。希望这篇内容能让你少走几步弯路。