如何评估 RAG 的分块Chunking策略-海口c网

如何评估 RAG 的分块策略

我对 RAG（检索增强生成模型）进行了深入研究，深知分块在任何 RAG 流水线中都至关重要。

我接触过的许多人坚信更好的模型能够提升 RAG 的性能。有些人则对向量数据库寄予厚望。即便那些认同分块重要性的人，也认为它对系统的改进幅度有限。

大多数人认为，大上下文窗口会取代分块策略的需求。

但分块技术是长期存在的，它非常有效，是任何 RAG 项目的必备要素。

然而，一个关键问题仍未得到解答：如何为项目挑选最佳的分块策略呢？

常用的分块策略有很多：递归字符分割、语义分块以及代理分块，llm分块等等，甚至还主张将聚类作为代理分块的快速且廉价的替代方案。

但残酷的事实是，没有任何一种策略总是表现良好。你无法仅凭项目的性质来猜测哪种策略会奏效。

唯一的办法就是将它们都试一遍，找出最适合的那一个。

所以，这就是我的方法。

首先对文档进行抽样

如果你正在开发一个生产级应用，可能需要处理数百 GB 的数据。

但你不可能用所有这些数据来试验分块策略。

成本是一个显而易见的原因。评估过程会使用 LLM（大型语言模型）推理，无论是通过 API 调用还是本地托管的 LLM 推理，都需要花费一定的成本。

另一个问题是评估的耗时。你的项目通常都有时间限制，你可不想浪费数周时间来评估分块技术。

想低成本且快速地推进？那就抽样吧。

但一定要确保你的样本能够准确地代表总体。

说起来容易做起来难。但鉴于项目的性质，分层抽样可能是个不错的选择。

例如，如果你有 100 份客户交付的 PPT、50 份案例研究 PDF 文档以及 300 份会议记录，你可以从每个类别中各抽取 10% 作为样本。这样就能确保每个群体都在样本中有所体现。

一旦有了样本，就可以进入下一步了。

创建测试集

进行评估需要问题，LLM 将根据这些问题来评估答案。

最好是由人类专家手动创建这些问题集。然而，并非每个组织都有这样的条件。领域专家的时间宝贵，通常无法用于研发工作。

在这种情况下，你可以借助 LLM 来帮忙。

LLM 可以根据样本文档生成问题。尽管我一直说 RAG 中的增强功能不需要强大的 LLM，但这个步骤却需要。选择一个推理能力出色的模型来创建高质量的问题集。

下面的代码可以帮助你开始，根据具体需求进行调整。

def create_test_questions(documents: List[Document], num_questions: int = 10) -> List[Dict[str, str]]:"""从文档中生成测试问题"""questions = []# 为问题生成采样文本sample_texts = []for doc in documents[:3]:  # 使用前 3 份文档words = doc.page_content.split()[:500]  # 前 500 个单词sample_texts.append(" ".join(words))combined_text = "\n\n".join(sample_texts)prompt = f"""根据以下文本，生成 {num_questions} 个多样化的问题，这些问题可以通过提供的信息来回答。文本：{combined_text}只返回问题，每个问题占一行，不要编号或添加额外文本。"""response = llm.invoke(prompt)question_lines = [q.strip() for q in response.content.split('\n') if q.strip()]for i, question in enumerate(question_lines[:num_questions]):questions.append({"question": question,"question_id": f"q_{i+1}"})return questions

上面的代码使用了 Langchain 框架。该函数将文档分成每段 500 个单词的部分，并提示 LLM 生成问题。

准备评估

LLM 评估是一个广泛的话题，我们在这里只是浅尝辄止。

我将使用 RAGAS，这是一个用于评估 RAG 的框架。它功能丰富，但我们只需要其中一小部分指标来完成这项任务。

你可以使用以下命令安装 RAGAS。

pip install ragas
# uv add ragas

如果遇到错误，尝试将 RAGAs 降级到 ragas==0.1.9。

