# 008 Agent 的记忆系统：短期记忆、长期记忆、向量数据库（Chroma、Pinecone）

一、一个让我凌晨三点还在查日志的 Bug

去年做某个工业质检 Agent 项目，Agent 在连续处理 200 张图片后突然开始“失忆”——明明前 5 分钟刚标记过的缺陷区域，它转头就重新检测一遍，还把结果写进数据库。排查到凌晨三点，发现是短期记忆的滑动窗口写死了固定长度，上下文窗口溢出后，旧信息被直接丢弃，而长期记忆的写入策略又没触发。

这个场景让我意识到：没有记忆系统的 Agent，本质上就是个无状态的 API 调用器。今天这篇笔记，就聊聊我在实际项目中怎么搭 Agent 的记忆系统，以及那些让我摔过跟头的坑。

二、短期记忆：别把它当缓存用

很多人把短期记忆等同于 Redis 缓存，这是第一个认知误区。Agent 的短期记忆核心是维持对话上下文的连贯性，不是简单的键值对存储。

2.1 滑动窗口的陷阱

早期我直接用固定长度的 deque：

fromcollectionsimportdeque# 别这样写！固定窗口会丢失关键上下文short_term_memory=deque(maxlen=10)

这个写法在简单问答场景没问题，但遇到多轮推理任务就崩了。比如用户先问“帮我分析这段代码的性能瓶颈”，接着问“那怎么优化”，再问“用 Python 实现看看”——如果窗口只保留最近 10 条，第三条指令可能已经把第一条的代码上下文挤出去了。

正确的做法是动态窗口 + 重要性评分：

classDynamicShortTermMemory:def__init__(self,max_tokens=4096):self.buffer=[]# 这里用列表，不用dequeself.max_tokens=max_tokens self.current_tokens=0defadd(self,message,importance=0.5):# 计算这条消息的token数（这里踩过坑：不同模型tokenizer不一样）msg_tokens=self._estimate_tokens(message)# 如果超了，先尝试丢弃低重要性消息whileself.current_tokens+msg_tokens>self.max_tokens:ifnotself._evict_low_importance():break# 实在腾不出空间，只能截断self.buffer.append({'content':message,'importance':importance,'timestamp':time.time()})self.current_tokens+=msg_tokens

关键点：重要性评分不是固定的。我通常让 Agent 在每次回复后，对自己刚刚处理的信息做一次重要性标记——比如涉及用户身份、任务目标、关键参数的消息，重要性自动+0.3。

2.2 短期记忆的序列化问题

Agent 重启后短期记忆丢失是常态，但有些场景需要持久化。比如工业质检中，Agent 正在处理一个复杂的检测流程，突然断电重启——如果短期记忆完全丢失，操作员得从头开始。

我的做法是定期 checkpoint：

defcheckpoint_short_term(self,interval=30):# 每30秒或每处理5条消息，把短期记忆序列化到本地文件# 这里用pickle，但注意安全风险（别在生产环境反序列化不可信数据）withopen(f'/tmp/agent_memory_{self.agent_id}.pkl','wb')asf:pickle.dump(self.buffer[-50:],f)# 只保存最近50条

三、长期记忆：别把所有东西都往里塞

长期记忆的设计原则是精而不多。我见过最离谱的设计是把 Agent 每次对话的完整日志都塞进向量数据库，结果检索时召回的全是噪声。

3.1 记忆的抽象与压缩

原始对话文本直接存长期记忆是灾难。我通常做三层抽象：

defabstract_memory(raw_conversation):# 第一层：提取关键实体和关系entities=extract_entities(raw_conversation)# 比如用户ID、设备型号、故障代码# 第二层：生成摘要（这里踩过坑：直接用LLM摘要太慢，改用textrank）summary=generate_summary(raw_conversation,max_length=200)# 第三层：提取可复用的经验规则rules=extract_rules(raw_conversation)# 比如“当温度>85度时，优先检查散热模块”return{'entities':entities,'summary':summary,'rules':rules,'raw_length':len(raw_conversation)}

重要经验：长期记忆的写入频率要远低于短期记忆。我一般设置阈值——只有当对话涉及新的实体、产生新的规则、或者解决了之前未遇到过的问题时，才触发写入。

3.2 记忆的遗忘机制

没有遗忘机制的长期记忆，最终会变成垃圾堆。我参考了 Ebbinghaus 遗忘曲线，给每条记忆一个衰减因子：

defcalculate_recall_score(memory,current_time):# 初始分数1.0，每过一天衰减10%days_passed=(current_time-memory['timestamp']).days base_score=1.0*(0.9**days_passed)# 但每次被成功召回，分数重置并增加ifmemory['last_recalled']:recall_boost=0.2*(1-base_score)# 越不常用的记忆，召回后提升越大base_score=min(1.0,base_score+recall_boost)# 如果记忆被标记为“重要”，保底0.5ifmemory.get('important'):base_score=max(0.5,base_score)returnbase_score

当分数低于 0.1 时，这条记忆会被标记为“待归档”，一周后如果仍未被召回，直接删除。

四、向量数据库：Chroma vs Pinecone 的实战选择

