EP06. "RAG 检索增强生成实战：从切块到混合搜索"

首页 › AI 工具 › Claude · Developer and Platform · Claude API · EP06

约 21 分钟· #EP06#Claude#Claude API

文档太大塞不进一个 prompt 时，与其把整份文件硬塞进去（撞到长度上限、拖慢速度、推高成本），不如先切块、只挑出跟问题相关的几块喂给 Claude——这就是 RAG（Retrieval Augmented Generation）
切块方式（chunking）选不好，AI 答案就容易出错——句子被腰斩、标题跟内容被拆散，都是常见问题；官方课程比较了四种切法，生产环境最常用的反而是最朴素的「按固定长度切」
语义搜索（找"意思相近"的内容）靠 text embedding + cosine similarity；但遇到工单编号、专有名词这种要「精确命中」的查询，语义搜索反而找不到，得靠 BM25 这种关键词搜索来补
最佳实践是两条腿走路：语义搜索 + BM25 关键词搜索一起跑，用 Reciprocal Rank Fusion（RRF）把两边的排名合并成一份最终结果
前置知识：读过 [EP01](./EP01_claude-api基础入门.md)，知道基本的 messages.create() 用法；这篇笔记涉及向量数据库、embedding API，属于 RAG pipeline 的完整拆解

重点内容

为什么不能把整份文档直接塞进 prompt

两条路：

做法	问题
整份文档塞进 prompt	撞到 prompt 长度硬上限；prompt 太长时 Claude 的表现会下降；处理成本更高；响应更慢
RAG：先切块，只挑相关的塞进去	需要额外的预处理和搜索机制，工程复杂度更高，但能处理超大文档、多份文档，同时省成本、省时间

RAG 换来的是"只处理真正相关的内容"，能撑住超大文档、能同时处理多份文档。代价是：需要一个切块的预处理步骤,需要一套找出"哪几块相关"的搜索机制,还要注意——挑出来的那一小块内容,有可能刚好缺了理解它需要的上下文。

📌 RAG 涉及更多技术决策、要花更多工程功夫——документ非常大、有多份文档要处理、或者在乎成本和性能的时候才值得上这套，不是所有场景都该无脑上 RAG。

切块策略（Chunking）怎么选

📌 切块是搭建 RAG pipeline 最关键的一步——切不好，会把不相关的上下文塞进答案里，直接导致 AI 给错答案。

方法一：按固定长度切（Size-Based Chunking）

把文本切成等长的字符串。比如一份 325 字符的文档，切成三块约 108 字符的片段。

缺点：

词可能被从句子中间硬切断
每块可能丢失重要上下文
章节标题会跟对应内容分家

实现（带 overlap，让相邻块之间有重叠，缓解硬切的问题）：

def chunk_by_char(text, chunk_size=150, chunk_overlap=20):
    chunks = []
    start_idx = 0

    while start_idx < len(text):
        end_idx = min(start_idx + chunk_size, len(text))
        chunk_text = text[start_idx:end_idx]
        chunks.append(chunk_text)

        start_idx = (
            end_idx - chunk_overlap if end_idx < len(text) else len(text)
        )

    return chunks

方法二：按文档结构切（Structure-Based Chunking）

按文档天然的结构（标题、段落、章节）切。适合格式规整的文档，比如 Markdown。

def chunk_by_section(document_text):
    pattern = r"\n## "
    return re.split(pattern, document_text)

方法三：按语义切（Semantic-Based Chunking）

用 NLP 把意思相关的句子归到一组。四种方法里最精细，但计算成本也最高。

方法四：按句子切（Sentence-Based Chunking）

把若干个句子归成一块，块之间可以有重叠句子——介于「简单」和「精细」之间的折中方案。

