☰
用Milvus为AI搭建长期记忆:从金鱼脑到图书馆的实战指南
2026/10/17 8:39:07 网站建设 项目流程

上篇我们聊到,普通大模型用过就忘,上下文窗口一翻篇,前面的对话就像金鱼记忆一样全部蒸发。这个系列走到下篇,目标很明确:把 Milvus 向量数据库真正端起来,给AI建一座能长期调用的"图书馆"。如果你在做 AI Agent、智能客服、个人知识库助手,或者只是想给聊天机器人加上"我记得你上次说过什么"的机制,这篇文章基本能覆盖你会踩到的大半条路。

我不会把 Milvus 讲成一张参数说明书,而是按"为什么选它 → 怎么跑起来 → 怎么把记忆写进去再捞出来 → 遇到问题怎么办"的顺序,直接给你一条可复现的实操链路。代码都用 Python 生态,环境用 Docker Compose,在 Windows、macOS、Linux 上都能照着做。

1. 为什么需要给AI建一座"图书馆"

1.1 AI的"金鱼记忆"到底卡在哪

先别急着聊向量,我们把痛点聊透。大语言模型本身是无状态的,它每一轮回答都只基于当前 prompt 里的内容。你说"我叫小明,我喜欢摄影",它记住了;下一轮你新开一个会话,或者上下文窗口被长文本挤爆,它立刻不记得你是谁。做过客服机器人的朋友应该深有体会:用户好不容易在上一轮提供了订单号,下一轮换个话题再问,模型就开始一本正经地胡说八道。

这就像金鱼,游过一圈就忘了刚才发生的事。所谓给AI建图书馆,本质上是把"发生过的事情、积累的知识"从模型外部找一个持久化载体,需要的时候再把相关内容捞回来,塞进当前对话里。这个载体不能是普通的 MySQL 表,因为用户的问题往往是模糊的、口语化的,你没法用"WHERE content = '摄影'"去匹配"我想拍夜景有什么建议"这种语义相近但字面完全不同的需求。

于是向量数据库站出来了。它存的不是原始文本,而是把文本"翻译"成一串高维数字,也就是向量。判断两条内容相不相关,就看这两个向量在空间里离得近不近。这样一来,哪怕提问和原文一个同义词都没有,只要语义相似,照样能检索得到。这就是"图书馆"和"杂物间"的本质区别:图书馆是按语义索引的,杂物间只按标签堆放。

1.2 向量数据库不是数据库,是"记忆皮层"

很多人第一次接触向量数据库时容易把它想复杂,其实可以这样理解:普通数据库是"精确查表",你说身份证号我就能找到人;向量数据库是"模糊找感觉",你描述一个人的气质,我把跟你描述最接近的那几个人都叫出来给你挑。

具体到技术层面,一条文本会被 embedding 模型转成一个固定维度的浮点数组,比如 768 维、1024 维。Milvus 负责把海量这样的向量存下来,并在查询时用余弦距离、内积或欧氏距离算出"谁和谁更像"。更重要的是,Milvus 还支持标量字段与向量字段的混合过滤,比如"只要张三在 2024 年之后发的消息,按语义相关性排序返回"。这一下就把单纯的向量检索推进到了可以实际工程落地的程度。

我们经常听到的 RAG(检索增强生成)就是建立在向量数据库之上的典型玩法。你不再指望大模型记住所有知识,而是先从一个可靠的"外部记忆"里把相关片段捞出来,再交给大模型组织语言。这样模型每回答一个问题,都像考试时先翻开笔记而不是全靠脑子硬背,准确率自然高出一大截。

1.3 RAG架构让"记忆"真正落地

RAG 的全流程其实只有三步:

  • 用户问题进来,先用同样的 embedding 模型把问题转成向量。
  • 拿这个向量去 Milvus 里做相似度搜索,返回最相关的几条历史记录或知识片段。
  • 把这些问题片段和用户问题一起塞进 prompt,交给大模型生成最终答案。

听起来简单,但这里有一个关键细节:用于检索的 embedding 模型,必须和入库时的 embedding 模型保持一致。你入库用的是 A 模型,检索时换了 B 模型,两个模型生成的向量空间根本不通用,检索质量会直接崩盘。这是新手最容易踩的坑,我先提个醒,后面实操还会再强调。

