RAG 课程笔记 - 离线索引阶段

离线索引流概览

RAG 系统的两条数据流

离线索引流的目标

输入：PDF、TXT、Markdown 等原始文档
输出：可检索的向量索引文件（.index + .json）
特点：
- ✅ 只需执行一次（或知识库更新时重新执行）
- ✅ 不涉及 LLM，仅需 Embedding 模型
- ✅ 追求效率和质量，不要求实时性

完整流程图

原始文档
   ↓
[Document Loader]  ← 统一格式加载
   ↓
完整文本
   ↓
[Text Splitter]    ← 切分成合适片段
   ↓
Chunks[]
   ↓
[Embedding API]    ← 批量向量化
   ↓
Vectors[][]
   ↓
[FAISS Index]      ← 构建向量索引
   ↓
持久化文件 (.index + .json)

Embedding 模型基础

什么是 Embedding？

定义：将文本转换为高维浮点向量的过程，使得语义相近的文本在向量空间中距离更近。

直觉理解：
- "如何申请年假" 和 "请假流程是什么" → 向量空间中的坐标非常接近
- "如何申请年假" 和 "今天天气怎么样" → 向量空间中的坐标相距很远

向量维度示例

# 阿里云 text-embedding-v3 输出 1024 维向量
vec = [
    0.028, -0.042, 0.156, ...,  # 前 3 个维度
    ...                         # 中间 1018 个维度
    ..., -0.091, 0.234          # 最后 2 个维度
]
len(vec) == 1024  # 固定维度

主流 Embedding 模型对比

API 接入实战

环境配置（.env 文件）

# 阿里云百炼 API 密钥（推荐）
DASHSCOPE_API_KEY=sk-your-api-key-here

# OpenAI API 密钥（可选）
OPENAI_API_KEY=sk-your-openai-api-key-here

初始化客户端

from openai import OpenAI
import os
from dotenv import load_dotenv

load_dotenv(override=True)

# 阿里云客户端
client_qwen = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

# OpenAI 客户端（可选）
client_openai = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

单文本 vs 批量向量化

# 单文本向量化（用于实时查询）
def get_embedding(text: str, client, model: str) -> list[float]:
    response = client.embeddings.create(input=[text], model=model)
    return response.data[0].embedding

# 批量向量化（用于离线索引）⭐ 性能提升 10-20 倍
def get_embeddings_batch(texts: list[str], client, model: str) -> list[list[float]]:
    response = client.embeddings.create(input=texts, model=model)
    return [item.embedding for item in response.data]

# 性能对比
texts = ["文本 1", "文本 2", ..., "文本 100"]

# ❌ 慢：循环单文本调用（100 次网络请求）
for text in texts:
    vec = get_embedding(text, client, model)  # 约 30 秒

# ✅ 快：批量调用（1 次网络请求）
vecs = get_embeddings_batch(texts, client, model)  # 约 2 秒

文档加载（Document Loader）

设计目标

统一输出结构：无论输入是 PDF、TXT 还是 Markdown，都输出相同格式的字典对象。

{
    "content": "完整的文本内容...",
    "metadata": {
        "source": "data/company_leave_policy.pdf",
        "type": "pdf",
        "pages": 4
    }
}

手写 Document Loader

import os
from pypdf import PdfReader

def load_txt(file_path: str) -> dict:
    """加载 TXT / Markdown 文件"""
    with open(file_path, 'r', encoding='utf-8') as f:
        return {
            "content": f.read(),
            "metadata": {"source": file_path, "type": "txt"}
        }

def load_pdf(file_path: str) -> dict:
    """加载 PDF 文件：逐页提取文本"""
    reader = PdfReader(file_path)
    text = "\n".join(page.extract_text() or "" for page in reader.pages)
    return {
        "content": text,
        "metadata": {
            "source": file_path,
            "type": "pdf",
            "pages": len(reader.pages)
        }
    }

