Milvus 3.0实战：从零搭建企业级RAG知识库完整指南-平芜编程栈

当业务侧希望把企业内部文档、产品手册、技术规范变成可对话的智能问答系统时，RAG（检索增强生成）几乎是目前性价比最高、落地最快的方案。而选择 Milvus 作为向量数据库，又是这套方案里非常关键的一环。网上关于 Milvus 和 RAG 的资料虽然很多，但大多分散在官方文档、个人博客、开源项目 README 里，真正能从零开始、把环境搭建、数据入库、检索问答、效果评估全部串起来的完整教程并不多。

本文就围绕“Milvus 3.0 实战”这条主线，手把手带你搭一套企业级 RAG 知识库。内容包括核心概念、Docker 部署、Python 操作 Milvus、文档向量化全流程、检索问答、RAG 效果评测方法和常见问题排查。无论你是刚接触向量数据库的新手，还是准备在公司内部落地知识库后端服务的开发者，都可以直接参考这套流程。

1. Milvus 与 RAG 的核心概念

在动手之前，先花几分钟把 Milvus 和 RAG 这两个核心概念理清楚。这不是为了堆术语，而是后面所有配置、代码和排错都建立在这些基础认知上。

1.1 Milvus 3.0 是什么

Milvus 是一款开源的分布式向量数据库，专门用于存储、索引和管理海量高维向量数据。传统的 MySQL、PostgreSQL 擅长存储结构化数据，但在“根据内容相似度检索”这个场景下表现一般；Milvus 这类向量数据库则把数据转换成向量后，通过最近邻搜索算法快速找到语义最接近的向量。

“Milvus 3.0”是 Milvus 产品演进中的重要版本方向。相比早期版本，它更强调云原生架构、多租户支持和更灵活的数据管理能力。对普通开发者来说，最直观的感受是部署方式更规范了，使用 Python SDK 或 RESTful API 时接口也更稳定。

这里需要特别说明：Milvus 的版本迭代比较快，很多教程里写的 API 接口在新版本中可能已经调整。本文标题虽然写的是 Milvus 3.0 实战，但正文中的操作方式同时兼容当前主流稳定版本。你在自己环境里安装时，建议直接使用官方最新稳定版，不要盲目追求“最新开发版”。

1.2 RAG 检索增强生成

RAG，全称 Retrieval-Augmented Generation，检索增强生成。它解决的痛点非常明确：大语言模型的知识截止时间是固定的，而且对私有业务数据完全不了解。你问它“公司最新发布的设备维护流程是什么”，如果没有检索模块，模型大概率只能给出一段泛泛而谈的内容。

RAG 的基本原理是“先检索，后生成”。用户提问后，系统先从知识库中检索出最相关的文本片段，把这些片段作为上下文，连同用户问题一起交给大模型生成最终答案。这样做的好处有三个：

答案有据可依，可以减少模型“一本正经地胡说八道”。
知识库可以实时更新，不需要频繁微调模型。
企业私有数据不用上传给模型厂商，数据可控性更高。

1.3 一个完整 RAG 流程包含哪些环节

一个标准 RAG 系统可以拆成两个阶段。

离线索引阶段（也叫索引 Pipeline）：

文档加载：读取 PDF、Word、Markdown、HTML 等不同格式的文件。
文本清洗：去掉页眉页脚、特殊字符、无意义内容。
文本分块（Chunking）：把长文档切成固定长度或按语义切分的文本块。
向量化（Embedding）：用文本嵌入模型把每个文本块变成向量。
写入向量数据库：把向量和原始文本一起写入 Milvus。

在线推理阶段（也叫查询 Pipeline）：

用户提问。
对问题做同样的向量化处理。
在 Milvus 中执行相似度检索，召回 Top-K 文本块。
把召回结果拼接成 Prompt。
调用大模型生成答案。

1.4 为什么选择 Milvus 做 RAG 知识库

市面上向量数据库不少，比如 Chroma、Weaviate、Qdrant、Pinecone 等。Milvus 的定位更偏“企业级生产环境”：它支持分布式部署、多种索引类型、数据持久化和权限管理，而且有活跃的开源社区。对要落地到真实业务系统的团队来说，Milvus 在性能和稳定性上的表现更让人放心。

