从零搭建一个能落地的 RAG 知识问答系统

~~# 从零搭建一个能落地的 RAG 知识问答系统

摘要：大模型很强，但它不知道你的私有文档。这篇文章用一套可运行的最小实现，讲清楚 RAG（检索增强生成）的完整链路：文档解析 → 切片 → 向量化 → 召回 → 重排 → 提示词组装 → 生成，并给出每一步真正影响效果的工程细节。

一、为什么需要 RAG

大模型的知识来自训练语料，天然存在三个硬伤：

1. 知识截止：训练数据之后发生的事它不知道。
2. 私有盲区：公司内部文档、产品手册、会议纪要，模型从未见过。
3. 幻觉：遇到不知道的问题，它会一本正经地编一个看起来合理的答案。

微调能解决一部分问题，但代价高、更新慢——每改一次知识就得重新训练。RAG 提供了另一条路：把知识放在外面，让模型带着资料去答题。知识以文档形式存在，随时更新，改一行文档，下一次问答立刻生效。

一句话概括 RAG 的本质：

在生成答案之前，先从你的知识库里检索出相关资料，把它和问题一起交给模型。

二、整体架构

一条完整的 RAG 链路包含两个阶段：

离线索引阶段（做一次，可增量）

原始文档 → 解析清洗 → 分块 → 向量化 → 写入向量库

在线问答阶段（每次提问）

用户提问 → 查询向量化 → 召回Top-K → 重排序 → 截取Top-N → 组装Prompt → 生成答案

很多人的 RAG 效果不好，问题往往不在模型，而在中间这几步被省略了：没清洗、没重排、切片一刀切。下面逐个拆开讲。

三、文档解析与清洗

文档质量直接决定天花板。常见问题：PDF 解析后段落断行混乱、页眉页脚污染正文、表格变成一堆乱码字符。

几个容易忽略但收益很大的处理：

• 去掉页眉页脚：按出现频次统计，跨页重复出现的短行基本都是噪声。
• 修复断行：中文场景直接按句末标点（。！？）重新切句，比按 \n 切稳得多。
• 保留结构信息：把标题层级写进 chunk 的元数据里，形如 "产品手册 > 第三章 > 计费规则"，这会让召回的片段自带上下文，模型更容易判断归属。

四、分块策略：最容易被低估的一步

分块（chunking）的目标很朴素：让每一个块语义自洽，同时尽量小。这两句话是矛盾的，工程上找平衡点。

四种常见切法：

实践里最有效的组合是：递归切分 + 重叠（overlap）+ 结构前缀。

CHUNK_SIZE = 512
OVERLAP = 64


def split_markdown(text: str):
    """按 Markdown 标题层级切分，同时给每块带上标题路径。"""
    blocks, path, buf = [], [], []

    for line in text.split("\n"):
        if line.startswith("#"):
            if buf:
                blocks.append((" > ".join(path), "\n".join(buf)))
                buf = []
            level = len(line) - len(line.lstrip("#"))
            path = path[: level - 1] + [line.lstrip("# ").strip()]
        else:
            buf.append(line)

    if buf:
        blocks.append((" > ".join(path), "\n".join(buf)))
    return blocks


def chunk_with_overlap(body: str, size=CHUNK_SIZE, overlap=OVERLAP):
    """对单个语义块做滑窗切分，避免硬切断句子。"""
    if len(body) <= size:
        return [body]

    chunks, start = [], 0
    while start < len(body):
        chunks.append(body[start : start + size])
        if start + size >= len(body):
            break
        start += size - overlap
    return chunks

使用方式：把每个 （标题路径，正文） 的正文用滑窗切开，再把标题路径拼回每个 chunk 的前面。最终存进向量库的是 "【产品手册 > 计费规则】" + chunk 这样的字符串。

关于 size 的经验值：中文场景下 300–600 字比较稳妥。太小则信息碎片化，模型看不到完整上下文；太大则召回的答案里混进大量无关内容，反而稀释精度。

五、向量化与检索

选 embedding 模型时注意三点：中文能力、维度、是否与部署成本匹配。开源方案里 bge、gte 系列的中文表现都不错；如果希望一个模型覆盖多语言，bge-m3 类的方案省心。

import faiss
import numpy as np
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("BAAI/bge-small-zh-v1.5")


def embed(texts, is_query=False):
    """bge 系列模型要求查询侧加 instruction 前缀。"""
    if is_query:
        texts = ["为这个句子生成表示以用于检索相关文章：" + t for t in texts]
    vecs = model.encode(texts, normalize_embeddings=True, batch_size=32)
    return np.asarray(vecs, dtype="float32")


