深入 ‘Automated Benchmarking’：建立一套从检索召回率到生成信实度的自动化回归测试流水线

各位来宾，各位技术同仁，大家好！

今天，我们齐聚一堂，共同探讨一个在当前AI浪潮中至关重要的话题：如何建立一套从检索召回率到生成信实度的自动化回归测试流水线，以深入实现对RAG（Retrieval-Augmented Generation，检索增强生成）系统或类似生成式AI系统的自动化基准测试。

在人工智能，特别是生成式AI飞速发展的今天，我们欣喜地看到大型语言模型（LLM）在理解、生成自然语言方面展现出惊人的能力。然而，这种能力并非没有代价。LLM的“幻觉”（hallucination）、信息过时、难以控制输出风格等问题，使得它们在实际应用中，尤其是在需要高精度、高可靠性的企业级场景中，面临巨大的挑战。RAG架构应运而生，它通过外部知识检索来增强LLM的生成能力，有效缓解了上述部分问题，让模型能够基于实时、准确的私有数据进行回答。

然而，RAG系统并非一劳永逸。它的性能受到检索模块、生成模块、以及两者之间协同作用的复杂影响。任何一环的改动，无论是模型更新、数据索引变更、提示词工程优化，都可能带来意想不到的退化。传统的、依赖人工的评估方式效率低下、成本高昂且主观性强，难以满足快速迭代和持续部署的需求。

这就是我们今天的主题——自动化基准测试流水线——所要解决的核心问题。我们将构建一个系统，它能够像一个不知疲倦、客观公正的质量工程师，在每次代码提交、模型更新或数据变更时，自动运行一系列测试，量化评估系统的性能，并及时发现潜在的回归问题。

自动化基准测试的挑战与必要性

在深入技术细节之前，我们首先要理解为什么自动化基准测试对生成式AI如此重要，以及它面临哪些独特挑战。

挑战：

1. 缺乏明确的“正确答案”： 与传统软件测试不同，生成式AI的输出往往具有多样性和创造性。对于一个给定的查询，可能存在多种合理且正确的回答，这使得定义一个单一的“黄金标准”变得困难。
2. 评估的复杂性与主观性： 衡量生成质量涉及多个维度，如准确性（Accuracy）、信实度（Faithfulness）、相关性（Relevance）、连贯性（Coherence）、流畅性（Fluency）等。这些维度往往难以用简单的数值指标来完全捕捉，且常常带有主观判断的色彩。
3. RAG系统的多阶段特性： RAG系统是检索与生成的串联。检索阶段的错误会直接影响生成阶段的质量。这意味着我们需要分阶段评估，并理解错误是如何在系统内部传播的。
4. 计算资源消耗： LLM推理通常是计算密集型的。大规模的自动化测试可能带来显著的计算成本和时间开销。
5. 数据漂移与模型漂移： 随着时间的推移，用户查询模式、外部知识库内容或底层LLM本身都可能发生变化，导致模型性能下降，需要持续更新和重新评估。

必要性：

1. 保证质量与可靠性： 自动化测试是确保系统在持续迭代中保持或提升质量的唯一途径。
2. 加速开发与迭代： 快速反馈机制让开发者能够信心十足地进行实验和优化。
3. 量化改进与退化： 提供客观数据来衡量每次改动的效果，为决策提供依据。
4. 降低人工评估成本： 显著减少对昂贵且耗时的人工评估的依赖。
5. 提升用户信任： 一个经过严格测试的系统能够提供更稳定、更准确的服务，从而赢得用户信任。

核心评估维度与指标

为了构建一个全面的自动化基准测试流水线，我们需要明确在RAG系统的不同阶段关注哪些评估维度和指标。

1. 检索阶段评估

检索模块的目标是根据用户查询，从庞大的知识库中召回最相关、最权威的文档片段。

维度	描述	典型指标
召回率	有多少相关的文档被成功检索到。	Recall@k (前k个结果中相关文档的比例)
精确率	检索到的文档中，有多少是真正相关的。	Precision@k (前k个结果中相关文档的比例)
排序质量	检索到的相关文档在结果列表中的排名是否靠前。	Mean Reciprocal Rank (MRR), Normalized Discounted Cumulative Gain (NDCG)
时效性	对于时效性强的查询，检索到的信息是否最新。	(需结合人工判断或时间戳元数据)

2. 生成阶段评估

生成模块的目标是根据检索到的文档和用户查询，生成一个高质量、准确、无幻觉的答案。

