告别检索幻觉！手把手搭建企业级 RAG 数据管道（附 Docker 一键部署） · devcfg

痛点：直接用通用 Embedding 模型做 RAG，遇到垂直领域总是答非所问？

你有没有遇到过这种情况：

• 用 OpenAI 的 text-embedding-3-small 做医疗文档检索，问”糖尿病的并发症有哪些”，返回的是”感冒的症状”
• 用通用的 BGE 模型检索法律合同，它把”违约责任”和”不可抗力”混为一谈
• 花了几天搭好 LangChain + ChromaDB，一扔真实业务数据进去，效果惨不忍睹

核心原因不是模型不够好，而是你的 RAG 数据管道没建对。

一个生产级的 RAG 系统，远不止”切分文本 → 向量化 → 存入向量库”这三步。你需要：

1. 多格式文件提取（PDF/Word/PPT/Excel/HTML…）
2. 智能分块策略（不是简单的按字数切）
3. 领域 Embedding 模型（BGE-M3 本地部署）
4. 双写存储架构（MySQL 元数据 + Milvus 向量库）
5. 表格特殊处理（传统方案的最大盲区）

读完本文，你将获得：

• ✅ 一个可运行的完整 RAG 数据处理 Pipeline
• ✅ Docker Compose 一键启动全栈环境
• ✅ 9 种文件格式的自动提取能力
• ✅ BGE-M3 1024 维本地向量化
• ✅ Milvus + MySQL + MinIO 三存储双写

---

前置检查清单

在开始之前，请确认你的环境满足以下要求：

硬件要求

组件	最低配置	推荐配置	说明
CPU	4 核	8 核+	PaddleOCR 和 BGE-M3 推理需要
内存	16 GB	32 GB+	PaddleOCR-VL 加载约需 4-6GB
GPU 显存	无（CPU 可跑）	≥12 GB	BGE-M3 微调需要，纯推理 CPU 也能用
磁盘	20 GB 可用	50 GB+	Milvus 数据 + 模型文件 + 原始文件

💡 技巧：没有 GPU？没问题。本项目支持 CPU 模式运行 BGE-M3 推理，只是速度慢一些。如果只有 CPU，建议将 EMBEDDING_DEVICE 设为 "cpu"。

软件环境

软件	版本要求	验证命令
Python	3.10.x	python --version
Docker	≥24.0	docker --version
Docker Compose	≥2.20	docker compose version
Git	最新版	git --version

# 快速检查所有环境
echo "=== Python ===" && python --version
echo "=== Docker ===" && docker --version
echo "=== Docker Compose ===" && docker compose version

---

架构全景图

先看整体架构，了解我们要搭建什么：

图 1：企业级 RAG 离线数据处理全架构

---

参见站内：《RAG 离线部分：多源异构数据清洗与去重策略》 — 摄入前的清洗、去重与质量闸门

步骤 1：Docker 一键拉起全栈环境

我们使用 Docker Compose 同时启动三个存储服务：

docker-compose.yml

version: '3.8'

services:
  milvus:
    image: milvusdb/milvus:v2.4-latest
    container_name: rag-milvus
    ports:
      - "19530:19530"
      - "9091:9091"
    environment:
      ETCD_USE_EMBED: "true"
      ETCD_DATA_DIR: "/var/lib/milvus/etcd"
      COMMON_STORAGETYPE: "local"
    volumes:
      - milvus_data:/var/lib/milvus
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:9091/healthz"]
      interval: 30s
      timeout: 10s
      retries: 3

  mysql:
    image: mysql:8.0
    container_name: rag-mysql
    ports:
      - "3306:3306"
    environment:
      MYSQL_ROOT_PASSWORD: "rag_password_2024"
      MYSQL_DATABASE: "rag_base_multimodal"
      MYSQL_CHARSET: "utf8mb4"
    command:
      - --character-set-server=utf8mb4
      - --collation-server=utf8mb4_unicode_ci
      - --max_connections=500
      - --innodb_buffer_pool_size=512M
    volumes:
      - mysql_data:/var/lib/mysql
    healthcheck:
      test: ["CMD", "mysqladmin", "ping", "-h", "localhost", "-proot"]
      interval: 30s
      timeout: 10s
      retries: 3

  minio:
    image: minio/minio:latest
    container_name: rag-minio
    ports:
      - "9000:9000"
      - "9001:9001"
    environment:
      MINIO_ROOT_USER: "rag_minio_admin"
      MINIO_ROOT_PASSWORD: "rag_minio_password_2024"
    command: server /data --console-address ":9001"
    volumes:
      - minio_data:/data
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
      interval: 30s
      timeout: 10s
      retries: 3