def chunk_by_sentence(text, max_sentences_per_chunk=5, overlap_sentences=1):
    sentences = re.split(r"(?<=[.!?])\s+", text)

    chunks = []
    start_idx = 0

    while start_idx < len(sentences):
        end_idx = min(start_idx + max_sentences_per_chunk, len(sentences))
        current_chunk = sentences[start_idx:end_idx]
        chunks.append(" ".join(current_chunk))

        start_idx += max_sentences_per_chunk - overlap_sentences

        if start_idx < 0:
            start_idx = 0

    return chunks

方法怎么选

切法	适合场景	备注
按结构切	文档格式受控、规整（如 Markdown）	章节边界清楚，效果最自然
按句子切	通用场景	不错的默认选项
按固定长度切	任何内容类型的保底方案	📌 简单可靠，是"生产环境里最常用的选择"，即使没有前两种精细

Text Embedding：把意思变成数字

语义搜索（semantic search）跟关键词匹配不一样——它靠 embedding 理解「用户问题」和「每个文本块」背后的意思,不是只比对字面。

Embedding 就是「一段文字所包含的意思」的数字化表示：把语言转成一串 -1 到 +1 之间的数字。

环境搭建：

%pip install voyageai

.env 加上：

VOYAGE_API_KEY="your_key_here"

生成 embedding：

from dotenv import load_dotenv
import voyageai

load_dotenv()
client = voyageai.Client()

def generate_embedding(text, model="voyage-3-large", input_type="query"):
    result = client.embed([text], model=model, input_type=input_type)
    return result.embeddings[0]

📌 一个容易被忽略的认知：没人能精确说出某一维数字具体代表什么。虽然理论上某些维度可能对应到文字的某种特质，但每一维实际的含义,是模型在训练过程中自己学出来的,不是人工设计的。

完整 RAG 流程（六步）

切块：把源文档切成块
生成 embedding：用 embedding 模型把每块转成向量
存进向量数据库：把这些向量存起来
embed 用户问题：用同一个 embedding 模型处理用户的问题
查询向量数据库：用 cosine similarity 算法，找出跟问题的 embedding 最相似的已存向量
生成答案：把找出来的相关文本块 + 用户问题一起组进 prompt，发给 Claude 生成最终答案

关于 embedding 和归一化：embedding 模型把文字转成数字向量后，会被归一化（normalize）成长度 1.0，方便统一比较。

Cosine Similarity（余弦相似度）：向量数据库用它来判断哪些 embedding 最相似。分数范围 -1 到 1，越接近 1 代表越相似，0 代表没有关联。

Cosine Distance（余弦距离）：算法是 1 - cosine similarity。这样换算的目的是让数字更好理解——距离越低，代表越相似（分数方向反过来，跟"距离"这个直觉一致）。

动手实现一遍完整流程

五个步骤：按章节切块 → 给每块生成 embedding → 建一个向量库、把每个 embedding 加进去 → 给用户问题生成 embedding → 到库里搜出最相关的几块。

第一步：切块

with open("./report.md", "r") as f:
    text = f.read()

chunks = chunk_by_section(text)

第二步：批量生成 embedding

embeddings = generate_embedding(chunks)

💡 这个函数同时支持单个字符串和列表输入，传列表就能批量处理，效率更高。

第三步：存进向量库

📌 存的时候不能只存向量本身——要连同生成这个 embedding 用的原始文字一起存,不然搜到向量也不知道对应哪段内容：

store = VectorIndex()

for embedding, chunk in zip(embeddings, chunks):
    store.add_vector(embedding, {"content": chunk})

第四步：把用户问题也 embed 一次

user_embedding = generate_embedding("What did the software engineering dept do last year?")

第五步：搜索最相关的块

results = store.search(user_embedding, 2)

for doc, distance in results:
    print(distance, "\n", doc["content"][0:200], "\n")

示范查询返回两条结果，distance 分别是 0.71、0.72——记住数字越低代表越相似。

语义搜索的盲区：BM25 关键词搜索来补位

语义搜索不是万能的。举例：搜一个具体的工单编号「INC-2023-Q4-011」，语义搜索可能只会找到"意思相关"但没提到这个精确编号的内容——因为它找的是"意思"，不是"字面"。

