RAG 教程:从零搭建知识库问答系统(50 行代码)

🎯 本节目标

学完这一节,你能:

• 用 50 行代码搭一个最小可用的 RAG(从零写,不用框架,而且分四步讲清楚)
• 理解 RAG 四步流程:加载 → 分块 → Embedding → 检索生成,且每一步都能对上"开卷考试"
• 说清楚为什么 RAG 能治幻觉(这是面试高频题)

RAG(检索增强生成)在国内 FDE JD 出现率 52%,是腾讯/蚂蚁/用友的硬性要求。这是本教程最重要的一节之一。

🍎 生活化类比

RAG = 开卷考试。

想象你进考场,有两种考法:

• 闭卷(普通大模型):只准带脑子。题目问"光合作用",你只能凭记忆写,记不清就编一个看起来像的——这就是幻觉。
• 开卷(RAG):可以带书。题目问"光合作用",你先翻到讲光合作用那页,照着书上写——答得又准又能引经据典。

RAG 干的就是这件事,只不过"翻书"这一步交给程序自动完成。把开卷考试的每一步对应过来:

记住这张表,下面四步讲解全在重复它。看不懂术语时,回来瞄一眼"在干嘛"那一列。

📖 概念讲解:RAG 四步

🖼 图：RAG 开卷考试四步流程（原图未收录）

1. 加载(Load)= 把资料搬进考场

把 PDF、Word、网页、数据库……全部变成纯文本。难点是扫描件 PDF 要先 OCR(就是用程序把图片里的字"认"出来),表格要单独处理,不然数字会乱。

回到类比:这步就像考试前一天,你把老师的讲义、课本、同学的笔记,统统装进书包。

2. 分块(Chunk)= 把整本书按页拆开

为什么不能整本塞?因为模型一次看不了那么多字(有"上下文窗口"限制),而且塞太多它反而会"忘"前面的、抓不住重点。所以要把长文档切成小段,通常 400-600 字一段。

回到类比:考试时你不会一页页从头翻到尾(太慢),你会按问题找对应那几页。前提是书得先按页拆开,你才能精准翻。

分块大小有讲究(这是 RAG 效果的第一杀手):

• 技术文档:400-600 字。一个知识点短,小切更精准。
• 法律条文:1200-1500 字。条款之间互相引用,大切才不会割断上下文。
• FAQ 问答:200-400 字。一条问答天然就是一小块,直接切。

3. Embedding(意思标签)= 给每页贴标签

这是 RAG 最"玄"的一步,但用类比一说就懂:给拆好的每一段文字,贴一个**"意思标签"。这个标签其实是一串数字(向量),但你可以把它想成——"这页讲的是光合作用、植物、阳光"**这种意思描述。

为什么要贴标签?因为程序不会像人一样"读懂"文字,但它会比数字。意思相近的两段,标签里的数字也接近。这样以后问"光合作用",程序就能算出"问题标签"和"哪页标签"最接近,直接定位。

回到类比:你给每页都贴了张便利贴,写着"这页讲啥"。考试时不用翻,只要扫一眼便利贴,就知道答案在哪几页。Embedding 就是那张自动生成的便利贴。

4. 检索 + 生成 = 翻到对应页,照着答题

用户提问 → 程序把问题也贴个"意思标签" → 拿这个标签去和所有页的标签比 → 挑出最像的几段 → 把这几段连同问题一起递给大模型 → 大模型看着资料写答案。

回到类比:你翻到 3 页相关内容,摊在桌上,看着它们答题。这就是 RAG 的最后一步。

为什么 RAG 能治幻觉?(面试高频)

普通大模型答题是凭记忆——它的"记忆"是训练时看过的海量文本,但记不清的细节,它会"编"出一个看起来合理的答案,这就是幻觉。

RAG 治幻觉的原理一句话:不让他凭记忆,强制他看着资料答。

具体怎么做?在 Prompt(给模型的指令)里把找到的资料塞进去,并要求"只基于资料回答,找不到就说不知道"。模型从"闭卷脑补"变成"开卷抄书",编造的概率大幅下降。