volumes:
  milvus_data:
  mysql_data:
  minio_data:

启动服务

# 进入项目目录
cd /path/to/rag-base-data

# 后台启动全部服务
docker compose up -d

# 查看服务状态（等待 healthy 状态）
docker compose ps

# 查看日志（排查问题时使用）
docker compose logs -f milvus
docker compose logs -f mysql

预期输出（healthy 表示就绪）：

NAME         IMAGE                    STATUS                    PORTS
rag-milvus   milvusdb/milvus:v2.4     Up Healthy (healthy)       0.0.0.0:19530->19530/tcp, 0.0.0.0:9091->9091/tcp
rag-mysql    mysql:8.0                Up Healthy (healthy)       0.0.0.0:3306->3306/tcp
rag-minio    minio/minio:latest       Up Healthy (healthy)       0.0.0.0:9000->9000/tcp, 0.0.0.0:9001->9001/tcp

---

步骤 2：克隆项目与安装依赖

# 克隆项目
cd rag-base-data

# 创建虚拟环境（推荐 Python 3.10）
python -m venv .venv
source .venv/bin/activate  # Linux/Mac
# .venv\Scripts\activate   # Windows

# 升级 pip
pip install --upgrade pip

# 安装依赖
pip install -r requirements.txt

核心依赖说明

依赖包	版本	用途
llama-index	≥0.11.0	RAG 框架核心
pymilvus	≥2.3.0	Milvus 向量数据库客户端
paddleocr[doc-parser]	≥3.3.0	PDF OCR 引擎（图像型）
PyMuPDF	≥1.23.0	PDF 文本快速提取
pdfplumber	≥0.10.0	PDF 表格精确提取
sqlalchemy	≥2.0.0	MySQL ORM
pymysql	≥1.1.0	MySQL 驱动
sentence-transformers	≥2.7.0	BGE-M3 Embedding 模型

⚠️ 注意：PaddleOCR 安装包较大（约 500MB），安装时间可能较长。如果不需要 PDF OCR 功能，可以从 requirements.txt 中移除 paddleocr 相关依赖。

---

步骤 3：配置环境变量

项目使用 Pydantic Settings 管理配置，支持环境变量覆盖。创建 .env 文件：

# .env 文件（放在项目根目录）

# Milvus 配置
MILVUS_HOST=localhost
MILVUS_PORT=19530
MILVUS_COLLECTION_NAME=rag_documents
MILVUS_DIMENSION=1024

# MySQL 配置
MYSQL_HOST=localhost
MYSQL_PORT=3306
MYSQL_USER=root
MYSQL_PASSWORD=rag_password_2024
MYSQL_DATABASE=rag_base_multimodal

# MinIO 配置
MINIO_ENDPOINT=localhost:9000
MINIO_ACCESS_KEY=rag_minio_admin
MINIO_SECRET_KEY=rag_minio_password_2024
MINIO_BUCKET=knowledge-bucket

# Embedding 配置
EMBEDDING_MODEL_PATH=./data_base/llm_models/bge-m3
EMBEDDING_DEVICE=cpu

---

参见站内：《RAG 落地：生产环境部署与性能监控实践》 — Pipeline 上线后的监控与容量规划

步骤 4：下载 BGE-M3 模型

BGE-M3 是智源研究院（BAAI）开发的多语言嵌入模型，支持 Dense、Sparse、ColBERT 三种检索模式。

# 方式 1：使用 HuggingFace CLI 下载（推荐）
pip install huggingface_hub
huggingface-cli download BAAI/bge-m3 \
    --local-dir ./data_base/llm_models/bge-m3

