告别Agent失忆:基于Milvus向量库的生产级长期记忆模块设计
2026/9/18 3:15:41 网站建设 项目流程

我见过不少Agent项目最后都死在同一个地方:不是模型能力不够,不是工具链不完善,而是Agent聊了几轮之后,把几分钟前自己刚刚确认过的关键信息忘得一干二净。为了让上下文能塞进模型窗口,大家都习惯用截断来处理超长历史,但截断的本质是把记忆丢进碎纸机,问题并没有被解决,只是被推迟,而且埋得更深。这里说的“失忆”,不止是丢一句用户偏好,还包括丢工具调用结论、丢中间判断、丢已经验证过的边界条件。靠无脑加大上下文窗口也救不回来,我在这方面栽过跟头,后来把记忆模块从简单的对话截断重构为基于Milvus向量库的生产级方案,才算是真正把“记忆”这件事做扎实。这篇文章就把我的完整思路、代码、踩坑过程和选型理由写出来,给正在搭Agent的朋友一个可以直接落地的参考。

1. 截断是把记忆丢进碎纸机:朴素方案的代价

先聊聊大多数团队都在用的方案。对话历史一长,为了不超过上下文窗口上限,最常见的就是滑动窗口截断,也就是只保留最近N轮对话。这种做法在Demo阶段没问题,但一旦Agent需要跨多轮完成任务,问题就立刻暴露。比如用户在第3轮说过“我吃素”,到第19轮你安排餐厅推荐时,这已经是20轮之前的信息,早就被截掉了。用户会非常困惑:为什么刚说过的要求你转头就忘?这不是模型傻,是我们压根没把该记的东西交给模型。

1.1 截断方案的三种“失忆”症状

我自己总结过,朴素截断至少会带来三类非常典型的故障。

第一类是前置依赖丢失。Agent在中间某一步调用了工具,拿到了一个重要的中间结果,比如“当前服务器API只支持HTTP/2”,后面几轮要做方案设计时,结果已经被截掉。Agent只能重新去调工具,浪费Token不说,还可能因为拿到的状态不一致,给出冲突建议。

第二类是用户画像消失。用户对输出风格的偏好、回答语言的偏好、某些领域的擅长程度,往往是在对话早期建立起来的。截断把这些偏好删掉之后,Agent会在同一轮会话里出现前后风格不一致,一会儿详细一会儿简略,让用户怀疑背后不是同一个人设。

第三类是任务上下文闭环失败。多步骤任务里面,第5步往往依赖第1步产出的参数,比如“前面已经创建了项目test-ai,现在可以继续用这个项目ID”。如果第一步信息被截断,后续所有步骤都会陷入“重新寻找ID”的循环,工具调用反复失败,整个执行链基本就废了。

这三种情况我都在真实项目里见过,而且排查起来特别头痛,因为表面上看是模型能力问题,实际上是我们用截断强行抹掉了必要信息。

1.2 为什么“总结历史”也救不回来

有人会说:截断太粗暴,那我用LLM把历史对话总结成摘要,只放摘要进上下文不就行了?我试过,这条路也走不顺畅。

一方面,摘要本身是损失很大的压缩过程。你让模型总结“用户查询了三次商品价格,第二次显示库存不足”,摘要很可能只留下一句“用户查询商品价格后遇到库存问题”。等下一轮需要精确知道第二次查询时返回的具体错误码时,摘要里根本找不到。

另一方面,摘要会逐渐积累出幻觉风险。总结一次没问题,但滚动式多层总结后,最初的细节会越变越模糊,模型甚至会在摘要里补充一些原本不存在的“合理信息”,把整个记忆链污染掉。你根本说不清哪个结论是原话,哪个是摘要自己脑补的。

所以我的结论很明确:摘要适合做“提炼过的粗线索”,但不能替代原始记忆条目。真正生产级的Agent记忆,要做的是让每条记忆都保留原始语义、时间戳、来源类型,再通过检索按需取回,而不是一股脑把所有历史塞进窗口。

