Part 7 · RAG 检索增强生成

Section 1 · 为什么需要 RAG（概念日）

Subsection 1 · 亲眼看到 LLM 的缺陷 用你的 llm_client.py 问 DeepSeek 两个问题：

我们公司2025年发布的《内部差旅报销制度》里，超标住宿怎么处理？
我昨天在会上说的那个项目代号是什么？

预期看到：模型一脸懵，或者编一个答案（这就是幻觉）。
记住：LLM 只记得训练数据截止时间之前的事，且可能胡说。RAG 就是来解决这个的。

Subsection 2 · 记 RAG 定义 备忘录写：

RAG（Retrieval-Augmented Generation，检索增强生成）= 先"检索"相关资料 → 把资料拼进 Prompt → 让模型"只依据资料"回答结果：能答私有知识 + 能引用来源 + 少幻觉

Subsection 3 · 画整体流程 画：

文档 → 切块 → 转成向量 → 存进向量库                                   ↓ 用户提问 → 把问题转成向量 → 搜出最相关的几块                                   ↓ 把这几块 + 问题 拼进 Prompt → 模型回答

过关：能复述 RAG 流程 = Section 1 完成。

Section 2 · RAG 三大范式 + 三大部件（概念日）

Subsection 1 · 记三大范式

Naive RAG     朴素版：索引→检索→生成（先学这个）Advanced RAG  进阶版：检索前后加优化（Part 8 学）Modular RAG   模块版：检索组件可插拔组合（Part 8 学）

Subsection 2 · 记三大部件

Retriever 检索器   负责从知识库找出相关文本（向量搜索/关键词）Generator 生成器   就是 LLM，负责根据检索结果写答案Augmentation 增强   把检索结果"加工"成更好的上下文（重排/压缩）

Subsection 3 · 过关自测 能说出：① 三范式名字 ② 三大部件各干嘛 = Section 2 完成。

Section 3 · 搭 RAG 环境（装 4 个库）

Subsection 1 · 装库 cmd 依次输入：

pip install chromadbpip install sentence-transformerspip install -U langchain-communitypip install beautifulsoup4

每一条预期看到 Successfully installed ...
慢：每条命令后加 -i https://pypi.tuna.tsinghua.edu.cn/simple

Subsection 2 · 准备测试资料

在 ai-lab 里建一个文件夹 rag_data
在里面新建 company.md，粘上你自己的内容（至少 5 段），比如：

# 公司制度手册（示例）## 考勤员工每天 9:30 前打卡上班，迟到 3 次以上扣发当日补贴。## 报销单笔 500 元以下凭发票直接报销，以上需附审批单。...

这段内容是你自己写的"私有知识"，RAG 就是让它能回答这些。

过关：4 个库装好 + 资料文件建好 = Section 3 完成。

Section 4 · 文档加载与分块（RAG 第一步）

Subsection 1 · 跑加载 + 分块代码 VS Code 新建 rag_step1_load.py：

python

from pathlib import Path# 1. 读文件text = Path("rag_data/company.md").read_text(encoding="utf-8")print("文档总字符数:", len(text))# 2. 简单分块：按空行分blocks = [b.strip() for b in text.split("\\") if b.strip()]print("分成", len(blocks), "块")for i, b in enumerate(blocks):    print(f"[块{i}] {b[:40]}...")

运行
预期看到：文档被拆成若干块，每块是一小段。
分块大小很讲究：太大→命中不准（夹带无关内容）；太小→上下文割裂。你现在先按空行分，后面实验对比。

Subsection 2 · 过关自测 能说出"为什么不能把整篇文档直接丢给模型"（因为超出上下文/检索不准/浪费 token）= Section 4 完成。

Section 5 · 把文字变成向量（Embedding）

Subsection 1 · 跑向量化 VS Code 新建 rag_step2_embed.py：

python

