用 Python 从零搭一套 RAG 系统：40 行跑通，再谈优化

大模型接个 API 就能聊天，但真到业务里你会发现一个尴尬的事实：它很聪明，却对你公司的资料一无所知。员工手册、产品文档、项目复盘，它统统没见过。你问它，它不会说 "我不知道"，而是开始一本正经地编。

RAG（Retrieval-Augmented Generation，检索增强生成）就是这件事的工程解法。说白了就一句：回答之前先翻资料，再照着资料说话。

这篇文章按真实开发的顺序走：先用 40 行代码把闭环跑通，再讲清楚关键参数，最后加上混合检索与权限隔离这些真正让系统能上线的东西。

一、RAG 到底在修什么 bug

大模型本质是个超大的概率模型，生成每个词都在猜 "下一个最可能的词"。它的知识来自训练语料，压在几万亿参数里。这带来三个短板：知识有截止日期、私有知识它没有、而且它不知道自己不知道。最后这条最要命 —— 面对盲区，它会用非常流畅的语言编一个看起来合理的答案。

RAG 的价值可以列成一张对照表：

表格

可追溯性经常被低估，但它是 RAG 能在企业活下来的根本 —— 答错了，你能点开引用判断是检索错了还是总结错了。这决定了系统能不能被追责、被信任。

二、骨架：两个阶段，五个环节

【离线建库】 文档加载 → 切分 → 向量化 → 存库
【在线问答】 提问 → 向量化 → 检索 TopK → 拼 Prompt → 生成

这个划分很重要，多数性能问题源于没分清。建库慢没关系，跑一次存下来；问答必须快。我见过最典型的翻车：每次问答都重新向量化整个知识库，单次响应三十多秒。

依赖清单其实很短：

pip install numpy requests pypdf
# 想用生态再装
pip install langchain langchain-community faiss-cpu chromadb

RAG 拆到底层就是 "文本处理 + 向量运算 + API 调用" 三件事。那为什么还有那么多人用 LangChain？因为它把五个环节封装成可串联模块，五分钟出 demo。但 demo 跑通和系统能维护是两件事 —— 换个场景、换批文档，框架甩出来的往往是抽象的内部异常，你连从哪查都不知道。

建议第一版手写，跑通后再决定要不要上框架：手写能打印每个中间结果，排错是线性的；等你能用百余行跑通，再看框架文档就会一眼看出它替你封装了什么。

三、40 行跑通最小闭环

import re, numpy as np, requests

EMBED_URL = "https://api.your-provider.com/v1/embeddings"
CHAT_URL  = "https://api.your-provider.com/v1/chat/completions"
HEADERS   = {"Authorization": "Bearer YOUR_KEY"}

def embed(texts):
    r = requests.post(EMBED_URL, headers=HEADERS, timeout=30,
                      json={"model": "bge-m3", "input": texts})
    return [d["embedding"] for d in r.json()["data"]]

def split(text, size=500, overlap=80):
    paras = [p.strip() for p in re.split(r"\n\s*\n", text) if p.strip()]
    chunks, cur = [], ""
    for p in paras:
        if len(cur) + len(p) > size and cur:
            chunks.append(cur)
            cur = cur[-overlap:] + "\n" + p
        else:
            cur = f"{cur}\n{p}" if cur else p
    if cur: chunks.append(cur)
    return chunks

chunks = split(open("knowledge.txt", encoding="utf-8").read())
vecs   = np.array(embed(chunks), dtype="float32")
vecs  /= np.linalg.norm(vecs, axis=1, keepdims=True)   # 归一化后内积即余弦

PROMPT = """严格依据【资料】回答【问题】。资料中没有的内容请直接回答"知识库中未找到相关信息"，不要编造。
【资料】
{context}
【问题】
{question}"""