解法：混合搜索（Hybrid Search）

语义搜索：靠 embedding 找概念上相关的内容
词汇搜索（Lexical Search）：靠文本匹配找精确出现的词
合并结果：两边一起用，取长补短

BM25（Best Match 25）算法怎么跑，四步：

Tokenize：把问题切成一个个词
算词频：统计每个词在各文档里出现的频率
按重要性加权：稀有、具体的词权重更高；常见词权重更低
找匹配：返回那些包含较多"高权重词"的文档

# 1. Chunk your text by sections
chunks = chunk_by_section(text)

# 2. Create a BM25 store and add documents
store = BM25Index()
for chunk in chunks:
    store.add_document({"content": chunk})

# 3. Search the store
results = store.search("What happened with INC-2023-Q4-011?", 3)

# Print results
for doc, distance in results:
    print(distance, "\n", doc["content"][:200], "\n----\n")

💡 BM25 的强项：给稀有词更高权重、自动忽略常见虚词、专注词频统计，对技术术语和 ID 编号这类精确匹配特别管用——刚好补上语义搜索的短板。

多索引 RAG Pipeline：融合两种搜索

VectorIndex 跟 BM25Index 两个类的 API 几乎一样——都有 add_document() 和 search() 方法。这让两者能被一个统一的 Retriever 类协调起来：同时查两个索引，再用 Reciprocal Rank Fusion（RRF，倒数排名融合） 把结果合并成一份最终排名。

RRF 公式：

RRF_score(d) = Σ(1 / (k + rank_i(d)))

k 是一个常数（业界常用 60，官方课程为了方便理解示范时用 1），rank_i(d) 是某份文档在第 i 个搜索结果里的排名位置。

手算示范（k=1）：

Section 2：1.0/(1+1) + 1.0/(1+2) = 0.833
Section 7：1.0/(1+2) + 1.0/(1+3) = 0.583
Section 6：1.0/(1+3) + 1.0/(1+1) = 0.75

最终排名：Section 2（0.833）> Section 6（0.75）> Section 7（0.583）。

实现骨架：

class Retriever:
    def __init__(self, *indexes: SearchIndex):
        if len(indexes) == 0:
            raise ValueError("At least one index must be provided")
        self._indexes = list(indexes)

    def add_document(self, document: Dict[str, Any]):
        for index in self._indexes:
            index.add_document(document)

    def search(self, query_text: str, k: int = 1, k_rrf: int = 60):
        # 从所有索引拿结果
        # 记录每份文档在各索引里的排名
        # 套用 RRF 公式算分
        # 返回合并排序后的结果

💡 用同一个「INC-2023-Q4-011」查询测试，混合搜索的效果比单用语义搜索或单用 BM25 都更好——各自的强项被保留下来了。这套架构也方便扩展：任何新的索引实现，只要符合 SearchIndex 这个协议（都有 add_document()/search()），就能直接插进 Retriever，自动参与融合。

常见错误

❌ 向量库里只存 embedding、不存对应的原始文字——搜到相似向量却不知道它对应哪段内容，等于白搜
❌ 不管什么文档都用同一种切块策略——格式规整的 Markdown 用按结构切效果最好，硬套按字符切反而把标题和内容拆散
❌ 只依赖语义搜索处理所有查询——工单编号、专有名词、精确 ID 这类查询，语义搜索经常找不到，要靠 BM25 关键词搜索补位
❌ 混淆 cosine similarity 和 cosine distance 的数字方向——similarity 越高越相似（接近 1），distance 相反，越低越相似，两个混着看容易会错意
❌ 切块时不留 overlap——按固定长度硬切、块与块之间完全不重叠，句子被腰斩后两边都读不出完整意思

Sources

官方文档：

Building with the Claude API（Claude Academy 课程）— https://academy.claude.com/courses/building-with-the-claude-api