1.3 记忆和上下文不是一回事

这里有个认知要掰开:上下文和记忆是两个不同层级的东西。上下文是当前这一步模型能看到的“工作区”,工作区大小由窗口决定;记忆是Agent长期的“存档”,它可以很大很长,只是在需要用的时候按需加载到工作区里。

我们不需要让所有记忆都同时在上下文中,我们需要的是一个存储系统和一个检索机制,把和当前目标最相关的记忆挑出来,放进工作区。这跟人脑很相似,你不会随时回忆起所有事,但你做事时会主动想起相关经验。所以Agent记忆模块的实质,是解决“存什么、怎么存、怎么取”这三个问题。下一节我就从这几个问题入手做设计。

2. 动手之前,先想清生产级记忆模块到底要记什么

在设计具体方案前,我先列了一堆需求。生产级记忆模块和毕业论文里的玩具Demo不一样,它必须能支撑真实业务,条目可能上百万,查询延迟要可控,还要能应对多Agent实例共享的情况。因此我不能只做一个“把聊天记录存进向量库”的简单脚本,而是要把记忆分成不同类型,再针对不同类型设计存储结构。

2.1 记忆也不是一锅炖:情节、语义、程序性记忆

我把Agent记忆分为三类,这个分法参考了认知科学里常见的分类,但做了工程化改造。

第一类是情节记忆,记录具体发生过的事件,比如“用户在某年某月某日反馈过注册流程卡顿”“上一次工具调用返回了HTTP 500”。这类记忆的特点是带时间、带场景、带具体结果,适合用事件流的方式追加写入。

第二类是语义记忆,记录抽出来的通用事实,比如“用户倾向于先看结论再看细节”“用户使用的是Python技术栈”“项目的生产环境数据库是PostgreSQL”。这类记忆是长期稳定的偏好,跨任务、跨会话都有效。

第三类是程序性记忆,记录任务执行模式,比如“每次创建新服务前需要先检查命名空间是否已存在”“用户提出需求时,agent习惯先拆解成3个步骤再开始”。这类记忆往往从历史工具调用链中提取出来,对提升Agent的稳定性和执行力帮助最大。

在真正落地时,三类记忆可以共用一套存储结构,但写入来源和检索权重有所不同。情节记忆随对话实时写入,语义记忆需要定期从情节记忆中提炼,程序性记忆则由工具调用链结构生成。这个设计帮我后面省了很多事,因为存储层统一了,API设计也简单。

2.2 一条记忆条目应该长什么样

我最终给记忆条目设计了一组字段,包含内容本身和检索所需的元数据。核心字段如下,每个字段都是后来生产实践中逐步加上的。

字段类型说明例子
memory_idint64 / varchar全局唯一ID,推荐确定性生成,用来去重agent_ctx_hash_bucket
agent_idvarchar所属Agent实例标识,多实例隔离的最重要字段customer-support-v3
event_typevarchar记忆类型,情节/语义/程序性,对应枚举episodic, semantic, procedural
contentvarchar格式化后的文本内容,也是嵌入模型输入用户反馈:注册流程卡顿,重试3次才成功
vectorfloat_vectorcontent的嵌入向量1024维向量
sourcevarchar来源,如user/tool/llm_summarytool
tsint64事件发生时间戳,用于时间过滤与排序1720000000
expire_atint64过期时间,0表示永不过期1722600000
importancefloat重要性权重,用于召回重排0.9

这个表看起来平平无奇,但实战中有几个不容易注意的细节。一是agent_id必须作为标量过滤条件,否则多个Agent共用一个collection时,召回会互相污染。二是expire_at必须单独设计,因为向量库不像Redis那样天然支持TTL,应用层需要考虑过期清理。三是content不要直接存原始JSON日志,而是要存格式化过的、适合模型阅读和嵌入模型的文本,这个在一开始就做规范化,后续检索质量会好很多。

2.3 接口设计越小越好

我最后只保留了四个接口,编程上的抽象也尽量轻量。