# 方式 2：使用 git lfs
git lfs install
git clone https://huggingface.co/BAAI/bge-m3 ./data_base/llm_models/bge-m3

下载完成后，模型目录结构如下：

data_base/llm_models/bge-m3/
├── config.json              # 模型配置（隐藏层维度等）
├── pytorch_model.bin        # 模型权重（约 2.2GB）
├── tokenizer.json           # 分词器
├── tokenizer_config.json
├── special_tokens_map.json
├── sentence_bert_config.json # Sentence-Transformers 配置
├── modules.json             # 模块定义
├── 1_Pooling/
│   └── config.json          # Pooling 层配置（mean pooling）
├── colbert_linear.pt        # ColBERT 线性层权重
└── sparse_linear.pt         # Sparse 权重

💡 为什么选 BGE-M3？

• 支持 100+ 语言（中文效果优秀）
• 输出 1024 维向量（精度高）
• 最大上下文长度 8192 tokens（长文档友好）
• MTEB 多语言榜单 Top 级表现

---

步骤 5：扔一个 PDF 进去，看完整数据流

现在来实际运行！准备一份测试 PDF 文件：

# 复制测试文件到 data 目录
cp your-test-file.pdf data_base/data/raw/

# 运行完整 Pipeline（双写入库模式）
python main.py \
    --file data_base/data/raw/test001.pdf \
    --ingest \
    --business-tag "AI技术底座" \
    --chunk-profile default

CLI 参数说明

参数	说明	示例
--file	单个文件路径	/path/to/doc.pdf
--dir	目录路径（批量）	/path/to/data/
--ingest	启用双写入库模式	（flag）
--business-tag	业务标签（数据隔离）	"医疗知识库"
--chunk-profile	分块策略	default / source_first / precision
--force-vl	强制 PDF 使用 OCR	（flag）
--extract-only	仅提取不入库	（flag）

实际运行日志输出示例

2026-04-23 00:30:35 | INFO  | 日志系统初始化完成 | 日志目录: .../logs
2026-04-23 00:30:36 | INFO  | LlamaRAGPipeline 初始化完成
2026-04-23 00:30:36 | INFO  | CustomRAGLoader 初始化: force_vl=False
2026-04-23 00:30:37 | INFO  | 提取文件: test001.pdf (格式: PDF)
2026-04-23 00:30:38 | INFO  | PDF 类型检测: text_pdf (使用 PyMuPDF 快速模式)
2026-04-23 00:30:39 | INFO  | 提取完成: 3 个文本块, 2 个表格, 1 张图片
2026-04-23 00:30:40 | INFO  | TableParseTransform: 2 个表格验证通过
2026-04-23 00:30:40 | INFO  | SymbolMapTransform: 替换符号 5 个 (±,≥,≤...)
2026-04-23 00:30:41 | INFO  | ChunkRouter 初始化: profile=default, 默认策略=parent_child
2026-04-23 00:30:42 | INFO  | 分块路由完成: 6 个文档 → 28 个chunk (parent=5, child=18, table=5)
2026-04-23 00:30:43 | INFO  | Embedding 模型构建完成: ./data_base/llm_models/bge-m3
2026-04-23 00:31:05 | INFO  | Milvus 插入 28 条向量
2026-04-23 00:31:06 | INFO  | MySQL 写入 28 条记录
2026-04-23 00:31:06 | INFO  | 执行结果: {"status": "success", "mysql_count": 28, "milvus_count": 28}

从日志中可以看到完整的处理流程：

1. 提取阶段：识别 PDF 为文本型 → PyMuPDF 提取 → 得到 3 文本 + 2 表格 + 1 图片
2. 清洗阶段：表格验证通过 + 符号映射替换
3. 分块阶段：父子分块策略 → 6 个文档变成 28 个 chunk（5 父块 + 18 子块 + 5 表格）
4. 入库阶段：BGE-M3 向量化 → 28 条写入 Milvus + 28 条写入 MySQL

---

步骤 6：验证 Milvus 向量是否写入成功

写入成功后，我们来验证一下向量检索是否正常工作：