def ask(q, k=5):
    qv = np.array(embed([q])[0], dtype="float32"); qv /= np.linalg.norm(qv)
    idx = np.argsort(vecs @ qv)[::-1][:k]
    ctx = "\n\n".join(f"[{i+1}] {chunks[i]}" for i in idx)
    r = requests.post(CHAT_URL, headers=HEADERS, timeout=60, json={
        "model": "your-chat-model", "temperature": 0.2,
        "messages": [{"role": "user", "content": PROMPT.format(context=ctx, question=q)}]})
    return r.json()["choices"][0]["message"]["content"]

四十行，一个能用的问答系统。当然还粗糙 —— 没有持久化、没有阈值过滤、没有关键词召回，但链路通了。这一步的意义远大于它本身的价值：后面所有优化都是在这条链上做局部替换，而不是推倒重来。有了参照系，你才知道变化来自哪里。

四、文档处理：决定系统上限的脏活

企业文档的真实状态是脏乱差：PDF 带页眉页脚、表格被拍平成一行字、扫描件提不出文字。噪音会一路传下去，最后表现为 "检索到的全是废话"。这时候换更强模型、调 Prompt 全是白费力气。数据质量排在模型前面，这是我反复验证过的铁律。

def clean(text):
    text = re.sub(r"\r\n?", "\n", text)
    text = re.sub(r"[ \t]+", " ", text)
    text = re.sub(r"\n{3,}", "\n\n", text)
    text = re.sub(r"(?<=[a-zA-Z])-\n(?=[a-zA-Z])", "", text)  # 修 PDF 断词
    return text.strip()

第四行容易被忽略。PDF 解析出的英文单词常被换行断开，embedd-ing 这样进嵌入模型，语义直接坏掉。表格图片多的文档别用正则硬扛，上 unstructured 或 OCR。这块的投入产出比比后面调十个参数都高。

切块：chunk_size 怎么定

整篇文档直接 embedding 是最常见的错误 —— 两万字压成一个向量，语义被平均化，检索精度低得离谱。

表格

默认起点 500 字符 + 80–100 重叠。术语密集的文档（技术规格书）降到 400–500；叙述性长文可放到 700–800。重叠的作用是防止知识点被切成两半，512 + 64 是性价比很高的组合。

实现上一个必须注意的细节：硬切的步长要是 chunk_size - overlap，否则你设多大重叠都等于没有。这个 bug 不报错，只是效果莫名其妙地差。

切完每条 chunk 要带元数据，ID 建议用 "来源 + 序号 + 内容摘要" 的哈希，别用自增序号 —— 文档一更新，增删段落会让后面所有序号错位，存好的向量全对不上。哈希 ID 是做增量更新的前提。

五、向量化与存储：别被 "向量数据库" 唬住

商用 API（text-embedding-3-small 1536 维、BGE-M3 1024 维）开箱即用；本地开源（bge-m3、bge-large-zh-v1.5 768 维）数据不出内网、无 token 费用，CPU 上几千 chunk 一两分钟跑完。

一条要记死的规则：建库和检索必须用同一个嵌入模型。换模型等于换了坐标系，新旧向量算出来的相似度全是噪音。建议在索引文件里存模型名和维度，加载时校验，不一致直接报错，别让它静默返回垃圾。

数据量在几千到几万条时，numpy 完全够用：

import json
class MiniStore:
    def __init__(self): self.vectors, self.records = None, []
    def add(self, records, vecs):
        v = np.asarray(vecs, dtype="float32"); v /= np.linalg.norm(v, axis=1, keepdims=True)
        self.vectors = v if self.vectors is None else np.vstack([self.vectors, v])
        self.records.extend(records)
    def search(self, qvec, k=5):
        q = np.asarray(qvec, dtype="float32"); q /= np.linalg.norm(q)
        s = self.vectors @ q
        return [{**self.records[i], "score": float(s[i])} for i in np.argsort(s)[::-1][:k]]
    def save(self, p):
        np.save(f"{p}.npy", self.vectors)
        json.dump(self.records, open(f"{p}.json", "w"), ensure_ascii=False)