class MemoryStore(Protocol): def write_memory(self, memory: MemoryItem) -> None: ... def recall(self, agent_id: str, query_text: str, top_k: int, filters: dict | None) -> list[MemoryItem]: ... def forget(self, agent_id: str, memory_ids: list[str]) -> None: ... def clean_expired(self) -> int: ...

为什么接口要这么精简?因为Agent项目的核心复杂度在于推理与工具调度,记忆模块如果做成一堆复杂API,开发根本用不顺手。小而清晰的接口意味着存储后端随时可以替换,今天用Milvus,明天想换成别的向量库,只要实现这4个方法就行。同时,Agent主程序和记忆存储解耦后,测试时我用本地内存实现,线上切到Milvus,一套逻辑跑两边,减少了环境依赖问题。

接口定完之后,才开始选存储引擎。这一段我做了不少对比,下一节详细说。

3. 存储选型:为什么最后落在Milvus上

Memory模块的存储选型,决定了下半年的运维体验,所以我花了不少时间对比。我把每个候选方案都画过简单原型,最后只留下了Milvus。下面说说我的判断依据和实际对比结果。

3.1 传统存储做向量检索是真的别扭

我一开始试过Redis和SQLite做记忆存储,并不是因为它们不能存,而是因为“语义检索”这个核心需求它们给不了。

用Redis,我能想到的方案是把对话按Key存成ZSet,按时间戳做范围查询。但这样只能做关键词匹配或严格顺序回放,做不到“和当前意图相似”的召回。比如用户说“上次那个权限问题后来怎么解决的”,你如果只按关键词“权限”去精确匹配Redis里的原始文本,可能匹配到完全无关的权限讨论,也可能漏掉一条表述不同但语义高度相关的记录。

用SQLite加全文索引,也类似。FTS能解决关键词匹配,但中文分词的piece问题会让召回质量非常不稳定。而且随着记忆量增长,SQLite单机的性能和并发能力很快成为瓶颈。用pgvector这样的PostgreSQL扩展会好一些,毕竟它是正经数据库,但我在项目里还需要处理亿级以上的向量数据、持续写入和按Agent隔离的标量过滤,单机PostgreSQL的容量规划和运维成本也要认真对待。

3.2 Milvus的哪个点真正打动人

我最终选Milvus,倒不是因为它在“向量检索速度排行”上一定第一名,而是因为它把“向量检索”和“标量过滤”这两个能力结合得比较顺手。

Milvus里的核心概念其实和数据库很像: - collection = 表 - entity = 一条记录 - field = 字段 - partition = 按某个字段做物理分区 - index = 针对向量的ANN索引 - consistency = 读写一致性级别

用这套模型,我可以把Agent记忆模型直接映射成一张表:每一条记忆是一个entity,包含agent_id这样的标量字段和vector这样的向量字段。检索时,我可以在同一个search请求里指定filter="agent_id == 'customer-support-v3'",先按Agent隔离,再在隔离结果里做向量近邻搜索,这个能力太关键了。

对比之下,纯向量数据库如果标量过滤能力弱,我就得把每个Agent拆成一个独立collection,等Agent数量多了,运维直接爆炸。Milvus的“一个collection + 标量过滤 + 分区”模式,让我用一套资源管理所有Agent,却能在数据层做隔离,这对我来说就是最直接的收益。

3.3 跑起来成本没有那么高

有些团队听说Milvus就觉得要上Kubernetes,有点被吓到。实际上从单机版开始完全够用,一个Docker Compose就能把Milvus Standalone和它的依赖等组件一起拉起来,适合开发测试,也适合中小规模的Agent业务。

我在开发环境起服务的步骤非常简单,核心流程如下:

  1. 准备好docker-compose.yml,定义milvus-standalone服务,暴露19530端口给客户端,暴露9091端口给监控。
  2. 额外挂载etcd和MinIO作为元数据和对象存储依赖,这是Milvus启动的必要组件。
  3. 执行docker compose up -d,等待容器健康后,用pymilvus连一下http://localhost:19530即可。

