05 · RAG 从零实战

本章目标：不依赖任何框架，用最少的库实现一个完整可用的 RAG 问答系统。你将在这一章亲手走完"加载 → 切分 → 向量化 → 检索 → 生成"全链路。

本章你将学到

1. 手写一个中文友好的文本切分器（为什么不能按固定字数硬切）；
2. 用 numpy 手算余弦相似度做 Top-K 检索，再升级到 FAISS 高性能版；
3. 把检索结果组装进 Prompt，让模型"先查资料再回答"；
4. 封装成一个可复用的 MiniRAG 类。

前置知识

• 已按第 02 章搭好环境、跑通过 LLM 调用；
• 了解 Embedding、分块、向量检索的概念（不熟先看第 04 章）。

5.1 为什么"从零写一遍"值得

用 LangChain 的 RetrievalQA 一条链就能实现 RAG，为什么还要手写？

• 理解检索在干什么：框架会替你完成切分和向量化，但"检索不到好答案"时你根本不知道是切分的问题、Embedding 的问题还是 K 值的问题；
• 知道框架在封装什么：等你读到第 7、8 章用 LangChain 重写时，会看得懂它每一步在做什么；
• 可定制：很多生产系统最终就是"手写骨架 + 局部用框架"，理解底层才能安全地做局部替换。

一句话：先能徒手拆装，再用工具。

5.2 项目结构

rag-tutorial/
├── .env                  # OPENAI_API_KEY / OPENAI_BASE_URL
├── requirements.txt
├── data/
│   └── rules.txt         # 知识库源文档（我们提供内容）
├── mini_rag.py           # MiniRAG 类（核心）
└── main.py               # 演示入口

requirements.txt：

openai
numpy
faiss-cpu
python-dotenv

本系列后续章节会用到 langchain 系列，但本章刻意不用——全部用 openai SDK + numpy + faiss 徒手实现。

5.3 准备知识库数据

创建 data/rules.txt，内容是一份《国内航班退改签规则》（为便于演示，模拟生成，不代表真实政策）：

国内航班自愿退票手续费规则
航班起飞前 24 小时以上申请退票，收取票面价 10% 的手续费。
航班起飞前 4 小时至 24 小时内申请退票，收取票面价 20% 的手续费。
航班起飞前 4 小时内申请退票，收取票面价 40% 的手续费。
航班起飞后申请退票，收取票面价 60% 的手续费。
婴儿票退票免收手续费；儿童票按成人票相应退票费率的 50% 收取。
由不可抗力（如台风、地震、军事活动）导致的航班取消，可全额退款，免收手续费。
非自愿退票：因航司原因（航班取消、超售、时刻调整）导致无法成行，可免费退改。
自愿改签规则：起飞前 24 小时以上改签，免收改签费；起飞前 4 至 24 小时改签，收取 15% 改签费；起飞前 4 小时内改签，收取 30% 改签费。
改签仅限同一航司、同一行程区间；改签后如产生票价差额，需补足差价。
自愿退票与改签均需在航班起飞前完成，误机后不可再自助办理，需到柜台处理。
误机处理：未按时办理登机且航班已起飞，按自愿退票最高档费率收取手续费。
航班延误处理：延误 2 小时以上，旅客可免费改签至当日同航线其他航班。
航班取消处理：航司将自动为您改签至后续航班，或全额退款；如选择改签，以航司短信通知为准。
退票款项到账时间：官网与 App 提交退票后，退款一般 7 至 15 个工作日原路退回。
联程航班退改：联程航段中任意一段发生变动，全程均可免费改签或退款。
行李退运：已办理行李托运后申请退票，需先联系值机柜台取消行李托运。
团体票退改：10 人及以上团队票不支持线上自助退改，请拨打 95599 客服专线办理。
特殊旅客：无人陪伴儿童、担架旅客等特殊服务申请，退改需人工审核，电话 95599。
里程兑换机票：自愿退票退回里程，改签按航空里程规则的 15% 收取改签费。

一共 19 条规则，覆盖退票、改签、延误、特殊旅客等常见查询。你可以换成自己的业务文档。

5.4 手写中文切分器

先读文件，再切分。

# mini_rag.py（第一部分：切分）
import re
from typing import List