from data_base.storage.milvus_store import milvus_store
from llama_index.embeddings.huggingface import HuggingFaceEmbedding

# 初始化
milvus_store.init()
milvus_store.create_collection()

# 构建 Embedding 模型
embed_model = HuggingFaceEmbedding(
    model_name="./data_base/llm_models/bge-m3",
    device="cpu",
)

# 生成查询向量
query = "Transformer 的注意力机制是怎么工作的？"
query_vector = embed_model.get_query_embedding(query)

# 执行检索
results = milvus_store.search(
    query_vector=query_vector,
    limit=5,
    business_tag="AI技术底座",
)

# 打印结果
for i, r in enumerate(results):
    print(f"[{i+1}] score={r['distance']:.4f} | doc_id={r['doc_id']}")

预期输出：

[1] score=0.8923 | doc_id=1723847192837647001
[2] score=0.8541 | doc_id=1723847192837647002
[3] score=0.8217 | doc_id=1723847192837647003
[4] score=0.7892 | doc_id=1723847192837647004
[5] score=0.7534 | doc_id=1723847192837647005

🔑 score 越接近 1，表示相似度越高。Milvus 使用 COSINE（余弦相似度），范围 [-1, 1]，对于同类型文本通常在 [0.7, 1.0] 区间。

---

参见站内：《RAG 评估：全链路指标设计与效果评测体系》 — 如何衡量检索与生成是否「幻觉」

核心代码解读

主入口 main.py

项目的入口非常简洁，通过 argparse 提供 CLI 接口：

import argparse
from data_base.extensions.llama_pipeline import LlamaRAGPipeline

def main():
    parser = argparse.ArgumentParser(description="RAG 数据层 Pipeline")
    parser.add_argument("--file", type=str, help="单个文件路径")
    parser.add_argument("--dir", type=str, help="目录路径")
    parser.add_argument("--ingest", action="store_true", help="双写入库")
    parser.add_argument("--business-tag", type=str, default="", help="业务标签")
    parser.add_argument("--chunk-profile", type=str, default="default",
                        choices=["default", "source_first", "precision"])
    parser.add_argument("--force-vl", action="store_true", help="强制PDF使用OCR")

    args = parser.parse_args()
    
    pipeline = LlamaRAGPipeline(
        collection_name=args.collection,
        force_vl=args.force_vl,
    )
    
    result = pipeline.run_dual_write_ingest(
        file_path=args.file,
        directory=args.dir,
        business_tag=args.business_tag,
        chunk_profile=args.chunk_profile,
    )

Pipeline 核心流程

LlamaRAGPipeline 是整个数据处理的总编排器，内部流程如下：

class LlamaRAGPipeline:
    def run_dual_write_ingest(self, file_path, directory, 
                               business_tag="", chunk_profile="default"):
        # Step 1: 文件提取（9种格式自动路由）
        loader = CustomRAGLoader(force_vl=self.force_vl)
        documents = loader.load_data(file_path=file_path)
        
        # Step 2: Transform 清洗（5个处理器链式调用）
        transforms = [
            TableParseTransform(),    # 表格验证
            SymbolMapTransform(),     # 符号映射 ±→正负公差
            OCRCorrectTransform(),    # OCR纠错
            DataCleanTransform(),     # 噪声清洗
            DataNormalizeTransform(), # 标准化
        ]
        for t in transforms:
            documents = t(documents)
        
        # Step 3: 分块路由（按内容类型自动选择策略）
        router = ChunkRouter(profile=get_chunk_profile(chunk_profile))
        chunks = router.route(documents)
        
        # Step 4: 双写入库（MySQL + Milvus + MinIO）
        dual_write = DualWriteService()
        result = dual_write.store_documents(
            documents=documents,
            business_tag=business_tag,
            embedding_fn=self._build_embed_model(),
            chunk_profile_name=chunk_profile,
        )
        return result

---

支持的文件格式一览

项目内置了 9 种格式提取器，覆盖企业场景绝大多数文件类型：