维度	描述	典型指标 (及其局限性)
信实度	生成的答案是否完全基于所提供的检索文档，没有引入外部信息或“幻觉”。	LLM-as-a-Judge (基于提示词的评分), RAGAS中的Faithfulness
相关性	生成的答案是否直接且完整地回答了用户查询。	LLM-as-a-Judge (基于提示词的评分), RAGAS中的Relevance
连贯性与流畅性	答案是否逻辑清晰、语言自然、无语法错误。	LLM-as-a-Judge (基于提示词的评分), BLEU/ROUGE (对开放域生成评估效果有限)
完整性	答案是否包含了用户查询所需的所有关键信息。	LLM-as-a-Judge (基于提示词的评分), (需结合人工判断或人工参考答案进行比较，如ROUGE-L)
简洁性	答案是否避免了冗余信息，直截了当。	答案长度，LLM-as-a-Judge (基于提示词的评分)
安全性	答案是否包含有害、偏见或不当内容。	LLM-as-a-Judge (基于提示词的评分), 特定安全模型（如Guardrails）进行过滤，或关键词匹配

关于LLM-as-a-Judge：
这是当前评估生成质量，特别是信实度和相关性的最有效方法之一。我们使用一个能力更强（通常是更大或微调过的）LLM作为“评委”，通过精心设计的提示词，让它对模型生成的答案进行评分或判断。这种方法避免了人工评估的成本，同时克服了传统基于文本匹配指标（如BLEU, ROUGE）在开放域生成中表现不佳的局限性。

构建自动化回归测试流水线：技术栈与架构

我们将构建的流水线将包含以下核心组件：

1. 测试数据集管理： 存储和管理测试查询、预期答案、相关文档ID等。
2. RAG系统集成： 能够调用被测试的RAG系统，获取检索结果和生成结果。
3. 检索评估模块： 计算检索阶段的各项指标。
4. 生成评估模块： 利用LLM-as-a-Judge或其他方法评估生成质量。
5. 结果存储与报告： 存储评估结果，生成可视化报告，并支持趋势分析。
6. CI/CD集成： 将整个流水线集成到开发流程中，实现自动化触发。

推荐技术栈：

• Python: 核心编程语言。
• Pandas/Pydantic: 数据结构与测试数据管理。
• LangChain/LlamaIndex: RAG系统集成框架，简化LLM和检索器的交互。
• OpenAI API / Hugging Face Transformers: LLM交互。
• RAGAS: 专门用于RAG评估的Python库，提供开箱即用的LLM-as-a-Judge评估器。
• Pytest: 测试框架，用于组织和运行测试。
• MLflow/Weights & Biases: 实验跟踪和结果管理（可选，但推荐用于生产环境）。
• GitHub Actions / Jenkins: CI/CD工具。

流水线架构概览：

graph TD
    A[代码提交/模型更新/数据变更] --> B(CI/CD Trigger)
    B --> C{自动化基准测试流水线}

    C --> D[加载测试数据集]
    D --> E[调用RAG系统]

    E --> F[RAG系统 - 检索阶段]
    F --> G[获取检索结果]
    G --> H[检索评估模块]
    H --> I[检索指标]

    E --> J[RAG系统 - 生成阶段]
    J --> K[获取生成结果]
    K --> L[生成评估模块 (LLM-as-a-Judge)]
    L --> M[生成指标 (信实度, 相关性等)]

    I & M --> N[结果聚合与存储]
    N --> O[报告生成与可视化]
    O --> P{阈值检查}
    P -- 通过 --> Q[部署/合并]
    P -- 未通过 --> R[失败通知/回滚]

深入实战：构建自动化回归测试流水线

现在，我们将通过具体的代码示例，一步步构建这个自动化基准测试流水线。

1. 环境准备

# 创建虚拟环境
python -m venv venv
source venv/bin/activate # macOS/Linux
# venvScriptsactivate # Windows

# 安装所需库
pip install pandas pydantic langchain openai chromadb transformers scikit-learn numpy ragas pytest

2. 定义数据模型

首先，我们需要一个清晰的数据模型来表示我们的测试用例和评估结果。

# test_models.py
from pydantic import BaseModel, Field
from typing import List, Dict, Any, Optional

class Document(BaseModel):
    """表示一个知识库中的文档片段。"""
    page_content: str
    metadata: Dict[str, Any] = Field(default_factory=dict)
    doc_id: Optional[str] = None # 方便检索时进行匹配

class TestCase(BaseModel):
    """一个独立的测试用例。"""
    test_id: str
    query: str
    ground_truth_answer: str
    # 用于检索评估：预期应该召回的文档的ID列表
    expected_retrieval_doc_ids: Optional[List[str]] = None
    # 用于生成评估：作为参考的完整相关文档内容
    ground_truth_docs: Optional[List[Document]] = None

class RetrievalMetrics(BaseModel):
    """检索评估结果。"""
    precision_at_k: Optional[float] = None
    recall_at_k: Optional[float] = None
    mrr: Optional[float] = None
    # 可以添加更多，如ndcg_at_k

