RAG评估与优化

1. RAG评估指标

1.1 核心评估维度

RAG系统的评估需要从三个维度进行：

维度	评估对象	关键问题
上下文相关性	检索结果	检索到的文档与查询相关吗？
忠实度	生成结果	生成的回答忠于检索到的上下文吗？
答案相关性	最终输出	回答真正解决了用户的问题吗？

1.2 上下文相关性 (Context Relevance)

衡量检索到的文档与用户查询的匹配程度：

• Precision@K: 前K个检索结果中相关文档的比例
• Recall@K: 所有相关文档中被检索到的比例
• MRR (Mean Reciprocal Rank): 第一个相关文档的排名倒数的平均值
• NDCG (Normalized Discounted Cumulative Gain): 考虑排序位置的相关性评分

1.3 忠实度 (Faithfulness)

衡量生成的回答是否忠于检索到的上下文（不编造信息）：

评估流程：
1. 将回答分解为独立的陈述（claims）
2. 对每个陈述，检查是否能在上下文中找到支持
3. 忠实度 = 有支持的陈述数 / 总陈述数

1.4 答案相关性 (Answer Relevance)

衡量生成的回答是否真正回答了用户的问题：

评估方法：
1. 从回答生成多个可能的问题
2. 计算生成的问题与原始问题的相似度
3. 答案相关性 = 平均相似度

2. RAGAS框架

2.1 RAGAS概述

RAGAS（Retrieval Augmented Generation Assessment）是一个专门用于评估RAG系统的开源框架。

2.2 安装和使用

from ragas import evaluate
from ragas.metrics import (
    context_precision,
    context_recall,
    faithfulness,
    answer_relevancy,
)
from datasets import Dataset

# 准备评估数据
eval_data = {
    "question": ["什么是Transformer？", "RAG的优势是什么？"],
    "answer": ["Transformer是一种...", "RAG的主要优势包括..."],
    "contexts": [
        ["Transformer是Google在2017年提出的..."],
        ["RAG通过检索外部知识来增强..."],
    ],
    "ground_truth": ["Transformer是一种基于自注意力的...", "RAG的优势在于..."],
}

dataset = Dataset.from_dict(eval_data)

# 运行评估
results = evaluate(
    dataset,
    metrics=[
        context_precision,
        context_recall,
        faithfulness,
        answer_relevancy,
    ],
)

print(results)
# {'context_precision': 0.85, 'context_recall': 0.78, 
#  'faithfulness': 0.92, 'answer_relevancy': 0.88}

2.3 RAGAS指标详解

指标	范围	含义	不需要ground truth
context_precision	0-1	检索结果中相关文档排名靠前	否
context_recall	0-1	ground truth中的信息被检索到	否
faithfulness	0-1	回答忠于检索上下文	是
answer_relevancy	0-1	回答与问题相关	是
answer_correctness	0-1	回答与ground truth一致	否

3. Chunk Size优化

3.1 Chunk Size对检索的影响

小Chunk (128-256 tokens):
+ 检索精度高（更精准匹配）
- 上下文信息不完整
- 检索开销大（更多向量）

大Chunk (1024-2048 tokens):
+ 上下文信息完整
- 检索精度低（噪声多）
- 可能超出LLM上下文窗口

推荐Chunk (256-512 tokens):
✓ 精度和上下文的平衡点

3.2 优化实验