另外，Milvus 生态中还有一个可视化工具 Attu，可以像使用 Navicat 操作 MySQL 一样，在图形界面里查看 Collection 数据、执行查询、管理索引。这对调试 RAG 知识库非常有帮助，所以建议从最开始就安装上。

2. 环境准备与版本说明

下面进入实操环节。整个环境准备分为四块：Docker 环境、Milvus 服务端、Attu 客户端、Python 开发环境。

2.1 环境清单

本文的示例环境如下：

环境项	推荐配置
操作系统	Ubuntu 22.04 / CentOS 7+ / macOS
Docker	Docker Engine 20.10+，Docker Compose v2
Milvus	官方最新稳定版（本文以 standalone 模式部署）
Attu	与 Milvus 服务端匹配的最新版本
Python	3.10+
IDE	VS Code 或 PyCharm

如果你的机器配置比较有限，standalone 模式完全够用。所谓 standalone 模式，就是把 Milvus 所需的 etcd、minio 和 milvus 三个服务组件用 Docker Compose 一起启动，适合本地开发和中小规模知识库。

版本注意事项：Milvus 的 Docker 镜像名一般是milvusdb/milvus，etcd 和 minio 是它依赖的底层组件。安装前建议到 Milvus 官网查看当前推荐版本号，避免直接用latest标签导致上下游依赖不一致。如果你使用的是老版本 Milvus 2.x 项目，要升级到 3.x 时，需要先确认 SDK 和索引格式的兼容性，再做数据迁移和验证，不要直接在线上环境强制执行。

2.2 Docker 部署 Milvus

第一步，准备docker-compose.yml文件。以下是一个标准的 Milvus standalone 部署配置：

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:v2.4.0 command: ["milvus", "run", "standalone"] 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" attu: container_name: attu image: zilliz/attu:v2.4 ports: - "8000:3000" environment: MILVUS_URL: milvus-standalone:19530 depends_on: - "milvus"

注意：上面的镜像版本号是示例。实际操作时，请以 Milvus 官方文档当前推荐的版本为准。镜像版本差异会导致docker-compose.yml中某些环境变量或启动命令略有不同。

第二步，启动服务：

docker-compose up -d

启动后查看容器状态：

docker-compose ps

正常情况下，etcd、minio、milvus 三个容器都处于 Up 状态。Milvus 服务会监听 19530 端口，这是客户端 SDK 连接端口；9091 是健康检查端口；Attu 会监听本地 8000 端口。

第三步，验证 Milvus 健康状态：

curl http://localhost:9091/healthz

如果返回OK，说明 Milvus 启动成功。

2.3 安装 Python 依赖

本文的代码示例使用pymilvus连接 Milvus，使用langchain-huggingface做文档加载、分块和向量化。如果你希望走纯原生 API，不引入 LangChain 也可以，后面我会给出两种写法。

创建项目目录并准备虚拟环境：

mkdir milvus-rag-demo && cd milvus-rag-demo python -m venv venv source venv/bin/activate

创建requirements.txt：

pymilvus>=2.4.0 langchain>=0.3.0 langchain-community>=0.3.0 langchain-huggingface>=0.1.0 langchain-text-splitters>=0.3.0 sentence-transformers>=3.0.0 pymupdf>=1.24.0 python-dotenv>=1.0.0

安装依赖：

pip install -r requirements.txt

说明一下为什么要用sentence-transformers：它负责把文本转换成向量。示例里我会使用 Hugging Face 上的BAAI/bge-small-zh-v1.5模型，这是一个中文语义向量模型，对中文知识库效果不错。首次运行时会自动下载模型权重，需要保持网络可以访问 Hugging Face，或者提前配置好国内镜像。

3. Milvus 核心概念与基础操作

在写 RAG 代码之前，先熟悉 Milvus 的几个核心概念。这部分直接关系到你能否正确设计数据模型。

3.1 核心概念：Collection、Partition、Field、Index

Milvus 的逻辑结构如下：

Database：数据库，一个实例可以创建多个 Database，类似 MySQL 中的 database。
Collection：集合，类似 MySQL 中的 table，用于存储一批向量和对应的标量字段。
Field：字段，Collection 由字段构成，至少有一个主键字段和一个向量字段。
Partition：分区，可以把 Collection 按某个规则切分成多个物理分区，查询时可以指定分区，减少扫描范围。
Index：索引，为向量字段创建的加速结构，是 Milvus 能够快速检索海量向量的关键。