class GenerationMetrics(BaseModel):
    """生成评估结果。"""
    faithfulness: Optional[float] = None # 信实度 (LLM-as-a-Judge)
    answer_relevance: Optional[float] = None # 答案相关性 (LLM-as-a-Judge)
    context_recall: Optional[float] = None # 上下文召回率 (RAGAS)
    context_precision: Optional[float] = None # 上下文精确率 (RAGAS)
    # 可以添加更多，如coherence, toxicity, response_similarity_with_gt (如果存在人工参考答案)

class EvaluationResult(BaseModel):
    """单个测试用例的完整评估结果。"""
    test_case: TestCase
    retrieved_docs: List[Document] # 实际检索到的文档
    generated_response: str # 实际生成的答案
    retrieval_metrics: RetrievalMetrics
    generation_metrics: GenerationMetrics
    timestamp: str # 记录评估时间

class BenchmarkReport(BaseModel):
    """整个基准测试的报告。"""
    report_id: str
    start_time: str
    end_time: str
    total_test_cases: int
    overall_retrieval_metrics: Optional[Dict[str, float]] = None
    overall_generation_metrics: Optional[Dict[str, float]] = None
    individual_results: List[EvaluationResult]
    passed_thresholds: Dict[str, bool] = Field(default_factory=dict)

3. 模拟RAG系统与知识库

为了演示，我们先构建一个简化的模拟RAG系统和知识库。在实际应用中，这将是您自己的RAG实现。

# rag_system.py
import time
from typing import List, Dict, Any
from langchain.schema import Document as LCDocument
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chat_models import ChatOpenAI
from langchain.chains import RetrievalQA
from test_models import Document

# 假设的知识库内容
MOCK_KNOWLEDGE_BASE = [
    Document(doc_id="doc1", page_content="Python是一种高级的、解释型的编程语言。它由Guido van Rossum于1991年首次发布。"),
    Document(doc_id="doc2", page_content="Python的特点包括：易于学习、代码可读性强、拥有庞大的标准库和第三方库生态系统。"),
    Document(doc_id="doc3", page_content="RAG（Retrieval-Augmented Generation）是一种结合了信息检索和文本生成的技术。"),
    Document(doc_id="doc4", page_content="RAG系统通过从外部知识库中检索相关信息来增强大型语言模型（LLM）的生成能力。"),
    Document(doc_id="doc5", page_content="自动化基准测试对于确保AI系统质量和发现回归至关重要。"),
    Document(doc_id="doc6", page_content="LLM的幻觉问题是生成式AI面临的主要挑战之一。"),
    Document(doc_id="doc7", page_content="Chromadb是一个开源的嵌入式向量数据库，常用于存储和检索文档嵌入。"),
    Document(doc_id="doc8", page_content="LangChain是一个用于开发由语言模型驱动的应用程序的框架。")
]

class MockRAGSystem:
    def __init__(self, api_key: str):
        self.llm = ChatOpenAI(temperature=0.0, openai_api_key=api_key)
        self.embeddings = OpenAIEmbeddings(openai_api_key=api_key)
        self.vectorstore = self._build_vectorstore(MOCK_KNOWLEDGE_BASE)
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            retriever=self.vectorstore.as_retriever(search_kwargs={"k": 3}),
            return_source_documents=True
        )

    def _build_vectorstore(self, documents: List[Document]) -> Chroma:
        """构建 ChromaDB 向量存储。"""
        lc_docs = [LCDocument(page_content=doc.page_content, metadata=doc.metadata) for doc in documents]
        # 注意：此处使用in-memory的ChromaDB，实际应用中可能需要持久化或使用其他向量数据库
        vectorstore = Chroma.from_documents(lc_docs, self.embeddings)
        return vectorstore

    def query(self, query_text: str) -> Dict[str, Any]:
        """
        模拟RAG系统的查询接口。
        返回检索到的文档和生成的答案。
        """
        response = self.qa_chain({"query": query_text})

        retrieved_lc_docs = response.get("source_documents", [])
        retrieved_docs = []
        for doc in retrieved_lc_docs:
            # 尝试从metadata中提取doc_id，或者基于内容生成一个
            doc_id = doc.metadata.get('doc_id') # 如果原始Document有id，这里应该能传递过来
            if not doc_id:
                # 简单的hash作为fallback，实际生产中需要更鲁棒的ID管理
                doc_id = str(hash(doc.page_content)) 
            retrieved_docs.append(Document(page_content=doc.page_content, metadata=doc.metadata, doc_id=doc_id))

        generated_answer = response.get("result", "")

        return {
            "retrieved_docs": retrieved_docs,
            "generated_answer": generated_answer
        }

# 示例用法 (需要在 .env 文件中设置 OPENAI_API_KEY 或直接传入)
# if __name__ == "__main__":
#     import os
#     from dotenv import load_dotenv
#     load_dotenv()
#     openai_api_key = os.getenv("OPENAI_API_KEY")
#     if not openai_api_key:
#         raise ValueError("OPENAI_API_KEY not set in environment variables.")