Windows和macOS上,只要Docker Desktop配置够用,基本没什么差别。我平时在Windows开发机上也这样跑,没有遇到诡异问题。唯一建议是给Docker多分配点内存,至少8GB,因为索引构建会吃内存。

跑起来之后,我逐步把记忆模块的代码补全,接下来进入核心实现部分。

4. 核心实现:从建表到与Agent循环对接

这节我直接给出完整的代码思路,你可以把它当模板用。我使用的是当前新版pymilvus的MilvusClient接口,整体写法比早期版本简洁很多。

4.1 创建Collection的Schema

我的collection名称叫agent_memory,核心字段和前一节的表一致。创建schema之前,最好先确认好向量维度和VARCHAR长度,因为后期修改字段类型在Milvus里比较麻烦。

from pymilvus import MilvusClient, DataType client = MilvusClient( uri="http://localhost:19530", token="root:Milvus" ) schema = client.create_schema(auto_id=False, enable_dynamic_field=False) schema.add_field(field_name="memory_id", datatype=DataType.INT64, is_primary=True) schema.add_field(field_name="agent_id", datatype=DataType.VARCHAR, max_length=256) schema.add_field(field_name="event_type", datatype=DataType.VARCHAR, max_length=32) schema.add_field(field_name="content", datatype=DataType.VARCHAR, max_length=8192) schema.add_field(field_name="vector", datatype=DataType.FLOAT_VECTOR, dim=1024) schema.add_field(field_name="source", datatype=DataType.VARCHAR, max_length=32) schema.add_field(field_name="ts", datatype=DataType.INT64) schema.add_field(field_name="expire_at", datatype=DataType.INT64) schema.add_field(field_name="importance", datatype=DataType.FLOAT) index_params = client.prepare_index_params() index_params.add_index( field_name="vector", index_type="HNSW", metric_type="IP", params={"M": 16, "efConstruction": 256} ) client.create_collection( collection_name="agent_memory", schema=schema, index_params=index_params, consistency_level="Bounded" )

这里有几个点要特别说明。第一,memory_id我故意没有用auto_id=True,而是自己生成确定性ID,这是为了后面做写入去重,这个坑我后面会详聊。第二,metric_type我用了IP内积相似度,使用前提是写入的向量都要做归一化,这样才能保证内积等价于余弦相似度。第三,consistency_level我建议用Bounded而不是默认的Strong,因为Agent记忆场景对毫秒级一致性的要求没那么高,Bounded能明显降低写入延迟。

4.2 写入路径:把事件变成可召回的记忆

存储结构搭好之后,下面写写入函数。核心流程是:拿到原始事件,格式化成文本,生成向量,和元数据一起插入Milvus。

import hashlib import time import numpy as np from sentence_transformers import SentenceTransformer embedder = SentenceTransformer("BAAI/bge-m3") def gen_memory_id(agent_id: str, content: str, bucket_seconds: int = 600) -> int: bucket = int(time.time()) // bucket_seconds raw = f"{agent_id}|{content}|{bucket}".encode("utf-8") return int(hashlib.sha256(raw).hexdigest()[:16], 16) def write_memory(agent_id: str, event_type: str, content: str, source: str = "user", importance: float = 0.5, expire_at: int = 0) -> int: vec = embedder.encode(content, normalize_embeddings=True).astype(np.float32) memory_id = gen_memory_id(agent_id, content) client.upsert( collection_name="agent_memory", data=[{ "memory_id": memory_id, "agent_id": agent_id, "event_type": event_type, "content": content, "vector": vec.tolist(), "source": source, "ts": int(time.time()), "expire_at": expire_at, "importance": importance, }], ) return memory_id

这里的工程细节比代码本身更重要。

一是normalize_embeddings=True必须打开,否则你后面用IP内积算相似度,结果会非常失真。二是gen_memory_id里我把内容哈希和时间桶结合,这样短时间内同一条内容重复写入时,memory_id不变,配合upsert就能天然去重。三是这里没有直接同步等待Milvus刷盘,Milvus默认是异步写入架构,对Agent场景完全够用。

