RAG 实战：搭建一个能回答你文档的本地知识库

从根本上理解 RAG

LLM 的知识有两个硬边界：训练截止日期之前发生的事它可能知道、私有数据它一无所知。想让它回答"我们公司的报销流程"或"上周的周会纪要说了什么"，要么做微调（成本高且数据更新就要重训），要么就用 RAG（Retrieval-Augmented Generation，检索增强生成）。

RAG 的思想朴素到一句话能说清：在把问题扔给 LLM 之前，先从你自己的文档里检索出最相关的几段，和问题一起塞进 Prompt。模型不需要"记住"你的知识，它只需要在回答时能看到相关片段就够了。

这一篇我们从头到尾搭一个能跑的 RAG 系统：读入一个目录下的 markdown 文档，切分成块，用第 05 篇的 embedding 编码，存到 Chroma 向量库，提问时做检索 + 生成。整个流程不依赖任何重量级框架（不用 LangChain），代码量控制在 200 行以内，但涉及的所有核心概念都是工业级的。

RAG 管线的四个阶段

一个标准的 RAG 系统有两条数据流：

索引阶段（离线，一次性或定期跑）

1. 加载：从 PDF / markdown / 数据库等来源读入原始文本
2. 切分：把长文本按合适的粒度切成块（chunk）
3. 向量化：用 embedding 模型把每个块编码成向量
4. 入库：把块文本 + 向量 + 元数据存入向量数据库

查询阶段（在线，每次提问触发）

1. 编码查询：把用户问题编码成向量
2. 检索：在向量库里找出最相似的 top-k 块
3. 组装 Prompt：把检索到的块作为上下文拼到问题前
4. 生成：调用 LLM 生成最终回答

下面按这个顺序一步一步搭。

准备环境和数据

pip install chromadb sentence-transformers openai python-dotenv pypdf

准备一个 docs/ 目录，放几篇 markdown 文档或 PDF。这里我放几个虚构的公司文档作为示例：

docs/
├── 报销流程.md
├── 年假政策.md
└── 技术栈规范.md

阶段 1：文档加载

针对不同来源写对应的 loader。为简单起见这篇只处理 markdown 和 PDF 两种：

# loader.py
from pathlib import Path
from pypdf import PdfReader


def load_markdown(path: Path) -> str:
    return path.read_text(encoding="utf-8")


def load_pdf(path: Path) -> str:
    reader = PdfReader(str(path))
    return "\n\n".join(page.extract_text() or "" for page in reader.pages)


def load_docs(directory: Path) -> list[dict]:
    """返回 [{source, text}, ...]"""
    docs = []
    for path in directory.rglob("*"):
        if path.suffix == ".md":
            docs.append({"source": str(path), "text": load_markdown(path)})
        elif path.suffix == ".pdf":
            docs.append({"source": str(path), "text": load_pdf(path)})
    return docs

真实项目里可能还要处理 Word、Excel、HTML、Confluence 导出等。unstructured 库能一次搞定大多数格式，但依赖比较重，学习阶段用上面的极简版就好。

阶段 2：文本切分

这一步决定了 RAG 效果的下限。切得太大，单个块塞进 Prompt 占太多 Token、噪声多；切得太小，上下文被割裂，模型看到半句话答不出问题。

常见策略：

• 固定长度切分——按字符数或 token 数切。简单但会在句子中间断开
• 按段落切分——自然分隔，但段落长度差异大
• 递归切分——先按段落，段落过长再按句子，仍过长按字符。这是最常用的妥协方案
• 语义切分——用 embedding 找"语义突变点"切分。效果最好但最复杂

对大多数场景，先用一个固定块大小 + 重叠窗口的简单方案已经足够：

# chunker.py
def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:
    """按字符切分，相邻块有 overlap 个字符重叠。"""
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunks.append(text[start:end])
        start += chunk_size - overlap
    return chunks

