企业级RAG落地：Milvus混合检索与PDF结构化解析实战

做企业级 RAG 落地时，最怕的不是模型能力不够，而是数据进不去、检索召不回、结果不靠谱。很多团队在 Demo 阶段跑得通一两个 PDF，生产环境一接真实文档就崩。这套 20 集实战内容，我把它压缩成一篇完整教程，覆盖 Milvus 向量库、PDF 结构化解析、双路混合检索与 RRF 重排，目标是让你不再写玩具级 RAG，而是能交付一套可扩展、可排查、可上线的知识库问答方案。

1. 别再写玩具级 RAG：先想清楚它要解决什么问题

在写代码之前，先回答一个关键问题：RAG 到底是用来解决什么的？

1.1 RAG 到底解决了什么问题

RAG 全称 Retrieval-Augmented Generation，即“检索增强生成”。它的核心思路不是让大模型记住所有知识，而是把外部知识先检索出来，再连同用户问题一起交给大模型生成答案。

直接微调大模型需要高质量训练数据、GPU 资源，而且知识更新一次就要重新训练一次。RAG 不需要重新训练模型，它把知识放在外部存储里，涨知识就是往库里加文档，改知识就是删掉旧文档重新切片导入，成本远低于微调。

从业务视角看，RAG 主要解决三类问题：

• 大模型不了解企业内部文档和私有数据；
• 大模型回答容易“幻觉”，需要引用真实来源约束生成；
• 知识更新频繁，需要低成本、低延迟地同步最新内容。

1.2 玩具级 RAG 的常见病与瓶颈

很多团队第一个 RAG Demo 是这样的：读 PDF、按 500 字切块、调一个 embedding、全塞进向量库、用户提问时召回 top 5 丢给大模型。Demo 看着没问题，生产环境一跑就暴露出一堆瓶颈。

第一个瓶颈是 PDF 解析质量差。PDF 不是纯文本文件，它内部是页面坐标、字体、图片、表格指令的混合体。直接按页抽取文本，表格结构会乱，公式会丢，多栏排版会被拼接成无意义的长句，最终向量化的是一堆脏文本，检索效果自然差。

第二个瓶颈是单路向量检索召回不准。向量检索擅长处理“语义接近但字面不同”的查询，比如用户问“怎么把数据库连接池调大”，文档里写的是“增大 maximumPoolSize”。但向量检索对专有名词、编号、精确型号不敏感，查询“MySQL 8.0.36 的 bug”时，包含“8.0.36”的文档未必排得上。

第三个瓶颈是缺少评测与重排。检索返回 20 条结果，直接塞给大模型，有用片段被淹没在噪声里。没有重排机制，没有引用来源，回答错误时无从排查。

一句话总结：玩具级 RAG 把“检索”想得太简单了，生产级 RAG 必须把解析、切片、召回、重排、评测当成一个系统工程来做。

1.3 RAG 知识库、结构化知识库、KG 知识库怎么选

很多读者分不清 RAG 知识库和知识图谱（KG）知识库的区别，这里给一张对比表：

三种知识库不是互斥的，企业里经常混用：

• 查“上个月订单总额”用结构化数据库；
• 查“售后服务条款中的第三条”用 RAG 文档库；
• 查“这个客户和供应商之间存在哪些关联控制关系”用 KG 知识库。

还有一个偏学术的词叫 Ontology RAG，意思是把本体（Ontology）和 RAG 结合，用领域概念体系约束文档切片和检索入口。它解决的是“文档领域性太强、通用向量检索管不住实体关系”的问题，适合医疗、法律、工业标准这类专业场景。这个概念不要在入门阶段死磕，先跑通基础链路，再考虑用本体约束。

2. 企业级 RAG 的总体架构与方案选型

这一节先给出整体架构，再解释每个环节为什么这么选。只有把架构讲清楚，后面写代码时才不会迷路。

2.1 一条完整的落地链路

一个可落到生产环境的 RAG 系统，至少包含下面几个环节：