def split_text_into_chunks(
    text: str, chunk_size: int = 300, overlap: int = 50
) -> List[str]:
    """简化版 RecursiveCharacterTextSplitter（中文友好）。

    思路：先按中文标点（。！？；和换行）切成"句"，
    再把句子顺序拼装进 chunk，拼到接近 chunk_size 就切一刀；
    新 chunk 复用上一块末尾 overlap 长度的内容，避免边界信息被切断。
    """
    sentences = [
        s.strip()
        for s in re.split(r"(?<=[。！？；\n])", text)
        if s.strip()
    ]

    chunks: List[str] = []
    buffer = ""
    for sentence in sentences:
        # 拼上新句会超长 → 先落盘当前块，再带 overlap 开新块
        if buffer and len(buffer) + len(sentence) > chunk_size:
            chunks.append(buffer)
            buffer = buffer[-overlap:] if overlap > 0 else ""
        buffer += sentence
    if buffer:
        chunks.append(buffer)
    return chunks

关键点：

• 按标点切，而不是数到 N 个字符硬切——保证一个 chunk 内是完整的句子，语义不被切碎（第 4 章 4.6 节讲过）；
• overlap 复用上一块末尾：如果某条规则正好横跨两块，overlap 能保证两块都带完整上下文；
• 简单起见按字符数控制长度；生产环境可以按 token 数控制（中文字符 ≈ 0.6~1 token）。

5.5 Embedding：文本 → 向量

# mini_rag.py（第二部分：Embedding）
import os

from dotenv import load_dotenv
from openai import OpenAI

load_dotenv()

_client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),  # 国产端点：DeepSeek/Qwen/GLM 等
)

EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "text-embedding-3-small")
# 国产端点若不支持该模型名，改为 bge-m3 等厂商支持的模型（以官方文档为准）


def get_embeddings(texts: List[str]) -> List[List[float]]:
    """批量向量化。注意分批，单次请求不要塞太多条。"""
    vectors = []
    batch_size = 32
    for i in range(0, len(texts), batch_size):
        batch = texts[i : i + batch_size]
        resp = _client.embeddings.create(model=EMBEDDING_MODEL, input=batch)
        vectors.extend([d.embedding for d in resp.data])
        # resp.usage 里有 token 用量，生产环境要记录成本
    return vectors

5.6 向量存储与检索

5.6.1 方案 A：numpy 手算余弦相似度（先理解原理）

import numpy as np

def cosine_topk(query_vec, matrix: np.ndarray, k: int = 3):
    """对 query_vec 与 matrix 中每一行算余弦相似度，返回 Top-K 下标。

    余弦相似度 = A·B / (|A| * |B|)。向量已归一化时等价于点积。
    """
    q = np.asarray(query_vec, dtype=np.float32)
    q = q / np.linalg.norm(q)                 # 归一化 query
    norms = np.linalg.norm(matrix, axis=1, keepdims=True)
    matrix_normed = matrix / np.maximum(norms, 1e-9)   # 归一化每一行
    scores = matrix_normed @ q                 # 点积 = 余弦相似度
    top_indices = np.argsort(-scores)[:k]      # 降序取前 k
    return top_indices, scores[top_indices]

• matrix 是 (N, dim) 的向量矩阵，scores[i] 表示 query 与第 i 个 chunk 的相似度；
• 全量扫描是 O(N * dim)，数据量小（几千条）完全够用；量大再上 ANN（FAISS，见方案 B）。

5.6.2 方案 B：FAISS（生产更常用）

FAISS 的 IndexFlatIP 是"精确内积"索引——只要把向量归一化，内积就是余弦相似度，且性能远好于纯 numpy 循环：

import faiss

class VectorIndex:
    """基于 FAISS 的向量索引，支持增量添加与 Top-K 检索。"""

    def __init__(self, dim: int):
        self.index = faiss.IndexFlatIP(dim)   # 内积索引（向量需归一化）
        self.texts: List[str] = []

    def add(self, vectors: List[List[float]], texts: List[str]) -> None:
        mat = np.asarray(vectors, dtype=np.float32)
        faiss.normalize_L2(mat)                # 归一化到单位长度
        self.index.add(mat)
        self.texts.extend(texts)

    def search(self, query_vec, k: int = 3):
        q = np.asarray([query_vec], dtype=np.float32)
        faiss.normalize_L2(q)
        scores, ids = self.index.search(q, k)
        ids = ids[0]
        return [(self.texts[i], float(scores[0][j])) for j, i in enumerate(ids) if i != -1]

归一化后 IndexFlatIP 的内积 = 余弦相似度。千万记住：query 与库向量要用同一个 Embedding 模型、做同样的归一化，否则分数没有可比性。

5.7 组装 Prompt 与生成

检索到了，接下来把结果"塞进"Prompt，让模型基于资料回答。

# mini_rag.py（第三部分：生成）
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI

_chat = ChatOpenAI(
    model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
    temperature=0,
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),
)


def build_rag_prompt(question: str, chunks: List[str]) -> List[dict]:
    """把检索到的资料与问题组装成给模型的消息。"""
    context = "\n\n".join(f"[{i + 1}] {c}" for i, c in enumerate(chunks))
    system = SystemMessage(
        content=(
            "你是客服助手。请只根据提供的资料回答用户问题。\n"
            "如果资料中没有答案，明确回答'资料中没有相关信息'，不要编造。\n"
            "回答时在末尾列出你依据的资料编号，如（依据 [1][2]）。"
        )
    )
    user = HumanMessage(content=f"【资料】\n{context}\n\n【问题】\n{question}")
    return [system, user]

要点：

• 约束来源：明确要求"只根据资料回答、查不到就说不知道"——这是抑制幻觉的第一道防线（第 4 章 4.8 节）；
• 引用编号：要求模型标注依据的资料编号，方便人工核对答案来源，这就是"可溯源"。

5.8 封装成 MiniRAG

把上面的零件拼成完整类：

# mini_rag.py（完整封装）
class MiniRAG:
    """不依赖框架的迷你 RAG：build_index(建库) / search(检索) / ask(问答)。"""

    def __init__(self, chunk_size: int = 300, overlap: int = 50):
        self.chunk_size = chunk_size
        self.overlap = overlap
        self.index = None
        self.chunks: List[str] = []

    def build_index(self, raw_text: str) -> None:
        """加载全文 → 切分 → 向量化 → 建索引。"""
        self.chunks = split_text_into_chunks(raw_text, self.chunk_size, self.overlap)
        print(f"共切分成 {len(self.chunks)} 块")
        vectors = get_embeddings(self.chunks)
        self.index = VectorIndex(dim=len(vectors[0]))
        self.index.add(vectors, self.chunks)

    def search(self, question: str, k: int = 3):
        """检索与问题最相关的 k 个块。"""
        q_vec = get_embeddings([question])[0]
        return self.index.search(q_vec, k)

    def ask(self, question: str, k: int = 3) -> str:
        """检索 + 生成：返回带引用的回答。"""
        results = self.search(question, k)
        chunks = [text for text, _ in results]
        print(f"检索到 Top-{k}：")
        for i, (text, score) in enumerate(results, 1):
            print(f"  [{i}] 相似度 {score:.3f} | {text[:40]}...")
        messages = build_rag_prompt(question, chunks)
        return _chat.invoke(messages).content

main.py：

from pathlib import Path

from mini_rag import MiniRAG

if __name__ == "__main__":
    text = Path("data/rules.txt").read_text(encoding="utf-8")

    rag = MiniRAG(chunk_size=300, overlap=50)
    rag.build_index(text)

    while True:
        q = input("\n你的问题（输入 q 退出）：").strip()
        if q.lower() == "q":
            break
        print("助手：", rag.ask(q))

5.9 跑起来

python main.py

试几个问题：

你的问题：航班起飞前4小时内退票，手续费是多少？
检索到 Top-3：
  [1] 相似度 0.63xx | 航班起飞前 4 小时内申请退票，收取票面价 40% 的手续费。...
  [2] 相似度 0.58xx | 航班起飞前 4 小时至 24 小时内申请退票，收取票面价 20% 的手续费。...
  [3] 相似度 0.55xx | 航班起飞后申请退票，收取票面价 60% 的手续费。...
助手： 航班起飞前 4 小时内申请退票，收取票面价 40% 的手续费。（依据 [1]）

注意第 2、3 条检索结果在语义上也相关（都在讲退票手续费），但排序把最精确的规则放在了第一——这正是"语义检索"的价值：不是搜关键字，而是搜"最接近的一句话"。

5.10 实验：chunk_size 对回答的影响

把 chunk_size 分别改成 100 和 800 重跑同一问题，观察差异：

诊断方法：先打印 search() 返回的相似度分数——如果该命中的没命中（分数低/排序靠后），优先怀疑切分粒度；如果分数高但回答还是错，再怀疑生成侧 Prompt。

5.11 小结

• 完整 RAG 链路 = 切分 → Embedding → 向量索引 → Top-K 检索 → 组装 Prompt → 生成，这一章你全部徒手实现了；
• 中文切分优先按标点，chunk_size 与 overlap 是核心调参点；
• 向量归一化后，内积 = 余弦相似度，FAISS IndexFlatIP 即可高性能检索；
• 抑制幻觉第一招：约束模型只依据资料回答，并标注引用编号；
• 下一站：把检索质量做上去（第 06 章——混合检索、重排序、评估）；再用 LangChain 把这一切变成一条可组合的链（第 07 章、第 08 章）。