为什么要重叠：纯粹的硬切分会把一个完整语义单元（比如一个定义）切成两半，某个块里只保留了前半句。重叠保证关键信息大概率完整出现在至少一个块里。经验值：中文按字符切，块大小 400~800，重叠 50~100；英文按 token 数切，块 512，重叠 50。

阶段 3：向量化与入库

用 Chroma 做存储。Chroma 最友好的地方是它能内置一个 embedding 函数，你只要存文本，它自动调模型编码。这里我们把自定义的 bge-m3 模型传给它：

# indexer.py
import chromadb
from chromadb.utils import embedding_functions
from pathlib import Path
from loader import load_docs
from chunker import chunk_text

# 用持久化模式，数据存在本地目录
client = chromadb.PersistentClient(path="./chroma_db")

# 用 sentence-transformers 的 bge-m3 做嵌入
embedder = embedding_functions.SentenceTransformerEmbeddingFunction(
    model_name="BAAI/bge-m3",
    normalize_embeddings=True,
)

# 获取或创建集合（相当于一张表）
collection = client.get_or_create_collection(
    name="company_docs",
    embedding_function=embedder,
    metadata={"hnsw:space": "cosine"},  # 用余弦相似度
)


def index_directory(docs_dir: Path):
    raw_docs = load_docs(docs_dir)

    chunks, metadatas, ids = [], [], []
    counter = 0
    for doc in raw_docs:
        for i, chunk in enumerate(chunk_text(doc["text"])):
            chunks.append(chunk)
            metadatas.append({"source": doc["source"], "chunk_index": i})
            ids.append(f"chunk_{counter}")
            counter += 1

    # 批量写入，Chroma 会自动调 embedder 编码
    collection.add(documents=chunks, metadatas=metadatas, ids=ids)
    print(f"已索引 {len(chunks)} 个块")


if __name__ == "__main__":
    index_directory(Path("./docs"))

跑一次 python indexer.py，chroma_db/ 目录下会生成持久化的数据库文件，下次启动不需要重新索引。

工程提醒：如果文档会更新，ids 应该用一个稳定的哈希（比如 hash(source + chunk_index)）而不是自增计数，这样重新索引时可以 upsert 而不是重复插入。生产系统还要处理文档删除（源文件没了但库里还在）。

阶段 4：检索

最简单的检索就是语义相似度 top-k：

# retriever.py
import chromadb
from chromadb.utils import embedding_functions

client = chromadb.PersistentClient(path="./chroma_db")
embedder = embedding_functions.SentenceTransformerEmbeddingFunction(
    model_name="BAAI/bge-m3",
    normalize_embeddings=True,
)
collection = client.get_collection(name="company_docs", embedding_function=embedder)


def retrieve(query: str, top_k: int = 5) -> list[dict]:
    result = collection.query(query_texts=[query], n_results=top_k)
    return [
        {
            "text": doc,
            "source": meta["source"],
            "score": 1 - dist,  # Chroma 返回距离，转为相似度
        }
        for doc, meta, dist in zip(
            result["documents"][0],
            result["metadatas"][0],
            result["distances"][0],
        )
    ]


if __name__ == "__main__":
    for hit in retrieve("年假可以累计到明年吗"):
        print(f"[{hit['score']:.3f}] {hit['source']}\n{hit['text'][:100]}...\n")

阶段 5：生成最终回答

把检索到的块拼到 Prompt 里，送给 LLM：

# rag.py
from ai import chat  # 沿用 02 篇的 chat 封装，指向 DeepSeek
from retriever import retrieve


SYSTEM_PROMPT = """\
你是一个企业内部知识库助手。严格按以下规则回答：

1. 只能基于给定的「参考资料」回答
2. 如果参考资料里没有相关信息，直接说"在现有资料中未找到相关信息"，不要编造
3. 回答末尾用 [来源：文件名] 的格式标出每条信息的出处
4. 回答要简洁，优先使用要点列举
"""