def evaluate_chunk_size(chunk_sizes, test_queries, ground_truths):
    """评估不同chunk size的效果"""
    results = {}

    for size in chunk_sizes:
        # 重新分块和索引
        chunks = split_documents(documents, chunk_size=size, overlap=size//5)
        vectorstore = build_index(chunks)
        retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

        # 评估检索质量
        precision_scores = []
        for query, truth in zip(test_queries, ground_truths):
            docs = retriever.get_relevant_documents(query)
            precision = calculate_precision(docs, truth)
            precision_scores.append(precision)

        results[size] = {
            "avg_precision": sum(precision_scores) / len(precision_scores),
            "num_chunks": len(chunks),
        }

    return results

# 实验不同的chunk size
sizes = [128, 256, 512, 1024, 2048]
results = evaluate_chunk_size(sizes, test_queries, ground_truths)

3.3 Parent-Child策略

检索时用小Chunk（高精度）
返回时用大Chunk（完整上下文）

Parent Chunk (1024 tokens): [整段文本]
  ├── Child Chunk 1 (256 tokens): [第一部分]  ← 用于检索
  ├── Child Chunk 2 (256 tokens): [第二部分]  ← 用于检索
  └── Child Chunk 3 (256 tokens): [第三部分]  ← 用于检索

匹配到Child Chunk 2 → 返回整个Parent Chunk

4. 重排序 (Reranking)

4.1 为什么需要重排序

向量检索（双编码器）速度快但精度有限，重排序（交叉编码器）精度高但速度慢：

第一阶段（召回）: 向量检索 → 从百万文档中召回Top 100
第二阶段（精排）: 重排序 → 从100个结果中精排Top 5

4.2 Cohere Rerank

import cohere

co = cohere.Client("your-api-key")

results = co.rerank(
    model="rerank-english-v3.0",
    query="What is the Transformer architecture?",
    documents=[doc.page_content for doc in retrieved_docs],
    top_n=5
)

for result in results.results:
    print(f"Score: {result.relevance_score:.4f} | {result.document.text[:100]}")

4.3 Cross-Encoder重排序

from sentence_transformers import CrossEncoder

# 加载交叉编码器
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

# 准备查询-文档对
pairs = [(query, doc.page_content) for doc in retrieved_docs]

# 计算相关性分数
scores = reranker.predict(pairs)

# 排序
ranked_results = sorted(
    zip(retrieved_docs, scores), 
    key=lambda x: x[1], 
    reverse=True
)[:5]

4.4 重排序模型对比

模型	类型	速度	精度	适用场景
Cohere Rerank v3	API	快	高	生产环境、免部署
cross-encoder/ms-marco	本地	中	中高	英文场景
bge-reranker-large	本地	中	高	中文场景
ColBERT	本地	快	中高	需要低延迟

5. 混合搜索 (Hybrid Search)

5.1 Dense + Sparse检索

结合语义搜索（Dense）和关键词搜索（Sparse）的优势：

from rank_bm25 import BM25Okapi

class HybridRetriever:
    def __init__(self, vectorstore, documents, alpha=0.5):
        self.vectorstore = vectorstore
        self.alpha = alpha  # dense权重

        # 构建BM25索引
        tokenized = [doc.split() for doc in documents]
        self.bm25 = BM25Okapi(tokenized)
        self.documents = documents

    def search(self, query, k=5):
        # Dense检索
        dense_results = self.vectorstore.similarity_search_with_score(query, k=k*2)

        # Sparse检索 (BM25)
        tokenized_query = query.split()
        bm25_scores = self.bm25.get_scores(tokenized_query)

        # 融合分数 (Reciprocal Rank Fusion)
        combined_scores = {}
        for doc, score in dense_results:
            combined_scores[doc.page_content] = self.alpha * score

        for idx, score in enumerate(bm25_scores):
            content = self.documents[idx]
            if content in combined_scores:
                combined_scores[content] += (1 - self.alpha) * score
            else:
                combined_scores[content] = (1 - self.alpha) * score

        # 排序返回Top K
        sorted_results = sorted(combined_scores.items(), key=lambda x: x[1], reverse=True)
        return sorted_results[:k]

5.2 混合搜索的优势

场景	Dense搜索	Sparse搜索	混合搜索
语义相似查询	优	差	优
精确关键词匹配	差	优	优
专业术语	中	优	优
长尾查询	中	中	优

5.3 RRF (Reciprocal Rank Fusion)

def reciprocal_rank_fusion(results_lists, k=60):
    """融合多个排序结果列表"""
    fused_scores = {}

    for results in results_lists:
        for rank, (doc_id, _) in enumerate(results):
            if doc_id not in fused_scores:
                fused_scores[doc_id] = 0
            fused_scores[doc_id] += 1 / (rank + k)

    sorted_results = sorted(fused_scores.items(), key=lambda x: x[1], reverse=True)
    return sorted_results

6. 评估Pipeline

6.1 端到端评估流程

1. 准备评估数据集
   - 问题集（覆盖不同类型和难度）
   - Ground truth答案
   - 相关文档标注

2. 分模块评估
   - 检索评估: Recall@K, Precision@K, MRR
   - 生成评估: Faithfulness, Answer Relevancy
   - 端到端评估: Answer Correctness

3. 失败分析
   - 检索失败: 相关文档未被检索到
   - 排序失败: 相关文档排名靠后
   - 生成失败: 有上下文但回答错误
   - 幻觉: 生成了上下文中没有的信息

4. 针对性优化
   - 检索失败 → 调整Chunk策略/嵌入模型
   - 排序失败 → 添加重排序
   - 生成失败 → 优化Prompt/更换模型
   - 幻觉 → 增加忠实度约束

6.2 失败分析框架

def analyze_failures(eval_results):
    """分析RAG系统的失败模式"""
    failures = {
        "retrieval_miss": [],      # 未检索到相关文档
        "ranking_error": [],       # 相关文档排名靠后
        "generation_error": [],    # 有上下文但回答错误
        "hallucination": [],       # 生成幻觉内容
    }

    for item in eval_results:
        if item["context_recall"] < 0.3:
            failures["retrieval_miss"].append(item)
        elif item["context_precision"] < 0.3:
            failures["ranking_error"].append(item)
        elif item["faithfulness"] < 0.5:
            failures["hallucination"].append(item)
        elif item["answer_correctness"] < 0.5:
            failures["generation_error"].append(item)

    return failures

7. 优化清单

7.1 检索优化

• [ ] 实验不同Chunk大小（256/512/1024）
• [ ] 尝试Parent-Child分块策略
• [ ] 添加混合搜索（Dense + BM25）
• [ ] 实现查询改写（HyDE/Multi-Query）
• [ ] 添加重排序模块

7.2 生成优化

• [ ] 优化RAG Prompt模板
• [ ] 实现上下文压缩
• [ ] 添加来源引用要求
• [ ] 实现Self-RAG验证

7.3 系统优化

• [ ] 建立自动评估Pipeline
• [ ] 设置持续监控
• [ ] 实现增量索引更新
• [ ] 优化查询延迟

参考资料

• Es et al., "RAGAS: Automated Evaluation of Retrieval Augmented Generation", 2023
• RAG架构设计 — RAG Pipeline整体架构
• 向量数据库实战 — 嵌入模型和向量数据库选择

---

评论