☰
用Chroma为Agent实现短期记忆:向量检索与Prompt注入
2026/10/19 2:40:51 网站建设 项目流程

先把结论放在前面:如果你正在做本地化部署的Agent,想让它具备“记得住前面聊过什么”的能力,又不想引入重型的数据库服务,Chroma是一个非常值得试的轻量方案。这篇文章就围绕“用Chroma给Agent做短期记忆”这条主线,把从环境搭建到检索注入Prompt的完整链路拆开讲清楚。

我自己在做多轮对话Agent的时候,最头疼的就是“上下文丢失”问题。模型本身的上下文窗口是有限的,你不能把整场对话废话全部塞进每一轮请求里,成本高不说,还容易把有效信息淹没掉。后来用Chroma把关键对话片段向量化存到本地,在用户每次提问时召回最相关的历史片段注入Prompt,整个对话质量立马不一样。这篇文章会把我的整个实现过程、踩坑记录和参数调整心得都摊开来说,希望能帮你少走弯路。

1. 为什么用Chroma做Agent短期记忆:三个方案的对比

1.1 大模型对话的“失忆”问题到底出在哪

大模型本身是没有记忆这个概念的。从工程上看,每一次请求都是独立的“无状态”调用,模型只能看到你当前Prompt里拼进去的内容,至于上一轮说过什么、上上轮用户透露过什么偏好,它一概不知。想要让Agent表现得更聪明,本质上就是由你在应用层把“历史信息”抢救回来,再塞回Prompt。

这里说的短期记忆,通常指的是“同一会话内、较近几轮里出现过的关键信息”。比如用户在某轮说过“我住在南方,经常下雨”,后几轮问“推荐一下适合我通勤的鞋子”,如果你能把那句居住信息召回并拼进Prompt,模型立刻就能给出更贴近需求的回答。反过来,如果你什么历史都不带,模型只能给一个通用答案。

问题在于,怎么精准地“抢救”这些信息。最笨的办法是把历史消息全部拼接起来,从头到尾每一轮都发给大模型。这种做法在短对话里没问题,但一旦会话变长,Token消耗会指数级膨胀,很快就触到上下文窗口上限。很多时候用户说的那些“嗯”“好的”根本没价值,却白白占着窗口。

所以这里需要的是一个“检索式记忆”:不同对历史对话做全量携带,而是根据用户当前的提问,从已经存储的对话片段里挑出最相关的几条,再注入Prompt。那怎么判断“相关”?直接把文本拿去做相似度匹配不现实,你得先把文本转成向量,再走向量检索。这就是Chroma这类向量数据库要干的事。

1.2 为什么不是SQLite或Redis

很多人听到“存数据”第一反应是SQLite,觉得轻量、稳定、不用额外服务。确实,如果只是把对话记录逐条存下来,SQLite完全够用。但这里的关键差异在于“检索方式”。

SQLite和Redis擅长的是精确匹配和结构化查询:你给我一个关键词,我把包含这个词的记录捞出来。但对话里的“相关”往往不是字面上的关键词匹配。用户问题里说的是“适合南方通勤的鞋”,历史记录里可能写的是“我住在南方,经常下雨,路面湿滑”,这两个句子在字面上没有共同词,SQLite很难把它们关联起来。而向量检索看的是语义相似度,两个句子即使没有一个词是重复的,只要语义接近,向量距离就会很近,可以被正确召回。

我把三个方案放在一起做了一个对比:

方案擅长场景语义检索能力部署成本适合做Agent记忆吗
SQLite精确查询、结构化记录基本没有极低适合做长期结构化档案,不适合短期语义召回
Redis缓存、队列、KV读写需要额外配模块低适合存原始对话,不适合直接做相关片段召回
Chroma向量存储、相似度检索原生支持极低(进程内嵌)非常适合短期记忆的“写-检-取”循环

选择Chroma还有一个很实际的原因:它是一个嵌入式向量数据库,可以直接以库的形式嵌进你的Python进程里,不需要单独启动一个服务端。对本地项目、小团队验证、边缘设备上的Agent来说,这种模式最省心。你不需要去维护一个数据库服务,代码里一句PersistentClient(path="..."),数据就落盘了。