def answer(question: str, top_k: int = 5) -> str:
    hits = retrieve(question, top_k=top_k)

    context = "\n\n---\n\n".join(
        f"[来源：{h['source']}]\n{h['text']}" for h in hits
    )

    user_prompt = f"""\
参考资料：
{context}

问题：{question}
"""

    return chat(
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_prompt},
        ],
        temperature=0,
    )


if __name__ == "__main__":
    q = "年假可以累计到明年吗"
    print(f"问：{q}\n答：{answer(q)}")

到这里整个 RAG 系统已经能跑起来了。几十行核心代码，端到端打通。

让它更可靠：几个必须加的改进

上面是最小可行版本。拉到真实场景你会很快发现需要这几个改进：

重排（Rerank）：向量检索召回的 top-20 里不一定最相关的真排第一。再用一个更重的"交叉编码器"模型对这 20 条重排，精度通常能提升 10~20%。常用模型是 BAAI/bge-reranker-v2-m3：

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("BAAI/bge-reranker-v2-m3")

def retrieve_with_rerank(query: str, top_k: int = 5, candidate_k: int = 20):
    candidates = retrieve(query, top_k=candidate_k)
    pairs = [(query, c["text"]) for c in candidates]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(candidates, scores), key=lambda x: -x[1])
    return [c for c, _ in ranked[:top_k]]

混合检索：向量检索擅长语义，但对精确的专有名词、代码名、缩写反而容易漏。补一条 BM25（传统关键词搜索），把两路结果融合。Chroma 本身不带 BM25，可以配合 rank_bm25 库实现。

查询改写：用户的原问题可能很口语化（"这个怎么弄"），直接拿去检索效果差。让 LLM 先把问题改写成更完整的检索查询再去检。

相似度阈值：如果 top-1 的相似度都很低（比如 < 0.4），说明库里根本没有相关内容，这时候应该让系统直接回答"没找到"而不是硬编。

评估：怎么知道 RAG 好不好用

RAG 的评估比纯 LLM 更复杂，因为链路有两段都可能出错：检索没召回到相关内容（检索失败），或者检索到了但模型答错了（生成失败）。主流评估维度：

• Context Recall：标准答案所需的信息是否都被检索到了
• Context Precision：检索到的内容是否都和问题相关
• Faithfulness：生成的答案是否完全依据上下文（不幻觉）
• Answer Relevance：答案是否真的在回答用户的问题

有一个专门的 Python 库 ragas 能自动化计算这些指标，用法和 pytest 一样顺手：

pip install ragas

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision

# eval_dataset 是 [{question, answer, contexts, ground_truth}, ...]
result = evaluate(eval_dataset, metrics=[faithfulness, answer_relevancy, context_precision])
print(result)

做 RAG 项目建议从第一天就攒评测集——哪怕只有 20 条人工标注的高质量问答对。没有评测集就没法判断改动是改好还是改坏。

本篇要点

• RAG 是"检索相关片段拼进 Prompt"，没有魔法
• 标准管线四阶段：加载 → 切分 → 向量化 → 入库；查询阶段：编码查询 → 检索 → 组装 → 生成
• 切分策略简单的"固定块 + 重叠"对 80% 场景够用，重叠是关键
• 向量库选型：原型用 Chroma，规模上去再换 Qdrant/Milvus/pgvector
• 生产可靠性必须加：重排、混合检索、阈值兜底、查询改写
• 从第一天就建评测集，用 ragas 做自动化评估

下一篇

RAG 解决的是"让 LLM 能看到你的数据"；第 07 篇解决另一个维度——让 LLM 能调用你的代码。我们会进入 Function Calling：模型根据对话上下文决定调用哪个函数、传什么参数，然后拿函数结果继续生成。这是 Agent 的基础，也是让 LLM 真正参与系统运行的关键机制。

参考资料

• Chroma 文档
• LlamaIndex RAG 指南 — 即便不用 LlamaIndex，它的概念文档对 RAG 架构很有帮助
• ragas 评估框架
• BGE Reranker 模型卡
• Advanced RAG Techniques — GitHub 上收集的 RAG 进阶技巧集合