#     rag_system = MockRAGSystem(api_key=openai_api_key)
#     test_query = "Python是由谁开发的？"
#     result = rag_system.query(test_query)
#     print(f"Query: {test_query}")
#     print(f"Generated Answer: {result['generated_answer']}")
#     print("Retrieved Documents:")
#     for doc in result['retrieved_docs']:
#         print(f"- {doc.page_content[:50]}...")

请注意，MockRAGSystem中的 _build_vectorstore 部分，在实际应用中，doc_id 需要在构建向量库时就明确地与 Document 关联起来，以便在检索结果中准确获取。这里为了演示，我们假设原始 Document 的 doc_id 能够通过 metadata 传递。

4. 测试用例生成与管理

测试用例是基准测试的基石。我们可以手动创建，也可以利用LLM生成。

# test_data_manager.py
import json
from typing import List
from test_models import TestCase, Document, MOCK_KNOWLEDGE_BASE

def load_test_cases(file_path: str) -> List[TestCase]:
    """从JSON文件加载测试用例。"""
    with open(file_path, 'r', encoding='utf-8') as f:
        data = json.load(f)
    return [TestCase(**tc) for tc in data]

def save_test_cases(test_cases: List[TestCase], file_path: str):
    """将测试用例保存到JSON文件。"""
    with open(file_path, 'w', encoding='utf-8') as f:
        json.dump([tc.dict() for tc in test_cases], f, ensure_ascii=False, indent=4)

def generate_synthetic_test_cases(num_cases: int = 5) -> List[TestCase]:
    """
    (简化版) 模拟通过LLM生成合成测试用例。
    在真实场景中，你会使用一个LLM来基于知识库生成问题、答案和相关文档ID。
    """
    synthetic_cases = [
        TestCase(
            test_id="s_001",
            query="Python的创始人和发布时间是什么？",
            ground_truth_answer="Python由Guido van Rossum于1991年首次发布。",
            expected_retrieval_doc_ids=["doc1", "doc2"],
            ground_truth_docs=[MOCK_KNOWLEDGE_BASE[0], MOCK_KNOWLEDGE_BASE[1]]
        ),
        TestCase(
            test_id="s_002",
            query="RAG技术的主要目的是什么？",
            ground_truth_answer="RAG技术旨在结合信息检索和文本生成，通过从外部知识库检索信息来增强LLM的生成能力。",
            expected_retrieval_doc_ids=["doc3", "doc4"],
            ground_truth_docs=[MOCK_KNOWLEDGE_BASE[2], MOCK_KNOWLEDGE_BASE[3]]
        ),
        TestCase(
            test_id="s_003",
            query="自动化基准测试为何重要？",
            ground_truth_answer="自动化基准测试对于确保AI系统质量、发现回归问题以及加速开发迭代至关重要。",
            expected_retrieval_doc_ids=["doc5"],
            ground_truth_docs=[MOCK_KNOWLEDGE_BASE[4]]
        ),
        TestCase(
            test_id="s_004",
            query="LLM的幻觉问题指的是什么？",
            ground_truth_answer="LLM的幻觉问题是指模型生成不准确或虚假信息，这是生成式AI面临的主要挑战之一。",
            expected_retrieval_doc_ids=["doc6"],
            ground_truth_docs=[MOCK_KNOWLEDGE_BASE[5]]
        ),
        TestCase(
            test_id="s_005",
            query="Chromadb和LangChain分别是什么？",
            ground_truth_answer="Chromadb是一个开源的嵌入式向量数据库，LangChain是用于开发语言模型驱动应用的框架。",
            expected_retrieval_doc_ids=["doc7", "doc8"],
            ground_truth_docs=[MOCK_KNOWLEDGE_BASE[6], MOCK_KNOWLEDGE_BASE[7]]
        )
    ]
    return synthetic_cases[:num_cases]

# 创建一个示例测试文件
if __name__ == "__main__":
    generated_cases = generate_synthetic_test_cases(5)
    save_test_cases(generated_cases, "test_cases.json")
    print("Generated and saved test_cases.json")

    loaded_cases = load_test_cases("test_cases.json")
    print(f"Loaded {len(loaded_cases)} test cases.")
    print(loaded_cases[0].dict())

generate_synthetic_test_cases 在真实场景中会更复杂，例如使用LlamaIndex的QueryGenerator或RAGAS的SyntheticTestDataset功能来自动化创建。

5. 检索评估模块

实现计算Precision@k、Recall@k、MRR的逻辑。

# evaluators.py
import numpy as np
from typing import List, Dict, Any, Optional
from test_models import TestCase, Document, RetrievalMetrics
from ragas.metrics import faithfulness, answer_relevance, context_recall, context_precision
from ragas import evaluate
from langchain.schema import Document as LCDocument