1.3 Chroma在这条路径上的具体优势

单从功能上看,Chroma能做的事,市面上主流的向量数据库基本都能做。但考虑到“本地化Agent短期记忆”这个具体场景,它有几点很贴合需求。

第一,数据持久化足够简单。传统向量数据库要做数据备份、迁移、权限管理,一套流程下来半天过去了。Chroma用PersistentClient之后,所有数据都写到你指定的目录里,备份就是拷贝那个文件夹,换机器就是把这个目录搬过去。对个人项目和中小型产品来说,这种透明、可控的存储方式反而更省事。

第二,元数据过滤很强。Agent的记忆检索不能每次都全库扫描,不同用户、不同会话之间的数据要隔离。Chroma的每条记录可以附带metadatas字段,检索时用where参数做精确过滤。比如我只想查当前会话的对话记录,就在where里传{"session_id": "xxx"},这样向量检索是在一个子集里进行的,准确率和效率都有保证。

第三,API设计非常直观。我一个下午就能把写入、检索、删除、更新这套流程全部跑通,不需要翻太多文档。对于写Agent业务逻辑为主的开发者来说,这种学习成本几乎可以忽略不计。跟那些围绕服务端、集群、分片概念设计的重型数据库比,Chroma就像是“开箱即用的工具”,而不是“需要运维的基础设施”。

2. 环境准备:Chroma的安装与工程目录设计

2.1 安装Chroma及其依赖

Chroma是一个Python库,安装方式很常规,直接用pip装就行。建议你在项目里创建独立的虚拟环境,避免和全局Python环境互相干扰。

python -m venv venv source venv/bin/activate # Windows 下执行 venv\Scripts\activate pip install chromadb

安装之后可以验证一下版本,确保装成功了。

python -c "import chromadb; print(chromadb.__version__)"

需要注意版本差异,Chroma的API在不同版本之间有过调整,尤其是在PersistentClient、get_or_create_collection这些接口上,不同版本的默认行为和参数细节不太一样。我写这篇文章用的版本是基于1.x的接口规范。如果你装到的版本比这个新,建议以官方文档的迁移说明为准,代码逻辑大体相通。

安装的时候可能会顺带装上onnxruntime等依赖,这是Chroma内置Embedding模型运行需要的。默认情况下,Chroma会使用一个内置的Embedding函数,不需要你额外申请任何API Key,也不依赖外部网络服务,对本地化部署非常友好。

2.2 两种启动模式的区别与选择

Chroma有两种使用方式,这里特意拿出来说一下,因为曾经在这上面栽过跟头。

第一种是客户端-服务端模式。需要先启动一个Chroma服务,然后你的应用代码通过客户端连上去。这种方式适合多机部署、多进程共享的场景,但对本地Agent项目来说,多了一个服务进程就多了一份运维负担。

第二种是嵌入式模式,也是我更推荐的方式。直接在进程内创建Client,数据落盘到本地目录。这种方式没有网络开销,也不存在连接管理的问题,代码写起来最直接。

import chromadb # 嵌入式模式:数据持久化到本地目录 client = chromadb.PersistentClient(path="./chroma_data") # 如果想用临时模式(不落盘),用下面的方式 # client = chromadb.Client()

在Agent的短期记忆场景里,我建议直接用PersistentClient,因为对话记录是有价值的,进程退出后重新启动,还要能够找回上次的记忆。临时模式只适合用来做功能验证,跑一遍就扔。

2.3 我的项目目录结构参考

一个功能完整的Agent短期记忆项目,文件组织不能太乱。我自己的习惯是这样划分的:

agent_demo/ ├── main.py # 主程序入口,写完整Agent循环 ├── memory.py # 封装Chroma读写逻辑的模块 ├── requirements.txt # 依赖清单 ├── chroma_data/ # Chroma持久化目录 └── logs/ # 日志目录,方便排查问题