def load_document(file_path: str) -> dict:
    """统一入口：根据扩展名自动选择加载器"""
    ext = os.path.splitext(file_path)[1].lower()
    loaders = {".txt": load_txt, ".md": load_txt, ".pdf": load_pdf}
    loader = loaders.get(ext)
    if not loader:
        raise ValueError(f"不支持的文件格式：{ext}")
    return loader(file_path)

PDF 解析的两大陷阱

陷阱 1：页眉页脚混入正文

doc = load_document("data/company_leave_policy.pdf")

# ❌ 问题：前三行不是正文
print(repr(doc['content'].split('\n')[:3]))
# 输出：
# ['XX 科技有限公司 内部文件 — 保密级别：内部',
#  '第 1 页 / 共 3 页',
#  '文件编号：HR-2026-001  版本：v3.2']

影响：这些噪声会被切分进 chunk，干扰检索结果。

解决方案（生产环境）：
- 使用正则表达式过滤页眉页脚模式
- 用 pdfplumber 替代 pypdf，支持定位页眉页脚区域
- 对表格类 PDF 改用 OCR 提取

陷阱 2：表格结构丢失

# 原始 PDF 中的表格：
# | 假期类型 | 天数 | 适用条件 | 是否带薪 |
# |----------|------|----------|----------|
# | 年假     | 5-15天| 工作满 1 年 | 是      |

# ❌ pypdf 提取结果（结构完全丢失）：
"""
假期类型
天数
适用条件
是否带薪
年假
5-15 天
工作满 1 年
是
"""

影响：列与列之间的对应关系被破坏，LLM 无法准确回答表格相关问题。

解决方案：
- 对含表格的 PDF 改用 pdfplumber 或 pymupdf
- 将表格转为 Markdown 格式后再入库
- 使用专门的表格提取工具（如 Camelot）

调试技巧：用 repr() 看清特殊字符

# ❌ 直接打印看不出问题
print(doc['content'][:100])

# ✅ 用 repr() 显示所有特殊字符
print(repr(doc['content'][:100]))
# 输出：'XX 科技有限公司\n\n第 1 页\n\n  文件编号：HR-2026-001  \n\n...'
#        ↑ 看到换行符、空格、制表符等细节

文本切分（Text Splitter）

为什么要切分文本？

原因：
1. LLM 上下文窗口有限（通常 128K tokens，约 10 万汉字）
2. 检索时需要精准匹配相关片段，而不是整篇文档
3. 过大的 chunk 会引入无关信息，过小会割裂语义

两个关键参数

chunk_size = 500    # 每个片段的最大字符数
overlap = 100       # 相邻片段的重叠区域

chunk_size 的选择策略

overlap 的作用

防止关键概念恰好被切分点割裂：

Chunk 0: [......第 80-100 句......]
                   ↑↑↑↑↑
              重叠区域（100 字）
                   ↑↑↑↑↑
Chunk 1:           [......第 80-180 句......]

即使切分点落在一个概念的中间，重叠区域也能保证两个 chunk 都包含这个概念的部分信息。

递归分隔符策略

思想：优先在"语义边界"处切开，按粒度从大到小依次尝试。

separators = ["\n\n", "\n", ".", ".", " ", ""]
# 优先级：段落 → 行 → 句子 → 词 → 字符

手写 Text Splitter（完整版）

