经过前14天的学习，你已经掌握了RAG技术的各个核心环节：向量数据库原理、文档处理技巧、企业知识库搭建、Rerank重排技术。今天，我们把这些知识融会贯通，从项目规划、环境搭建、到功能实现、部署上线，完整构建一个生产级别的RAG知识库系统。

这不是玩具项目，是一个真正能上线使用的系统。

一、项目需求与架构设计

1.1 我们要做什么

为技术团队构建一个智能知识库系统，解决实际问题：

痛点场景：团队积累了大量技术文档（API文档、需求文档、故障排查手册），每次找答案要翻半天。百度/Google搜不到内部资料，大模型也不知道答案。

解决方案：基于RAG技术，让AI能理解并回答关于内部文档的问题。

核心功能清单：

1.2 系统架构

系统分5层，从上到下各司其职：

数据流向：用户提问 → 文档处理层解析 → 向量检索层初筛 → Rerank层精排 → 生成层整合 → 返回答案

1.3 技术选型理由

二、环境搭建

2.1 一键启动Milvus

Milvus是整个系统的核心，负责存储向量和执行检索。我们用Docker Compose一键启动整个Milvus集群（包含etcd存储元数据、MinIO存储文件、Milvus本身）。

为什么用Docker Compose：Milvus依赖多个组件（etcd注册发现、MinIO对象存储），手动一个个装很麻烦。Docker Compose一键启动，还能随时停用删除，不污染本地环境。

docker-compose.yml文件内容：

version: '3.8'
# 定义4个服务：etcd(元数据)、MinIO(文件存储)、Milvus(向量数据库)、knowledge-base(我们的应用)

services:
  # etcd：Milvus的注册中心，存储集群节点信息
  milvus-etcd:
    image: quay.io/coreos/etcd:v3.5.5
    environment:
      - ETCD_AUTO_COMPACTION_MODE=revision
      - ETCD_AUTO_COMPACTION_RETENTION=1000
    volumes:
      - ./data/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：对象存储，Milvus用它存索引文件（不是存向量，向量存Milvus自己）
  milvus-minio:
    image: minio/minio:RELEASE.2023-03-20T20-16-18Z
    environment:
      MINIO_ACCESS_KEY: minioadmin
      MINIO_SECRET_KEY: minioadmin
    volumes:
      - ./data/minio:/minio_data
    command: minio server /minio_data --console-address ":9001"

  # Milvus：向量数据库主服务
  milvus:
    image: milvusdb/milvus:v3.0.0
    ports:
      - "19530:19530"   # Milvus客户端连接端口
      - "9091:9091"     # Prometheus监控端口
    environment:
      ETCD_ENDPOINTS: milvus-etcd:2372
      MINIO_ADDRESS: milvus-minio:9000
    volumes:
      - ./data/milvus:/var/lib/milvus
    depends_on:
      - milvus-etcd
      - milvus-minio

  # 我们的RAG应用
  knowledge-base:
    build: .
    ports:
      - "7860:7860"    # Gradio界面端口
    environment:
      MILVUS_HOST: milvus
      MILVUS_PORT: 19530
    volumes:
      - ./uploads:/app/uploads
    depends_on:
      - milvus

启动命令（在docker-compose.yml所在目录执行）：

# 拉取镜像并启动（首次运行）
docker-compose up -d

# 查看服务状态
docker-compose ps

# 查看Milvus日志（确认启动成功）
docker-compose logs milvus | tail -20

# 停止所有服务
docker-compose down

# 清理数据重新开始
docker-compose down -v

验证Milvus是否启动成功：

# 尝试连接Milvus（需要先安装pymilvus）
python3 -c "from pymilvus import connections; connections.connect('default', host='localhost', port='19530'); print('Milvus连接成功！')"

2.2 Python依赖安装

把需要用到的Python包整理到一个文件里，一键安装：

# 创建虚拟环境（推荐）
python3 -m venv venv
source venv/bin/activate  # Windows用: venv\Scripts\activate

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

requirements.txt内容：

pymilvus>=2.3.0        # Milvus Python客户端，连接数据库用
langchain>=0.1.0       # RAG应用框架，封装好了很多RAG流程
langchain-community>=0.0.10  # LangChain社区组件（各种集成）
sentence-transformers>=2.2.0  # 加载嵌入模型和Rerank模型
transformers>=4.35.0   # HuggingFace Transformers库
torch>=2.0.0           # PyTorch深度学习框架
gradio>=4.0.0          # 快速构建Web界面
pypdf>=3.17.0          # PDF解析
python-docx>=1.0.0     # Word文档解析
pymupdf>=1.23.0        # 更好的PDF解析（可选）
beautifulsoup4>=4.12.0  # HTML/文本清洗

