RAG（Retrieval-Augmented Generation，检索增强生成）通过“先检索资料，再让大模型回答”的方式，为大模型补充私有知识和最新信息。本文基于我的 rag-demo 项目，搭建一套可以本地部署的中文 RAG 系统。

项目不是简单的向量检索示例，而是实现了一条较完整的 RAG 链路：文档解析与切块、Embedding、Dense 向量召回、BM25 关键词召回、RRF 融合、BGE Reranker 精排，最后由 Qwen 根据资料生成带引用的答案。

为什么需要RAG

直接使用大模型进行知识问答，通常会遇到几个问题：

1. 模型训练数据存在时间边界，无法了解最新资料
2. 企业文档、产品手册和内部制度不在公开训练数据中
3. 模型可能生成看似合理但实际错误的内容
4. 回答缺少来源，用户难以验证结论

微调可以改变模型的行为和表达风格，但不适合频繁更新大量事实知识。RAG 将知识保存在外部数据库中，文档变化时只需要重新入库，不需要重新训练模型。

系统整体流程如下：

文档 -> 解析 -> 切块 -> Embedding -> Milvus
                              +-> Dense Vector
                              +-> BM25 Sparse

问题 -> Dense召回 + BM25召回 -> RRF融合 -> Reranker精排
     -> 拼接上下文 -> LLM生成答案 -> 返回引用来源

技术选型

本项目全部组件都可以在本地运行：

• RAG API：FastAPI
• 大语言模型：Ollama + Qwen2.5 7B
• Embedding：BAAI/bge-large-zh-v1.5
• Reranker：BAAI/bge-reranker-v2-m3
• 模型服务：Hugging Face Text Embeddings Inference（TEI）
• 向量数据库：Milvus Standalone
• 对象存储与元数据：MinIO + etcd
• 可视化管理：Attu

Embedding 和 Reranker 默认使用 GPU。推荐 Linux、Docker Compose v2、NVIDIA Container Toolkit，以及 12GB 以上显存；当前配置更适合 16GB 显存环境。

先确认 Docker 容器可以使用 GPU：

docker run --rm --gpus all \
  nvidia/cuda:12.4.1-base-ubuntu22.04 nvidia-smi

项目结构

核心代码位于 app/ 目录：

app/
├── chunking.py       # 文本清洗与切块
├── clients.py        # Embedding、Reranker和LLM客户端
├── config.py         # 环境变量配置
├── main.py           # FastAPI接口与生命周期
├── models.py         # 请求和响应模型
├── parsers.py        # TXT、PDF、DOCX等文件解析
├── rag.py            # RAG主流程
└── vector_store.py   # Milvus集合、入库和混合检索

compose.yaml 定义 Milvus、etcd、MinIO、Ollama、Embedding、Reranker、API 和 Attu 服务。各模块通过 HTTP 或 Milvus SDK 通信，后续可以单独替换其中任意组件。

启动本地环境

复制配置并启动服务：

cp .env.example .env
make up
make pull-model

make up 会构建 API 镜像、下载 BGE 模型，并启动全部容器。第一次执行需要下载较大的模型文件，耗时取决于网络速度。

常用命令：

docker compose ps
make logs
make test
make lint
make down

服务启动后检查健康状态：

curl http://localhost:8000/health

API 文档地址为 http://localhost:8000/docs，Attu 管理页面为 http://localhost:3000。

文档解析

系统支持 TXT、Markdown、CSV、JSON、PDF 和 DOCX。不同格式最终都转换为普通文本，再进入统一的切块流程。

def parse_file(filename: str, content: bytes) -> str:
    suffix = Path(filename).suffix.lower()
    if suffix in {".txt", ".md", ".csv", ".json"}:
        try:
            return content.decode("utf-8")
        except UnicodeDecodeError:
            return content.decode("gb18030")

    if suffix == ".pdf":
        reader = PdfReader(io.BytesIO(content))
        return "\n\n".join(page.extract_text() or "" for page in reader.pages)

    if suffix == ".docx":
        document = Document(io.BytesIO(content))
        return "\n".join(p.text for p in document.paragraphs)

    raise ValueError("unsupported file type")