1. 文档接入：支持 PDF、Word、Markdown、HTML 等格式；
2. 文档解析：抽取文本、表格、图片中的信息，保留版面结构；
3. 切片（Chunking）：把长文本切成适合检索和向量化的片段；
4. 向量化：用 Embedding 模型把文本转为向量；
5. 存储与索引：向量入库到 Milvus，建立向量索引；
6. 混合检索：向量召回 + 关键词召回 + 重排；
7. 生成：把重排后的片段与用户问题组装成 Prompt，交给大模型；
8. 评测与监控：记录召回率、准确率、引用覆盖率。

本文重点落在 2 到 6，因为这是最容易踩坑、也最影响效果的部分。

2.2 为什么选 Milvus 作为向量库

Milvus 是开源向量数据库，当前主流的 2.x 版本已经支持 Standalone 和 Cluster 两种部署形态。选择它的原因主要有几个：

• 性能与规模：十亿级向量场景有成熟方案；
• 索引类型丰富：支持 IVF、HNSW、DiskANN，能平衡召回率与内存；
• 自带混合检索能力：在 2.4 版本中支持 Dense + Sparse 的 Hybrid Search 和 RRF 重排；
• 生态成熟：GitHub Star 多、社区文档完整，PyMilvus SDK 易用。

如果你的数据量在百万级以下，Milvus Standalone 单机模式完全够用，用 Docker Compose 就能拉起来，非常适合在 Mac 上先做企业级架构验证。

2.3 为什么 PDF 解析是命门

RAG 的第一步是文档解析，这一步的质量直接决定后续所有环节的上限。

PDF 的复杂性在于：它没有“段落”和“表格”这种语义概念，只有“坐标”和“绘制指令”。同一个 PDF 里可能有文本层、矢量图形、位图、注释，还可能被扫描成纯图片。解析时必须针对不同情况分别处理：

• 文本型 PDF：用 PyMuPDF 等库提取文本块，能保留大部分内容；
• 复杂表格 PDF：需要表格识别，不能只按坐标切文本；
• 扫描件 PDF：需要 OCR，先识别图片内容再转文本。

如果文档解析阶段就丢信息，之后切块、向量化、检索做得再漂亮，也都是从残缺的数据里找答案。

2.4 为什么需要混合检索而不是单路向量检索

单路向量检索的问题已经在前面提过：语义召回强、字面精确匹配弱。关键词检索则相反，它对精确词、型号、编号、人名非常敏感，但理解不了同义改写。

企业级检索一般做“双路召回 + 重排”：

• 向量检索负责语义召回；
• BM25 关键词检索负责字面召回；
• RRF（Reciprocal Rank Fusion，倒数排名融合）或 Rerank 模型负责把两路结果合并排序。

这样做的好处是，用户输入口语化问题时靠向量，输入精确型号时靠 BM25，两路互补，整体召回率明显提升。

3. 环境准备：在 Mac 上从零搭建 Milvus Standalone

本文示例环境以 macOS + Docker Desktop 为主，Windows 和 Linux 操作步骤基本一致，只是 Docker 安装方式略有差别。

3.1 版本与依赖说明

先明确本文使用的软件栈，避免读者因版本不一致而踩坑：

版本不是死的，生产环境请以你自己项目的实际约束为准，本文重点演示的是架构和配置思路。

3.2 用 Docker Compose 启动 Milvus Standalone

在项目目录下创建 docker-compose.yml，内容如下：

yaml复制version: '3.5'

services:
  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
    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
    restart: always

  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
    restart: always

  standalone:
    image: milvusdb/milvus:latest
    command: ["milvus", "run", "standalone"]
    ports:
      - "19530:19530"
      - "9091:9091"
    volumes:
      - ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/milvus:/var/lib/milvus
    depends_on:
      - etcd
      - minio
    restart: always

打开终端，进入 docker-compose.yml 所在目录，执行：

bash复制docker compose up -d

首次启动会拉取 Milvus、etcd、MinIO 三个镜像，耗时取决于网络状况。启动完成后用命令确认容器状态：

bash复制docker compose ps