4.1 Chroma：本地开发首选

Chroma 的优势是零配置、本地运行，特别适合原型验证和边缘设备。

importchromadbfromchromadb.configimportSettings# 初始化（这里踩过坑：默认的persist_directory在/tmp，重启就没了）client=chromadb.Client(Settings(chroma_db_impl="duckdb+parquet",persist_directory="./chroma_data"# 一定要指定持久化路径！))# 创建集合时指定embedding函数collection=client.create_collection(name="agent_memory",embedding_function=my_embedding_func,# 别用默认的all-MiniLM-L6-v2，中文效果差metadata={"hnsw:space":"cosine"}# 用cosine相似度，欧氏距离在高维空间效果不好)# 添加记忆（注意：id不能重复，否则会覆盖）collection.add(documents=[memory['summary']],metadatas=[{'timestamp':memory['timestamp'],'importance':memory['importance'],'type':memory['type']# 比如 'rule', 'entity', 'experience'}],ids=[f"mem_{int(time.time())}_{random.randint(1000,9999)}"])

Chroma 的坑：

• 默认 embedding 模型对中文支持差，我换成BAAI/bge-large-zh-v1.5
• 当数据量超过 10 万条时，查询速度明显下降，需要手动调hnsw:ef_construction参数
• 不支持分布式，多进程访问时要注意锁

4.2 Pinecone：生产环境的正确选择

Pinecone 是托管服务，省去了运维向量数据库的麻烦。但成本不低，我那个工业项目一个月花了 200 刀。

importpinecone# 初始化（别把API key硬编码在代码里！）pinecone.init(api_key=os.getenv('PINECONE_API_KEY'),environment='us-west1-gcp')# 创建索引（这里踩过坑：维度必须和embedding模型输出一致）index_name="agent-memory-prod"ifindex_namenotinpinecone.list_indexes():pinecone.create_index(name=index_name,dimension=1024,# bge-large-zh-v1.5输出1024维metric='cosine',pods=1,# 生产环境至少2个pod，保证高可用pod_type='p1.x2')index=pinecone.Index(index_name)# 批量写入（别一条一条insert，效率极低）defbatch_write_memories(memories,batch_size=100):foriinrange(0,len(memories),batch_size):batch=memories[i:i+batch_size]vectors=[]formeminbatch:vectors.append((mem['id'],mem['embedding'],{'summary':mem['summary'],'timestamp':str(mem['timestamp']),'importance':str(mem['importance'])}))index.upsert(vectors=vectors)

Pinecone 的坑：

• 查询时top_k不要设太大，我一般设 5-10，多了召回质量反而下降
• 元数据过滤很慢，如果经常按时间范围查询，建议在 embedding 向量里编码时间信息
• 免费版有 1 个 pod 限制，且 7 天不活跃会删除索引

4.3 混合检索：向量 + 关键词

纯向量检索在某些场景下效果很差。比如用户问“上次那个温度异常的案例”，如果 embedding 模型没把“温度异常”和具体设备型号关联起来，可能召回不相关的结果。

我的做法是向量检索 + BM25 关键词检索的混合方案：

defhybrid_search(query,vector_weight=0.7,keyword_weight=0.3):# 向量检索vector_results=index.query(vector=embed_query(query),top_k=20,include_metadata=True)# 关键词检索（用Whoosh或Elasticsearch）keyword_results=keyword_index.search(query,top_k=20)# 加权融合（这里踩过坑：分数归一化很重要）combined={}foriteminvector_results['matches']:combined[item['id']]=vector_weight*item['score']foriteminkeyword_results:ifitem['id']incombined:combined[item['id']]+=keyword_weight*item['score']else:combined[item['id']]=keyword_weight*item['score']# 按总分排序sorted_results=sorted(combined.items(),key=lambdax:x[1],reverse=True)return[item[0]foriteminsorted_results[:10]]

五、记忆系统的架构整合

最后分享一个我在生产环境验证过的架构：

短期记忆（内存） → 重要性评估 → 抽象压缩 → 长期记忆（向量数据库） ↑ ↓ └──────── 召回机制 ←─────────┘

关键流程：

1. 用户输入先进入短期记忆，同时触发重要性评估
2. 短期记忆满时，低重要性消息被丢弃，高重要性消息被抽象后写入长期记忆
3. Agent 每次推理前，从长期记忆中召回与当前上下文相关的历史记忆
4. 召回的记忆与短期记忆合并，形成完整的上下文窗口

个人经验建议：

• 别追求“记住所有东西”，好的记忆系统应该知道“该忘什么”
• 向量数据库不是银弹，对于结构化记忆（比如用户配置、设备参数），传统关系型数据库更合适
• 定期做记忆质量评估：随机抽取 100 条记忆，人工判断召回结果的相关性，低于 70% 就要调整 embedding 模型或检索策略
• 边缘设备上别用 Pinecone，Chroma + 本地 embedding 模型足够，但要注意模型大小和推理速度的平衡

记忆系统是 Agent 从“工具”进化为“助手”的关键。没有记忆的 Agent 就像金鱼，每次对话都是第一次见面——这样的 Agent，用户用两次就烦了。