PDF 文本提取只适用于包含文字层的文件。扫描件需要先接入 OCR；复杂表格、双栏排版和图片也需要更专业的解析器，否则切块前就可能丢失语义。

文本切块

文档不能不加处理地直接生成一个向量。一方面 Embedding 模型有输入长度限制，另一方面整篇文档只生成一个向量会丢失局部语义。

项目默认参数如下：

CHUNK_SIZE=500
CHUNK_OVERLAP=80

切块时优先按照中文标点、英文标点和换行分割，同时保留一定重叠内容：

def chunk_text(text: str, size: int = 500, overlap: int = 80) -> list[str]:
    text = normalize_text(text)
    if overlap >= size:
        raise ValueError("chunk overlap must be smaller than chunk size")

    units = [
        unit.strip()
        for unit in re.split(r"(?<=[。！？!?；;\.])|\n+", text)
        if unit.strip()
    ]

    chunks = []
    current = ""
    for unit in units:
        if len(unit) > size:
            if current:
                chunks.append(current)
                current = ""
            step = size - overlap
            chunks.extend(
                unit[start:start + size]
                for start in range(0, len(unit), step)
            )
            continue

        candidate = f"{current}\n{unit}".strip() if current else unit
        if len(candidate) <= size:
            current = candidate
            continue

        chunks.append(current)
        prefix = current[-overlap:] if overlap else ""
        current = f"{prefix}{unit}"

    if current:
        chunks.append(current)
    return [chunk for chunk in chunks if chunk.strip()]

Overlap 可以降低答案刚好跨越两个切片时的信息损失，但过大会产生重复召回并增加存储量。实际项目应根据文档类型、Embedding 模型长度和问答粒度进行评估，不能把固定的 500 字作为通用最优值。

生成Embedding

Embedding 将文本转换为浮点向量，语义相近的文本在向量空间中距离也更接近。项目通过 TEI 的 /embed 接口批量生成向量：

async def embed(self, texts: list[str]) -> list[list[float]]:
    vectors = []
    batch_size = self.settings.embedding_batch_size

    for start in range(0, len(texts), batch_size):
        response = await self.client.post(
            f"{self.settings.embedding_base_url.rstrip('/')}/embed",
            json={"inputs": texts[start:start + batch_size], "truncate": True},
            timeout=120,
        )
        response.raise_for_status()
        vectors.extend(response.json())

    return vectors

批量请求可以提高 GPU 吞吐量。批次过大可能导致显存不足，过小则不能充分利用 GPU，因此项目通过 EMBEDDING_BATCH_SIZE 控制批量大小。

Milvus数据结构

每个文档切片在 Milvus 中保存以下字段：

• id：根据来源、序号和文本生成的 SHA-256
• text：切片原文
• source：文件名或业务来源
• metadata：部门、内容类型和切片序号等信息
• vector：Dense 浮点向量
• sparse_vector：Milvus 根据文本生成的 BM25 稀疏向量

Dense 字段使用 COSINE 相似度，Sparse 字段使用 BM25，并为中文配置 jieba 分词器：

FieldSchema(
    "text",
    DataType.VARCHAR,
    max_length=65535,
    enable_analyzer=True,
    analyzer_params={"tokenizer": "jieba"},
)
FieldSchema("vector", DataType.FLOAT_VECTOR, dim=dimension)
FieldSchema("sparse_vector", DataType.SPARSE_FLOAT_VECTOR)

同一个 source 再次上传时，系统会先删除旧切片，再写入新切片，避免知识库中出现多个版本的重复内容。

更换 Embedding 模型时必须注意向量维度。如果新模型维度与原 Collection 不一致，应该修改 Collection 名称或重建数据：