看到 standalone、etcd、minio 三个服务处于 Up 状态，就说明环境基本就绪。

这里有两个生产环境注意事项：

• 示例用了 latest 标签，方便尝鲜；生产环境必须固定到具体版本号，例如 milvusdb/milvus:v2.4.15，避免不可控升级；
• Milvus 在 Mac 上默认使用本机 Docker Desktop 的资源，生产部署建议放到 Linux 物理机或 K8s 集群。

3.3 安装 Python 依赖

创建 requirements.txt：

txt复制pymilvus==2.4.9
pymupdf==1.24.10
sentence-transformers==3.0.1
jieba==0.42.1
bm25s==0.1.6
numpy==1.26.4

安装命令：

bash复制pip install -r requirements.txt

如果你所在环境访问 Hugging Face 网络受限，可以给 sentence-transformers 配置镜像地址，这属于正常的网络加速措施：

bash复制export HF_ENDPOINT=https://hf-mirror.com

3.4 验证 Milvus 连接

写一个最小脚本 check_milvus.py：

python复制from pymilvus import connections

connections.connect(host="127.0.0.1", port="19530")
print("Milvus 连接成功")

运行：

bash复制python check_milvus.py

看到 Milvus 连接成功 就说明客户端和服务端是通的。如果连接超时，先检查端口，再确认容器是否正常启动。

4. PDF 解析：不是简单抽文本，而是结构化文本

PDF 解析是 RAG 里最“脏”的环节。这一节用 PyMuPDF 实现一个生产级可用的解析模块，包括文本块和表格提取。

4.1 PyMuPDF 的两种解析方式

PyMuPDF 提供两种常见的文本获取方式：

• page.get_text("text")：输出纯文本，适合快速预览；
• page.get_text("dict")：输出结构化分组，包含页面上的 block、line、span，可以按版面位置重组内容。

企业级解析必须使用 dict 模式，因为我们可以从 block 里区分文本块和图片块，避免把图片描述或页眉页脚混进正文。

4.2 提取文本块与表格

创建 pdf_parser.py：

python复制# -*- coding: utf-8 -*-
import fitz  # PyMuPDF


def extract_pdf_page_blocks(pdf_path: str) -> list[dict]:
    """按页提取 PDF 文本块和表格，返回结构化 chunk 列表。

    chunk 字段：
    - text: 文本内容
    - file_name: 来源文件名
    - page: 页码，从 1 开始
    - source: 内容来源，text_block 表示文本块，table 表示表格
    """
    chunks = []
    doc = fitz.open(pdf_path)

    for page_idx, page in enumerate(doc):
        # 1. 提取文本块
        blocks = page.get_text("dict")
        for block in blocks.get("blocks", []):
            # block["type"] == 0 表示文本块，1 表示图片
            if block["type"] != 0:
                continue

            text = ""
            for line in block.get("lines", []):
                for span in line.get("spans", []):
                    text += span.get("text", "")
            text = text.strip()
            if len(text) < 20:
                continue

            chunks.append({
                "text": text,
                "file_name": pdf_path.split("/")[-1],
                "page": page_idx + 1,
                "source": "text_block"
            })

        # 2. 提取表格
        try:
            tables = page.find_tables()
            for table in tables:
                markdown_text = table.to_markdown()
                if markdown_text and len(markdown_text) > 20:
                    chunks.append({
                        "text": markdown_text,
                        "file_name": pdf_path.split("/")[-1],
                        "page": page_idx + 1,
                        "source": "table"
                    })
        except Exception as e:
            print(f"第 {page_idx + 1} 页表格解析失败: {e}")

    doc.close()
    return chunks

几个关键点：

• block["type"] != 0 时直接跳过，避免把图片位图数据当文本；
• len(text) < 20 用来过滤短噪声，比如页眉“第 1 页”、单字水印；
• 表格用 table.to_markdown() 转成 Markdown，能保留表格行列结构，对后续大模型理解很重要；
• 表格解析失败不要中断整个流程，用 try/except 捕获后继续解析其他页。

