Milvus：从零搭建百万级向量检索的 RAG 系统实战

引子：一个被"找不到"逼疯的项目

今年年初，我们给某制造业客户做内部知识库，10万+份技术文档、质检报告、维修手册散落在NAS和SharePoint里。业务痛点非常具体：新员工搜"轴承异响"能翻半小时Excel，老师傅离职带走了唯一的大脑。客户最初想用开源的anything-llm套壳，但试了之后发现两个问题：一是知识库的文档格式极其混乱（PDF扫描件、CAD截图、Word表格），二是员工提问方式非常口语化——"上次那个电机出啥事儿来着？"——模糊查询命中率不到40%。

最终我选择从零搭建一个RAG系统，核心引擎用Milvus。这篇文章是完整的落地复盘，包含了每一步的真实代码、参数选择、以及让我熬夜到三点的那几个坑。

需求拆解：把"智能问答"翻译成技术指标

和业务方聊完，我们把模糊的"智能问答"拆成了以下可量化的需求：

• 数据规模：10万份文档，预估切块后约80-120万个向量（按每文档分块+重叠计算，实际最终106万）；文档格式需兼容PDF/Word/图片（OCR）
• 性能目标：检索+回答首字响应＜3秒；Top-10召回准确率≥85%（用人工标注的200条测试集评估）
• 成本&部署：不引入GPU服务器（客户IT预算有限），纯CPU机器，内存≤32GB；需支持私有化部署，不能把文档发到外部API

方案选型：为什么是Milvus，而不是那三个"更火的"

老规矩，我先列了候选方案：FAISS、Elasticsearch（自带kNN）、pgvector、Milvus。

• FAISS：最灵活，但它是库不是服务，需要自己处理持久化、并发、多租户；10万+文档后索引更新很痛苦
• Elasticsearch：我们确实重度用了ES做全文检索，但它的向量检索性能在百万级上不如Milvus（实测M40-3.5倍差距）；更重要的是ES的内存占用太恐怖，32GB根本喂不饱
• pgvector：适合"先用起来"，但它的索引类型只有IVFFlat和HNSW，支撑百万级没问题，然而查询并发和动态建模能力偏弱；而且pgvector没有可视化监控面板，排障要靠pg_stat_statements盲人摸象

选Milvus最核心的三个理由：（1）原生支持百万级向量规模化且参数可调，我们最终选了HNSW索引，查询延迟稳定在百毫秒级；（2）支持混合检索（向量+标量过滤），未来做"按部门过滤文档"不需要建两套系统；（3）Milvus的配置项虽然多，但提供了Mivlus Insight可视化工具，对排查问题极其友好。

当然，选型也是有代价的：Milvus引入了一个独立的外部依赖组件，意味着我们又要多运维一个"数据库"。

落地实现：从PDF到Chatbot的完整链路

Step 1：数据准备——文档解析与清洗

第一步就踩坑了。PDF有扫描件和文本件，Word里还嵌着Excel。用PyMuPDF抽文本还算顺利，但扫描件必须走OCR，选了PaddleOCR（支持中英文，CPU推理也不算慢）。

# 文档解析流程（伪代码）
import fitz  # PyMuPDF
import paddleocr
from pathlib import Path

def extract_and_ocr(pdf_path, ocr_engine):
    doc = fitz.open(pdf_path)
    texts = []
    for page in doc:
        text = page.get_text()
        if len(text.strip()) 

OCR引擎直接用了PaddleOCR的CPU模式，平均每页耗时1.2秒，10万份文档分批跑了3天。这里给大家的建议是：如果预算允许，OCR一定要上GPU——这是项目早期最大的瓶颈。

Step 2：分块与向量化——决定RAG质量的关键一跳

分块策略我调了整整一个周末。最初图省事用LangChain的RecursiveCharacterTextSplitter，chunk_size=1000，结果问答准确率只有61%。后来发现：技术文档里的分块Size对检索效果影响巨大。

最终方案（左侧是实际配置）：

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=450,       # 略小于embedding模型最大输入
    chunk_overlap=100,    # 控制上下文断裂
    separators=["

", "
", "。", "；"], # 按中文标点优先切分
    strip_whitespace=True
)

docs = splitter.split_documents(pdf_documents)
print(f"切分完成: {len(docs)} chunks")  # 最终输出: 106万个chunks

Embedding模型选了bge-large-zh-v1.5，为什么不用OpenAI的text-embedding-ada-002？因为客户不允许文档出内网。BGE模型在MTEB中文检索榜单上比ada-002高2-3个点，且支持GPU（我们的机器有8核CPU+16GB内存，embedding阶段靠CPU跑不太现实，最后借了云端1块T4临时跑的）。

Step 3：构建Milvus Collection与索引

向量生成完毕后，就进入Milvus落地阶段。我们用Docker Compose一键起服务（单独Milvus Standalone模式）：

# docker-compose-milvus.yml
version: '3.5'
services:
  etcd:
    image: quay.io/coreos/etcd:v3.5.5
    environment:
      - ETCD_AUTO_COMPACTION_MODE=revision
      - ETCD_AUTO_COMPACTION_RETENTION=1000
  milvus:
    image: milvusdb/milvus:v2.3.4
    command: ["milvus", "run", "standalone"]
    ports:
      - "19530:19530"
    environment:
      ETCD_ENDPOINTS: etcd:2379
    volumes:
      - ./milvus_data:/var/lib/milvus
    depends_on:
      - etcd

创建Collection时，关键配置是索引参数和度量方式。我们对准确率优先，选了HNSW（层级导航小世界）索引，参数如下：

from pymilvus import (
    connections, Collection, CollectionSchema,
    FieldSchema, DataType, utility
)

connections.connect(alias="default", host="localhost", port="19530")

# Schema定义
docs_id = FieldSchema(name="doc_id", dtype=DataType.INT64, is_primary=True, auto_id=False)
embedding = FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1024)
content = FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=65535)
schema = CollectionSchema([docs_id, embedding, content], description="RAG文档库")
collection = Collection(name="knowledge_base", schema=schema)

# 索引参数（HNSW + 内积）
index_params = {
    "metric_type": "IP",               # 内积相似度，配合归一化向量效果更佳
    "index_type": "HNSW",
    "params": {"M": 16, "efConstruction": 200}
}
collection.create_index(field_name="embedding", index_params=index_params)
collection.load()
print("Collection loaded, entities:", collection.num_entities)

批量插入106万条向量耗时约17分钟（单线程，如果并发multi-thread可降到5分钟内）。

Step 4：部署与接入RAG流程

完整的RAG服务用FastAPI写的，流程如下：

@app.post("/ask")
async def ask(query: str):
    # 1. 查询向量化（同bge模型）
    query_vec = embed_single(query).tolist()
    
    # 2. Milvus检索 Top-10
    results = collection.search(
        data=[query_vec],
        limit=10,
        output_fields=["content"],
        param={"search_params": {"ef": 64}}  # 搜索时的动态参数
    )
    
    # 3. 拼装Prompt + LLM生成（此处用了本地部署的ChatGLM3-6B INT8量化版）
    context = "
".join([hit.entity.get('content') for hit in results[0]])
    prompt = f"基于以下资料回答问题:
{context}

问题: {query}"
    answer = llm.chat(prompt)
    
    return {"answer": answer, "sources": results[0]}

上线后的数据：响应时间、准确率、成本

上线前与上线后一周的对比数据（实测200条业务人工标注测试集）：