Agent记忆实现：从InMemory到Milvus的完整演进

前段时间我在做一个带长期对话能力的 Agent 项目碰到一个特别典型的问题模型每次都是“冷启动”用户上一轮刚说过的偏好这一轮又要重新问一遍用户昨天交代过的关键信息今天再聊时模型完全没反应。说白了大模型本身是不带记忆的所有的智能都依赖于当下这个请求里塞了多少上下文。而 Agent 要真正像一个“助手”而不是“接口”Memory 模块是绕不开的一块硬骨头。这篇文章就围绕我给 Agent 装记忆的完整过程来写。我会从最简单的 InMemory 内存实现开始先让流程跑通再落到文件持久化解决“进程重启就失忆”的问题最后升级到 Milvus 向量数据库实现跨会话的语义召回。整个过程我会把每一步的设计思路、代码实现、踩过的坑都摊开讲适合正在做 Agent 开发、想在项目里接入记忆能力的同学参考。不管你之前有没有接触过向量数据库照着做就能跑起来。1. Agent Memory 到底在解决什么问题1.1 大模型天生没有记忆先说本质问题大模型 API 是一个无状态函数。你传一段输入它给你一段输出仅此而已。它不会记得五分钟前你问过什么更不会记得昨天你让它记过什么。这就像一个每天上班都换一个新同事的团队你每次对接的人都对你的项目一无所知你必须把背景从头讲一遍。有人会说那我直接把历史对话拼进 prompt 不就行了确实可以但有两个硬限制。第一个是上下文窗口有限几百 K 的上下文也不可能无限塞第二个是无效信息会稀释注意力你塞进去 10 万 token 的历史记录模型找关键信息的难度反而会变大回答质量会下降响应延迟和成本也会一起涨。所以 Memory 模块的核心职责不是“存储”本身而是“在合适的时候把合适的记忆放回上下文里”。存储只是手段检索和组装上下文才是目的。想清楚这一点后面每一步设计都不会走偏。1.2 Memory 的三种类型与分工实际工程里Memory 不是单一组件而是分层的。我习惯把它拆成三种类型生命周期典型实现作用工作记忆短期当前会话InMemory、滑动窗口维持多轮对话的连贯性长期记忆情景跨会话持久文件、数据库、向量库记住用户偏好和历史结论语义记忆知识长期共享向量检索、知识库按语义召回相关背景信息这三个层次不是互相替代的关系而是配合使用。工作记忆解决的是“眼下聊到哪了”长期记忆解决的是“这个人以前说过什么”语义记忆解决的是“和当前问题最相关的背景知识是什么”。我后面要讲的从 InMemory 到文件到 Milvus 的演进本质上就是一层一层补齐这几种记忆能力。1.3 Memory 模块在 Agent 中的位置在 Agent 的典型架构里Memory 模块通常不是独立服务而是编排层的一部分。LLM 是大脑工具是手脚Memory 则是大脑的“海马体”。具体来说在每次调用 LLM 之前Agent 会先走一遍“记忆检索”流程从短期缓存里取最近的对话从长期存储里按当前 query 召回相关记忆然后把这两部分拼进 system prompt 或者作为额外的 context 注入。LLM 生成回复之后又会把这一段交互写入记忆系统。这个“先读后写”的流程就是 Memory 模块存在的意义。2. 第一层InMemory 内存实现先让流程跑起来2.1 一个最简 Memory 类的写法很多 Agent 项目一开始根本不考虑持久化先用一个列表把历史消息存下来。这个阶段最重要的不是“存得久”而是“接口要清晰”。我见过不少项目在早期随手写一堆全局变量来存对话结果后面重构的时候到处找引用非常痛苦。我的建议是哪怕只是内存存储也把它封装成一个独立类定义好 add、get_recent、build_context 这几个方法。这样后面替换成文件存储或者向量库存储时上层代码几乎不用动。一个最小实现长这样from typing import List, Dict from dataclasses import dataclass, field import time dataclass class Message: role: str content: str timestamp: float field(default_factorytime.time) class InMemoryMemory: def __init__(self, max_messages: int 20): self.messages: List[Message] [] self.max_messages max_messages def add(self, role: str, content: str) - None: self.messages.append(Message(rolerole, contentcontent)) if len(self.messages) self.max_messages: self.messages self.messages[-self.max_messages:] def get_recent(self, n: int 10) - List[Message]: return self.messages[-n:] def build_context(self, n: int 10) - str: recent self.get_recent(n) return \n.join(f{msg.role}: {msg.content} for msg in recent)这里有两个设计细节值得说。第一我用timestamp字段给每条消息打点这个字段在内存阶段看似没用但后面做文件持久化、按时间过滤时都用得上。尽早把时间维度纳入数据结构后期会省很多事。第二max_messages用滑动窗口裁剪超出窗口就把最早的消息丢掉。这里要注意裁剪时保留的是末尾的最近 N 条因为新消息永远追加在列表末尾。2.2 什么时候 InMemory 是够用的不是所有场景都需要上向量数据库。如果你的 Agent 只做单轮工具调用、或者对话轮次很少、或者用户压根就没有跨会话恢复上下文的需求那 InMemory 完全够用。我记得早期做一个“表格问答 Agent”时用户只会针对一张表连续问三到五个问题问完就结束不存在“下次再来”的场景。这个阶段用 InMemory 不仅简单而且响应快连序列化开销都省了。另外很多 Agent 框架本身也提供了内置的会话记忆实现。比如 LangChain 的ConversationBufferMemory就是典型的内存窗口实现它的作用范围也仅限于当前进程内的会话管理。如果你的项目能接受“重启即失忆、多实例不共享”这两个前提那直接用框架自带的内存实现就够了没必要自己造轮子。2.3 InMemory 的硬伤InMemory 最大的问题有三个。第一个是进程重启后记忆全丢。开发环境里无所谓但生产环境只要部署实例一滚动更新所有会话状态就归零了用户体验直接退回解放前。第二个是多实例部署时数据不共享。你有三个 Agent 实例在跑负载均衡用户第一次请求打到实例 A第二次请求打到实例 BB 手里没有 A 存的记忆于是又变成“冷启动”。这个问题不解决InMemory 永远只能活在单机演示环境里。第三个是上下文膨胀。滑动窗口只限制条数不限制 token。如果某条消息内容特别长20 条消息可能就能顶爆一次请求的上下文。后面真正做长期记忆时一定要考虑按 token 而不是按条数来裁剪。3. 第二层文件落地把记忆沉淀下来3.1 用 JSONL 做追加式持久化既然内存靠不住那就落盘。最简单的持久化方式是把对话按行追加到 JSONL 文件里。JSONL 比 JSON 好在它是 append-only 的每写一行都是一次独立操作即使中途崩溃最多丢最后一行不会整个文件损坏。我在项目里一般按 conversation_id 分文件存储一个会话一个文件。实现起来非常直接import json import time from pathlib import Path from typing import List, Dict class FileMemory: def __init__(self, storage_dir: str ./memory_store): self.storage_dir Path(storage_dir) self.storage_dir.mkdir(parentsTrue, exist_okTrue) def _path_for(self, conversation_id: str) - Path: safe_name .join(c for c in conversation_id if c.isalnum() or c in -_) return self.storage_dir / f{safe_name}.jsonl def append(self, conversation_id: str, role: str, content: str) - None: record { role: role, content: content, timestamp: time.time(), } path self._path_for(conversation_id) with open(path, a, encodingutf-8) as f: f.write(json.dumps(record, ensure_asciiFalse) \n) def load(self, conversation_id: str, limit: int 20) - List[Dict]: path self._path_for(conversation_id) if not path.exists(): return [] records [] with open(path, r, encodingutf-8) as f: for line in f: line line.strip() if line: records.append(json.loads(line)) return records[-limit:]这里有两个地方是实践后才知道要处理的。第一是文件名转义。conversation_id 如果由用户传入可能带/或..直接拼到路径里会有安全风险所以要做一次字符白名单过滤。第二是读取时对空行做防御文件尾部可能因为上次写入中断而残留半截行json.loads会直接抛异常。3.2 用 SQLite 替代散文件JSONL 文件用了一段时间你会发现两个痛点。一是文件数量越来越多目录管理变得麻烦二是按条件检索非常弱比如要查“某个用户今天的所有对话”你得遍历所有文件再做字符串匹配。这个阶段我推荐换成 SQLite。它本质还是单文件存储不需要额外部署服务但提供了 SQL 能力事务、索引、条件查询都齐了。一个简单的表结构就够了CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, conversation_id TEXT NOT NULL, user_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL, created_at REAL NOT NULL ); CREATE INDEX IF NOT EXISTS idx_conversation ON messages(conversation_id, created_at); CREATE INDEX IF NOT EXISTS idx_user ON messages(user_id, created_at);用 SQLite 之后按会话查、按用户查、按时间范围查都变成了普通的 SELECT 语句逻辑清晰很多。Python 里直接用内置的sqlite3模块就行不需要引入 ORM。3.3 文件存储解决不了“语义召回”文件存储解决了“持久化”问题但没有解决“召回”问题。它只是在记录层面做到了不丢数据但检索方式仍然停留在“按会话 id 精确匹配”或者“关键词 like 匹配”这个级别。什么叫语义召回举个例子。用户周一在对话里说过“我平时通勤单程大概四十分钟”周五他问“帮我估算下每周通勤时间”。用关键词检索根本匹配不上因为“通勤四十分钟”和“每周通勤时间”之间没有一个共同的关键词但语义上它们强相关。这就是必须要引入向量数据库的原因。文件存储负责“把记忆记下来”向量检索负责“把相关记忆找回来”。两者不是替代关系而是互补关系。我后来的方案就是SQLite 保留原始记录Milvus 存向量索引两边用同一条主键关联。4. 第三层上 Milvus 向量数据库让记忆可召回4.1 为什么要向量化存储向量化存储的核心思路是把一段文本转换成一组高维浮点数用余弦相似度或内积来衡量文本之间的语义距离。相似的内容在向量空间里离得近不相关的内容离得远。这样我们就能做到“我不管字面上是否包含同一个词只要语义相关就能召回”。这个阶段要先解决 embedding 模型选型。我在实际项目里试过几种OpenAI 的text-embedding-3-small效果好、维度适中适合快速验证但如果要本地部署、不想引入外部依赖可以用 BGE-M3 或者bge-large-zh-v1.5对中文支持也不错。维度决定了后面 Milvus 里 collection 的向量字段长度所以先定 embedding 模型再定 collection schema。4.2 Milvus 单机版安装与启动Milvus 是一个云原生向量数据库生产环境一般用集群模式但本地开发和中小项目直接用 Docker Compose 启动单机版就够了。官方推荐的单机版部署包含三个组件Milvus 主服务、etcd元数据存储、MinIO对象存储。用 Docker Compose 一次性启动version: 3.5 services: etcd: container_name: milvus-etcd image: quay.io/coreos/etcd:v3.5.5 environment: - ETCD_AUTO_COMPACTION_MODErevision - ETCD_AUTO_COMPACTION_RETENTION1000 - ETCD_QUOTA_BACKEND_BYTES4294967296 volumes: - ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/etcd:/etcd command: etcd -advertise-client-urlshttp://etcd:2379 -listen-client-urls http://0.0.0.0:2379 --data-dir /etcd minio: container_name: milvus-minio image: minio/minio:RELEASE.2023-03-20T20-16-18Z environment: MINIO_ACCESS_KEY: minioadmin MINIO_SECRET_KEY: minioadmin volumes: - ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/minio:/minio_data command: minio server /minio_data healthcheck: test: [CMD, curl, -f, http://localhost:9000/minio/health/live] interval: 30s timeout: 20s retries: 3 standalone: container_name: milvus-standalone image: milvusdb/milvus:v2.4.4 command: [milvus, run, standalone] environment: ETCD_ENDPOINTS: etcd:2379 MINIO_ADDRESS: minio:9000 ports: - 19530:19530 - 9091:9091 volumes: - ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/milvus:/var/lib/milvus depends_on: - etcd - minio把这段保存为docker-compose.yml然后在同目录执行docker compose up -d。等几个容器都变成 healthy 状态后再检查一下端口docker compose ps正常会看到 standalone 容器监听 19530 端口。Milvus 的客户端默认用 19530 做 gRPC 通信9091 是健康检查端口。启动失败时重点看docker compose logs standalone的输出大部分问题都出在 etcd 或 MinIO 没起来导致依赖失败。4.3 创建 Collection字段、主键、索引连上 Milvus 之后第一步是定义 collection 结构。一个 collection 就像关系数据库里的表每个字段都要明确类型。我在这里存的是记忆片段所以至少需要这几个字段id主键自增user_id用户标识用于多租户隔离conversation_id来源会话方便回溯text原始记忆文本embedding向量字段使用 pymilvus 创建 collection 的代码如下from pymilvus import ( connections, Collection, CollectionSchema, FieldSchema, DataType, utility, ) connections.connect(hostlocalhost, port19530) COLLECTION_NAME agent_memory DIM 1024 # 由 embedding 模型决定 # text-embedding-3-small 是 1536 维 # bge-large-zh-v1.5 是 1024 维 fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(nameuser_id, dtypeDataType.VARCHAR, max_length64), FieldSchema(nameconversation_id, dtypeDataType.VARCHAR, max_length64), FieldSchema(nametext, dtypeDataType.VARCHAR, max_length65535), FieldSchema(nameembedding, dtypeDataType.FLOAT_VECTOR, dimDIM), ] schema CollectionSchema(fieldsfields, descriptionAgent memory store) if not utility.has_collection(COLLECTION_NAME): collection Collection(nameCOLLECTION_NAME, schemaschema) index_params { index_type: HNSW, metric_type: COSINE, params: {M: 16, efConstruction: 200}, } collection.create_index(field_nameembedding, index_paramsindex_params)这里两个参数值得展开。第一metric_type选了COSINE而不是IP内积。如果 embedding 做了归一化两者结果一样但 COSINE 的分数对文本长度不敏感召回结果更稳定我建议默认用 COSINE。第二HNSW 的M和efConstruction是精度和索引构建速度的平衡M越大召回越准但内存占用越高efConstruction越大索引构建越慢但查询精度越高。开发环境用M16, efConstruction200足够数据量上了千万后再调参。4.4 写入与检索一段可复用的代码Collection 创建好之后接下来就是写入和检索这是整个 Memory 模块最核心的两条路径。写入路径把一段文本做 embedding然后 insert 进 collection。from pymilvus import Collection def get_embedding(text: str) - list[float]: # 这里是示意实际项目根据自己的 embedding 方案实现 # 可以用 OpenAI SDK也可以用本地 bge 模型 import openai client openai.OpenAI() resp client.embeddings.create( modeltext-embedding-3-small, inputtext, ) return resp.data[0].embedding def add_memory(user_id: str, conversation_id: str, text: str): collection Collection(agent_memory) embedding get_embedding(text) collection.insert([ [user_id], [conversation_id], [text], [embedding], ]) def search_memory(user_id: str, query: str, top_k: int 5): collection Collection(agent_memory) collection.load() query_embedding get_embedding(query) results collection.search( data[query_embedding], anns_fieldembedding, param{metric_type: COSINE, params: {ef: 64}}, limittop_k, exprfuser_id {user_id}, output_fields[text, conversation_id, user_id], ) hits [] for hit in results[0]: if hit.distance 0.7: # 相关度阈值低于这个分数视为无关 continue hits.append({ text: hit.entity.get(text), conversation_id: hit.entity.get(conversation_id), score: hit.distance, }) return hits检索路径里有几个细节容易踩坑。第一个是collection.load()。Collection 创建后默认未加载到内存必须在 search 前执行 load。但频繁 load/unload 会影响性能实际项目一般在服务启动时统一 load不要每条请求都 load 一次。第二个是expr参数。我用user_id xxx做粗过滤只搜当前用户的记忆。这比全局检索后过滤效率高得多也是多租户隔离的基础。注意 expr 里的字符串值要加双引号而且字段类型是 VARCHAR值最好做一次转义。第三个是相似度阈值。Milvus 返回的 distance 在 COSINE 下是余弦相似度范围在 -1 到 1 之间越大越相关。0.7 这个阈值是我根据项目数据调出来的经验值你可以根据自己的场景调整。如果阈值太高召回结果太少太低无关内容会污染 prompt。我习惯先把检索结果打印出来看几轮再定阈值。4.5 为什么最终选了 Milvus 而不是其他向量库在写进 Milvus 之前我还试过 Chroma 和 FAISS。Chroma 上手极快Python API 很友好适合做原型FAISS 只是一个索引库没有服务端需要自己处理持久化和并发。最终换成 Milvus 的原因有三个。第一Milvus 是独立服务多个 Agent 实例可以共享同一个记忆存储解决了 InMemory 阶段“多实例不共享”的硬伤。第二它自带标量字段过滤能力expr表达式可以同时按用户、会话、时间过滤向量检索结果这正是 Memory 场景里高频需要的。第三数据量大之后可以平滑扩展到分布式模式不需要推翻重来。5. 生产级 Memory 模块的进阶设计5.1 分层记忆短期、长期、摘要到了 Milvus 这一层存储已经不是问题真正考验设计的是“怎么把不同层级的记忆组合起来用”。我的做法是三段式组装短期上下文 长期相关记忆 全局摘要。短期上下文来自 InMemory 或者文件里的最近 N 轮保证对话连贯长期相关记忆从 Milvus 按当前 query 召回提供背景知识全局摘要是对历史对话做定期压缩模型在每轮对话前读一遍摘要对用户的整体情况有基本感知。这三段拼进 prompt 后要控制总量。我的经验是短期上下文控制在 2000 token 以内召回记忆控制在 3 到 5 条摘要控制在 500 token 以内。总量超过阈值时优先砍摘要再砍短期上下文最后才砍召回记忆。5.2 召回不是越多越好新手最容易犯的错误是“召回的条数越多越好”。实际不是这样。从 Milvus 里召回 20 条记忆里面可能有 5 条是高度相关的剩下 15 条是勉强沾边的。这些勉强沾边的记忆塞进 prompt不会让模型变聪明反而会让它分心甚至把模型带到错误的方向上。我现在的策略是限制top_k 5并且只保留distance 0.7的结果。同时做一个“来源去重”同一个 conversation_id 下最多取 2 条避免某一个会话的记忆成堆出现挤压了其他重要记忆的空间。5.3 写入时机与改写策略记忆写入不是“每轮对话都写”而是要有策略。每轮都写会堆积大量低价值信息比如“好的”“嗯嗯”这种废话占向量库空间不说还会降低召回精度。我的写入规则是用户显式要求记住时比如“记住我周一要开会”必须同步写入LLM 完成了关键任务节点时比如查完一家公司资料后给出结论把结论写入普通轮次对话用异步方式批量写入并且存储前先用 LLM 做一次“记忆提炼”把一段冗长的对话改写成一两句简洁的陈述。这个“改写后存储”的步骤非常重要。直接用原文存召回时经常匹配到无意义的重复改成摘要之后召回精度和 prompt 的可用性都会明显提升。5.4 多租户隔离怎么做只要你的 Agent 服务不止一个人用就必须考虑隔离。最简单的方案就是在 collection 里加一个user_id或tenant_id字段写入时带上检索时在expr里过滤。但要注意expr过滤是检索后置条件不是索引前置条件。也就是说如果你的向量索引是全库的哪怕只检索一个用户的数据复杂度和全库搜索一样。数据量大到一定程度后更合适的做法是按用户分 shard或者用 Milvus 的 partition 功能把不同用户的数据放进不同 partition检索时指定 partition_names性能会好很多。5.5 记忆的更新与遗忘好的记忆系统不仅要能存能查还要能“忘”。用户可能换了偏好可能改了地址旧记忆就成了干扰项。Milvus 支持按表达式删除数据collection.delete(exprconversation_id abc123)我依赖这个能力做定期清理每 N 天扫描一次写入时间超过 TTL 的会话删除过期记忆用户主动要求清除记忆时按user_id删除全部关联记录。另外要提醒一句删除操作和插入操作一样在批量执行时最好做节流避免高频删除引发 compaction 压力。6. 常见问题与排查技巧实录6.1 进程异常退出0xc0000005 内存访问冲突如果你在 Windows 本地开发环境跑 Agent偶尔会遇到进程直接退出报错process exited with code 3221225477 (0xC0000005 - memory access violation)这个错误本质是内存访问越界常见于底层 C/C 扩展库比如某些本地向量索引库、或者特定版本的 Milvus 客户端依赖。我自己遇到过一次排查路径是这样的先看崩溃是否稳定复现如果只在某个特定操作后出现多半是传入的数据有问题比如 embedding 维度与 collection 定义不一致如果不稳定复现优先升级 pymilvus 和依赖库版本或者换一台 Linux 环境验证。Windows 下 run Docker 的 Milvus 服务端本身一般没问题问题通常出在本地客户端库的兼容性上。6.2 连接不上 Milvus 服务最常见的问题是容器起来了但客户端连不上 19530 端口。先检查监听状态netstat -an | findstr 19530如果端口没监听大概率是容器崩了。然后docker compose logs standalone我踩过的一个坑是 Docker 磁盘空间不足导致 Milvus 写入元数据失败容器反复重启。清理 Docker 磁盘占用后一切正常。另一个坑是本地 19530 端口被其他程序占用把 compose 里的端口映射改成19531:19530可以快速绕过。6.3 召回质量差搜出来一堆不相关的内容很多人一上来就怀疑 Milvus 有问题其实大部分时候是 embedding 或者数据预处理的问题。排查顺序我建议是第一步把 query 的 embedding 和召回结果的 embedding 拿出来用余弦公式手算一遍相似度确认 Milvus 返回的结果不是错的第二步检查存储时是不是整段长文本直接 embedding长文本语义容易被稀释最好按 200 到 500 字切块再分别存储第三步看 query 改写用户的原话常常缺少关键信息先让 LLM 把 query 改写成一个更完整的检索语句召回效果会有明显提升。6.4 批量写入慢得离谱一条一条 insert 肯定慢。Milvus 的写入是按 batch 设计的一次 insert 几千条和几千次 insert 每次一条的性能差距是两个数量级。我的做法是先把记忆记录攒在内存里积累到 100 条或者 1 秒钟后统一做一次批量插入。批量插入时同时注意 embedding 接口的限流很多 embedding API 是按每分钟请求数限制的本地 bge 模型则要注意显存吃紧的问题。6.5 索引构建失败或查询变慢索引构建失败最常见的原因是内存不足。HNSW 构建时需要把所有向量加载到内存数据量大的机器内存不够就会出现 OOM。解法是调小M和efConstruction或者分批构建。查询变慢则要看expr过滤条件是否把候选集压得太小以及 collection 是否频繁 load/unload。还有一个容易忽略的点collection 里的 segment 数量过多会导致查询变慢定期执行collection.compact()合并 segment 能改善查询性能。我自己现在做 Agent 项目默认路径已经固定下来先用 InMemory 把流程跑通再用文件/SQLite 做持久化兜底等真正需要跨会话语义召回时再上 Milvus。这个顺序看着土但每一步都能帮你更清楚地认识到当前项目的瓶颈在哪儿。给 Agent 装上记忆不是一步到位的事而是一个从“能跑”到“跑得好”的渐进过程。