格式	提取器	特殊处理
PDF	PdfExtractor	自动判断文本/图像型，双引擎切换
Word (.docx/.doc)	WordExtractor	样式保留、标题层级识别
PPT (.pptx/.ppt)	PptExtractor	幻灯片逐页提取
Excel (.xlsx/.xls/.csv)	ExcelExtractor	多 Sheet 支持
HTML	HtmlExtractor	正文提取、标签过滤
TXT	TxtExtractor	编码自动检测
ZIP/RAR	ArchiveExtractor	解压后递归提取
图片	ImageExtractor	MLX-VLM 多模态理解
邮件	EmailExtractor	头部分析 + 正文提取

每种提取器的输出统一为标准 Document 对象，后续的 Transform 和 ChunkRouter 无需关心来源格式。

---

避坑指南与性能调优

问题 1：CUDA Out of Memory（GPU 显存不足）

现象：运行时报错 RuntimeError: CUDA out of memory

解决方案：

# 方案 A：切换到 CPU 模式（最简单）
# 在 .env 中设置：
EMBEDDING_DEVICE=cpu

# 方案 B：减小 batch size
# 在 settings.py 中修改：
EMBEDDING_BATCH_SIZE: int = 8   # 从 32 降到 8

# 方案 C：启用梯度检查点（微调时）
# 训练脚本中添加：
model.gradient_checkpointing_enable()

问题 2：Milvus 连接失败

现象：ConnectionError: Failed to connect to Milvus

排查步骤：

# 1. 确认容器正在运行
docker compose ps

# 2. 确认端口可访问
curl http://localhost:9091/healthz

# 3. 查看 Milvus 日志
docker compose logs milvus --tail=50

# 4. 检查防火墙是否阻止了 19530 端口
netstat -an | grep 19530

问题 3：PaddleOCR 安装失败或加载缓慢

现象：pip install paddleocr 报错，或首次加载耗时超过 5 分钟

解决方案：

# 如果不需要 PDF OCR 功能，可以跳过
# 在 PdfExtractor 中设置 use_ocr=False

# 或者仅安装轻量版本
pip install paddlepaddle>=3.2.1
pip install paddleocr[doc-parser] --no-deps
pip install paddleocr 的其他依赖手动安装

问题 4：MySQL 字符集乱码

现象：中文内容存入 MySQL 后显示为 ????

解决方案：确保以下三点：

1. MySQL 容器启动参数包含 --character-set-server=utf8mb4
2. SQLAlchemy 连接 URL 包含 ?charset=utf8mb4
3. 数据库和表都使用 utf8mb4 字符集

性能调优参考表

参数	默认值	显存小调低	显存大调高	影响
EMBEDDING_BATCH_SIZE	32	8	64	批量向量化速度
PARENT_CHUNK_SIZE	1024	512	2048	父块大小
CHILD_CHUNK_SIZE	128	64	256	子块粒度
MILVUS HNSW.M	16	8	32	索引召回率 vs 内存
MILVUS HNSW.efConstruction	200	100	500	建索引速度 vs 质量

---

FAQ

Q：没有 GPU 可以运行吗？

A：可以。本项目完全支持 CPU 模式运行 BGE-M3 推理。只需将 .env 中的 EMBEDDING_DEVICE 设为 cpu 即可。CPU 模式下速度约为 GPU 的 1/10~1/20，对于少量数据完全够用。如果数据量大（万级文档以上），建议租用云 GPU（如 AutoDL、阿里云 PAI）。

Q：支持哪些文件格式？

A：当前支持 9 种格式：PDF、Word（.docx/.doc）、PPT（.pptx/.ppt）、Excel（.xlsx/.xls/.csv）、HTML、TXT、ZIP/RAR 压缩包、图片、邮件。如果需要支持更多格式（如 Markdown、ePub），只需新增一个 Extractor 类并注册到 ExtractorDispatcher 即可。

Q：如何切换 Embedding 模型？

A：修改 .env 中的 EMBEDDING_MODEL_PATH 指向新模型目录即可。项目基于 Sentence-Transformers 接口，兼容所有支持该格式的模型（如 text-embedding-3-small、e5-mistral-7b-instruct 等）。注意新模型的输出维度需要与 MILVUS_DIMENSION 配置一致。