class RetrievalEvaluator:
    def __init__(self, k: int = 3):
        self.k = k

    def evaluate(self, test_case: TestCase, retrieved_docs: List[Document]) -> RetrievalMetrics:
        """
        评估检索性能。
        """
        if not test_case.expected_retrieval_doc_ids:
            return RetrievalMetrics() # 如果没有预期文档，则无法评估

        retrieved_doc_ids = [doc.doc_id for doc in retrieved_docs if doc.doc_id]

        # 计算 Precision@k
        relevant_retrieved_count = len(set(retrieved_doc_ids[:self.k]) & set(test_case.expected_retrieval_doc_ids))
        precision_at_k = relevant_retrieved_count / self.k if self.k > 0 else 0.0

        # 计算 Recall@k (注意：这里的Recall@k是针对整个expected_retrieval_doc_ids的召回，而不是前k个)
        total_relevant_docs = len(test_case.expected_retrieval_doc_ids)
        recall_at_k = relevant_retrieved_count / total_relevant_docs if total_relevant_docs > 0 else 0.0

        # 计算 MRR (Mean Reciprocal Rank)
        mrr = 0.0
        for i, doc_id in enumerate(retrieved_doc_ids[:self.k]):
            if doc_id in test_case.expected_retrieval_doc_ids:
                mrr = 1.0 / (i + 1)
                break # 找到第一个相关文档即停止

        return RetrievalMetrics(
            precision_at_k=precision_at_k,
            recall_at_k=recall_at_k,
            mrr=mrr
        )

class GenerationEvaluator:
    def __init__(self, llm_judge: Any):
        """
        初始化生成评估器。
        llm_judge 可以是 LangChain 的 ChatOpenAI 实例。
        """
        self.llm_judge = llm_judge
        self.metrics = [
            faithfulness,
            answer_relevance,
            context_recall,
            context_precision
        ]

    async def evaluate(self, query: str, retrieved_docs: List[Document],
                       generated_response: str, ground_truth_answer: str,
                       ground_truth_docs: Optional[List[Document]] = None) -> GenerationMetrics:
        """
        利用RAGAS库评估生成质量。
        """
        # 将自定义Document转换为LangChain Document格式
        lc_retrieved_docs = [LCDocument(page_content=doc.page_content, metadata=doc.metadata) for doc in retrieved_docs]
        lc_ground_truth_docs = [LCDocument(page_content=doc.page_content, metadata=doc.metadata) for doc in ground_truth_docs] if ground_truth_docs else []

        # RAGAS需要一个DatasetDict格式的数据
        # 注意：RAGAS的context_recall和context_precision需要ground_truth_docs作为reference_contexts
        data_row = {
            "question": [query],
            "answer": [generated_response],
            "contexts": [[d.page_content for d in lc_retrieved_docs]],
            "ground_truth": [ground_truth_answer],
            "reference_contexts": [[d.page_content for d in lc_ground_truth_docs]] if lc_ground_truth_docs else None
        }

        # RAGAS 评估
        # 必须传入 LangChain LLM 实例作为 argument
        result = await evaluate(
            dataset=data_row,
            metrics=self.metrics,
            llm=self.llm_judge,
            embeddings=self.llm_judge.embeddings if hasattr(self.llm_judge, 'embeddings') else None # RAGAS 0.1.x 可能需要
        )

        # 将结果转换为 GenerationMetrics
        metrics_dict = result.to_pandas().iloc[0].to_dict()
        return GenerationMetrics(
            faithfulness=metrics_dict.get('faithfulness'),
            answer_relevance=metrics_dict.get('answer_relevance'),
            context_recall=metrics_dict.get('context_recall'),
            context_precision=metrics_dict.get('context_precision')
        )

# 示例用法 (需要在 .env 文件中设置 OPENAI_API_KEY 或直接传入)
# if __name__ == "__main__":
#     import os
#     from dotenv import load_dotenv
#     import asyncio
#     load_dotenv()
#     openai_api_key = os.getenv("OPENAI_API_KEY")
#     if not openai_api_key:
#         raise ValueError("OPENAI_API_KEY not set in environment variables.")

#     # 模拟数据
#     test_case_ex = TestCase(
#         test_id="ex_001",
#         query="Python是什么？",
#         ground_truth_answer="Python是一种高级的、解释型的编程语言。",
#         expected_retrieval_doc_ids=["doc1"],
#         ground_truth_docs=[MOCK_KNOWLEDGE_BASE[0]]
#     )
#     retrieved_docs_ex = [MOCK_KNOWLEDGE_BASE[0], MOCK_KNOWLEDGE_BASE[2]] # 模拟检索结果
#     generated_response_ex = "Python是一种高级的、解释型的编程语言，它非常流行。"