在 RAG 知识库场景中，一个典型的 Collection 结构是这样的：

字段名	类型	说明
id	INT64	主键，自动生成
content	VARCHAR	原始文本块内容
source	VARCHAR	文档来源，例如文件名
embedding	FLOAT_VECTOR	向量字段，表示文本的语义向量

3.2 向量索引类型如何选择

Milvus 支持多种索引类型，常见的有 FLAT、IVF_FLAT、HNSW、DISKANN 等。对 RAG 场景来说，最推荐的是 HNSW。

HNSW（Hierarchical Navigable Small World）是一种基于图的近似最近邻索引。它的特点是检索速度快、召回率高，适合中等规模到大规模向量数据。对应的索引参数中，M表示每个节点的最大连接数，默认值一般是 16；efConstruction控制建索引时的动态列表长度，越大建索引越慢但质量越高，通常设置在 200 到 500 之间。

如果数据量在百万级以内，直接用 HNSW 基本没有问题。如果数据量达到千万甚至亿级，就需要考虑 IVF 系列索引或者分布式集群方案了。

3.3 Python 连接 Milvus

先用最简单的代码验证连接是否正常：

# 文件路径: milvus_rag_demo/connect.py from pymilvus import connections, utility connections.connect(alias="default", host="localhost", port="19530") print("Milvus 连接成功") print("数据库列表:", utility.list_databases())

运行：

python connect.py

如果输出正常，说明 Python 客户端已经和 Milvus 服务端打通了。

这里需要注意：连接成功后，后续所有 Collection 操作都可以基于alias="default"这个连接别名。如果你在同一段代码里连多个 Milvus 实例，需要为不同连接设置不同 alias。

3.4 创建 Collection 和索引

下面创建 RAG 知识库的核心 Collection：

# 文件路径: milvus_rag_demo/create_collection.py from pymilvus import ( connections, Collection, CollectionSchema, FieldSchema, DataType, utility, ) connections.connect(alias="default", host="localhost", port="19530") COLLECTION_NAME = "rag_knowledge" if utility.has_collection(COLLECTION_NAME): utility.drop_collection(COLLECTION_NAME) fields = [ FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True), FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=8192), FieldSchema(name="source", dtype=DataType.VARCHAR, max_length=512), FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=512), ] schema = CollectionSchema( fields=fields, description="RAG knowledge base collection", enable_dynamic_field=False, ) collection = Collection(name=COLLECTION_NAME, schema=schema) index_params = { "index_type": "HNSW", "metric_type": "IP", "params": {"M": 16, "efConstruction": 200}, } collection.create_index( field_name="embedding", index_params=index_params, ) print(f"Collection {COLLECTION_NAME} 创建成功")

这段代码做了三件事：

定义字段结构：主键id自动生成，content存原始文本，source存来源，embedding存 512 维向量。
使用auto_id=True，这样插入数据时不需要自己维护主键。
创建 HNSW 索引，距离度量方式选择IP（内积）。选用 IP 而不是余弦距离，是因为 bge 系列向量模型在生成向量时已经做了归一化处理，内积和余弦相似度计算结果等价，但 IP 的检索性能通常更好。

如果你使用的向量模型输出维度不是 512，比如 OpenAI 的text-embedding-3-small是 1536 维，就要把dim改成 1536。创建 Collection 之后再改维度是不可能的，所以这一项必须在建表前确认好。

4. RAG 知识库完整实战

环境通了，Collection 建好了，接下来进入正题：把文档灌进 Milvus，并实现一问一答的完整知识库。

4.1 项目结构设计

整个项目拆成四个文件，职责分离，方便后续替换组件：

milvus-rag-demo/ ├── requirements.txt ├── data/ │ └── 产品操作手册.md ├── src/ │ ├── ingest.py # 文档加载、分块、向量化、写入 Milvus │ ├── search.py # 检索函数 │ ├── query.py # 完整问答入口 │ └── config.py # 公共配置：Milvus 连接、模型、Collection 名称

提前创建好这些目录：

mkdir -p data src

准备一个示例文档data/产品操作手册.md，内容可以是你自己公司产品的任意说明。为了测试效果好，建议写 20 行以上，包含产品功能、配置步骤、常见问题等。