4.3 召回路径:混合搜索加上标量过滤

召回是记忆模块的生命线,写得好不好直接影响Agent能不能“想起来”。我的实现思路是:把当前用户问题或当前Agent目标作为query文本,转成向量后在collection内做向量搜索,同时加上agent_id过滤和时间范围过滤。

def recall(agent_id: str, query_text: str, top_k: int = 8, event_type: str | None = None) -> list[dict]: query_vec = embedder.encode(query_text, normalize_embeddings=True).astype(np.float32) filter_expr = f"agent_id == '{agent_id}'" if event_type: filter_expr += f" and event_type == '{event_type}'" results = client.search( collection_name="agent_memory", data=[query_vec.tolist()], filter=filter_expr, limit=top_k, output_fields=["content", "event_type", "source", "ts", "importance"], ) hits = [] for hit in results[0]: hits.append({ "content": hit["entity"]["content"], "event_type": hit["entity"]["event_type"], "source": hit["entity"]["source"], "ts": hit["entity"]["ts"], "score": hit["distance"], }) # 按重要性和时间做一次简单加权 hits.sort(key=lambda x: x["score"] * 0.7 + min(x["ts"] / 1e9, 1.0) * 0.3, reverse=True) return hits

召回时有个容易被忽略的点:Milvus的search默认按向量相似度排序,但Agent记忆场景里,单看相似度不够,还要考虑内容的重要性和时效性。我在这里做的加权排序比较粗糙,但已经能解决“相似但不重要”的干扰问题。如果你的场景更复杂,可以考虑加一个rerank模型,把top_k先放大到20,再对20条用LLM或cross-encoder精排,效果会更好。

4.4 和Agent循环对接:把记忆注入上下文

写好了存储和召回,最后一步是把记忆模块挂到Agent主循环里。我常用的是比较朴素的注入方式:每次Agent准备调用LLM生成回复前,先把当前目标作为query,召回若干条记忆,再放进系统提示词的固定区块。

def build_prompt_with_memory(agent_id: str, user_query: str, base_prompt: str) -> str: memory_hits = recall(agent_id, user_query, top_k=6) memory_block = "\n".join([ f"- [{item['source']}] {item['content']}" for item in memory_hits ]) return f"{base_prompt}\n\n# 可参考的历史记忆\n{memory_block}\n\n# 当前用户问题\n{user_query}"

这一步看似简单,却是整个记忆模块最重要的产品化动作。记忆注入太多,会让模型困惑;注入太少,又起不到参考作用。我经验是top_k在4到8之间效果较好,超过10条后模型容易开始关注那些不太相关的边角记忆。另外,记忆区要用明确的标题和列表分隔开,模型才能分辨哪些是“历史资料”,哪些是当前的直接指令。

5. 嵌入模型和索引参数:精度与速度的平衡

很多人在记忆模块里只关注向量库,却忽略了嵌入模型和索引配置对最终效果的决定性影响。实际上,召回质量的上下限,一半在嵌入模型,一半在索引和查询参数的配合。

5.1 嵌入模型怎么选

我在这套模块里用的是开源模型BAAI/bge-m3,主要原因是它对中文和英文都支持得比较好,生成的向量维度是1024,对Milvus来说完全能接受,而且可以用sentence-transformers一行加载,部署成本很低。

如果你对中文场景的精度要求特别高,可以考虑bge-large-zh系列,但向量维度更高,召回速度和存储成本上涨明显。如果团队有预算且需要多语言,也可以换云端embedding服务,但云端API的延迟和调用成本在记忆模块这种高频写入场景里会非常扎眼,我建议先用开源模型跑通,再根据业务压力做替换。