from sentence_transformers import SentenceTransformer# 加载中文向量模型（第一次运行会下载，约 100MB，稍等）model = SentenceTransformer("BAAI/bge-small-zh-v1.5")sentences = ["员工每天9点半前打卡", "单笔500元以上报销需要审批单"]vectors = model.encode(sentences)print("每句话变成一个向量，维度:", vectors.shape)   # (2, 512)print("第一句的向量前8位:", vectors[0][:8])

预期看到：(2, 512) 和一段小数数组。
关键理解：意思相近的句子，向量就越"接近"（余弦相似度接近1）。这就是能"语义搜索"的原因。

Subsection 2 · 验证相似度 在文件末尾加：

python

from sentence_transformers import utila = model.encode(["我今天迟到了"])b = model.encode(["员工考勤规定"])c = model.encode(["今天天气不错"])print("迟到↔考勤 相似度:", round(util.cos_sim(a, b).item(), 3))print("迟到↔天气 相似度:", round(util.cos_sim(a, c).item(), 3))

预期看到：第一个相似度明显比第二个高。

过关：跑通并验证"语义相似度" = Section 5 完成。

Section 6 · 存进向量数据库（Chroma）

Subsection 1 · 跑入库代码 VS Code 新建 rag_step3_store.py：

python

import chromadbfrom pathlib import Pathfrom sentence_transformers import SentenceTransformer# 加载模型 + 分块（复用 Section 4/5 的成果）model = SentenceTransformer("BAAI/bge-small-zh-v1.5")text = Path("rag_data/company.md").read_text(encoding="utf-8")blocks = [b.strip() for b in text.split("\\") if b.strip()]# 建一个持久化向量库（存到本地文件夹）client = chromadb.PersistentClient(path="chroma_db")collection = client.get_or_create_collection("company_rules")# 向量化 + 入库（id 用序号，metadata 存原文）vectors = model.encode(blocks).tolist()collection.upsert(    ids=[f"doc{i}" for i in range(len(blocks))],    embeddings=vectors,    documents=blocks,          # 存原文，方便取回    metadatas=[{"source": "company.md"} for _ in blocks],)print("已入库", collection.count(), "条")

预期看到：已入库 N 条
Chroma 是本地零配置向量库，最适起步。生产可用 Milvus/FAISS/Qdrant（Part 8 提）。

过关：入库成功 = Section 6 完成。

Section 7 · 检索（把问题变成向量去搜）

Subsection 1 · 跑检索代码 VS Code 新建 rag_step4_search.py：

python

import chromadbfrom sentence_transformers import SentenceTransformer, utilmodel = SentenceTransformer("BAAI/bge-small-zh-v1.5")client = chromadb.PersistentClient(path="chroma_db")collection = client.get_or_create_collection("company_rules")# 把用户问题向量化，去库里搜最接近的question = "迟到会怎么样？"q_vec = model.encode([question]).tolist()result = collection.query(query_embeddings=q_vec, n_results=2)for i, doc in enumerate(result["documents"][0]):    print(f"命中{i}: {doc[:60]}...")

预期看到：两条和"迟到"相关的原文被搜出来。
这就完成了 RAG 的"R"（Retrieval 检索）。

过关：能搜出相关原文 = Section 7 完成。

Section 8 · 第一个完整 RAG（检索 + 生成闭环）

Subsection 1 · 写完整 RAG VS Code 新建 rag_full.py：

python