def split_text(text: str, chunk_size: int = 500, overlap: int = 100,
               separators: list[str] = None, metadata: dict = None) -> list[dict]:
    """
    递归字符切分器（保真版）

    设计约束：
    - chunk 的最终长度不超过 chunk_size
    - overlap 包含在 chunk_size 内
    - 不吞掉原文中的分隔符和连续空白
    """
    # 参数验证
    assert chunk_size > 0, "chunk_size must be greater than 0"
    assert overlap >= 0, "overlap must be greater than or equal to 0"
    assert overlap < chunk_size, "overlap must be smaller than chunk_size"

    if separators is None:
        separators = ["\n\n", "\n", ".", ".", " ", ""]
    metadata = metadata or {}

    # 预留 overlap 空间
    payload_size = chunk_size - overlap if overlap > 0 else chunk_size

    def _split_keep_separator(value: str, sep: str) -> list[str]:
        """切分但保留分隔符在片段末尾"""
        pieces = []
        start = 0
        while True:
            index = value.find(sep, start)
            if index == -1:
                tail = value[start:]
                if tail:
                    pieces.append(tail)
                break
            end = index + len(sep)
            pieces.append(value[start:end])
            start = end
        return pieces or [value]

    def _split_recursive(value: str, seps: list[str]) -> list[str]:
        # 已足够小，直接返回
        if len(value) <= payload_size:
            return [value]

        # 所有分隔符用完，退化为固定宽度硬切
        if not seps:
            return [value[i:i + payload_size] for i in range(0, len(value), payload_size)]

        sep = seps[0]
        if sep == "":
            return [value[i:i + payload_size] for i in range(0, len(value), payload_size)]

        result = []
        for piece in _split_keep_separator(value, sep):
            if len(piece) <= payload_size:
                result.append(piece)
            else:
                result.extend(_split_recursive(piece, seps[1:]))
        return result

    def _merge_splits(splits: list[str]) -> list[str]:
        """贪心合并小片段"""
        chunks = []
        current = ""
        for split in splits:
            if not current:
                current = split
            elif len(current) + len(split) <= payload_size:
                current += split
            else:
                chunks.append(current)
                current = split
        if current:
            chunks.append(current)
        return chunks

    # 递归切分 + 贪心合并
    raw_chunks = _merge_splits(_split_recursive(text, separators))

    # 组装最终结果（包含 overlap 和 metadata）
    chunks = []
    current_start = 0
    for i, raw_chunk in enumerate(raw_chunks):
        if overlap == 0:
            content = raw_chunk
        else:
            start_with_overlap = max(0, current_start - overlap)
            current_end = current_start + len(raw_chunk)
            content = text[start_with_overlap:current_end]

        current_start += len(raw_chunk)

        chunks.append({
            "content": content,
            "metadata": {
                **metadata,
                "chunk_index": i,
                "chunk_total": len(raw_chunks)
            }
        })
        current_start += len(raw_chunk)

    return chunks

测试验证

# 测试用例
sample_text = """人工智能（AI）正在深刻改变各个行业。

机器学习是人工智能的核心子领域。它通过算法让计算机从数据中自动学习规律。

深度学习是机器学习的一个分支，使用多层神经网络来处理复杂的模式识别任务。"""

chunks = split_text(sample_text, chunk_size=100, overlap=20)

print(f"总片段数：{len(chunks)}")
print(f"长度范围：{min(len(c['content']) for c in chunks)} ~ "
      f"{max(len(c['content']) for c in chunks)} 字符")

# 验证 overlap 生效
if len(chunks) > 1:
    prev_tail = chunks[0]['content'][-20:]
    cur_head = chunks[1]['content'][:20]
    print(f"\nOverlap 验证：{prev_tail in chunks[1]['content']}")
    print(f"前一片段末尾：{repr(prev_tail)}")
    print(f"当前片段开头：{repr(cur_head)}")

相似度计算原理

余弦相似度（Cosine Similarity）

定义

衡量两个向量方向的接近程度，而不受向量长度影响。

公式

\cos(a, b) = \frac{a \cdot b}{|a| \times |b|}

其中：
- a \cdot b 是向量点积
- |a| 和 |b| 是向量的 L2 范数（模长）

取值范围与含义

手写实现

import numpy as np

def cosine_similarity(vec_a: list[float], vec_b: list[float]) -> float:
    """计算两个向量的余弦相似度"""
    a, b = np.array(vec_a), np.array(vec_b)

    dot_product = np.dot(a, b)           # 点积
    norm_a = np.linalg.norm(a)           # L2 范数
    norm_b = np.linalg.norm(b)

    return float(dot_product / (norm_a * norm_b))

实际测试

MODEL = "text-embedding-v3"