当然,这不是 100% 治愈——如果检索召回的资料本身就错或者不相关,模型还是会答错。所以后面的进阶课要讲混合检索 + Rerank(把召回准确率从 60% 拉到 90%)。

🛠️ 动手实操:50 行最小 RAG

别怕,我把 50 行拆成 4 个小步骤,每步单独一个代码块,每步都对应"开卷考试"的一步。最后再给完整版。

先准备环境:把上一节封装用的智谱 GLM 客户端建好。

# === 第 0 步:搭好工具台(就两行,跟上一节一模一样)===
import os, numpy as np
from openai import OpenAI

client = OpenAI(
    api_key=os.getenv("GLM_API_KEY"),                         # 你的智谱 key
    base_url="https://open.bigmodel.cn/api/paas/v4/"          # 智谱的接口地址
)

Step 1:准备几条"文档"(加载 + 分块)

先用最简单的列表模拟"知识库",别一上来就读 PDF,那样你会被文件操作绕晕。先看清"知识库"长什么样——就是几段文字而已。

对应开卷考试:这步就是"把资料搬进考场,并按页码摆好"。

# === Step 1:知识库(就是几段文字,模拟分块后的样子)===
documents = [
    "FDE 是前沿部署工程师,Palantir 首创,2026 年因大模型爆发而火。",
    "FDE 的核心是填补产品与客户真实需求的鸿沟,既要写代码也要见客户。",
    "RAG 是检索增强生成,让模型基于你的资料回答,能治幻觉。",
    "评估是 FDE 面试分水岭,49% 的 JD 要求,但最没人教。",
]
print(f"知识库搬好了,一共 {len(documents)} 段")

Step 2:把文档变成"意思标签"(Embedding)

这步就像给每一页贴便利贴。我们调一次智谱的 embedding-3 模型,它会返回一串数字(就是那个"意思标签")。

对应开卷考试:给每页贴上"这页讲啥"的便利贴。

# === Step 2:给每段文字贴"意思标签"(Embedding)===
def embed(text):
    """把一段文字变成一串数字(意思标签)"""
    resp = client.embeddings.create(model="embedding-3", input=text)
    return np.array(resp.data[0].embedding)   # 拿到一串数字,存成 numpy 数组方便后面算

doc_vectors = [embed(d) for d in documents]   # 给每段都贴上标签
print(f"贴好了,每段标签是 {len(doc_vectors[0])} 个数字")

Step 3:问问题时,先找最相关的几条(检索)

用户提问时,我们先把问题也贴个标签,再和所有文档的标签比一比——"意思有多像"。比的方法叫余弦相似度,你不用记名字,记住它就是"算两个标签意思有多接近"就行:结果越接近 1,越像。

对应开卷考试:扫一眼便利贴,翻到跟问题最对得上的那几页。

# === Step 3:检索——算"意思有多像",挑最相关的 top_k 段 ===
def retrieve(query, top_k=2):
    q_vec = embed(query)                                  # 给问题也贴个标签
    scores = [np.dot(q_vec, dv) for dv in doc_vectors]    # 算问题跟每段"意思有多像"
    # 小注:点积要"≈余弦相似度"有个前提——向量已归一化(长度归一)。
    # 大多数 Embedding API(含智谱)默认返回归一化向量,直接点积即可;
    # 如果你换了不归一化的模型,先做 np.dot(v, v)**0.5 归一或改用 cosine 函数
    top_idx = np.argsort(scores)[-top_k:][::-1]           # 按像不像排序,取最像的前 top_k 段
    return [documents[i] for i in top_idx]                # 把这几段原文拿出来

# 试试看:问"FDE 是干嘛的",它能不能翻到讲 FDE 的那两段
print("翻到这几段:", retrieve("FDE 是干嘛的?"))

Step 4:把找到的塞给大模型答题(生成)

最后一步,把检索到的那几段连同问题一起丢给 glm-4.7-flash,让它看着资料答。这里调的就是上一节我们封装过的 chat() 那种接口。

对应开卷考试:把翻到的几页摊开,照着答题。