嵌入模型最容易被忽略的一件事是版本锁定。一旦模型文件更新,同一句话的向量分布可能变化,之前存的所有历史向量和新向量就会存在“语义偏移”,召回质量明显下降。我后来做了个简单机制:在collection的description里写入嵌入模型名称和版本号,每次部署调度时校验,一旦发现版本不一致,就触发历史向量重建。

5.2 向量归一化与相似度度量

这部分必须单独强调,因为太多人在这里踩坑。我最终用的是IP内积,但使用内积有一个硬性前提,就是向量必须归一化。

vec = embedder.encode(content, normalize_embeddings=True)

不加这行,内积计算出来的分数会被向量模长干扰,有些长文本天然分高,导致召回偏向于“长度更大的记忆”而不一定是“语义最相关的记忆”。用COSINE距离虽然能在API层面规避一部分问题,但部分版本的索引在COSINE度量下会额外做归一化,提前自己normalize_embeddings总没有坏处,还能让结果更可解释。

5.3 HNSW参数怎么调

Milvus里最常用的索引是HNSW,我用在Agent记忆场景下的参数组合是M=16efConstruction=256,查询时设置ef=64

M控制每个节点的最大连接数,M越大,图越密集,召回越准,但内存和构图时间也越大。efConstruction控制建图时的搜索宽度,太大建图慢,太小索引质量差。我给出的组合是中庸配置,适合几百万到几千万条记忆的规模。如果你的记忆量特别大,超过一亿条,可以考虑IVF_FLATSCANN,但参数调优会更麻烦,前期我不建议一上来就用。

搜索时的ef查询参数也是关键,在client.search()search_params里传{"ef": 64},可以控制每次搜索的候选集大小。值越大,召回越精确,但延迟越高。我把64作为默认值,实际在线峰值时单次查询也就几毫秒,完全够用。如果客服类Agent对延迟要求极高,可以降到32,召回损失通常不是很明显。

6. 生产硬化:TTL、缓存和两层记忆架构

代码能跑起来和能上生产是两码事。我在把模块推到线上前,又做了几个层面的加固,包括短期缓冲、过期清理、幂等去重。这些操作不是炫技,而是把“偶尔失灵”变成“稳定可用”的关键。

6.1 短期缓冲加长期存储,搭配更合理

一次对话过程中,可能有非常多的小事件,如果每条都立刻写向量库,既浪费资源又增加查询延迟。我的思路是做一个“短期事件缓冲”层,在Agent进程内维护一个最近的事件列表,当列表达到一定长度或任务结束时,再批量写入Milvus形成长期记忆。

这个模式很像缓存加数据库的分层结构。短期缓冲里,Agent可以直接访问最近N条上下文,不需要做向量检索;而recall接口只负责长期记忆的查询。这样既解决了上下文窗口内的即时性,又保证了跨会话的持久性。实践中,我会在session开始时把缓冲清空,session结束时统一写入,这样记忆的边界非常清晰。

6.2 幂等写入和事件去重

Agent在高并发或工具频繁调用时,很容易产生重复记忆。例如一个工具被自动重试了3次,如果你不处理,系统里会留下3条几乎一样的记录。我用确定性memory_idupsert方法,从存储层就拦截了这种冗余。

去重之外,还要注意批量写入时的性能。我一般用client.upsert一次传入几十条数据,而不是一条一条循环调,这样写入速度会快很多。Milvus的批量写入能力很强,单个请求里带上千条也不是问题,但过大的批次会导致单次请求耗时变长,我会控制在200条以内,兼顾速度和稳定性。

6.3 TTL过期清理的正确姿势

长期记忆如果不清理,会随着时间积累越来越多,既占用存储,又会把召回结果“污染”了。有些记忆是临时的,比如“这几天用户正在测试某个新功能”,过几天就没价值了。我给这类记忆设置expire_at字段,并设计了一个后台清理任务。

Milvus新版collection本身支持TTL属性,可以在建表时设置collection.properties里的collection.ttl.seconds。但为了防止某个环境不支持,我在应用层做兜底,定时执行:

def clean_expired(client: MilvusClient, collection_name: str) -> int: now = int(time.time()) res = client.delete( collection_name=collection_name, filter=f"expire_at > 0 and expire_at < {now}", ) return res["delete_count"]

这种清理任务一般放在定时调度里,每天凌晨跑一次即可。注意别在Agent在线高峰期做全量清理,删除操作也会占用一定资源,如果数据量大,可能影响在线查询。

7. 真实排坑记录:这些坑我都是花了不少时间才爬出来

最后一部分,我把实际排障过程中遇到的问题集中整理出来。这些坑在官方文档里通常不会被专门写出来,但遇到时真的会折磨人。

7.1 召回结果全是最近的记忆,相关记忆反而排不上来

这个现象我排查了很久,最后定位到两个原因。第一个原因是filter表达式写错了。比如直接写成f"agent_id == {agent_id}",agent_id如果是字符串,就必须加引号,否则表达式不合法,Milvus通常会报错,但某些旧版本会静默退化成全部扫描,结果可想而知。第二个原因是output_fields里没把importance返回出来,我在排序时拿不到这个字段,就会出现“最相关但不够新”的记忆被挤出top_k。

解决办法也很简单:先确认filter表达式格式,再用client.query单独验证标量条件,最后再把注意力放到排序策略上。排查顺序切记不要反了。

7.2 VARCHAR长度设置太短导致内容被截断

记忆条目的content并不是一条固定长度的数据,工具调用结果可能非常长。我第一次建表时图省事,把contentmax_length设成了512,结果上线后经常发现有些记忆内容变成了一堆被截断的半截话,Agent拿到这些半截信息后,理解经常跑偏。

Milvus里VARCHAR字段长度在collection创建后就不好改了,所以我后来重新建了collection,一次性把max_length调到8192。建议你在设计阶段就预留充足空间,不要省这几个字节。如果确实遇到超长内容,应该在写入前做切分,先把一条长记忆拆成多个chunk,再分别写入,并保证每个chunk都带有同一批元数据。

7.3 多Agent实例并发写入的归属问题

当同一个Agent服务被部署成多个实例时,如果所有实例共享同一个Milvus collection,忘记在写入时设置agent_id就麻烦了。我出现过一次事故:测试环境多条记忆没有归属,结果生产Agent查询时把这些脏数据也带了出来。

我的防御措施包括:所有写入方法强制要求agent_id,如果为空直接抛异常;同时在collection级别增加一个约定,不允许插入空agent_id的数据。另外,还会用client.query定期抽查每个agent的记忆条数,发现异常立刻报警。

7.4 备份时别忘了带上嵌入模型

这个坑最隐蔽。有一次我误删了collection,准备从备份恢复数据。数据是恢复了,但嵌入模型版本恰好也更新过,导致旧向量和新向量之间映射关系对不上,整个记忆库几乎等于作废。

从那以后,我的备份策略分两层:一是Milvus数据本身的备份,比如用milvus_backup工具导出的文件;二是模型版本说明和代码版本号的归档。恢复时先确保模型版本完全一致,再恢复向量数据。这套流程虽然多了一步,但能避免把整个记忆库变成一堆没有意义的浮点数。

另外,Milvus里废弃的collection不用急着删除,可以先改名为_deprecated_xxx,保留一段时间,等确认新方案稳定后再物理删除。这种操作成本很低,却能给恢复留一条后路。

整套记忆模块跑到现在,我最深的体会是:Agent的长期记忆不在于“记住全部”,而在于“在正确的时候取出正确的那一小部分”。截断是最省事的方案,但也是最容易在关键时候掉链子的方案。把记忆存储和检索交给Milvus之后,Agent不再每轮都像一个刚入职就失忆的新人,它开始从历史里吸取经验,这种变化在产品体验上是能明显感受到的。如果各位正在被Agent失忆问题折腾,我推荐按照这套思路从设计记忆条目开始,一步步替换掉截断逻辑,你会发现排查起各种灵异问题都顺了很多。

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

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

立即咨询