4.2 配置项管理

把容易变化的内容统一放到config.py：

# 文件路径: milvus_rag_demo/src/config.py MILVUS_HOST = "localhost" MILVUS_PORT = "19530" COLLECTION_NAME = "rag_knowledge" EMBEDDING_DIM = 512 EMBEDDING_MODEL = "BAAI/bge-small-zh-v1.5" CHUNK_SIZE = 300 CHUNK_OVERLAP = 50 TOP_K = 5

这里的CHUNK_SIZE是每个文本块的最大字符数，CHUNK_OVERLAP是分块时相邻块之间的重叠字符数。重叠是为了避免“一个完整语义被切成两半”导致检索漏掉关键内容。

4.3 文档加载与分块

RAG 的效果很大程度取决于分块策略。块太短，上下文不完整；块太长，向量语义不聚焦，检索精度下降。

下面用 LangChain 的文本加载器和分块器实现：

# 文件路径: milvus_rag_demo/src/ingest.py import os import sys sys.path.append(os.path.dirname(__file__)) from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_huggingface import HuggingFaceEmbeddings from config import ( CHUNK_SIZE, CHUNK_OVERLAP, EMBEDDING_MODEL, EMBEDDING_DIM, ) def load_and_split_documents(file_path: str): loader = TextLoader(file_path, encoding="utf-8") documents = loader.load() text_splitter = RecursiveCharacterTextSplitter( chunk_size=CHUNK_SIZE, chunk_overlap=CHUNK_OVERLAP, separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""], keep_separator=True, ) chunks = text_splitter.split_documents(documents) print(f"原始文档数: {len(documents)}，切分后文本块数: {len(chunks)}") return chunks

RecursiveCharacterTextSplitter会按照分隔符优先级递归切分文本。先尝试用段落空行切，再按换行、句号、逗号、空格逐级降级。这样能尽量保留完整语句，避免产生大量语义碎片。

注意：示例中加载的是 Markdown 文件。如果你要处理 PDF，可以把TextLoader换成PyMuPDFLoader或UnstructuredPDFLoader。不同加载器解析复杂 PDF 的效果差异很大，生产环境要多测试几种方案。

4.4 向量化与写入 Milvus

接下来把切好的文本块变成向量，再写入 Milvus：

# 文件路径: milvus_rag_demo/src/ingest.py（追加） from pymilvus import Collection, connections from config import MILVUS_HOST, MILVUS_PORT, COLLECTION_NAME def build_embeddings(): embeddings = HuggingFaceEmbeddings( model_name=EMBEDDING_MODEL, encode_kwargs={"normalize_embeddings": True}, ) return embeddings def insert_into_milvus(chunks, embeddings): connections.connect(alias="default", host=MILVUS_HOST, port=MILVUS_PORT) collection = Collection(name=COLLECTION_NAME) contents = [chunk.page_content for chunk in chunks] sources = [] for chunk in chunks: source = chunk.metadata.get("source", "unknown") sources.append(source) text_vectors = embeddings.embed_documents(contents) data = [ contents, sources, text_vectors, ] collection.insert(data) collection.flush() print(f"成功写入 {len(contents)} 条数据") # 创建索引并加载到内存 collection.create_index( field_name="embedding", index_params={ "index_type": "HNSW", "metric_type": "IP", "params": {"M": 16, "efConstruction": 200}, }, ) collection.load() print("Collection 索引创建完成并已加载")

这里有个非常重要的细节：collection.insert(data)必须按照 Collection 字段定义的顺序传值。由于id是auto_id=True，所以插入数据时不需要提供 id，只需按照content、source、embedding的顺序传入即可。

collection.flush()把内存中的数据持久化到对象存储。如果是小批量测试，可以每次插入后执行。生产环境通常先批量写入，最后再统一flush，减少性能开销。

collection.load()则是把索引加载到内存，这一步执行之后查询才能跑起来。如果数据量很大，load时间会比较长，这是正常现象。

然后写一个main入口：

# 文件路径: milvus_rag_demo/src/ingest.py（追加） def main(): chunks = load_and_split_documents("data/产品操作手册.md") embeddings = build_embeddings() insert_into_milvus(chunks, embeddings) if __name__ == "__main__": main()