安装常见问题：

三、核心代码实现

3.1 文档处理模块

这段代码做什么：把PDF、Word、Markdown等各种格式的文档，转换成纯文本，然后切成小块（chunk），方便后续向量化和检索。

为什么需要分块：文档太长不利于向量检索——语义太杂、向量表示不精准。一般每个chunk 300-500字效果比较好。

from abc import ABC, abstractmethod
from pathlib import Path
from typing import List, Dict, Any
import fitz          # PyMuPDF，PDF解析库
import docx         # python-docx，Word文档解析
import markdown     # Markdown转HTML/文本

# 定义抽象基类：所有文档处理器都按这个接口来
class DocumentProcessor(ABC):
    @abstractmethod
    def process(self, file_path: str) -> List[Dict[str, Any]]:
        """处理文档，返回chunk列表"""
        pass

# PDF处理器：按段落分块
class PDFProcessor(DocumentProcessor):
    def __init__(self, chunk_size: int = 500, overlap: int = 50):
        """
        chunk_size: 每块多少字符（不要太长，500左右效果好）
        overlap: 块与块之间的重叠（防止上下文丢失）
        """
        self.chunk_size = chunk_size
        self.overlap = overlap
    
    def process(self, file_path: str) -> List[Dict[str, Any]]:
        doc = fitz.open(file_path)
        chunks = []
        
        for page_num, page in enumerate(doc):
            # 提取纯文本（不要表格按图片）
            text = page.get_text("text")
            if not text.strip():
                continue
            
            # 按换行分块
            lines = text.split("\n")
            current_chunk = ""
            
            for line in lines:
                # 如果当前块加上这行超过大小，就保存并开始新块
                if len(current_chunk) + len(line) > self.chunk_size:
                    if current_chunk:
                        chunks.append({
                            "content": current_chunk.strip(),
                            "metadata": {
                                "source": str(file_path),
                                "page": page_num + 1,
                                "type": "pdf"
                            }
                        })
                    # 新块从当前行开始
                    current_chunk = line
                else:
                    current_chunk += "\n" + line
            
            # 处理最后一页的剩余内容
            if current_chunk.strip():
                chunks.append({
                    "content": current_chunk.strip(),
                    "metadata": {
                        "source": str(file_path),
                        "page": page_num + 1,
                        "type": "pdf"
                    }
                })
        
        doc.close()
        return chunks

# Word处理器：按段落分块（逻辑和PDF类似）
class WordProcessor(DocumentProcessor):
    def process(self, file_path: str) -> List[Dict[str, Any]]:
        doc = docx.Document(file_path)
        chunks = []
        current_chunk = ""
        
        for para in doc.paragraphs:
            text = para.text.strip()
            if not text:
                continue
            
            if len(current_chunk) + len(text) > 500:
                if current_chunk:
                    chunks.append({
                        "content": current_chunk,
                        "metadata": {
                            "source": str(file_path),
                            "type": "docx"
                        }
                    })
                current_chunk = text
            else:
                current_chunk += "\n" + text
        
        if current_chunk:
            chunks.append({
                "content": current_chunk,
                "metadata": {
                    "source": str(file_path),
                    "type": "docx"
                }
            })
        
        return chunks

# 文档处理流水线：根据后缀自动选择处理器
class DocumentPipeline:
    def __init__(self):
        self.processors = {
            ".pdf": PDFProcessor(),
            ".docx": WordProcessor(),
            ".doc": WordProcessor(),
            ".txt": WordProcessor(),
            ".md": WordProcessor()
        }
    
    def process(self, file_path: str) -> List[Dict[str, Any]]:
        ext = Path(file_path).suffix.lower()
        processor = self.processors.get(ext)
        
        if not processor:
            print(f"不支持的文件格式: {ext}")
            return []
        
        return processor.process(file_path)

# 使用示例
if __name__ == "__main__":
    pipeline = DocumentPipeline()
    
    # 处理单个文件
    chunks = pipeline.process("docs/api-guide.pdf")
    print(f"解析出 {len(chunks)} 个文本块")
    
    # 批量处理目录
    from pathlib import Path
    for pdf_file in Path("docs").glob("*.pdf"):
        chunks = pipeline.process(str(pdf_file))
        print(f"{pdf_file.name}: {len(chunks)} chunks")

运行方法：