走完这套流程,AI 的"记忆"就不存在上下文窗口不够用的问题了。多轮对话可以归档到 Milvus,团队知识库可以切片后入库,Agent 在执行多步任务时也能随时把中间结果保存起来、下次继续。所谓长期记忆,本质就是"外部化 + 语义索引 + 及时召回"。

2. 选型拆解:为什么是Milvus

2.1 市面上那么多向量库,各有什么脾气

先说清楚,我没有收 Milvus 一分钱广告费,纯粹是几个项目对比下来,它在我的场景里最合适。为了让你也心里有数,我把当时重点对比的几个方案列一下:

方案类型优点注意点
Faiss向量检索库性能强,灵活只是库不是服务,索引、持久化、并发都要自己搞,适合算法工程师做实验
Chroma轻量级向量库上手快,代码简单大规模数据时稳定性一般,服务能力偏弱,适合原型项目
Weaviate向量数据库服务自带图式和模块化对中文社区资料相对少,embedding 组件耦合稍重
Qdrant向量数据库服务Rust 编写,性能好,API 简洁功能不错,但当时我要的过滤能力和生态完整度略逊
Milvus向量数据库服务云原生架构,功能全,中文社区活跃组件多,部署上手有门槛,熟悉后很顺

我这边选 Milvus 的理由主要是三个。一是数据规模弹性好,从几万条到几亿条向量都能扛,不会项目做大了就换数据库;二是标量过滤功能强,能够把"用户ID + 时间范围 + 语义相似度"揉在一条查询里,这在做 AI 记忆时几乎是刚需;三是文档和社区在国内比较活跃,遇到问题能搜到中文答案,这对团队协作很友好。

当然,如果你只是做一个 500 条样本的 demo,Chroma 完全够用,不必迷信 Milvus。选型永远看场景,没有银弹。

2.2 Milvus在"金鱼图书馆"架构里的位置

我把这个"金鱼图书馆"简化成三块:业务服务、Milvus、Embedding 模型。业务服务就是你的 AI 应用,负责调度对话和逻辑;Milvus 负责记忆内容的存储和相似度检索;Embedding 模型负责把所有文本转成向量。三者关系如下:

  • 写入链路:业务服务拿到一段新对话 → 调用 Embedding 模型生成向量 → 把文本和向量一起插入 Milvus。
  • 读取链路:用户新提问 → 调用同一个 Embedding 模型生成问题向量 → Milvus 返回最相关的几段历史 → 拼入 prompt → 大模型回答。

Milvus 在这里的角色就是"记忆皮层"。它不负责思考,也不负责生成语言,只负责两件事:准确记住写过的东西,以及在需要时快速想起相关的东西。把职责拆开,系统才能保持简单和可靠。

我还要强调一点:Milvus 是存向量和相关标量的地方,但不是唯一的存储。原始终端用户的聊天原始报文,建议同时备份到对象存储或用 ClickHouse 之类的做日志归档。向量库里只放经过切分和向量化的内容,这样万一向量索引坏了或者要换 embedding 模型,你还能从原始数据重建,不至于全盘归零。

3. 实操:把Milvus跑起来

3.1 第一次启动:用Docker Compose快速搭一个Standalone实例

Milvus 有 Standalone 和 Cluster 两种形态,个人开发、中小项目用 Standalone 就行。它虽然叫"单机",但内部还是依赖 etcd 做元数据管理、MinIO 做对象存储,所以用 Docker Compose 一次把三个组件拉起来最方便。

先确认你机器上装了 Docker 和 Docker Compose。我把当时的 docker-compose.yml 贴出来(基于 Milvus 2.4.x):

version: '3.5' services: etcd: container_name: milvus-etcd image: quay.io/coreos/etcd:v3.5.5 environment: - ETCD_AUTO_COMPACTION_MODE=revision - ETCD_AUTO_COMPACTION_RETENTION=1000 - ETCD_QUOTA_BACKEND_BYTES=4294967296 - ETCD_SNAPSHOT_COUNT=50000 volumes: - ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/etcd:/etcd command: etcd -advertise-client-urls=http://127.0.0.1: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 milvus: container_name: milvus-standalone image: milvusdb/milvus:2.4.4 command: ["milvus", "run", "standalone"] security_opt: - seccomp:unconfined environment: ETCD_ENDPOINTS: etcd:2379 MINIO_ADDRESS: minio:9000 volumes: - ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/milvus:/var/lib/milvus ports: - "19530:19530" - "9091:9091" depends_on: - etcd - minio