运行：

python src/ingest.py

如果一切正常，控制台会输出写入的文本块数量，之后你就可以在 Attu 中看到这些数据了。

4.5 用 Attu 可视化查看数据

打开浏览器访问http://localhost:8000，连接地址填写http://localhost:19530（Attu 会自动识别部署方式，如果连接失败可以换成milvus-standalone:19530的容器网络地址）。

在 Attu 界面里，你可以看到rag_knowledge这个 Collection，可以预览字段和行数据，也可以直接执行向量查询。排查数据是否成功写入时，Attu 是最直观的工具。

顺便说明一下“Attu 支持哪个 Milvus 版本”这个问题。Attu 的版本需要和 Milvus 服务端版本配套使用，如果版本差距过大，可能出现连接后 Collection 列表为空、页面报错等问题。建议优先使用与 Milvus 镜像同期的 Attu 版本，或者直接采用 Milvus 新版本内置的 Web UI 组件。总之，不要盲目升级其中一端而忽略另一端。

4.6 实现检索函数

向量数据入库后，开始写查询端逻辑。这里先把“检索”和“生成回答”拆开，因为在实际调试中，你要先确认检索结果对不对，再检查生成效果。

# 文件路径: milvus_rag_demo/src/search.py import os import sys sys.path.append(os.path.dirname(__file__)) from pymilvus import Collection, connections from langchain_huggingface import HuggingFaceEmbeddings from config import ( MILVUS_HOST, MILVUS_PORT, COLLECTION_NAME, EMBEDDING_MODEL, TOP_K, ) def search(query: str, top_k: int = TOP_K): connections.connect(alias="default", host=MILVUS_HOST, port=MILVUS_PORT) embeddings = HuggingFaceEmbeddings( model_name=EMBEDDING_MODEL, encode_kwargs={"normalize_embeddings": True}, ) query_vector = embeddings.embed_query(query) collection = Collection(name=COLLECTION_NAME) collection.load() results = collection.search( data=[query_vector], anns_field="embedding", param={"metric_type": "IP", "params": {"ef": 64}}, limit=top_k, output_fields=["content", "source"], ) for i, hit in enumerate(results[0]): print(f"Top {i + 1}: score={hit.score:.4f}, source={hit.entity.get('source')}") print(hit.entity.get("content")) print("-" * 60) return results[0] if __name__ == "__main__": search("产品支持哪些登录方式")

几个关键参数：

anns_field：指定在哪个向量字段上执行检索。
param.metric_type：必须和建索引时的度量方式一致，这里都是IP。
limit：召回数量，也就是要取回多少个最相关的文本块。
output_fields：返回哪些字段。如果你只拿到向量但拿不到原始文本，RAG 后续环节根本没法工作。
ef：HNSW 检索时的动态搜索范围，数值越大召回越准，但耗时越高。一般在 16 到 256 之间调优。

4.7 接入大模型，形成完整问答

最后一步，把检索结果拼进 Prompt，调用大模型。

为了不让本文绑定某个特定大模型供应商，下面用一个抽象的call_llm(messages)函数占位。在实际项目中，你可以替换成 OpenAI SDK、阿里云百炼、智谱 GLM、Ollama 本地模型等任意方案。

# 文件路径: milvus_rag_demo/src/query.py import os import sys sys.path.append(os.path.dirname(__file__)) from pymilvus import Collection, connections from langchain_huggingface import HuggingFaceEmbeddings from config import MILVUS_HOST, MILVUS_PORT, COLLECTION_NAME, EMBEDDING_MODEL, TOP_K def retrieve_context(query: str, top_k: int = TOP_K) -> str: connections.connect(alias="default", host=MILVUS_HOST, port=MILVUS_PORT) embeddings = HuggingFaceEmbeddings( model_name=EMBEDDING_MODEL, encode_kwargs={"normalize_embeddings": True}, ) collection = Collection(name=COLLECTION_NAME) collection.load() query_vector = embeddings.embed_query(query) results = collection.search( data=[query_vector], anns_field="embedding", param={"metric_type": "IP", "params": {"ef": 64}}, limit=top_k, output_fields=["content", "source"], ) context_parts = [] for i, hit in enumerate(results[0]): context_parts.append(f"【片段{i + 1}】来源: {hit.entity.get('source')}\n{hit.entity.get('content')}") return "\n\n".join(context_parts) def call_llm(prompt: str) -> str: # TODO: 替换成实际的大模型调用 # 例如 OpenAI 风格： # from openai import OpenAI # client = OpenAI() # resp = client.chat.completions.create(...) return "这是大模型生成的回答占位内容。" def rag_query(question: str) -> str: context = retrieve_context(question) prompt = f""" 你是一个企业内部知识库助手。请根据下面提供的知识库片段，回答用户问题。 回答时只基于给定的片段，不要编造知识库中不存在的信息。 如果片段内容不足，请直接回答“根据当前知识库无法回答该问题”。 知识库片段： {context} 用户问题：{question} """ answer = call_llm(prompt) print("完整 Prompt 如下：\n", prompt) print("\n最终回答：\n", answer) return answer if __name__ == "__main__": rag_query("产品支持哪些登录方式")