注意：与本文相关的所有代码都在这个笔记本中。我将在每个部分中仅讨论该代码库的一部分。

下面的代码创建并评估了一个简单的 RAG，使用了分块策略。它使用了 Langchain 框架和 Chroma DB 作为向量存储。但如果你使用的是其他向量存储，最好用你选择的向量存储来测试代码。

from typing import List, Dict
import time
import numpy as np
from tqdm import tqdm
from datasets import Dataset
from langchain_chroma import Chroma
from ragas import evaluate
from ragas.metrics import (answer_relevancy,faithfulness,context_precision,context_recall,answer_correctness
)def evaluate_strategy(strategy_name: str,strategy,documents: List,test_questions: List[Dict[str, str]],embeddings,llm,chroma_client,top_k: int = 5) -> 'EvaluationResult':"""评估单个分块策略参数：strategy_name：正在评估的策略名称strategy：具有 chunk_documents 方法的分块策略对象documents：要分块的文档列表test_questions：问题字典列表embeddings：向量存储的嵌入函数llm：用于答案生成的语言模型chroma_client：ChromaDB 客户端实例top_k：要检索的文档数量返回：包含指标和性能数据的 EvaluationResult 对象"""print(f"\n正在评估策略：{strategy_name}")# 1. 对文档进行分块chunks = strategy.chunk_documents(documents)print(f"创建了 {len(chunks)} 个分块")# 2. 创建向量存储collection_name = f"eval_{strategy_name}_{hash(str(time.time()))}"collection_name = collection_name.replace("-", "_").replace(" ", "_")vectorstore = Chroma(collection_name=collection_name,embedding_function=embeddings,client=chroma_client)# 将分块添加到向量存储vectorstore.add_documents(chunks)# 3. 使用测试问题进行评估evaluation_data = []for q_data in tqdm(test_questions, desc="处理问题"):question = q_data["question"]# 检索上下文start_time = time.time()retrieved_docs = vectorstore.similarity_search(question, k=top_k)retrieval_time = time.time() - start_timecontexts = [doc.page_content for doc in retrieved_docs]# 生成答案start_time = time.time()context_text = "\n\n".join(contexts)prompt = f"上下文：
{context_text}问题：{question}答案："response = llm.invoke(prompt)answer = response.contentgeneration_time = time.time() - start_time# 为 RAGAS 准备数据evaluation_data.append({"question": question,"answer": answer,"contexts": contexts,"ground_truth": answer,  # 使用生成的答案作为代理"retrieval_time": retrieval_time,"generation_time": generation_time})# 4. 运行 RAGAS 评估dataset = Dataset.from_list(evaluation_data)metrics = [answer_relevancy,faithfulness,context_precision,context_recall,answer_correctness]ragas_results = evaluate(dataset, metrics=metrics)# 5. 计算统计信息avg_chunk_size = np.mean([len(chunk.page_content) for chunk in chunks])avg_retrieval_time = np.mean([d["retrieval_time"] for d in evaluation_data])avg_generation_time = np.mean([d["generation_time"] for d in evaluation_data])# 计算总体得分（加权平均值）overall_score = (ragas_results["answer_relevancy"] * 0.25 +ragas_results["faithfulness"] * 0.25 +ragas_results["context_precision"] * 0.2 +ragas_results["context_recall"] * 0.2 +ragas_results["answer_correctness"] * 0.1)result = EvaluationResult(strategy_name=strategy_name,chunk_count=len(chunks),avg_chunk_size=avg_chunk_size,retrieval_time=avg_retrieval_time,generation_time=avg_generation_time,answer_relevancy=ragas_results["answer_relevancy"],faithfulness=ragas_results["faithfulness"],context_precision=ragas_results["context_precision"],context_recall=ragas_results["context_recall"],answer_correctness=ragas_results["answer_correctness"],overall_score=overall_score)# 清理try:chroma_client.delete_collection(collection_name)except:passreturn result