#     # 检索评估
#     ret_evaluator = RetrievalEvaluator(k=3)
#     ret_metrics = ret_evaluator.evaluate(test_case_ex, retrieved_docs_ex)
#     print(f"Retrieval Metrics: {ret_metrics.dict()}")

#     # 生成评估
#     llm_judge_instance = ChatOpenAI(temperature=0.0, openai_api_key=openai_api_key)
#     gen_evaluator = GenerationEvaluator(llm_judge=llm_judge_instance)
#     # RAGAS 0.1.x 的 evaluate 是异步的
#     gen_metrics = asyncio.run(gen_evaluator.evaluate(
#         query=test_case_ex.query,
#         retrieved_docs=retrieved_docs_ex,
#         generated_response=generated_response_ex,
#         ground_truth_answer=test_case_ex.ground_truth_answer,
#         ground_truth_docs=test_case_ex.ground_truth_docs
#     ))
#     print(f"Generation Metrics: {gen_metrics.dict()}")

注意： RAGAS库的evaluate函数是异步的，因此需要使用asyncio.run()来运行。

6. 自动化基准测试流水线核心逻辑

将所有组件整合起来，执行完整的测试流程。

# benchmark_pipeline.py
import os
import datetime
import asyncio
from typing import List, Dict, Any
from dotenv import load_dotenv
import pandas as pd
from test_models import TestCase, Document, RetrievalMetrics, GenerationMetrics, EvaluationResult, BenchmarkReport
from test_data_manager import load_test_cases, generate_synthetic_test_cases, save_test_cases
from rag_system import MockRAGSystem, MOCK_KNOWLEDGE_BASE
from evaluators import RetrievalEvaluator, GenerationEvaluator
from langchain.chat_models import ChatOpenAI
from langchain.embeddings import OpenAIEmbeddings