# === Step 4:生成——把找到的资料塞进 Prompt,让模型看着答 ===
def rag_answer(query):
    context = "\n".join(retrieve(query))                                       # 把翻到的几段拼起来
    prompt = f"基于以下资料回答问题,找不到就说不知道。\n资料:\n{context}\n\n问题:{query}"
    resp = client.chat.completions.create(
        model="glm-4.7-flash",
        messages=[{"role": "user", "content": prompt}]
    )
    return resp.choices[0].message.content

print(rag_answer("FDE 是干嘛的?"))
print(rag_answer("怎么治幻觉?"))

完整代码(4 步合起来)

把上面 4 步合起来,就是一个能跑的最小 RAG。下面是完整版,你可以直接复制运行。

import os, numpy as np
from openai import OpenAI

client = OpenAI(
    api_key=os.getenv("GLM_API_KEY"),
    base_url="https://open.bigmodel.cn/api/paas/v4/"
)

# 1. 知识库(模拟加载+分块后的结果)
documents = [
    "FDE 是前沿部署工程师,Palantir 首创,2026 年因大模型爆发而火。",
    "FDE 的核心是填补产品与客户真实需求的鸿沟,既要写代码也要见客户。",
    "RAG 是检索增强生成,让模型基于你的资料回答,能治幻觉。",
    "评估是 FDE 面试分水岭,49% 的 JD 要求,但最没人教。",
]

# 2. 贴"意思标签"(Embedding)
def embed(text):
    resp = client.embeddings.create(model="embedding-3", input=text)
    return np.array(resp.data[0].embedding)

doc_vectors = [embed(d) for d in documents]

# 3. 检索:算意思有多像,挑最像的前 top_k 段
def retrieve(query, top_k=2):
    q_vec = embed(query)
    scores = [np.dot(q_vec, dv) for dv in doc_vectors]
    top_idx = np.argsort(scores)[-top_k:][::-1]
    return [documents[i] for i in top_idx]

# 4. 生成:把资料塞进 Prompt,让模型看着答
def rag_answer(query):
    context = "\n".join(retrieve(query))
    prompt = f"基于以下资料回答问题,找不到就说不知道。\n资料:\n{context}\n\n问题:{query}"
    resp = client.chat.completions.create(
        model="glm-4.7-flash",
        messages=[{"role": "user", "content": prompt}]
    )
    return resp.choices[0].message.content

if __name__ == "__main__":
    print(rag_answer("FDE 是干嘛的?"))
    print(rag_answer("怎么治幻觉?"))

✅ 跑通后,你会发现模型答案明确基于那几条文档,而不是它自己脑补——这就是 RAG 治幻觉的原理在代码层面的体现。模块 4 第 2 节我们把它工程化(混合检索 + Rerank)。

⚠️ 常见错误

✅ 自测题

<summary>1. RAG 的四步流程是?为什么不能把整本书一次塞进模型?</summary>

加载 → 分块 → Embedding → 检索生成。不能整本塞是因为 ① 上下文窗口有长度限制 ② 塞太多模型会"忘"前面 ③ 检索精度下降。分块才能精准命中相关片段。

</details> <summary>2. 用"开卷考试"的类比,解释为什么 RAG 能治幻觉。</summary>

普通大模型是"闭卷",凭记忆答题,记不清就编(幻觉)。RAG 强制模型"开卷"——先把相关资料检索出来塞进 Prompt,要求"只基于资料答,找不到就说不知道"。模型从脑补变成照着抄,编造概率大幅下降。注意:如果检索召回的资料本身就错,模型还是会答错,所以还要靠 Rerank 提精度。

</details> <summary>3. 为什么技术文档分块通常切 400-600 字,法律切 1200-1500?</summary>

技术文档一个知识点短,小切块更精准。法律条文相互关联、互相引用,大切块才不会割裂上下文。分块大小要匹配内容的语义颗粒度——这是 RAG 效果的第一杀手。

</details>

🚀 实战小项目

把上面的最小 RAG 改成一个能回答"本教程里 FDE 是什么"的小应用:

1. 把 module-0-fde-intro/01-what-is-fde.md 里的关键段落(比如"FDE 的官方定义""FDE vs AI 工程师"两节)复制成 5-8 段,替换掉 documents 列表。
2. 把 documents 里的每段切成 300-500 字的小块(手动切就行,先别写自动分块)。
3. 跑一遍 rag_answer("FDE 和 AI 工程师有啥区别?"),看它能不能引用到你塞进去的内容。
4. 对比一下:同样的问题直接问 glm-4.7-flash(不塞资料),和 RAG 版的答案有啥区别——你会肉眼看到"治幻觉"的效果。

把你跑出来的两个答案截图存档,这是你简历项目里"RAG 演示"的第一手素材。

📚 总结 & 下一节

3 句话回顾:

1. RAG 就是"开卷考试":加载(搬资料)→ 分块(按页拆)→ Embedding(贴意思标签)→ 检索生成(翻页+照着答)。
2. 它治幻觉的原理:不让他凭记忆,强制他看着资料答。
3. 分块大小是 RAG 效果第一杀手,要按内容颗粒度调。

下一节:工程化——混合检索(关键词 + 向量)+ Rerank,把召回准确率从 60% 拉到 90%。

🏋️ 配套自练

• LlamaIndex 5 天 RAG 教程 — 官方"理解 RAG"系列,从零搭一个(免费)
• RAG From Scratch — LangChain 出品的 41 节 RAG 系列,配 YouTube 视频(免费)
• 把本节的 50 行 RAG 跑通,换成你自己的文档(如一份产品说明书)

🔗 延伸阅读

• LlamaIndex RAG 教程
• Chroma 向量库(本地最简)
• RAG From Scratch(langchain-ai)

---

← 上一节:MCP 协议 | 下一节:RAG 进阶 →

---

动手实验

实验改编自 ai-engineer-notebooks（MIT License）第 03 章 Embeddings & Retrieval（01-embeddings-retrieval.ipynb）· github.com/calmrocks/ai-engineer-notebooks 运行方式：Colab 或本地 Jupyter 任选；模型用 Groq 免费 API（OpenAI 兼容），换成任何 OpenAI 兼容端点（含公司内部网关）只需改 base_url 与 key。

实验目标

从零搭一条最小检索链路：切块 → 本地 embedding → numpy 暴力余弦检索 → top-k 塞进 prompt 生成带引用的答案；再亲手触发三个经典检索陷阱：相似不等于相关、词表失配、分数不是置信度。对应本课时讲义「embedding 把语义相近映射到向量邻域」与「答案质量的上限由检索决定」。

前置条件

1. 免费 GROQ_API_KEY（console.groq.com，无需信用卡）。
2. Colab 或本地 Jupyter；embedding 用本地 sentence-transformers 模型（首次下载约 90MB，CPU 即可）。
3. 安装 aien 包（自带 groq 客户端）。

实验步骤

步骤 1：安装与初始化

%pip install -q "git+https://github.com/calmrocks/ai-engineer-notebooks.git@v0.1.0" requests sentence-transformers numpy

from aien import setup
client, MODEL = setup()

步骤 2：语料与切块

下载 10 篇 IETF RFC 并切块。切块不可跳过：embedding 模型只读每段开头约 256 token，整篇塞进去等于只索引了开头（原 notebook 用按编号标题切分的完整版，这里为省篇幅换段落打包版，检索行为等价）：

import os, re, requests
RFC_NUMBERS = [791, 793, 1035, 2616, 4271, 5321, 6455, 6749, 7540, 9110]
os.makedirs('data/rfc', exist_ok=True)
corpus = {}
for num in RFC_NUMBERS:  # 缓存：重跑不重复下载
    p = f'data/rfc/rfc{num}.txt'
    if not os.path.exists(p):
        open(p, 'w').write(requests.get(
            f'https://www.rfc-editor.org/rfc/rfc{num}.txt', timeout=30).text)
    corpus[num] = open(p).read()

def clean_rfc(text):  # 去掉 RFC 文本的页眉页脚
    text = text.replace('\f', '\n')
    lines = [l for l in text.split('\n')
             if not re.match(r'^.*\[Page \d+\]\s*$', l)
             and not re.match(r'^RFC \d+\s+.*\S+ \d{4}\s*$', l)]
    return re.sub(r'\n{3,}', '\n\n', '\n'.join(lines)).strip()