4000 条 × 768 维 × 4 字节 ≈ 12MB，5000 条全量检索几十毫秒。这个规模引分布式向量库，属于用航母打蚊子。

什么时候该换真库？数据超 5 万条、需要按元数据过滤、需要高并发 —— 分别对应 Chroma/Qdrant 和 Milvus。好在 MiniStore 只有 add 和 search 两个核心方法，替换时重写这两个就行，上层代码一行不改。

还有一个必做动作：索引持久化。3000 个 chunk 在 CPU 上向量化约 8 分钟，每次重建的话根本没法调试。存成 .npy + .json，启动判断存在就加载 —— 这让你从 "脚本" 变成 "可用的小服务"。

六、检索：效果 80% 在这里决定

RAG 的效果 80% 取决于检索质量。检索错了，后面再强的模型也答不对。

裸向量检索有三个毛病：
一是缺阈值。 用户问的东西库里根本没有，相似度可能只有 0.5，TopK 照样把 "矮子里拔高个" 的塞给你。必须加分数过滤，让模型有资格回答 "没有相关信息"。阈值经验值 0.6–0.75，但不同模型分布差异很大，别硬抄，自己打印真实查询的分数看一眼。
二是精确匹配失灵。 这是最致命的。问 "设备 XK-2000 的防护等级"，向量模型可能召回一堆讲 "防护等级通用要求" 的文档 —— 语义很像，但答不对。型号、错误码、SKU、人名，这些精确 token 恰恰是向量最不擅长的。
三是候选里混着 "像但没用" 的片段，排序依据是语义距离，不是 "能不能回答问题"。

解法是双路召回，让两种检索各干擅长的事：

def normalize(s):
    lo, hi = min(s), max(s)
    return [0.5] * len(s) if hi == lo else [(x - lo) / (hi - lo) for x in s]

def hybrid(query, store, k=5, w_sem=0.7, w_bm25=0.3):
    sem = [0.0] * len(store.records)
    for h in store.search(embed([query])[0], k=k * 3):
        sem[store.records.index(h)] = h["score"]
    bm = bm25_scores(query, [r["text"] for r in store.records])
    sn, bn = normalize(sem), normalize(bm)
    fused = {i: w_sem * sn[i] + w_bm25 * bn[i] for i in range(len(sem)) if sn[i] or bn[i]}
    return [store.records[i] for i in sorted(fused, key=fused.get, reverse=True)[:k]]

BM25 自己写也就三十行：jieba 分词，k1=1.5、b=0.75：

idf = math.log((N - df + 0.5) / (df + 0.5) + 1)   # 末尾 +1 必须加

这个 +1 不是凑数。当某词出现在超过一半的文档里时，括号内小于 1，对数变负，这个高频词就会反向惩罚排序。教程里经常省略，但线上会结结实实咬你一口。

权重 0.7 / 0.3 是稳妥起点。语料里专名、编号多（设备手册、工单系统），BM25 提到 0.4–0.5 效果明显；叙述性文档调回 0.2。用 LangChain 的话，EnsembleRetriever 组合 BM25Retriever 和向量检索器即可，也可以改用 RRF 融合 —— 它不依赖分数分布，更鲁棒。

重排：把最能回答问题的排前面

向量检索是双塔结构，问题和文档分别编码再算距离，快但精度有上限。交叉编码器把 "问题 + 片段" 拼在一起送进模型直接打分，准得多，但慢 —— 只能对上几十条候选做。所以标准打法是两段式：召回 50 条，重排取前 5 条。

from sentence_transformers import CrossEncoder
reranker = CrossEncoder("BAAI/bge-reranker-base")   # 中文场景推荐
scores = reranker.predict([(query, c["text"]) for c in candidates])

值不值得上？文档量大、准确性要求高、能接受几十到上百毫秒额外延迟 —— 值得；库只有几百条、查询又简单，提升可能不值这个延迟。

七、生成：把片段变成让人信服的答案