class AutomatedBenchmarkingPipeline:
    def __init__(self, openai_api_key: str, retrieval_k: int = 3):
        self.rag_system = MockRAGSystem(api_key=openai_api_key)
        self.retrieval_evaluator = RetrievalEvaluator(k=retrieval_k)
        # RAGAS 需要一个 LLM 实例作为 judge
        self.llm_judge = ChatOpenAI(model_name="gpt-4", temperature=0.0, openai_api_key=openai_api_key) # 推荐使用更强的模型作为judge
        self.llm_judge.embeddings = OpenAIEmbeddings(openai_api_key=openai_api_key) # RAGAS 0.1.x 可能需要 embeddings
        self.generation_evaluator = GenerationEvaluator(llm_judge=self.llm_judge)
        self.evaluation_results: List[EvaluationResult] = []
        self.benchmark_report: Optional[BenchmarkReport] = None

    async def run_single_test_case(self, test_case: TestCase) -> EvaluationResult:
        """
        运行单个测试用例，执行检索和生成，并进行评估。
        """
        print(f"Running test case: {test_case.test_id} - Query: {test_case.query}")

        # 1. 调用RAG系统
        rag_output = self.rag_system.query(test_case.query)
        retrieved_docs = rag_output["retrieved_docs"]
        generated_response = rag_output["generated_answer"]

        # 2. 检索评估
        retrieval_metrics = self.retrieval_evaluator.evaluate(test_case, retrieved_docs)

        # 3. 生成评估
        generation_metrics = await self.generation_evaluator.evaluate(
            query=test_case.query,
            retrieved_docs=retrieved_docs,
            generated_response=generated_response,
            ground_truth_answer=test_case.ground_truth_answer,
            ground_truth_docs=test_case.ground_truth_docs
        )

        current_time = datetime.datetime.now().isoformat()
        result = EvaluationResult(
            test_case=test_case,
            retrieved_docs=retrieved_docs,
            generated_response=generated_response,
            retrieval_metrics=retrieval_metrics,
            generation_metrics=generation_metrics,
            timestamp=current_time
        )
        self.evaluation_results.append(result)
        return result

    async def run_benchmark(self, test_cases_file: str = "test_cases.json"):
        """
        运行整个基准测试。
        """
        start_time = datetime.datetime.now()
        print(f"Starting benchmark at {start_time.isoformat()}")

        # 确保测试用例文件存在，如果不存在则生成
        if not os.path.exists(test_cases_file):
            print(f"Test cases file '{test_cases_file}' not found. Generating synthetic cases.")
            synthetic_cases = generate_synthetic_test_cases()
            save_test_cases(synthetic_cases, test_cases_file)

        test_cases = load_test_cases(test_cases_file)
        print(f"Loaded {len(test_cases)} test cases.")

        self.evaluation_results = []
        # 并发运行测试用例 (RAGAS evaluate 是异步的)
        tasks = [self.run_single_test_case(tc) for tc in test_cases]
        await asyncio.gather(*tasks) # 等待所有异步任务完成

        end_time = datetime.datetime.now()
        print(f"Benchmark finished at {end_time.isoformat()}")

        # 聚合结果
        self._aggregate_results(start_time, end_time, len(test_cases))

        # 报告生成
        self.generate_report()

    def _aggregate_results(self, start_time: datetime.datetime, end_time: datetime.datetime, total_cases: int):
        """聚合所有测试用例的评估结果，计算平均值。"""
        df_retrieval = pd.DataFrame([res.retrieval_metrics.dict() for res in self.evaluation_results])
        df_generation = pd.DataFrame([res.generation_metrics.dict() for res in self.evaluation_results])

        overall_retrieval_metrics = df_retrieval.mean().to_dict()
        overall_generation_metrics = df_generation.mean().to_dict()

        self.benchmark_report = BenchmarkReport(
            report_id=f"benchmark_{start_time.strftime('%Y%m%d_%H%M%S')}",
            start_time=start_time.isoformat(),
            end_time=end_time.isoformat(),
            total_test_cases=total_cases,
            overall_retrieval_metrics=overall_retrieval_metrics,
            overall_generation_metrics=overall_generation_metrics,
            individual_results=self.evaluation_results
        )

    def generate_report(self, output_dir: str = "benchmark_reports"):
        """生成详细的HTML或JSON报告。"""
        if not self.benchmark_report:
            print("No benchmark report available. Run benchmark first.")
            return

        os.makedirs(output_dir, exist_ok=True)
        report_file = os.path.join(output_dir, f"{self.benchmark_report.report_id}.json")

        with open(report_file, 'w', encoding='utf-8') as f:
            json.dump(self.benchmark_report.dict(), f, ensure_ascii=False, indent=4)
        print(f"Benchmark report saved to {report_file}")

        # 打印简要报告
        print("n--- Benchmark Summary ---")
        print(f"Report ID: {self.benchmark_report.report_id}")
        print(f"Total Test Cases: {self.benchmark_report.total_test_cases}")
        print("nOverall Retrieval Metrics:")
        for metric, value in self.benchmark_report.overall_retrieval_metrics.items():
            print(f"  {metric}: {value:.4f}")
        print("nOverall Generation Metrics:")
        for metric, value in self.benchmark_report.overall_generation_metrics.items():
            print(f"  {metric}: {value:.4f}")
        print("-------------------------n")

    def check_thresholds(self, thresholds: Dict[str, float]) -> bool:
        """
        检查整体指标是否达到预设阈值。
        """
        if not self.benchmark_report:
            print("No benchmark report available to check thresholds.")
            return False

        all_passed = True
        self.benchmark_report.passed_thresholds = {}

        print("n--- Threshold Check ---")
        for metric, threshold in thresholds.items():
            current_value = None
            if metric in self.benchmark_report.overall_retrieval_metrics:
                current_value = self.benchmark_report.overall_retrieval_metrics[metric]
            elif metric in self.benchmark_report.overall_generation_metrics:
                current_value = self.benchmark_report.overall_generation_metrics[metric]

            if current_value is not None:
                passed = current_value >= threshold
                self.benchmark_report.passed_thresholds[metric] = passed
                print(f"  {metric}: Current={current_value:.4f}, Threshold={threshold:.4f} -> {'PASS' if passed else 'FAIL'}")
                if not passed:
                    all_passed = False
            else:
                print(f"  Warning: Metric '{metric}' not found in report.")

        print(f"Overall Threshold Check: {'PASS' if all_passed else 'FAIL'}")
        return all_passed

# 主执行逻辑
if __name__ == "__main__":
    load_dotenv()
    openai_api_key = os.getenv("OPENAI_API_KEY")
    if not openai_api_key:
        raise ValueError("OPENAI_API_KEY not set in environment variables.")

    pipeline = AutomatedBenchmarkingPipeline(openai_api_key=openai_api_key)

    # 运行基准测试
    asyncio.run(pipeline.run_benchmark(test_cases_file="test_cases.json"))

    # 定义通过阈值
    performance_thresholds = {
        "precision_at_k": 0.8,
        "recall_at_k": 0.7,
        "mrr": 0.85,
        "faithfulness": 4.0, # RAGAS 默认 1-5 分
        "answer_relevance": 4.0,
        "context_recall": 0.7,
        "context_precision": 0.8
    }

    # 检查阈值
    pipeline.check_thresholds(performance_thresholds)

7. CI/CD集成示例 (GitHub Actions)

将上述Python脚本集成到CI/CD流程中，可以确保每次代码提交或模型更新后自动运行基准测试。

# .github/workflows/benchmark.yml
name: RAG Automated Benchmark

on:
  push:
    branches:
      - main
      - develop
  pull_request:
    branches:
      - main
      - develop
  workflow_dispatch: # 允许手动触发