memory.py专门负责跟Chroma打交道,对外暴露remember、recall、clear_session这几个函数。main.py里的Agent逻辑不直接碰Chroma,只调用memory.py提供的接口。这样做的好处是,如果以后想换别的向量库,只需要改memory.py一个文件,业务层完全不用动。

3. 核心实现:短期记忆的写入、检索与注入

3.1 初始化向量库和Collection

进入正题,写代码。首先初始化Chroma Client并创建一个Collection。可以把Collection理解成一张表,里面存向量、原始文本和元数据。

import chromadb client = chromadb.PersistentClient(path="./chroma_data") collection = client.get_or_create_collection( name="agent_short_term_memory", metadata={"hnsw:space": "cosine"} # 指定距离度量方式 )

get_or_create_collection这个接口值得多说一句。它会先检查指定名称的集合是否存在,存在就直接返回,不存在才创建。相比create_collection,它能避免重复创建时报错,更适合长期运行的应用。

metadata里的hnsw:space是用来指定向量距离算法的。可选值有l2、cosine、ip,对应欧氏距离、余弦相似度、内积。对于文本语义检索类场景,cosine是默认选择,因为它在衡量语义相似度时受向量长度影响最小。这一点后面调优部分会再展开讲。

3.2 把每轮对话写进记忆库

现在核心问题来了:怎么把“用户问题”和“助手回答”存进向量库里。这里有一个设计选择:是把每一轮单独存,还是把一整段历史一次性存?

我的做法是把每一轮完整的问答(用户问题加助手回答)当作一个文档来存。这样检索出来的每条记录都包含完整的上下文,而不是只拿到半个句子,对后续拼Prompt更友好。

import datetime def remember(session_id, user_text, assistant_text, turn_id): doc_id = f"{session_id}_{turn_id}" content = f"用户:{user_text}\n助手:{assistant_text}" collection.upsert( ids=[doc_id], documents=[content], metadatas=[{ "session_id": session_id, "turn_id": turn_id, "time": int(datetime.datetime.now().timestamp()) }] )

注意这里用的是upsert而不是add。两者的区别在于:upsert会先检查传入的ID是否已存在,存在就更新,不存在才插入。这在对话场景中很有用,比如某轮回答因为生成中断需要重新生成,再次调用remember就能覆盖旧内容,而不会产生重复记录。

metadatas里我存了三个字段:session_id用于会话隔离,turn_id用于记录顺序,time用于后续按时间排序。这些字段在检索时都能作为过滤条件使用。

3.3 Agent如何检索记忆并拼装Prompt

存进去只是第一步,更重要的工作是“取”回来。当用户发出新问题时,我们把这个新问题作为查询向量,去Collection里找最相似的几条历史记录。

def recall(session_id, query_text, top_k=3): results = collection.query( query_texts=[query_text], n_results=top_k, where={"session_id": session_id} ) return results

这里有个很关键的细节:where参数保证了检索范围被限制在指定会话内。如果不加这个过滤条件,当多个用户共用同一个Collection时,A用户的问题可能会召回B用户的私人对话记录,这是隐私上的大坑,绝对不能忽视。

另一个细节是Chroma的默认行为是返回距离最小的n_results条记录,也就是最相似的记录。但你拿到的results是原始返回结构,里面包含了ids、documents、metadatas、distances多个列表。实际使用时要先提取文档和元数据,再按时间排序,因为单纯按相似度排序会把时间顺序打乱。

from typing import List def build_memory_context(session_id: str, query_text: str, top_k: int = 3) -> str: results = collection.query( query_texts=[query_text], n_results=top_k, where={"session_id": session_id} ) documents = results["documents"][0] metadatas = results["metadatas"][0] # 按时间戳排序,保证拼接顺序符合对话发生顺序 pairs = sorted( zip(documents, metadatas), key=lambda x: x[1].get("time", 0) ) context_parts = [] for doc, meta in pairs: context_parts.append(f"[第{meta.get('turn_id', '?')}轮]\n{doc}") return "\n\n---\n\n".join(context_parts)

这样返回的memory_context字符串,就是可以直接拼进Prompt的历史片段。拼接的时候放在系统提示词和当前用户问题之间,或者单独用一个段落标注这部分是“历史对话上下文”,让模型区分哪些是历史、哪些是当前问题。

3.4 完整代码:一个带短期记忆的Agent链路

把上面几个环节串起来,再加上大模型调用,就是一个具备短期记忆能力的Agent。下面这个示例为了聚焦记忆实现,把大模型调用部分简化为一个函数,实际使用时替换成你对接的模型接口。

import chromadb import datetime from typing import List, Dict # 初始化Chroma client = chromadb.PersistentClient(path="./chroma_data") collection = client.get_or_create_collection( name="agent_short_term_memory", metadata={"hnsw:space": "cosine"} ) def remember(session_id: str, user_text: str, assistant_text: str, turn_id: int): doc_id = f"{session_id}_{turn_id}" content = f"用户:{user_text}\n助手:{assistant_text}" collection.upsert( ids=[doc_id], documents=[content], metadatas=[{ "session_id": session_id, "turn_id": turn_id, "time": int(datetime.datetime.now().timestamp()) }] ) def build_memory_context(session_id: str, query_text: str, top_k: int = 3) -> str: results = collection.query( query_texts=[query_text], n_results=top_k, where={"session_id": session_id} ) documents = results["documents"][0] metadatas = results["metadatas"][0] pairs = sorted( zip(documents, metadatas), key=lambda x: x[1].get("time", 0) ) context_parts = [] for doc, meta in pairs: context_parts.append(f"[第{meta.get('turn_id', '?')}轮]\n{doc}") return "\n\n---\n\n".join(context_parts) def chat_with_agent(session_id: str, user_input: str, turn_id: int): memory_context = build_memory_context(session_id, user_input) prompt = f"""你是AI助手。请根据当前问题和历史对话上下文回答。 历史对话上下文: {memory_context} 当前用户问题:{user_input} 回答:""" # 这里替换成你实际使用的大模型调用 response = call_llm(prompt) remember(session_id, user_input, response, turn_id) return response

整个流程没有特别复杂的地方,但每一个细节都影响最终效果:用什么粒度存、检索时加什么过滤、按什么顺序拼接,这些都要跟实际场景匹配。

4. 调参和优化:让记忆检索更准、更快

4.1 距离度量的选择

Chroma的hnsw:space参数可以在创建Collection时指定。我在前面提到了三个选项,这里细说各自适用场景。

cosine是处理文本向量的默认首选。它计算两个向量夹角的余弦值,关注的是方向一致性而非长度。在大模型生成的文本向量里,向量长度通常与文本具体内容没有直接关系,所以用余弦相似度能比较公平地衡量语义相近程度。检索时返回distance值越小,表示相似度越高(注意这里是1 - cosine_similarity的转化关系)。

l2也就是欧氏距离,适合向量长度本身有意义的场景。比如某些Embedding模型把文本长度或情感强度编进了向量模长,这时候欧氏距离能保留这层信息。但大多数通用场景下,它的表现不会比余弦好。

ip表示内积,通常在向量维度经过归一化处理后才比较合适。如果向量没有归一化,内积结果会受向量长度干扰,导致长文本天然占优势。Agent记忆场景里对话文本长短不一,直接上内积很容易出现“长记录碾压短记录”的问题。

我的建议是:没有特别理由就直接用cosine,如果你用的Embedding模型有特殊要求,再参照模型文档调整。

4.2 top_k与相似度阈值怎么定

top_k决定了每次检索返回几条最相似的记录。这个值太大,会塞进大量无关内容,稀释Prompt中真正有用的信息;太小,又会漏掉关键记忆。我在实践中的经验是:3到5比较合适。

如果对话每轮信息量比较大,比如用户连续问多个复杂问题,3条可能不够覆盖全部关键信息,这时候可以适当提高到5。具体效果可以在调试时打印检索出来的历史片段,肉眼判断是否符合预期。

还有一个很容易忽略的问题:即使是最相似的那几条,也可能完全不相关。比如用户突然切换话题,问了一个跟前十轮毫无关系的新问题,此时检索出来的“最相似”记录实际上也不该被拼进Prompt。这种情况就需要一个相似度阈值来做兜底。

DISTANCE_THRESHOLD = 0.4 def build_memory_context_with_threshold(session_id: str, query_text: str, top_k: int = 3): results = collection.query( query_texts=[query_text], n_results=top_k, where={"session_id": session_id} ) documents = results["documents"][0] distances = results["distances"][0] metadatas = results["metadatas"][0] valid_pairs = [] for doc, distance, meta in zip(documents, distances, metadatas): if distance <= DISTANCE_THRESHOLD: valid_pairs.append((doc, distance, meta)) if not valid_pairs: return "" valid_pairs.sort(key=lambda x: x[2].get("time", 0)) return "\n\n---\n\n".join([f"[第{p[2].get('turn_id', '?')}轮]\n{p[0]}" for p in valid_pairs])

这个阈值需要根据你的Embedding模型和实际数据分布来调整。有一个相对笨但有效的方法:先把所有历史对话打印出来,人工标注哪些片段之间是相关的,然后分别计算它们的距离,取一个能区分“相关”和“不相关”的临界值。不同Embedding模型的输出分布差异很大,没有统一的万能阈值。

4.3 会话隔离与记忆清理策略

多用户场景下,会话隔离是硬要求。我强烈建议每个用户的每次会话都用独立的session_id,并且所有检索都带上where={"session_id": session_id}。这个前面已经强调过,这里是再加一个“清理”维度的提醒。

对话记录一直存下去会越来越臃肿。短期记忆的核心特征是“近”,时间太久远的记录不仅占用存储空间,检索时还可能干扰结果。因此需要定期清理或归档。

两个常用策略:

  • 按时间清理:定期删除超过N天的记录。比如只保留最近7天的对话数据,更早的转存到长期记忆库或者直接丢弃。
  • 按轮次清理:每个会话只保留最近N轮记录。超出滑窗范围的最老记录删除。

删除操作在Chroma里很简单:

collection.delete(where={"session_id": session_id, "time": {"$lt": cutoff_time}})

实际使用中我建议“按时间清理”更稳妥,因为用户可能会在一个长会话里聊很多内容,单纯按轮数删除容易误伤。时间维度更符合短期记忆的自然语义。

5. 常见问题与避坑实录

5.1 数据持久化不生效

这是比较常见的新手坑。代码里明明用了PersistentClient,路径也指定了,但重启项目后发现之前写入的数据全部消失。

排查方向:确认你是不是每次启动都调用了client = chromadb.Client()。如果是,那用的是临时模式,所有数据只存在内存里,进程退出就没了。改用chromadb.PersistentClient(path="./chroma_data")之后,还注意路径要统一,不要这次用相对路径,下次用绝对路径,否则指向的不是同一个数据目录。

另外一个容易忽略的点是:如果项目工作目录切换了,相对路径解析到的位置也会变化,看起来就像数据丢了。建议在配置里固定使用相对于项目根目录的路径,或者直接用绝对路径。

5.2 检索结果排序混乱

我前面在build_memory_context里专门做了按时间排序的处理,这个坑就是在那时候踩上的。

Chroma返回的检索结果是按相似度从高到低排列的,也就是说距离最小的记录排在前面。如果你直接把返回结果按原顺序拼进Prompt,会出现一种很奇怪的现象:模型看到的历史对话是乱序的,上一句还是第5轮的内容,下一句就跳到第2轮,再下一句是第8轮。这种上下文连贯性被打破之后,模型的回答质量会明显下降。

解决办法就是上面代码里写的:拿到结果后,根据metadatas里的time字段排序,让拼接到Prompt里的历史片段按时间顺序呈现。这比按turn_id排序更可靠,因为turn_id可能在重新生成回答时发生变化。

5.3 检索不到预期内容

有时候用户提问了一个问题,你觉得某个历史片段明显相关,但检索结果里就是没有它。这种情况大多是top_k太小,或者相似度阈值设置太严格。

可以先不做任何过滤,直接把n_results调大,比如调到10,然后把返回结果打印出来看。如果相关片段确实在10条以内但排得比较靠后,说明当前Embedding模型对该文本的语义表征不够细,可以考虑换更大的模型;如果完全没出现,说明文本被截断或者写入时内容格式有问题,需要返回检查remember里的内容拼接方式。

还有一个细节:Chroma的内置Embedding函数可能对中文理解有限,如果你的Agent场景里中文占比很高,可以尝试更换为效果更好的中文Embedding模型。做法是在创建Collection时传入自定义的embedding_function参数,具体更换方式需要根据你选的模型服务来定。

5.4 常见问题速查表

问题现象可能原因解决方案
重启后数据丢失用了临时模式或路径不一致统一PersistentClient路径,确认启动方式
检索结果乱序只按相似度排序按meta中的time字段重新排序
多条相同记录重复写入同一ID改用upsert,保证ID唯一
where过滤无效Collection里没有对应元数据字段写入时确认metadata字段名一致
检索结果为空top_k=0、阈值过严、Collection为空逐步放宽条件,打印中间结果定位
命中内容不相关Embedding模型不适合当前语料换Embedding模型,或调低阈值

6. 把短期记忆升级成完整记忆体系

6.1 从短期到长期:定时归纳摘要

短期记忆解决的是“记住最近几句”,但Agent在真实场景里还需要长期记忆。比如用户一周前告诉你他养了一只猫,今天问“我家猫最近总睡觉正常吗”,如果只靠短期记忆,那条信息早就被挤出去了。

比较省力的升级方案是:定期把一段短期记忆内的对话内容做摘要,把摘要作为一条新的记忆写入另一个Collection,专门存放长期记忆。这样短期记忆库保持轻量,长期记忆库承担“重要信息长期可查”的工作。

def summarize_and_store(session_id: str, start_time: int, end_time: int): # 1. 从短期记忆库中取出这个时间段的所有对话记录 # 2. 把记录拼接后交给大模型生成摘要 # 3. 把摘要写入长期记忆Collection,带上session_id和时间范围 pass

这个方案我实际跑通过,效果还不错。关键是摘要的生成时机,不能太频繁,否则成本高;也不能太久,否则信息丢失严重。我一般对话结束或每隔10轮触发一次摘要,具体频率可以根据业务场景调整。

6.2 记忆写回与遗忘机制

有了摘要之后,另一个问题是“什么时候把短期记忆写进长期记忆”。这里需要一个触发策略,比较常见的做法是“会话结束时统一总结”。用户关闭会话或会话空闲时间超过阈值时,触发总结任务,把这一整个会话的核心信息抽取出来,存进长期记忆。

还要考虑遗忘机制。如果用户的某个话题已经结束,相关的记忆就不要再频繁召回,否则会让Agent表现得很“纠缠”。最简单的做法是在长期记忆检索时加一个“最近活跃时间”过滤:只召回最近N天内更新过的记忆条目。这样Agent既保留了长期记忆,又能自然“淡忘”过期信息。

6.3 后续扩展的其他可能

短期内不必把所有记忆功能都做得很重。如果只是验证想法,跑通一个带短期记忆的最小闭环就够用了。但架构上要把接口留好,比如memory.py里的remember和recall函数签名,确保后续加长期记忆库时不用大改。

我自己在实际项目里的体会是:短期记忆和向量检索这种组合,真正难的不是代码,而是如何设计“记忆的写入粒度和召回时机”。写入粒度直接影响检索精度,召回时机直接影响模型读到的上下文是否自然。这两点没有标准答案,需要根据你自己的对话场景反复调试。

最后分享一个小技巧:调试阶段建议把每次检索出来的历史片段都打印到日志里。肉眼看看模型到底读到了什么,比只看最终回答是否变好要直观得多。很多时候Agent回答不好,不是因为模型能力不够,而是因为“喂进去的历史记忆不对”。先把检索结果调准,回答质量自然会跟上来。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询