# 准备好文档目录
mkdir -p docs && cp /path/to/your/docs/*.pdf docs/

# 测试解析
python3 document_pipeline.py

调优建议：

3.2 RAG检索流水线

这段代码做什么：串联向量检索、Rerank重排、答案生成三步。用户提问 → 向量数据库找到候选 → Rerank精排 → LLM生成答案。

为什么分三步：向量检索快但不精准（ANN算法是近似搜索），Rerank慢但精准（交叉编码器逐个深度比较）。先用向量检索快速海选（100个候选），再用Rerank精挑（选10个最相关的）。

from sentence_transformers import CrossEncoder, HuggingFaceBgeEmbeddings

class RAGPipeline:
    def __init__(self, milvus_client, embed_model, rerank_model):
        """
        milvus_client: Milvus连接实例
        embed_model: 嵌入模型（把文本转成向量）
        rerank_model: Rerank模型（精排候选文档）
        """
        self.milvus = milvus_client
        self.embed = embed_model
        self.rerank = rerank_model
    
    def retrieve(self, query: str, top_k: int = 100, rerank_k: int = 10):
        """
        两阶段检索：
        1. 向量检索：快速找出top_k个候选（这一步是近似搜索，可能把不太相关的也拉进来）
        2. Rerank：用交叉编码器精排（这一步慢但准，只保留最相关的rerank_k个）
        """
        # ====== 第一步：向量检索 ======
        # 把用户问题转成向量
        query_vector = self.embed.embed_query(query)
        
        # 在Milvus中搜索
        results = self.milvus.search(
            collection_name="knowledge",  # 之前创建好的Collection名
            data=[query_vector],
            limit=top_k,                   # 返回多少个候选
            output_fields=["content", "source", "page"]  # 返回哪些字段
        )
        
        if not results or not results[0]:
            return []
        
        # 提取所有候选文档内容
        candidates = [r["entity"]["content"] for r in results[0]]
        
        # ====== 第二步：Rerank精排 ======
        # 构建(query, document)对
        pairs = [(query, doc) for doc in candidates]
        
        # 批量预测：每个文档和问题的相关度分数
        scores = self.rerank.predict(pairs)
        
        # 按分数排序（从高到低）
        doc_scores = list(zip(candidates, scores))
        doc_scores.sort(key=lambda x: x[1], reverse=True)
        
        # 返回Top-K
        return [
            {
                "content": doc,
                "score": float(score),
                "source": results[0][i]["entity"]["source"],
                "page": results[0][i]["entity"]["page"]
            }
            for i, (doc, score) in enumerate(doc_scores[:rerank_k])
        ]
    
    def generate(self, query: str, context: list) -> str:
        """
        基于检索结果生成答案
        context: Rerank后的相关文档列表
        """
        # 拼接上下文，标注来源
        context_text = "\n\n".join([
            f"【文档{i+1}】(来源: {c['source']}, 页码: {c['page']})\n{c['content']}"
            for i, c in enumerate(context)
        ])
        
        # 构造Prompt
        prompt = f"""你是一个技术文档助手。请基于以下参考资料回答用户问题。
如果参考资料中没有相关信息，请如实说明。

参考资料：
{context_text}

用户问题：{query}

请给出回答，并标注参考来源："""
        
        # 这里调用LLM API（根据你的LLM选择调用方式）
        response = call_llm(prompt)
        return response
    
    def answer(self, query: str) -> dict:
        """
        完整RAG流程：检索 → 生成 → 返回
        """
        # 1. 检索相关文档
        context = self.retrieve(query)
        
        if not context:
            return {
                "answer": "抱歉，我在知识库中没有找到与您问题相关的内容。",
                "sources": []
            }
        
        # 2. 生成答案
        answer = self.generate(query, context)
        
        # 3. 返回答案和来源
        return {
            "answer": answer,
            "sources": [
                {
                    "content": c["content"][:200] + "...",  # 只显示前200字
                    "source": c["source"],
                    "page": c["page"],
                    "relevance": c["score"]  # 相关度分数
                }
                for c in context
            ]
        }

# 使用示例
if __name__ == "__main__":
    from pymilvus import connections
    
    # 1. 连接Milvus
    connections.connect('default', host='localhost', port='19530')
    
    # 2. 加载模型（首次运行会下载，需要等待）
    print("加载嵌入模型...")
    embed_model = HuggingFaceBgeEmbeddings(model_name="BAAI/bge-large-zh-v1.5")
    
    print("加载Rerank模型...")
    rerank_model = CrossEncoder("BAAI/bge-reranker-large")
    
    # 3. 创建流水线
    pipeline = RAGPipeline(None, embed_model, rerank_model)
    
    # 4. 测试检索
    query = "如何部署Milvus集群？"
    result = pipeline.answer(query)
    
    print(f"\n问题：{query}")
    print(f"\n回答：{result['answer']}")
    print(f"\n参考来源：{len(result['sources'])} 个文档")

运行方法：

# 确保Milvus正在运行
docker-compose ps

# 确保知识库Collection已创建并导入数据
# 运行测试
python3 rag_pipeline.py

参数调优指南：

3.3 Web界面开发

这段代码做什么：用Gradio构建一个好看的Web界面，让用户输入问题、查看答案和来源。不需要写HTML/CSS，Python原生。

为什么用Gradio：5分钟能出一个能看的界面，可以分享链接让同事测试，不需要写前端代码。适合快速验证和内部使用。

import gradio as gr

def create_web_interface(pipeline: RAGPipeline):
    """
    pipeline: 上面定义好的RAGPipeline实例
    """
    # 创建Blocks布局（比Interface更灵活）
    with gr.Blocks(title="技术知识库助手") as demo:
        # 页面标题
        gr.Markdown("# 技术知识库智能助手")
        gr.Markdown("基于RAG技术，支持文档理解、语义检索、智能问答")
        
        # 两列布局：左边输入，右边输出
        with gr.Row():
            with gr.Column(scale=1):
                query_input = gr.Textbox(
                    label="请输入您的问题",
                    placeholder="例如：如何部署Milvus集群？",
                    lines=3
                )
                submit_btn = gr.Button("🔍 搜索", variant="primary")
            
            with gr.Column(scale=2):
                answer_output = gr.Textbox(label="回答", lines=8)
                sources_output = gr.JSON(label="参考来源")
        
        # 示例问题（点击直接填充）
        gr.Examples(
            examples=[
                ["如何申请年假？"],
                ["公司报销流程是什么？"],
                ["如何重置账户密码？"]
            ],
            inputs=query_input
        )
        
        # 定义查询函数
        def handle_query(query):
            result = pipeline.answer(query)
            return result["answer"], result["sources"]
        
        # 绑定按钮事件
        submit_btn.click(
            fn=handle_query,
            inputs=query_input,
            outputs=[answer_output, sources_output]
        )
    
    return demo

# 启动应用
if __name__ == "__main__":
    from pymilvus import connections
    from sentence_transformers import CrossEncoder, HuggingFaceBgeEmbeddings
    
    # 初始化各组件
    print("连接Milvus...")
    connections.connect('default', host='localhost', port='19530')
    
    print("加载嵌入模型（首次需要下载，请耐心等待）...")
    embed_model = HuggingFaceBgeEmbeddings(model_name="BAAI/bge-large-zh-v1.5")
    
    print("加载Rerank模型...")
    rerank_model = CrossEncoder("BAAI/bge-reranker-large")
    
    # 创建流水线
    pipeline = RAGPipeline(None, embed_model, rerank_model)
    
    # 创建界面
    print("启动Web界面...")
    demo = create_web_interface(pipeline)
    
    # 0.0.0.0让局域网能访问，server_port指定端口
    demo.launch(server_name="0.0.0.0", server_port=7860)

访问方式：

界面效果预期：

• 顶部标题”技术知识库智能助手”
• 左侧输入框 + 搜索按钮
• 右侧显示答案
• 下方JSON显示参考来源列表

四、部署与运维

4.1 生产环境注意事项

Docker Compose适合开发和测试环境，生产环境建议用Kubernetes：

关键配置：

# Milvus资源配置（根据数据量调整）
milvus:
  resources:
    limits:
      memory: "16Gi"      # 16GB内存
      cpu: "4"           # 4核CPU
    requests:
      memory: "8Gi"
      cpu: "2"

4.2 性能调优建议

向量检索层优化：

Rerank层优化：

4.3 监控指标

建议监控这几个关键指标：

监控工具推荐：Prometheus + Grafana，Milvus自带exporter。

五、总结与行动建议

恭喜你完成了RAG进阶实战系列的第15天学习！

我们从向量数据库原理出发，逐步深入到文档处理、Rerank技术，最终完成了这个综合实战项目。现在你应该能够独立构建一个能上线的RAG知识库系统了。

今天学到的核心技能：

1. Milvus集群部署：Docker Compose一键启动生产级向量数据库

2. 多格式文档处理：PDF、Word、Markdown统一解析和分块

3. 两阶段检索：向量检索 + Rerank精排，兼顾速度和精度

4. Gradio界面开发：5分钟出原型，快速验证效果

下一步学习方向：

学完15天RAG知识，你现在有两个方向可以深入：

• Agent开发（Day16起）：让RAG系统从”被动问答”进化到”主动规划、自主执行”
• 垂直场景优化：医疗/法律/金融，不同行业有不同的优化技巧

今日行动建议：

1. 动手部署：按照今天的内容，一步步把系统跑起来

2. 上传自己团队的文档：替换掉示例文档，真正解决工作问题

3. 收集反馈：让同事测试，记录问题，持续优化

下期预告：Day16我们将进入《Agent智能体开发》系列，学习如何让AI不仅能回答问题，还能自主规划任务、调用工具、执行复杂操作。敬请期待！