jobs:
  run_benchmark:
    runs-on: ubuntu-latest

    env:
      OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }} # 从 GitHub Secrets 获取 API Key

    steps:
    - name: Checkout code
      uses: actions/checkout@v3

    - name: Set up Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.9' # 或您项目使用的版本

    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install -r requirements.txt # 确保所有依赖都列在 requirements.txt 中
        # 或者直接安装：
        # pip install pandas pydantic langchain openai chromadb transformers scikit-learn numpy ragas pytest python-dotenv

    - name: Run automated benchmark
      id: benchmark_run
      run: |
        python benchmark_pipeline.py

    - name: Check benchmark thresholds
      run: |
        python -c "
import json
import os
from test_models import BenchmarkReport

report_path = 'benchmark_reports/' # 假设报告生成在此目录
report_files = [f for f in os.listdir(report_path) if f.endswith('.json')]
if not report_files:
    print('No benchmark report found.')
    exit(1)

# 找到最新的报告
latest_report_file = max(report_files, key=lambda f: os.path.getmtime(os.path.join(report_path, f)))
with open(os.path.join(report_path, latest_report_file), 'r', encoding='utf-8') as f:
    report_data = json.load(f)

report = BenchmarkReport(**report_data)

performance_thresholds = {
    'precision_at_k': 0.8,
    'recall_at_k': 0.7,
    'mrr': 0.85,
    'faithfulness': 4.0,
    'answer_relevance': 4.0,
    'context_recall': 0.7,
    'context_precision': 0.8
}

all_passed = True
for metric, threshold in performance_thresholds.items():
    current_value = None
    if report.overall_retrieval_metrics and metric in report.overall_retrieval_metrics:
        current_value = report.overall_retrieval_metrics[metric]
    elif report.overall_generation_metrics and metric in report.overall_generation_metrics:
        current_value = report.overall_generation_metrics[metric]

    if current_value is not None:
        if current_value < threshold:
            print(f'::error::Metric {metric} failed: {current_value:.4f} < {threshold:.4f}')
            all_passed = False
        else:
            print(f'Metric {metric} passed: {current_value:.4f} >= {threshold:.4f}')
    else:
        print(f'::warning::Metric {metric} not found in report.')

if not all_passed:
    print('::error::Benchmark failed: One or more metrics fell below thresholds.')
    exit(1)
else:
    print('Benchmark passed all thresholds.')
"

在上述CI/CD配置中：

• 我们从GitHub Secrets中获取OPENAI_API_KEY，这是安全的做法。
• run_automated_benchmark步骤执行我们的Python脚本，生成报告。
• check_benchmark_thresholds步骤是一个Python单行脚本，它读取最新生成的报告，并根据预设阈值进行检查。如果任何指标未达标，它会使用::error::前缀输出错误信息，这会使GitHub Actions任务失败，从而阻止不合格的代码合并或部署。

进阶考量与未来展望

1. 大规模测试数据管理： 对于海量的测试用例，需要更专业的工具如数据湖、专用测试数据集管理系统，并结合版本控制（如DVC）对测试数据进行管理。
2. 更复杂的评估指标：
   • 毒性/偏见检测： 使用专门的模型或规则引擎检测生成内容中的有害或偏见信息。
   • 安全性： 评估对抗性攻击（如越狱）的鲁棒性。
   • 效率： 除了质量，推理时间、成本也是重要指标。
3. 人类反馈（Human-in-the-Loop）： 对于特别困难或边缘的测试用例，自动化评估可能不足。引入少量人工标注可以作为“黄金标准”来校准LLM-as-a-Judge，或处理模型难以判断的复杂场景。
4. A/B测试与渐进式发布： 将基准测试与在线A/B测试结合，确保新版本在真实用户流量下的表现。
5. 可视化与趋势分析： 利用MLflow、Weights & Biases、Grafana等工具，将基准测试结果可视化，并跟踪指标随时间变化的趋势，这对于识别长期退化和理解模型行为至关重要。
6. 多模态RAG的挑战： 当RAG系统处理图像、音频等多模态信息时，评估将变得更加复杂，需要新的评估维度和工具。
7. 自适应测试： 根据系统最近的表现，动态调整测试用例的生成策略，例如，针对经常失败的区域生成更多测试用例。

结语

自动化基准测试是构建健壮、可靠、高性能的RAG及其他生成式AI系统的基石。通过建立一套从检索召回率到生成信实度的自动化回归测试流水线，我们不仅能够显著提升开发效率，降低运营成本，更能确保我们的AI系统在快速变化的环境中持续提供高质量、高可靠度的服务。这使得我们能够信心十足地迭代创新，将生成式AI的巨大潜力真正转化为生产力。

希望今天的分享能为大家提供一些有益的思路和实践指导。谢谢大家！