从零构建生产级 RAG 知识库:架构、代码与部署实战
引言:RAG 为什么再次成为焦点
过去一年,大模型应用的重心正在从“提示词工程”转向“知识工程”。企业在落地 AI 时普遍遇到三个现实问题:模型不知道内部文档、回答缺乏可追溯依据、微调成本高且更新滞后。检索增强生成(Retrieval-Augmented Generation,RAG)正是针对这些问题的工程化解法——它把外部知识以向量化形式存入索引,在推理时检索相关片段注入上下文,让模型“带着资料回答”。
近期社区对 RAG 的讨论明显升温,原因有二:其一,长上下文模型并未消除检索需求,反而让“检索质量”成为决定回答质量的上限;其二,围绕文档解析、切分策略、混合检索、重排序的工程实践逐渐沉淀,形成了一套可复用的技术栈。本文面向开发者,从架构设计讲到可运行代码,再到 GitHub 源码部署与生产化调优,给出一条完整落地路径。
一、RAG 的核心架构与数据流
一个生产级 RAG 系统通常分为离线索引与在线推理两条链路。
离线索引链路:文档采集 → 格式解析(PDF/Word/HTML/Markdown)→ 文本清洗 → 分块(Chunking)→ 向量化(Embedding)→ 写入向量数据库,同时保留元数据(来源、页码、更新时间)用于引用与过滤。
在线推理链路:用户查询 → 查询改写(可选)→ 向量检索 + 关键词检索(混合检索)→ 重排序(Rerank)→ 上下文组装 → 大模型生成 → 引用标注与后处理。
两条链路共享同一套元数据规范,这是后续做权限过滤、增量更新和答案溯源的基础。很多教程只讲“向量化 + 相似度搜索”,但真正决定效果的是检索召回率与上下文精度的平衡,这也是本文重点展开的部分。
二、技术选型:一套务实的组合
面向开发者,推荐一套兼顾效果与可控性的开源组合:
- 文档解析:
unstructured或PyMuPDF,处理 PDF 版面与表格;Markdown/HTML 可直接用解析器。 - Embedding 模型:中文场景可选
BGE-M3(支持多语言、稠密+稀疏混合),英文可用text-embedding-3-small等 API。 - 向量数据库:中小规模用
Qdrant或Milvus Lite,单机即可跑通;大规模用Milvus集群。 - 编排框架:
LlamaIndex或LangChain,也可自研轻量管线,避免框架黑盒。 - 重排序模型:
BGE-Reranker系列,对 Top-K 结果二次打分。 - 生成模型:本地
Qwen系列或调用兼容 OpenAI 协议的 API。
选型原则是:解析与切分必须可控,检索层必须可观测,生成层可替换。不要把整条链路锁死在单一框架里。
三、关键工程细节:决定效果的四个环节
1. 文档解析与清洗
PDF 是最大的坑。扫描件需要 OCR;双栏排版需要按阅读顺序重排;表格建议转成 Markdown 保留结构。清洗阶段要统一空白字符、去除页眉页脚、合并断行。解析质量差,后面所有环节都会被拖累。
2. 分块策略
固定长度切分简单但会切断语义。推荐“结构化切分 + 重叠窗口”:先按标题层级切分章节,再在章节内按句子边界切到 300–600 token,相邻块保留 10%–15% 重叠。对表格、代码块单独成块并打标签。块太小召回碎片化,太大稀释相关性,需结合评测集调参。
3. 混合检索
纯向量检索对专有名词、编号、缩写不敏感。生产环境建议向量检索与 BM25 关键词检索并行,用 RRF(Reciprocal Rank Fusion)融合结果。查询侧可加轻量改写:对多轮对话补全指代,对短查询扩展同义词。
4. 重排序与上下文组装
召回 Top-50 后用 Reranker 精排取 Top-5,能显著提升上下文精度。组装时按相关性排序并保留来源元数据,同时做 token 预算控制,避免超出模型上下文窗口。提示词中明确要求“仅依据给定资料回答,无法回答时说明”,并输出引用编号。
四、最小可运行代码:索引与检索
以下示例使用 Python,向量库以 Qdrant 为例,Embedding 走兼容 OpenAI 协议的本地服务,便于替换。
import uuid
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
from openai import OpenAI
client = QdrantClient(url="http://localhost:6333")
llm = OpenAI(base_url="http://localhost:8000/v1", api_key="EMPTY")
COLLECTION = "kb_docs"
DIM = 1024 # 与 embedding 模型维度一致
def ensure_collection():
if COLLECTION not in [c.name for c in client.get_collections().collections]:
client.create_collection(
collection_name=COLLECTION,
vectors_config=VectorParams(size=DIM, distance=Distance.COSINE),
)
def embed(texts: list[str]) -> list[list[float]]:
resp = llm.embeddings.create(model="bge-m3", input=texts)
return [d.embedding for d in resp.data]
def index_chunks(chunks: list[dict]):
"""chunks: [{"text": ..., "source": ..., "page": ...}]"""
vectors = embed([c["text"] for c in chunks])
points = [
PointStruct(
id=str(uuid.uuid4()),
vector=vec,
payload={"text": c["text"], "source": c["source"], "page": c.get("page")},
)
for c, vec in zip(chunks, vectors)
]
client.upsert(collection_name=COLLECTION, points=points)
def retrieve(query: str, top_k: int = 5):
qvec = embed([query])[0]
hits = client.search(collection_name=COLLECTION, query_vector=qvec, limit=top_k)
return [{"text": h.payload["text"], "source": h.payload["source"], "score": h.score} for h in hits]
生成环节将检索结果拼接为上下文:
def answer(query: str) -> str:
docs = retrieve(query, top_k=5)
context = "\n\n".join(f"[{i+1}] {d['text']}" for i, d in enumerate(docs))
prompt = (
"你是知识库助手。仅依据以下资料回答,并在句末标注引用编号;"
"资料不足时明确说明。\n\n资料:\n" + context + f"\n\n问题:{query}"
)
resp = llm.chat.completions.create(
model="qwen2.5-7b-instruct",
messages=[{"role": "user", "content": prompt}],
temperature=0.1,
)
return resp.choices[0].message.content
这段代码刻意保持精简,便于替换组件。实际项目中应把解析、切分、索引拆成独立模块,并加入批量写入与失败重试。
五、GitHub 源码部署:从仓库到服务
生产部署推荐“仓库 + 容器 + 编排”的标准路径。以常见开源 RAG 项目为例,典型步骤如下:
1. 拉取源码与依赖
git clone https://github.com/<your-org>/rag-kb.git
cd rag-kb
python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
2. 配置环境变量
将模型地址、向量库连接、API Key 写入 .env,通过 pydantic-settings 或 python-dotenv 加载,禁止硬编码密钥。
3. 启动依赖服务
用 docker compose 拉起向量库与 Embedding 服务:
services:
qdrant:
image: qdrant/qdrant:latest
ports: ["6333:6333"]
volumes: ["./data/qdrant:/qdrant/storage"]
embedding:
image: ghcr.io/huggingface/text-embeddings-inference:latest
command: ["--model-id", "BAAI/bge-m3", "--port", "8000"]
ports: ["8000:8000"]
4. 构建索引与服务化
执行离线索引脚本 python scripts/build_index.py --dir ./docs,再用 uvicorn app.main:app --host 0.0.0.0 --port 8080 启动 API。对外暴露 /ingest 与 /query 两个端点,便于前端与内部系统接入。
5. 持续集成与发布
用 GitHub Actions 做三件事:代码