pairs = [
    ("如何申请年假", "请假流程是什么"),        # 中文同义，预期 > 0.8
    ("如何申请年假", "今天天气怎么样"),        # 语义无关，预期 < 0.5
    ("Python 列表排序", "Python list sort"),  # 中英文同义，预期 > 0.8
]

for text_a, text_b in pairs:
    vec_a = get_embedding(text_a, client_qwen, MODEL)
    vec_b = get_embedding(text_b, client_qwen, MODEL)
    score = cosine_similarity(vec_a, vec_b)
    print(f"{text_a!r:20s} vs {text_b!r:20s} → {score:.4f}")

向量点积（Dot Product）

定义

两个向量对应位置相乘再求和：

a \cdot b = \sum_{i} (a_i \times b_i)

示例

a = [1, 2, 3]
b = [4, 5, 6]
a · b = (1×4) + (2×5) + (3×6) = 4 + 10 + 18 = 32

几何意义

点积本身受向量长度和方向共同影响，不能直接用于语义相似度比较。

关键结论：当向量经过 L2 归一化后（‖a‖ = ‖b‖ = 1），点积等价于余弦相似度：

\text{归一化后：} a \cdot b = \cos(\theta)

欧氏距离（Euclidean Distance）

定义

两点之间的直线距离：

d(a, b) = \sqrt{\sum_{i} (a_i - b_i)^2}

与余弦相似度的关系

未归一化时：两者不等价，欧氏距离受长度影响
归一化后：存在数学关系 d^2 = 2 - 2\cos(\theta)，单调等价

为什么 RAG 首选余弦相似度？

示例：同样的语义，不同的表达长度

a = "如何申请年假"
b = "请问一下年假的申请流程具体是什么样的呢？"

# 两个句子的 Embedding 向量长度不同，但方向应该接近
# 余弦相似度能捕捉这种方向一致性，欧氏距离会被长度差异干扰

FAISS 向量索引构建

为什么需要 FAISS？

问题：如果用 NumPy 逐一计算余弦相似度，在海量向量中会很慢。

# ❌ 暴力搜索：O(N) 复杂度
query_vec = [...]
for doc_vec in all_vectors:  # 假设有 10 万个向量
    score = cosine_similarity(query_vec, doc_vec)
# 需要计算 10 万次，耗时数秒甚至更长

解决方案：FAISS 使用 ANN（Approximate Nearest Neighbor，近似最近邻）算法，实现毫秒级检索。

索引类型选择：IndexFlatIP + normalize_L2

import faiss
import numpy as np

def build_faiss_index(embeddings: list[list[float]]) -> faiss.IndexFlatIP:
    """构建 FAISS 内积索引"""
    dim = len(embeddings[0])
    vectors = np.array(embeddings, dtype=np.float32)

    # 关键：L2 归一化（让每个向量长度变为 1）
    faiss.normalize_L2(vectors)

    # 创建内积索引
    index = faiss.IndexFlatIP(dim)
    index.add(vectors)

    return index

为什么这样选择？

1. L2 归一化：消除向量长度影响
2. IndexFlatIP：内积索引，归一化后内积 = 余弦相似度
3. 返回分数直观：得分范围 [-1, 1]，越大越相似

💡 深入理解：L1 vs L2 归一化

命名的由来："范数"（Norm）

L1、L2 中的 "L" 来自 "Lebesgue"（法国数学家），表示向量的"长度"或"大小"的度量方式。

L1 vs L2 归一化对比

import numpy as np

# 示例向量
v = np.array([3, 4], dtype=np.float32)

# ──────────────────────────────────────────
# L1 归一化：除以绝对值之和
# ──────────────────────────────────────────
l1_norm = np.sum(np.abs(v))  # |3| + |4| = 7
v_l1_normalized = v / l1_norm  # [3/7, 4/7] ≈ [0.429, 0.571]

# 归一化后：L1 范数 = 1
print(f"L1 归一化后：{v_l1_normalized}")
print(f"L1 范数验证：{np.sum(np.abs(v_l1_normalized)):.4f}")  # 1.0