EMBEDDING_MODEL=BAAI/bge-m3
MILVUS_COLLECTION=rag_documents_hybrid_bge_m3

Dense与BM25混合检索

只使用 Dense 检索容易遗漏产品编号、错误码、姓名等精确关键词；只使用 BM25 又难以理解同义词和自然语言表达。因此项目同时发起两路召回：

dense_request = AnnSearchRequest(
    data=[query_vector],
    anns_field="vector",
    param={"metric_type": "COSINE", "params": {}},
    limit=top_k,
)

bm25_request = AnnSearchRequest(
    data=[question],
    anns_field="sparse_vector",
    param={"metric_type": "BM25", "params": {}},
    limit=top_k,
)

results = collection.hybrid_search(
    reqs=[dense_request, bm25_request],
    rerank=RRFRanker(k=60),
    limit=top_k,
    output_fields=["text", "source", "metadata"],
)

RRF（Reciprocal Rank Fusion）根据文档在不同结果列表中的排名进行融合。它不要求 Dense 分数和 BM25 分数位于同一数值范围，比直接对两种分数加权更加稳定。

使用Reranker精排

向量检索适合从大量数据中快速召回候选，但它通常只分别编码问题和文档。Cross-Encoder Reranker 会同时读取问题与候选文本，计算更准确的相关性分数。

项目先召回 top_k * 4 个候选，最多不超过 50 个，再精排并保留最终结果：

candidate_k = min(50, max(top_k, top_k * 4))
vector = (await embeddings.embed([question]))[0]
candidates = await store.search(question, vector, candidate_k)
hits = await reranker.rerank(question, candidates, top_k)

精排提高了准确率，但计算量明显高于向量检索。如果候选数量过多，延迟和 GPU 显存占用都会增加。实际服务需要在召回率、准确率和响应时间之间平衡。

构造上下文并调用LLM

检索结果按照编号拼接，并通过 MAX_CONTEXT_CHARS 限制上下文长度：

[1] 来源：产品手册
退款申请应在订单完成后七天内提交。

[2] 来源：客服制度
退款审核通常需要两个工作日。

System Prompt 明确限制模型只能依据资料回答，并要求引用资料编号：

system = (
    "你是一个严谨的知识库问答助手。只能依据给定的参考资料回答。"
    "如果资料不足，请明确回答‘根据当前知识库无法确定’，不要编造。"
    "回答应简洁清晰，并在相关陈述后引用资料编号，例如 [1]。"
)

LLM 统一使用 OpenAI-compatible /chat/completions 接口。默认连接 Ollama，也可以切换到 vLLM、Xinference 或其他兼容服务，而不需要修改 RAG 主流程。

RAG核心流程

RAGService 将入库和查询逻辑组合起来：

async def ingest(self, text, source, metadata):
    chunks = chunk_text(text, self.settings.chunk_size, self.settings.chunk_overlap)
    vectors = await self.embeddings.embed(chunks)
    return await asyncio.to_thread(
        self.store.replace_source, source, chunks, vectors, metadata
    )

async def query(self, question, top_k=None, threshold=None):
    hits = await self.retrieve(question, top_k, threshold)
    if not hits:
        return "根据当前知识库无法确定。请先上传相关资料。", []

    context = self.format_context(hits)
    answer = await self.llm.chat(question, context)
    return answer, hits

Milvus Python SDK 是同步接口，因此使用 asyncio.to_thread 避免阻塞 FastAPI 的事件循环。

文档入库接口

直接写入文本：

curl -X POST http://localhost:8000/v1/documents/text \
  -H 'Content-Type: application/json' \
  -d '{
    "source": "产品手册",
    "text": "退款申请应在订单完成后七天内提交。审核通常需要两个工作日。",
    "metadata": {"department": "客服"}
  }'

上传文件：

curl -X POST http://localhost:8000/v1/documents/file \
  -F 'file=@./your-document.pdf'

接口会返回生成的切片数量：

{
  "source": "产品手册",
  "chunks": 1
}