4.3 常见 PDF 解析失败场景

文本型 PDF 并不总是能顺利解析，常见情况有三种：

如果企业文档里有大量扫描件，建议在 PyMuPDF 之后串联 OCR 服务，而不是强行依赖文本层。

5. 设计 Milvus 集合：字段、索引与写入

Milvus 里的集合可以理解成 MySQL 里的一张表。这条链路中，向量库承担的是“检索索引”的角色，目的不是存储所有原始文档，而是存储切片文本和可定位文档来源的元数据。

5.1 集合与字段设计

创建 vector_store.py：

python复制# -*- coding: utf-8 -*-
from pymilvus import (
    connections,
    Collection,
    CollectionSchema,
    FieldSchema,
    DataType,
    utility
)

COLLECTION_NAME = "rag_pdf_demo"


def create_collection(dim: int = 1024) -> Collection:
    """创建知识库集合，dim 需要和 Embedding 模型输出维度一致。"""
    connections.connect(host="127.0.0.1", port="19530")

    # 如果集合已存在，直接加载返回
    if utility.has_collection(COLLECTION_NAME):
        return Collection(COLLECTION_NAME)

    fields = [
        FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
        FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=16000),
        FieldSchema(name="file_name", dtype=DataType.VARCHAR, max_length=512),
        FieldSchema(name="page", dtype=DataType.INT64),
        FieldSchema(name="chunk_type", dtype=DataType.VARCHAR, max_length=32),
        FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=dim),
    ]

    schema = CollectionSchema(
        fields=fields,
        description="企业 PDF 知识库文档"
    )

    collection = Collection(name=COLLECTION_NAME, schema=schema)

    # 创建 IVF_FLAT 向量索引，配合 COSINE 距离度量
    index_params = {
        "index_type": "IVF_FLAT",
        "metric_type": "COSINE",
        "params": {"nlist": 1024}
    }
    collection.create_index("embedding", index_params)
    return collection

字段设计思路：

• id 使用自增主键，程序不需要自己生成；
• text 保存切片后的原文，检索结果会展示给用户，也用来拼 Prompt；
• file_name、page 用于定位答案来自哪份文档哪一页，这是企业场景强制要求；
• chunk_type 记录该片段是普通文本还是表格，便于后续调试；
• embedding 是浮点稠密向量，维度必须与模型对齐。

5.2 创建集合和索引

上面代码中，索引类型选了 IVF_FLAT，它适合中等数据量，检索速度和精度比较均衡。在几万到几十万条切片场景下完全够用。

当数据量继续增长，可以切换 HNSW 或 DISKANN。HNSW 图索引召回率高但内存占用大，DISKANN 适合海量数据低成本存储，不过参数调优难度更高。

5.3 Embedding 模型选择与向量化

这里使用国产中英多语模型 BAAI/bge-m3，它支持 1024 维，中文检索效果好，在 sentence-transformers 生态里直接可用。

创建 build_index.py：

python复制# -*- coding: utf-8 -*-
import os

from sentence_transformers import SentenceTransformer

from pdf_parser import extract_pdf_page_blocks
from vector_store import create_collection


def main(pdf_dir: str):
    # 1. 准备集合
    collection = create_collection(dim=1024)
    collection.load()

    # 2. 加载 Embedding 模型
    model = SentenceTransformer("BAAI/bge-m3")

    # 3. 遍历目录下所有 PDF
    all_rows = []
    for file_name in os.listdir(pdf_dir):
        if not file_name.lower().endswith(".pdf"):
            continue

        pdf_path = os.path.join(pdf_dir, file_name)
        print(f"解析: {pdf_path}")

        chunks = extract_pdf_page_blocks(pdf_path)
        for chunk in chunks:
            text = chunk["text"]
            embedding = model.encode(text).tolist()
            all_rows.append([
                text,
                chunk["file_name"],
                chunk["page"],
                chunk["source"],
                embedding
            ])

    # 4. 批量写入 Milvus
    if all_rows:
        collection.insert(all_rows)
        collection.flush()
        print(f"写入完成，共 {len(all_rows)} 条数据")
    else:
        print("没有可写入的数据")