PROMPT = """你是严谨的问答助手，请严格基于【参考材料】回答【用户问题】。

<context>
{context}
</context>

【用户问题】{question}

【回答要求】
1. 只能使用参考材料中的信息，不得引入外部知识；
2. 材料不足时直接回答"根据现有知识库无法解答"，绝对不要推测或编造；
3. 末尾标注引用来源，格式 [来源:文件名#片段序号]；
4. 先给结论再给依据，不要整段复述原文。
"""

第 2 条最关键。不加这句，模型遇到盲区会强行作答；加上之后，说 "不知道" 的概率明显提高。企业场景里这是要紧的安全特性 —— 宁可它说不知道，也不要它编一个看起来很专业的错答案。第 4 条同样重要，不写 "不要复述"，模型会偷懒把片段原样吐出来。

参数上：temperature 设 0.1–0.3（RAG 要事实准确，不要文采，高于 0.5 时即使有约束编造率也会上升）；top_k 取 4–6；max_tokens 512–1024。

token 预算给个换算：中文 1 字符约 0.6–1 token，5 片段 × 500 字符约 2000–2500 token，主流模型都从容。超限时别截断字符串（会把句子切两半），按分数从低到高逐条剔除片段。

八、从 demo 到能上线：四件容易被忽略的事

权限隔离。 普通员工不能通过 AI 查到高管薪资。不处理这件事，RAG 会变成 "越权查询加速器"—— 比翻共享盘还方便。做法是在元数据打 department、security_level 标签，检索时按用户角色注入过滤。注意是先多召回再过滤，否则过滤后可能不够 k 条。

评估。 上线后最常见的对话是 "效果怎么样"——"感觉还行"。这不行。轻量方案半小时能搭：挑 20–30 个有明确答案的问题，标注答案所在文件段落；测检索命中率、答案正确率、引用准确率。命中率 0.8 是及格线，低于这条先调切分和阈值，别急着换模型。要更系统就用 Ragas，看 "上下文相关性" 和 "答案忠实度" 两个指标，后者能自动揪出模型自己加的料。

增量更新。 利用哈希 ID，索引里存每个 chunk 的内容哈希，启动时比对 —— 一致跳过，不一致才重算那几条。几十行代码，更新时间从几分钟降到几秒。真正需要复杂增量索引的，是几万以上文档的量级。

表格与多模态。 表格被拍平成 "一季度 100 200 300" 这种字符串，向量化直接失效。折中方案：表格转 Markdown 保留行列关系，图片走 OCR 生成描述，"描述 + 结构化内容" 作为 chunk 正文，原图路径放元数据。检索走文本，展示走原图。

九、七个高频坑速查

表格

十、优化顺序：别一上来就奔着大而全

很多人第一版就想把混合检索、重排、多轮记忆、知识图谱全加上，结果系统跑起来了，但每层都是抄来的默认值，出了问题不知道是哪层导致的。

正确顺序是先跑通，再按瓶颈逐个优化：

1. 最小可用版：加载 → 切块 → 向量化 → 裸向量检索 → 生成。跑通，能答对简单问题。
2. 修检索：加阈值过滤、加 BM25 混合召回。通常效果提升最大，成本最低。
3. 修生成：强化 Prompt 约束、加引用、调 temperature。再考虑重排与多模态。

每一步只改一个变量，改完用测试集验证，看得见变化，心里才踏实。

RAG 听起来有门槛，拆开看就是一个搜索系统 + 提示工程的融合体。你需要理解的其实只有四样：向量怎么表示语义、相似度怎么算、Prompt 怎么约束、数据怎么治理。

模型负责 "说话"，知识库负责 "内容"。你不给资料，它只能猜；给精准资料，它就能像个熟悉业务的老员工一样回答。

做完之后你会发现，最有成就感的不是它答得多流畅，而是答案末尾那行 [来源:员工手册.pdf#12]—— 点开，原文就在那儿。那一刻，这个系统才真正可信。