练习

1. 往 rules.txt 里加 3 条新规则，不重建索引前问一个相关问题，观察旧索引答不出——体会数据更新与重建索引的必要性（第 6 章会讲增量更新）。
2. 把 k 改成 1 和 5，对比回答质量与 token 消耗，找到你数据上的甜点位。
3. 在 ask() 里加一个断言：如果 Top-1 相似度低于某个阈值（如 0.5），直接回复"未检索到相关资料"，观察多少问题会被拦住。
4. 把 Embedding 模型换成本地免费的 bge-m3（pip install sentence-transformers），对比中文检索效果。

参考答案

<details>
<summary>练习1（数据更新与重建索引）</summary>

往 rules.txt 加 3 条新规则后、重建索引前提问，预期：旧索引里根本没有新规则的向量，检索不到 → 模型答"资料中没有"或给出旧信息。这不是 bug，是 RAG 的固有特性：

• 知识库的"记忆"在向量索引里，文本文件变了，索引不会自动跟着变；
• 正确做法：文档入库时给每个 chunk 记一个稳定的 doc_id（如"文件名+序号"），更新流程 = 新规则算 Embedding 入库、改过的 chunk 更新、删除的移除（第 6 章增量更新）；
• 演示版最简单：改完文本后重建索引（重跑入库代码）。生产绝不能整库重建，量大了扛不住。

这也解释了为什么"知识库更新频率"是选型与架构的关键输入（第 4 章练习 2）。
</details>

<details>
<summary>练习2（k 的甜点位）</summary>

• k=1：只取最相似一块。问题常见两种：该命中的分块信息不完整（如只命中"退票手续费按比例收取"却没命中百分比那半句）；Top-1 命中但缺上下文，模型凑不出答案。
• k=5：覆盖更全，但噪音变多——Top-5 里常有 1-2 块无关内容，模型可能被带偏，且 token 消耗上升。
• 甜点位：一般 k=3。判断标准不是"分数高低"，而是回答完整率：问 20 个问题，统计"答案引用编号是否都有效、是否漏关键信息"。k=1 完整率低，k=3 完整率达标，k=5 完整率不再涨——3 就是你的点。

进阶：k 大一点 + Rerank（第 6 章）是另一条路线——先召回 10~20，重排取 3，兼顾召回与精确。

</details>

<details>
<summary>练习3（相似度阈值拦截）</summary>

在 ask() 里加断言：

def ask(question: str, k: int = 3, min_score: float = 0.5) -> str:
    hits = search(question, k=k)
    if not hits or hits[0][1] < min_score:          # (chunk, score)
        return "未检索到相关资料，请换个问法或联系人工。"
    context = "\n".join(f"[{i}] {c}" for i, (c, _) in enumerate(hits, 1))
    # ... 组装 prompt 生成

观察点：

• 阈值 0.5 会拦下多少问题？若拦截率过高，说明你的问题集很多是开放闲聊（不该进 RAG）或 Embedding 质量差；若几乎从不拦截，说明阈值形同虚设。
• 这是"检索置信度兜底"的最小实现——宁可说"查不到"，也不要用低相关上下文编答案，生产环境这一条直接关系到幻觉率。
• 注意相似度分数分布和具体模型强相关，阈值要看自己数据上的分布再定，别照搬 0.5。
  </details>

<details>
<summary>练习4（换本地 bge-m3）</summary>

pip install sentence-transformers

from sentence_transformers import SentenceTransformer

model = SentenceTransformer("BAAI/bge-m3")   # 首次运行自动下载，之后离线可用
vec = model.encode(query)                     # 768 维

替换原 API Embedding 后的对比：

• 优点：免费、离线、中文效果好，bge-m3 同时支持稠密向量 + 稀疏向量（混合检索第 6 章有用）；
• 代价：本地 CPU 推理速度慢（首个请求要加载模型），高 QPS 生产环境要上 GPU 或改用 API；
• 关键约束：入库与检索必须用同一个模型。换成 bge-m3 后，原来用 API 向量入库的数据全部要重新 Embedding，否则新旧向量不可比（第 4.9.4 节的坑）。
  </details>

延伸阅读

• 第 04 章：Embedding / 分块 / 检索指标的完整原理
• 第 06 章：混合检索、重排序、RAG 评估
• 第 08 章：用 LangChain 重写本套 RAG
• FAISS 官方文档：faiss.normalize_L2 与各种索引类型