class VectorStore:
    def __init__(self, dim: int):
        self.index = faiss.IndexFlatIP(dim)   # 内积 = 余弦相似度（已归一化）
        self.chunks = []

    def add(self, chunks):
        self.chunks.extend(chunks)
        self.index.add(embed(chunks))

    def search(self, query, top_k=20):
        q = embed([query], is_query=True)
        scores, ids = self.index.search(q, top_k)
        return [(self.chunks[i], float(s)) for i, s in zip(ids[0], scores[0]) if i >= 0]

注意 normalize_embeddings=True 之后用 IndexFlatIP（内积）就等价于余弦相似度，比 IndexFlatL2 更省一次换算。这是很多人写错的地方。

六、重排序：精度提升最明显的一步

向量召回追求"别漏"，所以它倾向于多选；重排序（rerank）负责"排准"。这一步通常用 Cross-Encoder：它把问题和候选文档拼在一起过一遍模型，精度远高于双塔向量，代价是慢——所以只能用在召回后的少量候选上。

典型的漏斗：召回 Top-50 → 重排取 Top-3~5 → 送进生成。

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("BAAI/bge-reranker-base")


def rerank(query, candidates, top_n=3):
    pairs = [(query, text) for text, _ in candidates]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(scores, [c[0] for c in candidates]), reverse=True)
    return [text for _, text in ranked[:top_n]]

如果你的场景对延迟敏感，可以退而求其次做 向量+关键词的混合检索：用 BM25 兜住专有名词和编号（比如错误码 E-1024、型号 ABC-200），这类字符串语义模型往往抓不准。

def hybrid_search(query, store, bm25, chunks, top_k=20):
    dense = dict(store.search(query, top_k))
    sparse = dict(bm25.search(query, top_k))

    # 倒数排名融合：不依赖两路分数的量纲，比加权求和鲁棒
    fused = {}
    for rank, cid in enumerate(dense):
        fused[cid] = fused.get(cid, 0) + 1.0 / (60 + rank)
    for rank, cid in enumerate(sparse):
        fused[cid] = fused.get(cid, 0) + 1.0 / (60 + rank)

    top = sorted(fused.items(), key=lambda x: -x[1])[:top_k]
    return [(chunks[cid], dense.get(cid, 0.0)) for cid, _ in top]

RRF（Reciprocal Rank Fusion）的好处是不用纠结两路分数怎么归一化——它们本来就不可比。

七、提示词组装与生成

把候选片段塞进 prompt 时有几个关键点：

1. 带上出处。每段前面标注来源标题，方便模型引用，也方便人工核查。
2. 明确授权模型说"不知道"。这是抑制幻觉最有效的一句话。
3. 输出结构化。要求模型回答时附上引用编号。

SYSTEM = """你是企业知识库助手。只能依据下面提供的【参考资料】作答。

规则：
- 若参考资料中没有答案，直接回答"根据现有资料无法回答该问题"，不要编造。
- 回答末尾用 [1] [2] 标注引用的资料编号。
- 语言简洁，避免复述问题。

【参考资料】
{context}
"""


def build_messages(query, passages):
    context = "\n\n".join(
        f"[{i+1}] 来源：{meta}\n{text}" for i, (text, meta) in enumerate(passages)
    )
    return [
        {"role": "system", "content": SYSTEM.format(context=context)},
        {"role": "user", "content": query},
    ]

上线前务必测三类问题：库内有答案的、库内没有答案的、需要多段信息拼接才能回答的。第三类最能暴露切片的缺陷——如果一块内容被切到两个 chunk 里且 overlap 太小，模型就拼不起来。

八、怎么评估效果

不要凭感觉。至少准备 50 条真实问题，人工标注标准答案，然后分两个层面看：

• 检索层：Recall@K——正确答案所在的片段有没有进候选集合。如果这个数低，说明问题在切片或 embedding，调生成模型是没用的。
• 生成层：答案忠实度——回答内容是否能在给定片段中找到依据（可用另一个模型做打分器）。

绝大多数"模型答错了"的投诉，根因都在检索层。先把 Recall@K 提上去，再考虑换更大的生成模型，这个顺序能省下大量成本。

九、小结

一个能落地的 RAG 系统，重点不在选哪个大模型，而在这些细节：

• 清洗掉页眉页脚，修复 PDF 断行；
• 递归切分 + overlap + 标题路径前缀；
• embedding 做归一化，用内积做检索；
• 引入 rerank，或至少做向量+BM25 混合检索；
• prompt 里明确允许模型回答"不知道"；
• 用 Recall@K 定位问题出在检索还是生成。

把这几步做扎实，哪怕用的是参数量很小的模型，效果也常常好过直接把整篇文档塞给大模型。RAG 是一个系统工程，收益藏在细节里。
~~