# ──────────────────────────────────────────
# L2 归一化：除以平方和的平方根
# ──────────────────────────────────────────
l2_norm = np.sqrt(np.sum(v**2))  # √(3² + 4²) = √25 = 5
v_l2_normalized = v / l2_norm  # [3/5, 4/5] = [0.6, 0.8]

# 归一化后：L2 范数 = 1（单位圆上）
print(f"L2 归一化后：{v_l2_normalized}")
print(f"L2 范数验证：{np.sqrt(np.sum(v_l2_normalized**2)):.4f}")  # 1.0

核心区别总结

为什么 RAG 用 L2 而不是 L1？

关键原因：L2 归一化后，点积 = 余弦相似度

# 数学推导
设 a, b 为两个向量

# L2 归一化后：‖a‖₂ = 1, ‖b‖₂ = 1

# 点积公式：
a · b = ‖a‖₂ × ‖b‖₂ × cos(θ)
      = 1 × 1 × cos(θ)
      = cos(θ)  ← 直接等于余弦相似度！

# 但 L1 归一化后没有这个性质：
a · b ≠ cos(θ)  （因为 L1 范数不是欧氏空间的自然度量）

FAISS 的 normalize_L2() 做的事情：

# 对每个向量 v
faiss.normalize_L2(v) 
# 等价于：
v_normalized = v / np.linalg.norm(v, ord=2)
# 即：v / √(v₁² + v₂² + ... + vₙ²)

为什么不选 IndexFlatL2（欧氏距离索引）？

• 欧氏距离受长度影响，不适合语义检索
• 归一化后虽然与余弦相似度单调等价，但返回值是距离值（越小越好），不够直观

双文件持久化策略

import json

def save_index(index: faiss.IndexFlatIP, chunks: list[dict],
               index_path: str, metadata_path: str) -> None:
    """
    持久化索引与元数据双文件方案

    - index_path: FAISS 向量索引（二进制格式）
    - metadata_path: 文本块及元数据（JSON 格式）
    """
    faiss.write_index(index, index_path)

    with open(metadata_path, 'w', encoding='utf-8') as f:
        json.dump(chunks, f, ensure_ascii=False, indent=2)

    print(f"✅ 索引保存成功：{index_path}（包含 {index.ntotal} 条向量）")
    print(f"✅ 元数据保存成功：{metadata_path}")

def load_index(index_path: str, metadata_path: str) -> tuple[faiss.IndexFlatIP, list[dict]]:
    """从磁盘恢复索引和关联的文本块数据"""
    index = faiss.read_index(index_path)

    with open(metadata_path, 'r', encoding='utf-8') as f:
        chunks = json.load(f)

    return index, chunks

设计优势

1. 向量和元数据分离：FAISS 只存向量矩阵，JSON 存文本和 metadata
2. 独立更新：换了 Embedding 模型只需重建 .index 文件
3. 数据一致性验证：通过 assert 确保两个文件同步

# 构建 → 保存 → 加载验证
embeddings = get_embeddings_batch(texts, client_qwen, MODEL)
index = build_faiss_index(embeddings)
save_index(index, chunks, "faiss_data/rag.index", "faiss_data/rag_chunks.json")

# 重新加载并验证
index2, chunks2 = load_index("faiss_data/rag.index", "faiss_data/rag_chunks.json")
assert index2.ntotal == len(chunks2), "❌ 索引和 chunks 数量不匹配！"
print(f"✅ 索引持久化验证通过：{index2.ntotal} 条向量 = {len(chunks2)} 个 chunks")

关键知识点总结

核心概念

两类模型的区别

关键事实：
- ✅ 离线索引阶段只用Embedding 模型，不涉及 LLM
- ✅ LLM 只在在线查询阶段的最后一步才登场
- ✅ 索引和查询必须使用同一个Embedding 模型

离线索引流完整代码骨架

# Step 1: 加载文档
doc = load_document("data/company_leave_policy.pdf")

