一个Agent如果在每次对话结束后就把所有信息忘得一干二净，那它永远只能处理孤立的任务。人类助手之所以高效，是因为他记得你上周提过的偏好、上个月的项目进展。同样地，Agent需要一套长期记忆机制，而向量存储正是当前实现长期记忆最主流的技术路线。本文将完整拆解这套方案的原理与落地细节。
一、为什么Agent需要向量存储来做长期记忆
先看问题本身。Agent的对话历史本质上是文本序列，最朴素的做法是把所有历史对话塞进上下文窗口。这条路有三个致命缺陷：第一，成本随对话轮次线性增长，长对话场景下Token消耗非常可观；第二，大模型对上下文中间部分的注意力会明显衰减，也就是所谓lost in the middle现象，关键信息埋在中间反而容易被忽略；第三，上下文窗口再大也有上限，跨会话、跨天甚至跨月的记忆根本放不下。
向量存储的思路完全不同。它把每条记忆（可以是一段对话、一个事实、一个用户偏好）通过Embedding模型编码成一个高维向量，然后存入向量数据库。当Agent需要回忆时，把当前查询也编码成向量，通过相似度搜索找出语义最相关的若干条记忆，只把这些塞回上下文。这样每次召回的记忆量是可控的，成本恒定，而且天然支持跨会话的持久化存储。
从架构上看，一套完整的向量记忆系统包含四个环节：记忆写入、记忆索引、记忆检索、记忆更新与遗忘。很多教程只讲前三个，但第四个环节恰恰决定了长期记忆会不会越积越脏，后面会重点展开。
二、记忆写入与检索的实现细节
1. 记忆的写入流程
写入不是简单地存原文。一段对话里既有闲聊噪音，也有值得长期记住的事实，比如用户说我住在杭州、我对花生过敏。写入前通常要做一层信息抽取或分类，判断这段内容是否值得进入长期记忆，以及属于哪种记忆类型。常见的记忆类型包括：用户画像类（姓名、偏好、习惯）、事实类（项目截止日期、约定事项）、 episodic类（发生过的事件）。
下面是一段用Python实现的记忆写入核心逻辑，基于OpenAI的Embedding接口和Chroma向量库：
import chromadb
from openai import OpenAI
client = OpenAI()
db = chromadb.PersistentClient(path="./agent_memory")
collection = db.get_or_create_collection("long_term_memory")
def embed_text(text: str):
resp = client.embeddings.create(
model="text-embedding-3-small",
input=text
)
return resp.data[0].embedding
def write_memory(mem_id: str, content: str, meta: dict):
# content是抽取后的事实描述，meta存放类型、时间戳等元信息
collection.add(
ids=[mem_id],
embeddings=[embed_text(content)],
documents=[content],
metadatas=[meta]
)
write_memory(
mem_id="mem_001",
content="用户居住在杭州，从事后端开发工作",
meta={"type": "user_profile", "timestamp": 1718000000, "decay": 0.95}
)
注意这里的meta设计：type字段用于后续按类型过滤检索，timestamp和decay字段为后面的遗忘机制做准备。元数据设计得好不好，直接决定后期记忆管理的灵活程度。
2. 记忆的检索流程
检索时先用同样的Embedding模型把用户当前输入编码，再从向量库中找Top-K相似记忆。这里有个容易被忽略的点：不要只按相似度排序，还要结合时间因素做加权。一条三个月前的记忆和一条十分钟前的记忆即使相似度相同，价值也未必一样，具体取决于业务场景。
import time
def search_memory(query: str, top_k: int = 5, mem_type: str = None):
results = collection.query(
query_embeddings=[embed_text(query)],
n_results=top_k * 2, # 多召回一些，后面再过滤
where={"type": mem_type} if mem_type else None
)
now = time.time()
scored = []
for doc, meta, dist in zip(
results["documents"][0],
results["metadatas"][0],
results["distances"][0]
):
sim = 1 - dist # 距离转相似度
age_days = (now - meta["timestamp"]) / 86400
# 时间衰减因子，越新权重越高
time_weight = meta.get("decay", 0.95) ** age_days
scored.append((doc, sim * 0.7 + time_weight * 0.3))
scored.sort(key=lambda x: x[1], reverse=True)
return [d for d, _ in scored[:top_k]]
relevant = search_memory("用户所在城市和职业是什么", top_k=3)
这段代码里融合了相似度和时间新鲜度两个信号，权重0.7和0.3可以根据实际效果调整。召回后把记忆拼装成一段提示词注入上下文，例如格式化为已知用户信息：...的形式，模型就能自然地利用这些记忆回答问题了。
三、记忆更新、去重与遗忘机制
长期记忆系统跑久了必然面临两个问题：重复和过期。用户可能在不同时间说过类似的话，如果每次都新增一条记忆，库里的冗余会越来越多，检索质量随之下降。解决办法是在写入前先做一次相似度检查，如果找到相似度超过阈值（比如0.92）的已有记忆，就执行更新而不是新增。
def dedup_and_write(content: str, meta: dict, threshold: float = 0.92):
q = embed_text(content)
hits = collection.query(query_embeddings=[q], n_results=1)
if hits["distances"][0] and 1 - hits["distances"][0][0] > threshold:
old_id = hits["ids"][0][0]
# 覆盖旧记忆，刷新时间戳
meta["timestamp"] = int(time.time())
collection.update(
ids=[old_id],
documents=[content],
embeddings=[q],
metadatas=[meta]
)
return old_id
new_id = f"mem_{collection.count() + 1}"
write_memory(new_id, content, meta)
return new_id
遗忘机制则可以通过定时任务实现：定期扫描全库，计算每条记忆的当前有效权重，低于阈值的直接删除，或者降级归档到冷存储。归档而非硬删除有个好处，万一某条记忆其实还有用，可以从冷库找回。对于用户画像这类高价值记忆，可以设置decay为1（不衰减），只清理事实类和事件类记忆。
四、向量库与Embedding模型怎么选
选型上没有银弹，看场景。开发调试阶段，Chroma足够轻量，本地文件存储零运维；生产环境数据量大、并发高时，Milvus和Qdrant更合适，两者都支持分布式部署和标量过滤。如果团队已经在用PostgreSQL，pgvector扩展可以让你把记忆和业务数据放在同一个库里，减少运维组件。下表做个简单对比：
| 方案 | 适用场景 | 部署复杂度 |
|---|---|---|
| Chroma | 原型验证、单机小规模 | 极低 |
| pgvector | 已有PostgreSQL基础设施 | 低 |
| Qdrant | 中大规模、需要精细过滤 | 中 |
| Milvus | 海量数据、高并发生产 | 较高 |
Embedding模型方面，中文场景建议优先实测bge-large-zh或text-embedding-3-small，前者开源可私有化部署，后者效果好但需要调用外部API。无论选哪个，务必保证写入和检索用同一个模型，混用不同模型的向量做相似度计算是没有意义的。
最后提醒一个实践要点：记忆系统上线后要建立评估闭环。定期抽样检查召回的记忆是否相关、Agent回答是否真的用到了记忆。没有评估的记忆系统会静悄悄地劣化，等用户投诉时往往已经积累了大量脏数据。从最小可用版本起步，先跑通写入、检索、更新三个环节，再逐步加入分层记忆和遗忘策略，是比较稳妥的演进路径。