Q：--chunk-profile 三个选项有什么区别？

A：

• default（默认）：父子分块 — 平衡检索精度和上下文完整性，适合大多数场景
• source_first：不分块溯源 — 保持原文完整性，适合法律/合同/财报场景
• precision：精准小粒度 — 更细的分块粒度，适合 FAQ/客服问答场景

详细对比请阅读本系列第 3 篇《RAG 分块怎么做才不丢上下文？》

Q：--force-vl 什么时候需要用？

A：当你的 PDF 是扫描件或图片型 PDF（文字是嵌入在图片中的）时，普通文本提取器无法读取内容，此时需要加 --force-vl 强制使用 PaddleOCR-VL 进行视觉识别。代价是速度慢 5-10 倍、内存占用更高。

Q：BGE-M3 和 MTEB 榜单的关系是什么？

A：MTEB（Massive Text Embedding Benchmark）是目前最权威的 Embedding 模型评测基准。BGE-M3 在 MTEB 多语言榜单上排名前列，尤其在中文任务上表现优异。选择 MTEB 高排名模型可以保证你有一个好的起点，但垂直领域的微调仍然必要——这将在本系列第 4 篇详解。

Q：不做微调直接用 RAG 效果不好怎么办？

A：通用模型在垂直领域的表现确实会打折扣。建议按优先级尝试：

1. 先优化分块策略（换用父子分块）
2. 尝试不同的 top-k 和 similarity threshold
3. 加入关键词检索（BM25）做多路召回
4. 最后考虑领域微调（见第 4 篇教程）

Q：如何处理大量文件的批量导入？

A：使用 --dir 参数指定目录即可批量处理：

python main.py \
    --dir /path/to/your/documents/ \
    --ingest \
    --business-tag "企业知识库" \
    --chunk-profile default

Pipeline 会自动遍历目录下所有支持的文件格式，逐个提取和处理。

---

资源下载与互动

---

写在最后

你在搭建 RAG 系统的过程中遇到了什么问题？欢迎在评论区留言，我会逐一回复。

常见的问题方向：

• 你的业务场景是什么？（医疗/法律/金融/教育/客服？）
• 目前用的什么技术栈？
• 最头疼的问题是哪一块？（提取不准？分块不合理？检索不相关？）

下一篇文章预告：《PDF 提取总是丢表格？PyMuPDF + PaddleOCR-VL 混合方案实战》，我们将深入拆解 PDF 提取的双引擎架构，以及 Apple Silicon 用户独享的 MLX-VLM 加速方案。敬请关注！

---

专题导航与站内延伸

本文属于 **企业级 RAG 数据管道实战专题**（工程实战 8 篇，与 RAG 实战全链路理论系列 配套阅读）。

本专题篇章

篇章	标题
第 1 篇	告别检索幻觉！手把手搭建企业级 RAG 数据管道（附 Docker 一键部署）
第 2 篇	PDF 提取总是丢表格？PyMuPDF + PaddleOCR-VL 混合方案实战（含 MLX 加速）
第 3 篇	RAG 分块怎么做才不丢上下文？5 种策略从入门到生产级（附选型决策树）
第 4 篇	BGE-M3 本地微调实战：从零搭建到生产级部署（附完整代码）
第 5 篇	Milvus 生产环境 Collection 设计 + HNSW 调优实战指南
第 6 篇	表格 4 级向量化方案：让 RAG 系统真正理解结构化数据
第 7 篇	RRF 多路融合排序：让 RAG 检索精度提升 30%+ 的秘密武器
第 8 篇	MySQL+Milvus+MinIO 三存储双写架构：构建企业级 RAG 数据底座

站内理论延伸

以下文章来自 RAG 全链路理论系列，帮助理解本专题所依赖的概念与方法论：

• 《RAG 离线部分：多源异构数据清洗与去重策略》 — 摄入前的清洗、去重与质量闸门
• 《RAG 落地：生产环境部署与性能监控实践》 — Pipeline 上线后的监控与容量规划
• 《RAG 评估：全链路指标设计与效果评测体系》 — 如何衡量检索与生成是否「幻觉」