在 docker-compose.yml 所在目录执行:

docker compose up -d

启动完成后,验证一下三个容器状态:

docker compose ps

正常情况下,etcd、minio、milvus 三行都应该是 Up 状态。再检查 Milvus 的端口是否可用:

curl http://localhost:9091/healthz -v

如果返回OK或者 HTTP 200,说明 Milvus 服务已经起来了。这里有个小提醒:Milvus 首次拉镜像会比较慢,特别是 milvusdb/milvus 这个镜像有几个 GB,建议提前配好 Docker 镜像加速,别等到部署时干瞪眼。

3.2 安装客户端并建立第一个Collection

服务起来了,接下来用 Python 操作。先安装官方客户端 pymilvus:

pip install pymilvus==2.4.4

然后写一段最小连接代码:

from pymilvus import connections, utility connections.connect(alias="default", host="localhost", port="19530") print(utility.get_server_version())

如果打印出版本号,说明连接成功。接下来建 Collection。在 Milvus 2.x 的概念里,Collection 相当于关系型数据库的表。我们需要设计几个字段:

  • id:主键,用自增整数就可以。
  • content:原始文本内容,也就是"记忆"本身。
  • source:来源标识,比如user_profile、chat_history、knowledge_base。
  • user_id:用户标识,用于多用户场景下隔离数据。
  • created_at:时间戳,方便按时间范围过滤。
  • embedding:向量字段,这里先用 768 维演示,实际维度由你的 embedding 模型决定。

建集合并创建索引:

from pymilvus import FieldSchema, CollectionSchema, DataType, Collection fields = [ FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True), FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=65535), FieldSchema(name="source", dtype=DataType.VARCHAR, max_length=128), FieldSchema(name="user_id", dtype=DataType.VARCHAR, max_length=64), FieldSchema(name="created_at", dtype=DataType.INT64), FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=768), ] schema = CollectionSchema(fields=fields, description="AI Long-term Memory") collection_name = "ai_memory" collection = Collection(name=collection_name, schema=schema) index_params = { "index_type": "AUTOINDEX", "metric_type": "COSINE", "params": {} } collection.create_index(field_name="embedding", index_params=index_params) print("Collection created:", collection_name)

这里重点说明两个选择原因。

一是metric_type选了COSINE而不是L2。向量检索用什么度量取决于你的 embedding 模型。大部分中文 embedding 模型(比如 bge、text2vec)官方推荐余弦相似度,因为它在向量模长不一致时依然能保持稳定的语义相关性判断。如果你库里的向量都是归一化后的,用 L2 和 COSINE 排序结果差不多,但保险起见还是看模型文档。

二是index_type用了AUTOINDEX。这是 Milvus 2.4 里最省心的索引方式,系统会根据数据规模自动选择合适的索引结构,对中小项目来说完全够用。如果你追求极致性能,后面可以手动调成 IVF_FLAT、HNSW 这些,但那是另一个话题,初期不建议在这上面纠结。

4. 核心实现:让AI真正"记住"再"想起"

4.1 记忆写入链路:切分 → 嵌入 → 入库

服务跑起来、集合建好了,真正的重头戏在于怎么把一段对话变成一条可检索的记忆。

首先是切分。大模型一次能处理的文本长度有限,embedding 模型对超长文本生成的向量质量也会下降,所以原始文本必须先切片。切分不是简单按字符数截断,要考虑语义完整性。我常用的方案是 LangChain 的RecursiveCharacterTextSplitter,它在遇到换行、句号、空格时会优选中止点,避免把一个完整句子拦腰斩断。

from langchain_text_splitters import RecursiveCharacterTextSplitter text_splitter = RecursiveCharacterTextSplitter( chunk_size=512, chunk_overlap=64, separators=["\n\n", "\n", "。", "!", "?", ".", "!", "?", ";", ";"] ) chunks = text_splitter.split_text(long_text)