if __name__ == "__main__":
    main("./pdf_files")

运行前先在项目目录下建 pdf_files 文件夹，放入几个真实 PDF：

bash复制mkdir pdf_files
python build_index.py

5.4 批量写入与 flush

写入时注意两个操作：

• collection.insert(all_rows) 是批量写入，注意行内字段顺序必须和创建集合时的字段顺序一致；
• collection.flush() 触发数据落盘，让后续查询能看到刚写入的数据。

生产环境写入量较大时，建议分批插入，比如每 1000 条 flush 一次，避免一次性占用过多内存。写入完成后调用 collection.load() 加载到内存或显存，检索才能命中。

6. 混合检索实战：向量检索 + BM25 + RRF 重排

这一节实现最核心的检索模块。整体思路是：向量检索召回语义相似的文本，BM25 召回包含精确关键词的文本，然后用 RRF 把两路结果融合排序。

6.1 单路向量检索的问题

先看一个典型反例。用户查询：

text复制license key 过期时间在哪里查看

向量检索可能召回语义接近的“证书有效期管理说明”，却漏掉字面上有“license key”“过期时间”的精确文档片段。反之，纯 BM25 可能召回包含关键词但不理解语义的碎片。所以企业场景中要两条腿走路。

6.2 构建 BM25 稀疏索引

BM25 是经典的关键词排序算法，中文场景必须配合分词器使用，这里用 jieba 做粗粒度切词，用到 bm25s 库构建索引。

创建 hybrid_search.py：

python复制# -*- coding: utf-8 -*-
import jieba
import bm25s

from pymilvus import Collection


class HybridSearchEngine:
    def __init__(self, collection_name: str = "rag_pdf_demo"):
        self.collection = Collection(collection_name)
        self.collection.load()
        self.model = None  # 在 build_embedding_model 中加载
        self.bm25 = None
        self.corpus_texts = []
        self.text_meta_map = {}

    def build_embedding_model(self):
        from sentence_transformers import SentenceTransformer
        self.model = SentenceTransformer("BAAI/bge-m3")

    @staticmethod
    def _tokenize(text: str) -> list[str]:
        # 使用 jieba 分词，过滤空白字符
        return [w.strip() for w in jieba.lcut(text) if w.strip() and not w.isspace()]

    def build_sparse_index(self, records: list[dict]):
        """基于全量切片构建 BM25 索引。

        records: 从 Milvus 里导出的 [{"text":..., "file_name":..., "page":..., "chunk_type":...}]
        """
        self.corpus_texts = []
        self.text_meta_map = {}

        for record in records:
            text = record["text"]
            self.corpus_texts.append(text)
            self.text_meta_map[text] = {
                "file_name": record.get("file_name", ""),
                "page": record.get("page", 0),
                "chunk_type": record.get("chunk_type", "")
            }

        corpus_tokens = [self._tokenize(t) for t in self.corpus_texts]
        self.bm25 = bm25s.BM25Okapi()
        self.bm25.index(corpus_tokens)

    def dense_search(self, query: str, top_k: int = 20) -> list:
        """向量召回，返回 Milvus 的 hit 列表。"""
        if self.model is None:
            raise ValueError("请先调用 build_embedding_model()")

        query_embedding = self.model.encode(query).tolist()
        results = self.collection.search(
            data=[query_embedding],
            anns_field="embedding",
            param={"metric_type": "COSINE", "params": {"nprobe": 16}},
            limit=top_k,
            output_fields=["text", "file_name", "page", "chunk_type"]
        )
        return results[0]

    def sparse_search(self, query: str, top_k: int = 20) -> list[tuple[str, float]]:
        """BM25 关键词召回，返回 [(text, score)]。"""
        if self.bm25 is None:
            raise ValueError("请先调用 build_sparse_index()")

        tokenized_query = self._tokenize(query)
        hits = self.bm25.retrieve(tokenized_query, k=top_k)

        records = []
        for doc_id, score in zip(hits.documents[0], hits.scores[0]):
            if doc_id < 0:
                continue
            records.append((self.corpus_texts[doc_id], float(score)))

        return records