知识库问答接口

发送问题：

curl -X POST http://localhost:8000/v1/query \
  -H 'Content-Type: application/json' \
  -d '{"question": "退款审核需要多久？", "top_k": 5}'

响应同时返回答案和引用，便于前端展示来源或进行人工核验：

{
  "answer": "退款审核通常需要两个工作日。[1]",
  "citations": [
    {
      "source": "产品手册",
      "text": "退款申请应在订单完成后七天内提交。审核通常需要两个工作日。",
      "score": 0.98,
      "fusion_score": 0.0325,
      "rerank_score": 0.98,
      "metadata": {"department": "客服", "chunk_index": 0}
    }
  ]
}

如何评估RAG效果

RAG 优化不能只观察最终答案。应该把链路拆成多个阶段分别评估：

1. 解析质量：文本、表格和标题是否被正确提取
2. 切块质量：答案需要的上下文是否位于同一切片
3. 召回率：正确切片是否进入候选集合
4. 精排质量：正确切片是否出现在最终 Top K
5. 忠实度：模型回答是否完全来自检索资料
6. 引用准确性：答案中的编号是否指向对应证据
7. 性能：统计 Embedding、检索、精排和生成阶段的耗时

项目已经为文本清洗和切块编写 pytest 测试。生产环境还应建立固定的“问题—标准答案—相关文档”评测集，在修改模型、切块参数或检索策略后进行回归测试。

常见问题

检索结果看起来相关但无法回答问题

通常是切块粒度不合适，或者答案依赖的标题和正文被拆开。可以保留章节标题、调整 Chunk Size，或者使用按 Markdown 标题和文档结构切分的策略。

专有名词和编号无法召回

纯 Dense 检索对精确字符串不一定敏感，应保留 BM25 关键词召回。中文场景还需要检查分词结果，自定义词典可以改善产品名和行业术语的识别。

回答仍然出现幻觉

RAG 只能降低幻觉，不能完全消除。需要同时设置严格 Prompt、相关性阈值和上下文边界。没有可靠资料时，应让系统明确拒绝回答，而不是强行生成结论。

更新模型后Milvus报维度错误

不同 Embedding 模型的向量维度可能不同。更换模型后使用新的 Collection 名称，并重新生成全部文档向量，不能混用旧数据。

GPU显存不足

可以降低 Embedding Batch Size、换用更小的 LLM，或者将 Embedding、Reranker 和 LLM 分配到不同 GPU。资源有限时也可以关闭 Reranker，但需要重新评估检索准确率。

生产环境改进

当前项目已经包含完整的基础链路，正式部署还建议补充：

• 为入库和查询接口增加认证、权限和限流
• 使用异步任务队列处理大文件和批量文档
• 保存原始文件，并记录版本、租户和权限元数据
• 按租户或知识库增加 Milvus 过滤条件
• 对 Prompt Injection 和恶意文档内容进行防护
• 记录召回结果、引用、耗时和用户反馈
• 定期备份 Milvus、MinIO 和业务元数据
• 对重复文档、空文本和敏感信息进行检测

删除 Docker Volume 会同时删除知识库数据和已下载模型，执行下面的命令前需要确认已经备份：

docker compose down -v

总结

一套可用的 RAG 系统不只是“把文档转成向量”。文档解析、切块策略、混合召回、结果精排、上下文构造、引用返回和离线评测都会影响最终效果。

本文项目通过 BGE Embedding、Milvus Dense + BM25、RRF 和 BGE Reranker 提高中文资料的检索质量，再由兼容 OpenAI API 的本地模型生成有依据的回答。后续可以继续增加多租户权限、流式输出、查询改写、父子文档检索和自动化评测，让系统逐步具备生产使用能力。

参考

• https://milvus.io/docs
• https://huggingface.co/BAAI/bge-large-zh-v1.5
• https://huggingface.co/BAAI/bge-reranker-v2-m3
• https://ollama.com
• https://fastapi.tiangolo.com