chunk_size选 512 是我在中文场景下试过比较稳的数值,太大检索粒度太粗,太小则每条记忆信息量不足。chunk_overlap设为 64,是为了避免切分处正好切断语义联系,检索时相邻片段能有一些重叠覆盖。

然后是嵌入。我的经验是用本地部署的中文 embedding 模型,比如BAAI/bge-large-zh-v1.5,这样不依赖外部 API,隐私性更好,也不会因为接口限流导致写入失败。用sentence-transformers加载:

from sentence_transformers import SentenceTransformer embedding_model = SentenceTransformer("BAAI/bge-large-zh-v1.5") def embed_texts(texts): return embedding_model.encode(texts, normalize_embeddings=True).tolist()

normalize_embeddings=True这一步很重要,它把向量归一化成单位向量,配合 COSINE 度量会有更稳定的效果。

最后是入库。把切分好的 chunk、来源信息、用户 ID、时间戳一起插入 Milvus:

from pymilvus import Collection import time collection = Collection("ai_memory") def save_memory(texts, source, user_id): vectors = embed_texts(texts) now = int(time.time()) rows = [ [texts[i] for i in range(len(texts))], [source] * len(texts), [user_id] * len(texts), [now] * len(texts), vectors, ] collection.insert(rows) collection.flush() print(f"Saved {len(texts)} memory chunks")

flush()是强制把数据落盘,保证后续检索能立刻看到刚写入的内容。在写入频繁的场景可以攒一批再 flush,性能更好,但初期不建议为了性能牺牲一致性。

4.2 记忆读取链路:向量检索 + 上下文组装

记忆写入后,读取才是真正体现价值的地方。用户发来一个新问题,我们先把问题向量化,然后在 Milvus 里搜索最相关的历史记忆:

from pymilvus import Collection, utility collection = Collection("ai_memory") collection.load() def search_memory(query, user_id, top_k=5): query_vector = embed_texts([query]) results = collection.search( data=query_vector, anns_field="embedding", param={"metric_type": "COSINE", "params": {"ef": 128}}, limit=top_k, expr=f'user_id == "{user_id}"', output_fields=["content", "source", "created_at"], ) memories = [] for hit in results[0]: memories.append({ "content": hit.entity.get("content"), "source": hit.entity.get("source"), "score": hit.score, }) return memories

这里有两个容易被忽略的细节。

一是collection.load()。Milvus 在查询前需要把数据加载到内存,首次调用 load 后不是立刻就绪,如果遇到collection not loaded的报错,稍等一下再查。在服务启动阶段可以先调用一次load(),避免用户请求过来时还在加载。

二是expr参数。我们按user_id过滤,这保证了多用户系统里 A 用户查不到 B 用户的记忆。如果还想按时间过滤,可以追加and created_at > 1700000000这样的条件。

检索到相关记忆后,最关键的一步是把这些记忆组装进 prompt:

def build_prompt(query, memories): memory_text = "\n".join([ f"- {m['content']} (来源: {m['source']})" for m in memories ]) prompt = f"""你是一个拥有长期记忆的智能助手。 以下是这位用户过往对话中与你当前问题相关的记忆片段: {memory_text} 请你结合以上记忆,回答用户的问题: {query} """ return prompt

最后把 prompt 发给大模型。我用 OpenAI 兼容接口示例:

from openai import OpenAI client = OpenAI(base_url="http://localhost:8000/v1", api_key="EMPTY") memories = search_memory("我上次说想学什么来着?", user_id="user_123") prompt = build_prompt("我上次说想学什么来着?", memories) response = client.chat.completions.create( model="your-model-name", messages=[{"role": "user", "content": prompt}], ) print(response.choices[0].message.content)

到这里,一条完整的记忆闭环就跑通了:历史对话被切分、嵌入、存储,用户新问题触发检索,相关记忆被捞出来辅助大模型回答。AI 不再是金鱼,它开始拥有一个随时能翻开的笔记本。

5. 常见问题与排查技巧实录

5.1 服务起不来的那些坑

先说 Docker Compose 起不来。最常见的原因是内存不足,Milvus Standalone 至少需要 8GB 可用内存,如果机器只有 4GB,etcd 和 Milvus 进程很容易被系统杀掉。排查命令:

docker compose logs etcd | tail -50 docker compose logs milvus | tail -50

如果看到out of memory或者进程直接消失,先加内存或 swap,再重启。

第二个是端口冲突。19530 是 Milvus 的 gRPC 端口,9091 是健康检查端口,如果本机有别的程序占了,需要修改 compose 文件里的映射端口,或者停掉占用进程。

第三个是镜像拉取超时。etcd 镜像来自 quay.io,国内网络访问有时不太稳定,遇到超时多试几次,或者配置 Docker daemon 的 registry-mirror。

还有一个小问题:用了 2.4 的镜像,但 pymilvus 版本是 2.2 或者更老,连接时会报协议不兼容。我的建议是客户端与服务端版本尽量保持一致,升级的时候一起升,别混着用。

5.2 检索效果不好,先从这些方向排查

检索结果不相关,是向量数据库项目最让人头疼的问题。我会按顺序排查四个点:

  • 确认 embedding 模型一致。入库和查询必须用同一个模型,这是头号原因。很多人开发时先用了 A 模型测试,后来又换了 B 模型,历史数据没重新入库,检索效果自然一塌糊涂。
  • 检查切分粒度。chunk_size 过大或过小都会影响检索质量。如果你搜索出来的内容总是"相关但不够精确",试着把 chunk_size 从 512 降到 256;如果内容太碎、上下文不够完整,就调到 768。
  • 加 filter 条件。比如用户问题是在问最近的对话,但库里捞回来的是一个月前的记忆,这时就要在expr里加时间范围过滤,让时间成为硬性条件,而不是全交给向量相似度。
  • 检查 score 的绝对值。余弦相似度在 0.7 以上通常算相关,0.5 以下基本不靠谱。如果检索结果 top1 的分数都低于 0.5,说明你这个库里的内容离用户问题太远,要么增加更多记忆,要么换更强力的 embedding 模型,比如 bge-m3。

5.3 数据量变大的调优方向

项目跑一段时间后,Milvus 集合里可能积累了几十万条向量,你会发现插入和检索速度都有变化。这时候可以做一些调整:

  • 调整索引类型。从AUTOINDEX切到HNSW,在创建索引时设置M=16、efConstruction=256,检索时设ef=128,速度会明显提升。
  • 开启 Milvus 的压缩(Compaction)。频繁插入删除会产生很多小分段,定期执行手动压缩能提升查询性能。调用方式比较底层,我一般在运维脚本里定时触发。
  • 考虑集合分区(Partition)。如果数据天然按天或按用户群区分,可以用分区表减少每次扫描的范围。注意,分区不能替代标量过滤,但对于超大流量场景很有帮助。

我自己实际经历过一个比较典型的案例:某个知识库机器人最初只有 2 万条知识,检索飞快;后来涨到 80 万条,响应时间从十几毫秒涨到了四百多毫秒。后来把索引改成 HNSW,并加了时间过滤,延迟降回了七八十毫秒。这说明向量数据库的核心优化永远在"索引结构 + 过滤条件"这两个点上。

5.4 一个容易被忽略的运维习惯

最后分享一个我在实战中踩过坑后养成的习惯:每次修改 embedding 模型、切分参数或索引配置前,先把 Milvus 里的数据备份好。Milvus 的数据都在 Docker 挂载的 volumes 目录里,备份可以直接复制整个目录。别小看这一步,等你发现问题想回滚时,一个干净的历史副本能让你少熬夜。

另外,建议给 Milvus 加一个定时任务,每天自动检查数据量和磁盘占用。向量数据非常吃磁盘,一不留神就是几十 GB 的增长。别等磁盘满了再去清理,到时候索引文件损坏就得不偿失了。

我个人在实际项目里有一个偏执的做法:把用户原始对话按天归档到普通数据库和对象存储,Milvus 里只保留切分后的向量记忆。这样即使 Milvus 需要重建,我也可以从原始数据重新生成向量,不会出现"数据全丢"的灾难场景。作为给 AI 搭图书馆的人,你得时刻保证这座图书馆即使被烧了,底下的书稿还能再造一座新的。

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

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

立即咨询