这段代码的 Prompt 设计有三个要点：

明确告诉模型“只基于给定片段回答”，降低幻觉风险。
允许模型在知识不足时说“无法回答”，而不是强行编造。
把片段来源一起拼进去，方便你在调试阶段追踪答案出处。

4.8 运行和预期结果

按顺序执行：

python src/ingest.py python src/search.py python src/query.py

正常情况下，第一次运行ingest.py会下载 embedding 模型，耗时可能较长。后续再运行会使用本地缓存。

如果search.py输出的 Top 片段和问题语义高度相关，说明检索链路没有问题。如果召回结果不相关，优先检查两件事：embedding 模型是否适合中文场景；分块大小是否合适。

5. 进阶方向：Agentic RAG 与 Graph RAG

基础 RAG 流程跑通后，你会发现它仍有不少局限，比如无法处理多跳问题（需要经过多个知识片段推理）、无法主动判断“什么时候该检索、该检索什么”。这也是目前 RAG 技术演进的重要方向。

5.1 Agentic RAG

Agentic RAG 是把大模型 Agent 的规划能力与 RAG 检索结合。传统 RAG 的流程是“用户提问 -> 固定向量检索 -> 生成”，Agentic RAG 则允许模型自行决定：

当前问题是否需要检索知识库。
需要检索一个知识库还是多个知识库。
一次检索不够时，是否需要改写问题或补充检索条件。
如果知识库内容不足，是否要调用其他工具。

例如用户问“对比 A 和 B 两个产品的运维成本”，如果知识库里关于 A 和 B 的信息分散在不同文档中，单纯一次向量检索很难把关键信息全部召回。Agentic RAG 可以先拆解问题，分别检索 A 的运维说明和 B 的运维说明，再做汇总。

实现 Agentic RAG 的常用方式是使用 LangChain 的create_retriever_tool加 Agent 框架，或者使用 LlamaIndex 的 Agent 模块。本文不展开完整代码，因为不同 Agent 框架的 API 变化较快，等你基础 RAG 稳定后，再按官方文档接入即可。

5.2 Graph RAG

Graph RAG 是另一个热点方向。它的思路是在向量检索之外，额外建立一份知识图谱，把实体和实体间的关系结构化存储。

比如知识库中有一句话“Milvus 支持 HNSW 和 IVF 两种索引”，传统向量检索会把整句话作为一个文本块存储；Graph RAG 则会抽取实体 Milvus、HNSW、IVF，以及关系“支持”，并把它们存入图数据库或图结构索引中。

Graph RAG 的优势在于处理“多跳关系型问题”效果更好，比如“HNSW 索引适合什么场景”。但建设成本也明显更高，需要设计实体抽取方案、图存储方案和图查询逻辑。企业是否要上 Graph RAG，建议先评估已有 RAG 系统在关系型问题上的失败率，再决定投入产出比。

5.3 Java 技术栈参考

如果你所在的团队是 Java 后端，不需要把这套流程改成 Python 微服务。目前 Java 生态中，LangChain4j 和 Spring AI 都已经提供了 Milvus 的向量存储集成方案。

LangChain4j 中可以用MilvusEmbeddingStore直接对接 Milvus，Spring AI 也在持续推进 Milvus 的向量存储实现。核心流程和 Python 示例是一样的：文档分块、embedding、写入 Milvus、检索、调用大模型。底层通过 gRPC 与 Milvus 19530 端口通信，所以服务端部署方式完全通用，可以放心从这套教程迁移到 Java 技术栈。