# Step 2: 文本切分
chunks = split_text(doc["content"], chunk_size=500, overlap=100, metadata=doc["metadata"])

# Step 3: 批量向量化
texts = [c["content"] for c in chunks]
embeddings = get_embeddings_batch(texts, client_qwen, "text-embedding-v3")

# Step 4: 构建 FAISS 索引
index = build_faiss_index(embeddings)

# Step 5: 持久化
save_index(index, chunks, "faiss_data/rag.index", "faiss_data/rag_chunks.json")

# Step 6: 验证
index2, chunks2 = load_index("faiss_data/rag.index", "faiss_data/rag_chunks.json")
assert index2.ntotal == len(chunks2), "数据不一致！"

常见陷阱与解决方案

Embedding 相关

陷阱 1：索引和查询使用不同的模型

# ❌ 错误示范
索引时用：text-embedding-v3
查询时用：text-embedding-3-small

# 后果：两个模型的向量空间完全不同，检索结果随机且无意义
# 特征：系统不会报错，但返回的结果完全不相关

解决方案：
- 在配置文件或代码常量中统一声明模型名称
- 加载索引时验证模型一致性（可在 metadata 中记录使用的模型）

陷阱 2：忘记做 L2 归一化

# ❌ 错误
vectors = np.array(embeddings, dtype=np.float32)
index = faiss.IndexFlatIP(dim)
index.add(vectors)  # 未归一化的向量，内积≠余弦相似度

# ✅ 正确
faiss.normalize_L2(vectors)  # 关键！
index.add(vectors)

文档加载相关

陷阱 3：PDF 解析质量盲目信任

# ❌ 天真做法
doc = load_pdf("file.pdf")
print(f"提取了{len(doc['content'])}字")  # 看起来正常就以为没问题

# ✅ 谨慎做法
doc = load_pdf("file.pdf")
print(repr(doc['content'][:200]))  # 用 repr() 检查隐藏的特殊字符
# 手动查看前几行和后几行，确认没有页眉页脚污染

陷阱 4：编码问题导致乱码

# ❌ 可能报错或乱码
with open(file_path, 'r') as f:
    text = f.read()

# ✅ 明确指定 UTF-8 编码
with open(file_path, 'r', encoding='utf-8') as f:
    text = f.read()

文本切分相关

陷阱 5：chunk_size 设置过大或过小

# ❌ 过大：引入大量无关信息
chunk_size = 5000  # 检索到的内容大部分用不上

# ❌ 过小：语义割裂
chunk_size = 50  # 一个完整概念被切成多段

# ✅ 合理范围
chunk_size = 500  # 通用场景
chunk_size = 800  # 技术文档

陷阱 6：忽略 overlap 的重要性

# ❌ 不设 overlap
chunks = split_text(text, chunk_size=500, overlap=0)
# 切分点恰好落在"年假申请流程"中间，两个 chunk 都不完整

# ✅ 设置 10%-20% overlap
chunks = split_text(text, chunk_size=500, overlap=100)
# 两个 chunk 都包含"年假申请"的部分信息，检索时都能命中

参数验证相关

陷阱 7：缺少参数校验导致隐式 Bug

# ❌ 没有验证
def split_text(text, chunk_size, overlap):
    # 如果 overlap >= chunk_size，会导致负数的 payload_size
    payload_size = chunk_size - overlap
    ...

# ✅ 加上 assert 验证
def split_text(text, chunk_size, overlap):
    assert chunk_size > 0
    assert overlap >= 0
    assert overlap < chunk_size  # 关键！防止逻辑错误
    ...

数据一致性相关

陷阱 8：索引和 metadata 数量不匹配

# ❌ 直接使用，可能导致索引越界
index, chunks = load_index(...)
for idx in indices:
    content = chunks[idx]["content"]  # 如果 idx >= len(chunks) 会报错

# ✅ 加载后立即验证
index, chunks = load_index(...)
assert index.ntotal == len(chunks), "数据不一致！"