def chunk_paragraphs(text, max_chars=2400):  # 段落打包切块：chunk 是检索单元
    paras = [p for p in text.split('\n\n') if p.strip()]
    chunks, cur = [], ''
    for p in paras:
        if cur and len(cur) + len(p) + 2 > max_chars:
            chunks.append(cur); cur = p
        else:
            cur = cur + '\n\n' + p if cur else p
    if cur: chunks.append(cur)
    return chunks

chunk_texts, chunk_meta = [], []
for num in RFC_NUMBERS:
    for c in chunk_paragraphs(clean_rfc(corpus[num])):
        chunk_texts.append(c); chunk_meta.append({'rfc': num})
print(len(chunk_texts), 'chunks')

步骤 3：embedding 与暴力余弦检索

from sentence_transformers import SentenceTransformer
import numpy as np

embedder = SentenceTransformer('all-MiniLM-L6-v2')  # 本地免费，384 维
embeddings = embedder.encode(chunk_texts, normalize_embeddings=True)

def search(query, k=5):
    q = embedder.encode([query], normalize_embeddings=True)[0]
    scores = embeddings @ q                    # 已归一化：余弦相似度 = 点积
    top = np.argsort(-scores)[:k]
    return [(float(scores[i]), int(i)) for i in top]

def show(query, k=5):
    print(f'Q: {query}')
    for score, i in search(query, k):
        print(f'  {score:.3f}  RFC {chunk_meta[i]["rfc"]:>4}  {chunk_texts[i].splitlines()[0][:66]}')
    print()

show('How does the TCP three-way handshake work?')  # 语义检索舒适区，头部应落在 RFC 793

千级语料用暴力检索不是偷懒而是正确做法：精确、毫秒级、没有索引会失配；近似索引（HNSW 等）是上亿向量才需要的东西。

步骤 4：亲手触发三个陷阱

# 陷阱 1 相似≠相关：问 HTTP 超时，排前面的却是 TCP 重传超时
show('how long should I wait before timing out a connection?')
show('in HTTP, how long should I wait before timing out a connection?')  # 补上已知上下文

# 陷阱 2 词表失配：精确标识符被 384 维压缩抹平，substring grep 反而全中
show('Retry-After')
hits = [i for i, t in enumerate(chunk_texts) if 'retry-after' in t.lower()]
print('grep Retry-After:', len(hits), 'chunks contain it')

# 陷阱 3 分数不是置信度：语料里根本没有 Kubernetes，top 分数照样不低
for q in ['How does TCP recover from a lost segment?',
          'How does Kubernetes schedule pods onto nodes?']:
    s, i = search(q, 1)[0]
    print(f'{s:.3f}  RFC {chunk_meta[i]["rfc"]}  <- {q}')

步骤 5：检索接生成，闭环 RAG

def answer(question, k=5):
    hits = search(question, k)
    context = '\n\n'.join(
        f'[{n + 1}] (RFC {chunk_meta[i]["rfc"]})\n{chunk_texts[i][:1500]}'
        for n, (_, i) in enumerate(hits))
    prompt = ('Answer the question using only the numbered passages below. '
              'Cite the passages you used like [1] or [2][3].\n\n'
              f'{context}\n\nQuestion: {question}')
    resp = client.chat.completions.create(
        model=MODEL, max_tokens=1024,
        messages=[{'role': 'user', 'content': prompt}])
    return resp.choices[0].message.content

print(answer('What does an HTTP 503 response mean, and which header can tell '
             'the client when to retry?'))  # 对照检查：引用是否真来自取回的 chunk

自查问题

1. 陷阱 1 里，为什么补上「in HTTP」前缀能修好检索？另外两种修法（元数据过滤、query 改写）各适合什么场景？
2. 生成步为什么不可能引用一个检索没取回来的 chunk？这决定了排查答案质量问题时应该先看什么？
3. 为什么不能给余弦分数设阈值（比如低于 0.5 就拒答）来识别「答不了」的问题？