6. RAG 效果评测怎么做

很多团队把 RAG 系统搭起来后，第一个问题不是“能不能跑”，而是“效果到底好不好”。RAG 的评测需要分两层：检索层的质量和生成层的质量。

6.1 检索层评估指标

检索层的核心任务是：从知识库中召回尽可能多、尽可能靠前的相关片段。

常用指标包括：

指标	含义	说明
Recall@K	前 K 个结果中包含相关片段的比例	衡量“有没有找到对的资料”
Precision@K	前 K 个结果中相关片段占比	衡量“找到的资料里有多少是相关的”
MRR	第一个相关结果在结果列表中的位置倒数	强调“最相关的内容是否排在最前面”
NDCG	归一化折损累计增益	衡量排序整体质量

在 Milvus 场景中，你可以在测试集上准备一批“问题 -> 对应文档片段”的标注数据，然后调用检索接口，计算上述指标。如果 Recall@K 很低，说明分块策略或 embedding 模型需要调整。

6.2 生成层评估指标

生成层的评估更偏主观，通常需要人工或大模型辅助打分。常见维度有：

忠实度（Faithfulness）：回答是否严格基于检索到的知识片段，有没有幻觉。
相关性（Relevance）：回答是否准确命中用户问题，是否答非所问。
完整度（Completeness）：知识库中有充分信息时，回答是否覆盖了所有要点。

人工评测可以建立一个小型测试集，包含典型问题、边界问题、知识库外问题三类，然后逐条打分。如果团队比较大，也可以用 RAGAS 这类开源评测框架，用大模型自动打分，降低人工成本。

6.3 评测集怎么构建

构造评测集建议从真实业务问题出发，不要凭空编造。收集渠道可以是：

客服聊天记录。
用户支持工单。
产品试用团队的高频提问。
根据文档章节反向生成“如果我是用户，我会怎么问”。

评测集数量不用多，质量比数量更重要。初期 50 到 100 条高质量测试问题，足以暴露 RAG 链路中的大部分问题。

7. 常见问题与排查思路

实操过程中，最影响体验的就是各种环境问题和数据问题。这里把高频问题整理成表格，再详细展开其中几个。

问题现象	常见原因	解决思路
Docker 启动后 Milvus 容器退出	端口被占用或镜像版本与依赖不兼容	查看`docker logs`，检查端口占用
Attu 页面打不开	Attu 端口未映射或版本不匹配	检查 Docker 端口映射，换版本
Python 连接 Milvus 超时	网络不通、防火墙拦截	先 curl 健康检查接口
检索结果为空	Collection 没有 load 或数据未写入	确认`flush`和`load`已执行
检索结果不相关	embedding 模型不适合或分块不合理	换中文 embedding 模型，调整 chunk
向量维度报错	embedding 模型维度与 Collection dim 不一致	建表前确认模型输出维度
插入数据报错字段不匹配	data 列表顺序与 schema 不一致	对照字段定义顺序传值
大模型回答不完整	Prompt 中知识片段被截断或 Prompt 指令不清	优化 Prompt，增加输出格式约束

7.1 Attu 连接本地 Milvus 失败

这个问题出现频率很高，尤其是在 Docker 部署 Milvus、本地启动 Attu 的场景中。

先确认 Milvus 容器是否正常：

curl http://localhost:9091/healthz

如果健康检查正常，说明 Milvus 服务端没问题。再看 Attu 的MILVUS_URL环境变量。如果你是用桌面版 Attu 连接本地 Milvus，地址一般是http://localhost:19530；如果 Attu 也跑在 Docker 容器里，就不能写localhost，而要写 Milvus 容器的服务名或 IP。

另外，确认 Milvus 容器的 19530 端口确实映射到了宿主机。执行docker-compose ps查看端口映射情况。

7.2 版本不匹配导致异常

Milvus、pymilvus、Attu、LangChain 四者都有版本要求，乱配很容易出问题。其中最常见的是 pymilvus 版本过旧、连接新版 Milvus 失败。

排查顺序：

查看 Milvus 服务端版本：docker-compose exec milvus milvus version。
查看 pymilvus 版本：pip show pymilvus。
查阅官方版本兼容表，确认两端匹配。