import chromadbfrom dotenv import load_dotenvfrom openai import OpenAIfrom sentence_transformers import SentenceTransformerimport osload_dotenv()model = SentenceTransformer("BAAI/bge-small-zh-v1.5")client = chromadb.PersistentClient(path="chroma_db")collection = client.get_or_create_collection("company_rules")llm = OpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com")def ask(question: str):    # 1. 检索：找最相关的 3 块    q_vec = model.encode([question]).tolist()    result = collection.query(query_embeddings=q_vec, n_results=3)    context = "\\".join(result["documents"][0])    # 2. 拼 Prompt：把资料给模型，并要求"只依据资料"    prompt = f"""请根据以下资料回答用户问题。如果资料里没有答案，就老实说"资料中没有相关内容"，不要编造。【资料】{context}【问题】{question}"""    # 3. 生成    resp = llm.chat.completions.create(        model="deepseek-chat",        messages=[{"role": "user", "content": prompt}],    )    return resp.choices[0].message.contentif __name__ == "__main__":    while True:        q = input("提问（输入 exit 退出）：")        if q == "exit":            break        print("回答:", ask(q), "\")

运行，连续问：迟到会怎么样、报销有什么规定、公司食堂在哪（最后这个资料里没有）
预期看到：前两个回答引用了资料内容；最后一个回答"资料中没有相关内容"——这就是 RAG 防幻觉的表现。

过关：能跑通"资料外问题不乱编" = Section 8 完成。你已经做出第一个 RAG 了。

Section 9 · 实验：不同 chunk 大小对比

Subsection 1 · 跑对比实验 新建 experiment_chunk.py：

python

# 目的：同样一个问题，不同的"切块大小"检索效果差多少# 用 langchain 的文本切分器做两种分块from langchain_text_splitters import RecursiveCharacterTextSplitterfrom pathlib import Pathtext = Path("rag_data/company.md").read_text(encoding="utf-8")small = RecursiveCharacterTextSplitter(chunk_size=50, chunk_overlap=10).split_text(text)big = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=50).split_text(text)print("小块数量:", len(small), "| 大块数量:", len(big))# 看看"迟到"在哪几块里被提到for label, blocks in [("小块", small), ("大块", big)]:    hits = [b for b in blocks if "迟到" in b]    print(f"{label} 命中'迟到'的块数: {len(hits)}")

运行
预期看到：小块命中更多、更精准；大块命中少但上下文更全。
结论写进笔记：块小→准，块大→全，折中一般 200~500 字 + 少量重叠（overlap）。

过关：能说出块大块小的取舍 = Section 9 完成。

Section 10 · 实验：TopK 和相似度算法

Subsection 1 · 跑 TopK 对比 修改 rag_step4_search.py，把 n_results 分别设为 1、3、5，各打印命中内容。

预期看到：K 越大，带进来的资料越多，回答可能更全但也可能混入噪音。
一般 TopK=3~5 够用。

Subsection 2 · 换相似度算法 Chroma 默认用余弦距离。看官方文档把 collection.query 加参数 distance="l2" 试试（欧式距离）。

对比两次检索结果是否变化，写一句心得。

过关：能说出 TopK 影响和两种距离的区别 = Section 10 完成。

Section 11 · 中文向量模型选型（bge 家族）

Subsection 1 · 记选型表

英文/通用   OpenAI text-embedding-3-small 等中文        百度文心 Embedding-V1、智谱 embedding-2、阿里 text-embedding-v2开源中文     BAAI/bge-large-zh-v1.5（强但大）、bge-small-zh-v1.5（快，你在用）

你现在的 bge-small 已够学习用；生产更追求效果就换 bge-large 或厂商 API。

Subsection 2 · 验证中文效果 在 Section 5 的代码里，用 bge-large-zh-v1.5 再跑一次相似度对比，感受差别（可能更准但更慢）。

过关：能说出"中文场景选什么 Embedding" = Section 11 完成。

Section 12 · 把 RAG 重构成可复用工具

Subsection 1 · 重构 新建 rag_tool.py（把整个 RAG 打包成类，方便以后到处用）：

python

import chromadb, osfrom dotenv import load_dotenvfrom openai import OpenAIfrom sentence_transformers import SentenceTransformerload_dotenv()class RagKB:    """一个简单的知识库问答工具"""    def __init__(self, db_path="chroma_db", collection="company_rules"):        self.model = SentenceTransformer("BAAI/bge-small-zh-v1.5")        self.client = chromadb.PersistentClient(path=db_path)        self.collection = self.client.get_or_create_collection(collection)        self.llm = OpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com")    def add_document(self, file_path: str):        from pathlib import Path        text = Path(file_path).read_text(encoding="utf-8")        blocks = [b.strip() for b in text.split("\\") if b.strip()]        vectors = self.model.encode(blocks).tolist()        self.collection.upsert(            ids=[f"doc{i}" for i in range(len(blocks))],            embeddings=vectors,            documents=blocks,            metadatas=[{"source": file_path} for _ in blocks],        )        return len(blocks)    def ask(self, question: str, top_k: int = 3):        q_vec = self.model.encode([question]).tolist()        result = self.collection.query(query_embeddings=q_vec, n_results=top_k)        context = "\\".join(result["documents"][0])        prompt = f"请根据以下资料回答，资料没有就说'资料中没有相关内容'。\【资料】\{context}\【问题】\{question}"        resp = self.llm.chat.completions.create(model="deepseek-chat", messages=[{"role": "user", "content": prompt}])        return resp.choices[0].message.contentif __name__ == "__main__":    kb = RagKB()    print("入库条数:", kb.add_document("rag_data/company.md"))    while True:        q = input("提问（exit退出）：")        if q == "exit":            break        print("回答:", kb.ask(q))

过关：重构跑通，能 add 能 ask = Section 12 完成。

Section 13 · 用 FastAPI 把它变成 Web 接口

Subsection 1 · 装 FastAPI cmd：pip install fastapi uvicorn

Subsection 2 · 写 API 服务 新建 rag_server.py：

python

from fastapi import FastAPIfrom pydantic import BaseModelfrom rag_tool import RagKBapp = FastAPI()kb = RagKB()class AskReq(BaseModel):    question: str@app.post("/ask")def ask(req: AskReq):    return {"answer": kb.ask(req.question)}

cmd 运行：uvicorn rag_server:app --port 8000
预期看到：Uvicorn running on http://127.0.0.1:8000
浏览器打开 http://127.0.0.1:8000/docs → 能看到接口文档 → 试一次 /ask
你已把一个"知识库问答"变成了标准 Web API——这是后面所有产品的底座。

过关：能通过浏览器 /docs 调通接口 = Section 13 完成。

Section 14 · 综合测试与问题排查

Subsection 1 · 列 10 个测试问题 用 rag_tool.py 测 10 个问题，覆盖：资料内有答案 / 资料内没有 / 模糊问法 / 长问题。

记录每个回答是否：准确 / 引用资料 / 没乱编。

Subsection 2 · 记常见坑 在备忘录写：

问题1 答非所问 → TopK 太小 or chunk 太大，调大 K / 调小 chunk问题2 资料有却答不出 → 检索没召回，检查 Embedding 语言是否匹配（中文要中文模型）问题3 幻觉/乱编 → Prompt 里"资料没有就说没有"写死问题4 速度慢 → bge-large 换 small；或加缓存

过关：10 题测试完 + 常见坑写进笔记 = Section 14 完成。

Section 15 · 站 5 验收

勾选

能说出"为什么需要 RAG"和完整流程
能说出三大范式、三大部件
跑通加载→分块→向量化→入库→检索（Section 4~7）
rag_full.py 完整闭环，资料外问题不乱编（Section 8）
做完 chunk 大小实验，能说取舍（Section 9）
做过 TopK 对比，知道影响（Section 10）
能说出中文 Embedding 选型（Section 11）
rag_tool.py 封装成类（Section 12）
FastAPI 接口调通（Section 13）
10 题测试 + 常见坑笔记（Section 14）

写 300 字Part 5 总结：RAG 完整链路 + 你踩过的坑。

全勾选 = Part 7 通过 → 进入下一站 Part 8 · RAG 优化与评估，10 天：三大范式 + 重排 + RAGAS 打分）。+