做Agent的朋友应该都有过这种体验对话一多模型就开始“失忆”前面说的关键信息后面全不认账。硬把所有历史一股脑塞进prompttoken消耗肉眼可见地涨响应也变得越来越慢稍微长一点的会话基本没法用。这期还是接着Agent实践系列往下走聊一个轻量但很实用的解法——用Chroma搭建本地向量库给Agent补上短期记忆。我会从为什么需要记忆开始讲到安装、写入、查询、接入Agent的完整流程最后把实际操作里踩过的坑一并列出来。适合在用LangChain这类框架做Agent、又不想为了记忆功能直接上重型数据库的开发者参考当然你完全不用框架、只调原生API也一样能套用。1. 先想明白Agent的短期记忆到底缺什么1.1 大模型本身不是“记忆体”很多人刚上手做Agent时都会下意识觉得模型什么都知道肯定记得住刚才的话。这里有个本质误区——大模型本质上是一个“每次输入都独立处理”的函数它没有对话历史的概念。API接口里就算带了上下文也是你手动在拼参数而不是模型自己“想起来的”。你每一次调用模型它看到的都只是本次请求里的那段prompt。上一轮的问答结果如果不拼进去模型这一轮就是全新状态。所以所谓“记忆”在工程上就是一套“把该留的信息存下来在合适的时机取出来再塞回prompt里去”的机制它和人类记忆的运作方式完全不一样更像是一个外挂的记事本。这个记事本分两种。长期记忆存的是跨多个会话稳定的知识比如用户的工作背景、偏好设置短期记忆则集中在当前会话内部的上下文比如用户几分钟前提到的一个需求细节、一个待办事项。这套方案要解决的就是短期记忆——说得直白点就是如何把最近的对话历史管理好让Agent在长对话里不丢关键信息。1.2 短期记忆为什么偏要用向量库短期记忆的第一个笨办法是滑动窗口只保留最近N轮对话直接拼进prompt。这个方案实现最简单很多新手一开始都会这么写。但它有两个很明显的瓶颈一个是token浪费对话到了50轮、100轮时就算只留最近20轮长度也可能非常可观另一个是丢失率太高如果用户在第30轮提起一个需求第60轮再问相关问题时这段历史早被移出窗口了。向量库解决的是后一个问题。它的核心思路是把每条用户消息和助手回复都转成向量存起来每次收到新问题时先做相似度检索把和当前问题最相关的历史片段捞出来而不是无脑保留最近几条。这样既不用为“窗口太小丢记忆”发愁也不用为了保留全部历史而付出巨大的token代价。这个方案对比缓存式存储有一个关键优势召回和语义相关。滑动窗口只能按时间顺序硬切但用户往往会在很久之后回头询问某个细节比如“刚才你说的那个方案二报价是多少”这时候“方案二”其实是几十轮之前的内容了按时间窗口早被丢掉按向量相似度却能在历史里精准捞回来。提示这里说“短期记忆”但不要误会成只能存几轮。它的核心特征是“按话题检索、随会话生命周期管理”而不是“只能存几条”。2. 向量库选型Chroma凭什么是轻量场景的省心选择2.1 和同类方案的简单对比市面上能做向量检索的组件非常多没选型经验时容易看花眼。先给一张我实际对比过的表把关键差异列出来方案部署方式轻量程度本地持久化适合场景Chromapip安装内嵌运行很高无需独立服务原生支持个人项目、中小型Agent、原型验证FAISSpip安装需要自己管理索引高但维护成本略高手动管理索引文件大规模纯向量检索、研究场景Milvus独立服务需要Docker或集群低支持生产级海量向量检索、多租户Qdrant独立服务或嵌入式模式均可中支持需要丰富过滤能力的生产场景直接用Redis/mysql拼相似度已有中间件取决于现有架构支持简单查询但不适合语义检索这个项目选Chroma核心原因是两个。第一个是轻它不需要单独起服务在Python进程里直接跑pip装完就能用这对绝大多数Agent项目来说省掉了大量运维成本第二个是完整它自带embedding处理、集合管理、metadata过滤、持久化存储单库就能覆盖“存向量、算相似度、按条件过滤、落盘重启不丢”这几个核心需求不用自己拼组件。我见过不少团队为了“正规化”上来就部署重量级向量库结果Agent本身只有几千条对话记忆服务倒是开了一堆无非是给运维添负担。工具选型应该跟着数据量级走初期一个Chroma完全够用。2.2 Chroma的核心API要先摸清楚Chroma的使用模型其实很简单核心概念只有三个client、collection、document。client管理数据库连接和持久化位置collection类似于数据库里的表存放一组带有向量的文档document就是一条具体的记忆可以附带metadata元信息比如角色、时间、会话编号。先看最常用的初始化方式import chromadb # 持久化客户端数据会落盘到指定目录 client chromadb.PersistentClient(path./chroma_data)这里要记住一个关键区别PersistentClient才是本地落盘的数据会保存到磁盘目录直接用chromadb.Client()其实是临时内存模式进程一结束数据就没了。很多人踩过这个坑代码写在Jupyter里跑得好好的重启后一切归零原因就在这里。我实际推荐的做法是把所有使用同一份数据的模块都指向同一个持久化路径不要有的用临时客户端有的用持久化客户端。Collection层的操作有三个创建/获取集合、往集合里加数据、从集合里查数据。初次接触时容易绕的点是collection的名字、embedding_function、以及查询时的过滤条件这三个字段看似简单实际组合起来就能覆盖大部分记忆管理需求。注意Chroma的数据结构设计让我想起把书按标签放进书架的思路。document是书本身metadata是书封上的分类标签向量是书的“内容指纹”添加时你告诉它书放哪、标签写什么查询时它会找出内容最相关的几本还能按标签筛选。先把这个类比记在脑子里后面所有操作都顺了。3. 本地向量库搭建从安装到第一轮写入3.1 安装与客户端初始化安装非常简单用pip直接装官方包就行。建议装在项目独立的虚拟环境里避免和别的依赖打架。pip install chromadb装完之后可以确认一下版本是否正常import chromadb print(chromadb.__version__)然后初始化持久化客户端。我习惯把数据目录放在项目根目录下一个固定的子文件夹这样后面无论是备份、迁移还是排查问题路径都是明确的。client chromadb.PersistentClient(path./data/chroma)这一步看似简单但请务必想清楚一个问题这个目录就是整个短期记忆的“存盘点”。如果你的Agent部署在多台机器上或者容器每次重启都是新的文件系统那最终记忆会失效。本地方案适用于单机、单进程场景这是它的能力边界也是它的简单之处。3.2 创建Collection并写入第一批记忆创建collection时有几个参数值得认真对待。第一个是name建议起得有点含义比如agent_memory第二个是embedding_function默认值会自动下载一个小型的句子向量模型对中文场景效果普通但对“快速验证流程”来说完全够用第三个是metadata里的距离算法默认是L2我可以更推荐cosine因为语义相似度场景中cosine对文本长度不那么敏感。from chromadb.utils import embedding_functions embed_fn embedding_functions.DefaultEmbeddingFunction() collection client.get_or_create_collection( nameagent_memory, embedding_functionembed_fn, metadata{hnsw:space: cosine} )用get_or_create_collection而不是create_collection是因为前者天然幂等跑了第二次不会报“集合已存在”的错误这在开发期反复执行脚本时能省很多心。接下来写入记忆。每条记忆至少要有三个部分唯一id、文本内容、metadata元数据。id一定要保证唯一且稳定我常用的策略是“会话编号_角色_序号”这种组合例如s1_user_3既方便排查数据问题又能从id里直接看出这条记忆属于谁、扮演什么角色。collection.add( ids[s1_user_1, s1_assistant_1], documents[ 用户说我想找一款能离线使用的笔记软件, 助手说可以看一下本地优先的笔记应用比如支持端到端加密的那些 ], metadatas[ {session_id: s1, role: user, seq: 1, time: 1710000001}, {session_id: s1, role: assistant, seq: 2, time: 1710000005} ] )这里有个细节值得展开metadata里我加了session_id这是整个短期记忆架构的关键设计。没有这个字段所有会话的记忆会混在一起Agent给A用户提供服务时可能会把B用户的历史对话搜出来这在多用户场景下是严重的逻辑漏洞。有了这个字段查询时用where{session_id: s1}就能把检索范围严格控制在当前会话内。3.3 查询出来看看效果写入之后第一件事是做一个简单查询确认数据真的进去了检索行为符合预期。results collection.query( query_texts[有什么离线笔记工具], n_results2, where{session_id: s1} ) print(results[documents])看到的结果应该类似[ [ 用户说我想找一款能离线使用的笔记软件, 助手说可以看一下本地优先的笔记应用比如支持端到端加密的那些 ] ]可能有人会问为什么query_texts可以直接传原始文本不用自己先转向量因为Chroma内部会自动调用collection创建时指定的embedding_function来生成查询向量。如果你后续想手动控制embedding也可以用query_embeddings传入已经算好的向量两者只能二选一。初次跑通这个流程后我建议你先做一个小实验来“找手感”写几条不同话题的对话然后分别用不同措辞来查询看看返回结果的相关性差异。Chroma返回结果里还有一个distances字段它表示相似度距离数值越小说明越相近。我会很在意这个值的变化因为后面调优检索质量时它是量化判断的重要依据。提示如果results[documents]返回的结构让你觉得绕不用急这是嵌套列表外层对应你的每个查询词内层对应返回的n_results条结果。大多数情况下一条一条查询时取[0]这一层就行。4. 把向量库接进Agent短期记忆的完整回路4.1 记忆管理器的基本设计单独会调Chroma API不算什么真正有价值的是把它封装成Agent能用的记忆管理器。我习惯把短期记忆的能力收敛到一个类里对外只暴露两个方法remember用来记recall用来回忆。这样Agent主体逻辑里不用到处出现Chroma的调用细节以后想换存储后端也只改一个类。设计这个类的几个关键决策会话隔离初始化时传入session_id所有写入和查询都带这个过滤条件。id自增每条记录的id要单调递增避免重复也方便按顺序重建上下文。metadata统一每条记忆必须包含角色、会话、序号、时间这是后续过滤的基石。召回条数可调top_k参数暴露出来由Agent层根据当前问题复杂度决定。具体代码可以这么写import time import uuid import chromadb from chromadb.utils import embedding_functions class ShortTermMemory: def __init__(self, persist_dir./data/chroma, session_iddefault, collection_nameagent_memory, top_k4): self.client chromadb.PersistentClient(pathpersist_dir) self.session_id session_id self.embed_fn embedding_functions.DefaultEmbeddingFunction() self.collection self.client.get_or_create_collection( namecollection_name, embedding_functionself.embed_fn, metadata{hnsw:space: cosine} ) self.top_k top_k self.seq 0 def remember(self, role: str, content: str): self.seq 1 record_id f{self.session_id}_{role}_{self.seq}_{uuid.uuid4().hex[:8]} self.collection.add( ids[record_id], documents[content], metadatas[{ session_id: self.session_id, role: role, seq: self.seq, time: time.time() }] ) def recall(self, query: str, top_k: int None): k top_k or self.top_k results self.collection.query( query_texts[query], n_resultsk, where{session_id: self.session_id} ) if not results[documents] or not results[documents][0]: return [] return results[documents][0]这个封装里uuid.uuid4().hex[:8]是防止极端情况下同秒多条记录id冲突的保险。4.2 一个可以跑起来的带记忆Agent有了记忆管理器下一步就是把LLM调用和记忆循环串起来。这里我用一个伪llm_call函数代替真实模型调用实际使用时你可以接任意模型API或本地开源模型。核心流程分四步从记忆里召回相关内容组装成带有历史上下文的prompt调用模型得到回答把用户输入和模型回答都写回记忆。def llm_call(prompt: str) - str: # 实际使用时替换为你的模型调用 return 已收到你的需求我来安排。 def chat(): memory ShortTermMemory(session_ids1, top_k4) while True: user_input input(用户) if user_input.strip() exit: break history memory.recall(user_input) context_block \n.join(history) if history else 暂无相关历史 prompt f 请根据以下历史对话上下文回答用户的新问题。 历史对话 {context_block} 当前用户输入{user_input} reply llm_call(prompt) memory.remember(user, user_input) memory.remember(assistant, reply) print(f助手{reply}) if __name__ __main__: chat()跑通这个主循环后短期记忆的雏形就出来了。每次用户输入时Agent先“回想”相关历史再基于历史生成回答结束后把这段新经验存回“记忆库”。整个过程和人类对话时的状态非常接近——不是记住每一句话而是在必要时想起关键内容。这里有个容易被忽略的设计点写回记忆的动作发生在模型回答之后而且用户输入和助手输出都要写入。如果只记用户输入不记助手回答Agent下次回忆时看到的是一堆用户的独白缺少上下文另一半回答质量会大打折扣。4.3 短期记忆的几个工程化细节完整跑通之后想做得更扎实一点有几个工程细节我从实践里总结出来按优先级排列。第一个是记忆上限管理。短期记忆虽然比滑动窗口灵活但也不能无限增长。每条对话都写入几百轮下来向量库一样会膨胀检索速度也会下降。一个比较实用的策略是每个session最多保留最近N条记录比如300条超过后删除最旧的记录。实现方式也比较直接先查一下当前记录数再把seq最小的那批删掉。def evict_oldest(self, max_records: int 300): current self.collection.count() if current max_records: return # 取当前collection里所有记录的metadata data self.collection.get(where{session_id: self.session_id}) seqs sorted(data[metadatas], keylambda m: m[seq]) overflow current - max_records to_delete [item[seq] for item in seqs[:overflow]] # 根据seq构造id列表并删除第二个是时间维度上的分级。短期记忆的“短”除了数量限制还可以有时间维度。比如超过24小时的历史对话可能不再适合作为短期记忆召回这时候可以把时间戳作为过滤条件追加到where里。Chroma的where条件支持$gt这类比较操作符可以做到“只召回最近一天以内的内容”。第三个是总结压缩。当一段对话特别长但话题又高度集中时直接全文存向量会产生大量冗余检索时容易返回多条高度相似的记录挤占top_k名额。一种改进方案是定期把最近若干轮对话做一个摘要把摘要作为一条记忆存进去而不是一直让原始记录堆着。这个思路相当于在短期记忆和长期记忆之间加了一个“压缩层”我后来在项目里实际验证过对长会话的场景改善非常明显。注意短期记忆的落点不是“存得越多越好”而是“在prompt里放进去的有用信息越多越好”。检索回来的低相关内容稀释作用远比想象中大。如果发现模型回答变散先检查召回回来的历史里有没有大量无关内容。5. 实操里容易踩的坑和排查实录5.1 常见问题速查表做了几轮Demo之后我把最容易遇到的问题整理成了一份速查表你可以直接对照排查现象原因处理重启后数据全没了用了临时客户端或不同模块的持久化路径不一致统一使用PersistentClient并固定同一个path写入时报ID重复错误同批次id重复或复用了旧id用“会话_role_序号_随机后缀”保证唯一性查询返回空结果metadata过滤条件太严格或session_id不匹配先用不带where的查询验证数据是否存在再逐步加过滤检索结果明显不相关默认embedding模型对领域语言理解不足换用领域相关的embedding模型或调整top_k数据能查询但相似度分数异常距离函数选择不合适在collection metadata中指定cosine并重新构建集合多轮后响应越来越慢记忆库膨胀严重检索量太大实施记录数上限管理增加时间维度过滤容器部署后记忆不可用持久化目录是临时文件系统挂载持久化存储卷或将数据目录映射到稳定磁盘这里面最让我印象深刻的是第一个内存模式和持久化模式的区别。有一次我在测试脚本里用了chromadb.Client()数据一切正常然后换到另一个脚本里用PersistentClient查同一份数据发现查不到。排查半天才反应过来两个脚本各用了不同的client一个在内存里一个在磁盘上数据根本不在一处。后面我就强制自己只在项目入口处创建一次client再往各处传递避免这种低级问题。5.2 检索效果不好先别急着换模型遇到检索效果差时多数人的第一反应是换更大的embedding模型。这个方向不一定错但应该放在最后。我实际调试时的顺序是先检查数据本身再看过滤条件然后调召回参数最后才考虑换模型。先说数据层面最常出问题的是存入的文本太碎或太长。碎到一句话只有几个字语义信息太少怎么换模型都救不回来长到一段话1000多字向量只能表示一个笼统的语义中心检索到的段落可能有一半内容不相关。比较合理的实践是每条记忆控制在50到200字之间超出就切分。再说过滤条件where里字段名一定要和写入时metadata里的key完全一致。Chroma对metadata的过滤是精确匹配session_id和sessionId不通用值类型也要一致写的时候是字符串查的时候就不能传数字。然后调召回参数n_results并不是越大越好之前说过召回太多低相关内容反而会把有用信息的比重拖低。我建议从2开始往上试观察回答质量曲线的拐点多数场景4到6是比较舒服的值。最后再考虑embedding模型。默认的DefaultEmbeddingFunction对中文支持一般如果你发现换了好几个方案都不满意可以换成支持中文的本地向量模型比如bge系列用HuggingFace的sentence-transformers加载后传给collection即可。要注意embedding模型换了之前写入的向量就失效了需要重建collection并重新写入全部数据所以最好在项目初期就定下来。from chromadb.utils import embedding_functions # 使用本地中文embedding模型的写法示例 hf_embed_fn embedding_functions.HuggingFaceEmbeddingFunction( api_keyNone, model_nameBAAI/bge-small-zh-v1.5 )用这个embedding_function替换掉之前的DefaultEmbeddingFunction()再重建collection即可。提示embedding模型切换导致旧数据失效这个坑很隐蔽。我当时只换模型不重建数据查询结果一直很奇怪还以为是新模型有问题后来才发现是旧向量和新向量不在同一个语义空间相似度计算完全失真。写在实操之后这套“Chroma Agent”的组合我前后在几个不同项目里用过从最早的对话机器人到后来的需求分析Agent本质上都是同一套记忆管理逻辑。我的体会有两条。第一条是短期记忆真的不用一开始就上复杂架构向量库能存能查能过滤把这三件事跑顺就已经能解决80%的长对话失忆问题第二条是schema设计要趁早id怎么生成、metadata里放哪些字段、会话ID用什么格式这些一旦上线再改成本会成倍上升。最后分享一个小技巧开发阶段可以在recall结果里顺手把documents和distances一起打印出来通过相似度分数观察同一批数据不同查询词下的变化这对理解向量检索的行为非常有帮助甚至比直接看模型回答更直观。