建议在requirements.txt中固定 pymilvus 大版本，比如pymilvus>=2.4.0，不要直接安装最新开发版。

7.3 检索效果差，如何系统性优化

如果检索结果不相关，不要一上来就调索引参数，先按下面顺序排查：

检查数据质量：原始文档是否包含大量无关内容？有没有全角半角混乱、乱码？
检查分块策略：300 字符还好，但如果你的文档有大量短句或列表，可能需要调整分块大小和重叠值。
检查 embedding 模型：中文知识库尽量使用专门的中文向量模型，比如BAAI/bge-m3或BAAI/bge-large-zh-v1.5。
检查查询方式：短问题直接用embed_query；长问题或复杂问题，可以先用大模型改写后再检索。
调索引参数：HNSW 的ef可以适当调高，比如从 64 调到 128。

如果做了这些优化还是不行，可以考虑混合检索方案：同时跑向量检索和 BM25 关键词检索，再用 RRF（Reciprocal Rank Fusion）合并排序。这样对“专业术语精确匹配”和“语义相近表达”都能兼顾。

8. 最佳实践与工程建议

最后这部分，把我的实战经验浓缩成几条可以直接落地的建议。

8.1 数据管理建议

生产环境的 RAG 知识库不是“一次性导入就完事”，文档会持续更新。建议在 Collection schema 中增加版本号或更新时间字段，每次导入数据时记录批次，方便后续定向清理。

Milvus 支持按表达式删除数据，比如：

from pymilvus import Collection collection = Collection(name="rag_knowledge") collection.delete("source in ['2024年运维手册.pdf']")

但在生产环境执行删除前，务必先备份索引和原始文档，确认影响范围。删除后要重新flush和load，否则查询结果可能不对。

8.2 安全与权限

企业知识库通常包含敏感信息，要注意：

Milvus 默认没有开启认证，生产环境必须配置用户名密码，或通过内网防火墙限制 19530 端口访问。
不要直接让前端业务请求 Milvus，应该封装一层后端服务，做用户鉴权、数据权限过滤和接口限流。
embedding 模型和大模型 API 的密钥不要写在代码里，统一通过环境变量或配置中心管理。

8.3 性能优化建议

大量写入时关闭flush自动触发，等批量写完成后再统一flush。
设置合理的 HNSW 参数：M不用太大，16 到 32 足够；efConstruction400 以内即可。
查询接口设置超时时间，避免慢查询拖垮整个服务。
如果 Collection 数据量持续增长到单个节点瓶颈，再考虑 Milvus 集群模式或按业务域拆 Collection。

8.4 可维护性与监控

部署后需要记录三类日志：

写入日志：记录每次批量导入的数据量、耗时、来源文件。
查询日志：记录用户问题、召回片段、召回耗时、最终回答。
错误日志：记录模型调用失败、Milvus 连接异常、分块异常。

有了这些日志，你才能在用户反馈“回答变差了”的时候，快速定位到底是检索问题、模型问题还是知识库数据过期。

9. 总结与下一步学习路线

到这里，你已经从零完成了一套基于 Milvus 的 RAG 知识库：用 Docker 部署了 Milvus 和 Attu，创建了带 HNSW 索引的 Collection，用 LangChain 完成了文档加载、分块、向量化、入库，并实现了检索问答链路。也了解了 RAG 效果评测的基本方法和常见问题排查思路。

下一步建议按这个顺序继续深入：

把示例中的call_llm替换成真实大模型接口，完成线上可用的问答服务。
准备一份真实业务文档，按本文流程跑一遍，积累检索效果数据。
尝试构建 50 条评测集，对当前系统做量化评估，找到最弱的环节。
再根据评估结果决定是否需要引入混合检索、重排序、Agentic RAG 或 Graph RAG。
如果是 Java 技术栈，则重点研究 LangChain4j 和 Spring AI 的 Milvus 集成方式。

Milvus 3.0 和 RAG 技术仍然在快速演进，版本、接口、模型选择都会有更新。但只要掌握了这套完整流程和排查方法论，后续无论技术栈怎么切换，你都能快速适应。建议把这篇文章收藏备用，在实际搭建时对照着操作，遇到问题也可以直接在评论区留言，大家一起讨论。