6.3 双路召回与 RRF 融合

RRF 的核心思想是：不直接比较向量分数和 BM25 分数，而是按“排名”融合。每一路里排第一的文档给最高分，排名越靠后分数越低，最终把两路的分数加总。

继续在 HybridSearchEngine 中加入融合方法：

python复制    def rrf_merge(
        self,
        dense_hits: list,
        sparse_hits: list[tuple[str, float]],
        k: int = 60,
        top_k: int = 20
    ) -> list[dict]:
        """RRF 融合两路召回结果。k 是平滑因子，一般取 60。"""
        score_map = {}

        # 向量路
        for rank, hit in enumerate(dense_hits):
            text = hit.entity.get("text")
            if text not in score_map:
                score_map[text] = 0.0
            score_map[text] += 1.0 / (k + rank + 1)

        # BM25 路
        for rank, (text, _score) in enumerate(sparse_hits):
            if text not in score_map:
                score_map[text] = 0.0
            score_map[text] += 1.0 / (k + rank + 1)

        # 按融合分数降序排序
        sorted_items = sorted(score_map.items(), key=lambda x: x[1], reverse=True)

        merged_results = []
        for text, rrf_score in sorted_items[:top_k]:
            meta = self.text_meta_map.get(text, {})
            merged_results.append({
                "text": text,
                "rrf_score": round(rrf_score, 6),
                "file_name": meta.get("file_name", ""),
                "page": meta.get("page", 0),
                "chunk_type": meta.get("chunk_type", "")
            })

        return merged_results

需要说明的是，两路召回结果可能包含相同的文本，RRF 正是通过将两路排名分数相加，让同时被两路命中的文档更容易排到前面。这比直接加权平均更稳定，不需要反复调节向量分数和 BM25 分数之间的权重比例。

6.4 一个完整的检索 demo

创建 query_demo.py：

python复制# -*- coding: utf-8 -*-
from pymilvus import Collection

from hybrid_search import HybridSearchEngine


def load_all_records(collection_name: str = "rag_pdf_demo", limit: int = 10000):
    """从 Milvus 拉取全量切片，用于构建 BM25 索引。

    注意：真实生产环境数据量很大时，不应直接全量拉取，而是使用离线导出或增量同步。
    """
    collection = Collection(collection_name)
    collection.load()
    return collection.query(
        expr="file_name != ''",
        output_fields=["text", "file_name", "page", "chunk_type"],
        limit=limit
    )


def main():
    engine = HybridSearchEngine()
    engine.build_embedding_model()

    # 构建稀疏索引
    records = load_all_records()
    engine.build_sparse_index(records)
    print(f"BM25 索引构建完成，共 {len(records)} 条切片")

    query = "license key 过期时间在哪里查看"
    dense_hits = engine.dense_search(query, top_k=20)
    sparse_hits = engine.sparse_search(query, top_k=20)

    final_results = engine.rrf_merge(dense_hits, sparse_hits, top_k=5)

    print("\n混排结果：")
    for idx, item in enumerate(final_results, 1):
        print(f"Top {idx} 分数={item['rrf_score']} 来源={item['file_name']} 第{item['page']}页")
        print(item["text"][:180])
        print("-" * 60)


if __name__ == "__main__":
    main()

运行结果会打印混排后的 Top 5 片段。把这些片段作为上下文，与用户问题一起组装 Prompt，交给大模型即可生成带引用的回答。

这里给一个生成阶段的 Prompt 模板：

text复制请根据以下参考资料回答用户问题。如果资料不足以回答，请明确说“资料中没有相关信息”。

参考资料：
[1] 文件名：产品手册.pdf，第 12 页
内容：...

[2] 文件名：FAQ.docx，第 3 页
内容：...

用户问题：...

注意，引用来源不是可选项，企业级 RAG 必须让回答可追溯。

7. 常见问题与排查思路

这一节整理我在落地过程中遇到过的高频问题，按“现象 - 原因 - 解决思路”描述。

排查时可以按下面顺序走一遍：

1. 先看数据源头：PDF 解析出的 chunk 是否完整、可读；
2. 再看向量化：query 和文档是否使用同一个 Embedding 模型；
3. 然后看召回：把向量结果和 BM25 结果分别打印，确认是哪一路出了问题；
4. 最后看融合：确认 RRF 参数 k 和 top_k 是否符合预期。

8. 企业级落地的最佳实践

代码跑通只完成了 30%，剩下 70% 是工程化和运维问题。

8.1 PDF 解析层面

• 不要一刀切用同一个解析策略，先按 PDF 是否含文本层分桶；
• 扫描件先 OCR 再解析，否则文本全是空；
• 表格统一转换成 Markdown 或 HTML 后再入库，而不是保存为散落文本；
• 对解析质量做抽检，建立“解析失败率”指标，比如每 100 页中有多少页无法提取有效文本。

8.2 切片策略

切片大小没有标准答案，常见取值范围在 200 到 800 字之间，但更关键的是“不要切断语义”。

推荐优先级：

• 按版面片段切，例如 PyMuPDF 解析出的 block 天然保留文本块边界；
• 如果片段过长，再按句子边界二次切分；
• 相邻切片保留少量重叠，避免跨切片语义断裂；
• 表格单独作为一个切片单元，不要和其他正文混在一起。

8.3 检索与重排层面

混合检索中，RRF 适合作为通用融合方案。如果追求更高精度，可以增加一个 Rerank 模型，对两路召回结果做精确打分排序。Rerank 模型一般比普通 Embedding 模型更重，但只对 Top 50 以内的结果打分，性能开销可接受。

除此之外，企业场景还会遇到“查最近更新的文档”这类时间敏感需求。建议在集合里增加 update_time 字段，在 RRF 融合后叠加时间衰减因子，让旧数据不会长期霸占 Top 结果。

8.4 数据安全与权限

这是企业级 RAG 最容易忽略的部分。很多团队辛辛苦苦把知识库建好，直接把检索能力暴露给所有员工，这是很危险的做法。

需要注意：

• 文档接入前要做权限分级，敏感文档不能进入公共检索集合；
• 鉴权必须放在检索入口，而不是只依赖大模型的 Prompt 约束；
• 提问日志和检索记录要脱敏保存，便于安全审计；
• 对生产环境的集合删除、清空操作，必须先备份并在测试环境验证。

不要把内部文档不加区分地全部灌入同一个集合，这是“玩具级”和“企业级”的重要分界线。

8.5 生产环境的版本与运维

• Milvus 镜像固定版本，不用 latest；
• 定期备份 etcd 和 MinIO 数据，Milvus 的元数据和数据文件分别存在这两个组件中；
• 监控 Milvus 资源占用，重点关注集合加载后的内存变化；
• 建立文档更新的全量/增量同步机制，避免知识库内容越用越旧。

9. 总结与学习路线

这篇教程从 RAG 的瓶颈与知识库选型开始，走了一遍企业级 RAG 的核心链路：Milvus Standalone 环境搭建、PDF 结构化解析、切片与向量化、Milvus 集合设计、混合检索与 RRF 重排。到这里，你已经拥有一套可复现的 RAG 工程骨架，不再是单点的小 Demo。

接下来可以按下面顺序继续深入：

1. 替换更专业的 PDF 解析服务，比如 OCR、版面分析、表格识别模型；
2. 引入 Rerank 模型，评测它对最后生成效果的提升幅度；
3. 设计评测集，覆盖“精确词查询、同义查询、跨文档查询、无答案查询”四类问题；
4. 调研 Milvus 2.4 的原生 Hybrid Search 能力，用稀疏向量替代外部 BM25 索引；
5. 在权限隔离、多租户、增量同步三个方向上做工程化改造。

如果这篇教程帮助你把 RAG 从玩具级往前推进了一步，可以先收藏备用，等真正开始设计生产架构时再打开对照落地。