面向大语言模型应用的数据驱动评估

概述

continuous-eval 是一个开源工具包，专为基于大语言模型的应用程序提供数据驱动的评估而设计。

continuous-eval 有何不同？

• 模块化评估：使用量身定制的指标来衡量流水线中的每个模块。

• 全面的指标库：涵盖检索增强生成（RAG）、代码生成、智能体工具使用、分类等多种大语言模型应用场景。支持混合使用确定性、语义和基于大语言模型的指标。

• 概率性评估：通过概率性指标对您的流水线进行评估。

快速入门

该代码以 PyPI 包的形式提供。要安装它，请运行以下命令：

python3 -m pip install continuous-eval

如果您想从源码安装：

git clone https://github.com/relari-ai/continuous-eval.git && cd continuous-eval
poetry install --all-extras

要运行基于大语言模型的指标，代码需要在 .env 文件中至少配置一个大语言模型的 API 密钥。请参阅示例 .env.example 文件。

运行单个指标

以下是针对单个数据点运行单个指标的方法。所有可用指标请参见：链接

from continuous_eval.metrics.retrieval import PrecisionRecallF1

datum = {
    "question": "法国的首都是什么？",
    "retrieved_context": [
        "巴黎是法国的首都和最大城市。",
        "里昂是法国的主要城市。",
    ],
    "ground_truth_context": ["巴黎是法国的首都。"],
    "answer": "巴黎",
    "ground_truths": ["巴黎"],
}

metric = PrecisionRecallF1()

print(metric(**datum))

运行评估

如果您想对一个数据集进行评估，可以使用 EvaluationRunner 类。

from time import perf_counter

from continuous_eval.data_downloader import example_data_downloader
from continuous_eval.eval import EvaluationRunner, SingleModulePipeline
from continuous_eval.eval.tests import GreaterOrEqualThan
from continuous_eval.metrics.retrieval import (
    PrecisionRecallF1,
    RankedRetrievalMetrics,
)


def main():
    # 下载检索数据集示例
    dataset = example_data_downloader("retrieval")

    # 设置评估流水线（即数据集、指标和测试）
    pipeline = SingleModulePipeline(
        dataset=dataset,
        eval=[
            PrecisionRecallF1().use(
                retrieved_context=dataset.retrieved_contexts,
                ground_truth_context=dataset.ground_truth_contexts,
            ),
            RankedRetrievalMetrics().use(
                retrieved_context=dataset.retrieved_contexts,
                ground_truth_context=dataset.ground_truth_contexts,
            ),
        ],
        tests=[
            GreaterOrEqualThan(
                test_name="召回率", metric_name="context_recall", min_value=0.8
            ),
        ],
    )

    # 启动评估管理器并运行指标（及测试）
    tic = perf_counter()
    runner = EvaluationRunner(pipeline)
    eval_results = runner.evaluate()
    toc = perf_counter()
    print("评估结果：")
    print(eval_results.aggregate())
    print(f"耗时：{toc - tic:.2f}秒\n")

    print("正在运行测试...")
    test_results = runner.test(eval_results)
    print(test_results)


if __name__ == "__main__":
    # 为了避免多进程问题，务必在新进程中运行此脚本
    main()

在流水线上运行评估（模块化评估）

有时，系统由多个模块组成，每个模块都有自己的指标和测试。 Continuous-eval 通过允许你在流水线中定义模块并选择相应的指标来支持这种用例。

from typing import Any, Dict, List

from continuous_eval.data_downloader import example_data_downloader
from continuous_eval.eval import (
    Dataset,
    EvaluationRunner,
    Module,
    ModuleOutput,
    Pipeline,
)
from continuous_eval.eval.result_types import PipelineResults
from continuous_eval.metrics.generation.text import AnswerCorrectness
from continuous_eval.metrics.retrieval import PrecisionRecallF1, RankedRetrievalMetrics


def page_content(docs: List[Dict[str, Any]]) -> List[str]:
    # 从流水线结果中提取检索到的文档内容
    return [doc["page_content"] for doc in docs]


def main():
    dataset: Dataset = example_data_downloader("graham_essays/small/dataset")
    results: Dict = example_data_downloader("graham_essays/small/results")

    # 简单的三步 RAG 流水线：检索器->重排序器->生成
    retriever = Module(
        name="retriever",
        input=dataset.question,
        output=List[str],
        eval=[
            PrecisionRecallF1().use(
                retrieved_context=ModuleOutput(page_content),  # 指定如何提取所需内容（即 page_content）
                ground_truth_context=dataset.ground_truth_context,
            ),
        ],
    )

    reranker = Module(
        name="reranker",
        input=retriever,
        output=List[Dict[str, str]],
        eval=[
            RankedRetrievalMetrics().use(
                retrieved_context=ModuleOutput(page_content),
                ground_truth_context=dataset.ground_truth_context,
            ),
        ],
    )

    llm = Module(
        name="llm",
        input=reranker,
        output=str,
        eval=[
            AnswerCorrectness().use(
                question=dataset.question,
                answer=ModuleOutput(),
                ground_truth_answers=dataset.ground_truth_answers,
            ),
        ],
    )

    pipeline = Pipeline([retriever, reranker, llm], dataset=dataset)
    print(pipeline.graph_repr())  # 以 Marmaid 格式可视化流水线

    runner = EvaluationRunner(pipeline)
    eval_results = runner.evaluate(PipelineResults.from_dict(results))
    print(eval_results.aggregate())


if __name__ == "__main__":
    main()

注意：将代码包裹在 main 函数中（使用 if __name__ == "__main__": 保护）非常重要，以确保并行化正常工作。

自定义指标

创建自定义指标有多种方法，请参阅文档中的自定义指标部分。

最简单的方法是利用 CustomMetric 类来创建一个“LLM 作为裁判”的自定义指标。

from continuous_eval.metrics.base.metric import Arg, Field
from continuous_eval.metrics.custom import CustomMetric
from typing import List

criteria = "检查生成的答案是否包含 PII 或其他敏感信息。"
rubric = """请根据答案的简洁性，使用以下评分标准对答案进行打分：
- 是：答案包含 PII 或其他敏感信息。
- 否：答案不包含 PII 或其他敏感信息。
"""

metric = CustomMetric(
    name="PIICheck",
    criteria=criteria,
    rubric=rubric,
    arguments={"answer": Arg(type=str, description="待评估的答案。")},
    response_format={
        "reasoning": Field(
            type=str,
            description="给出该答案评分的理由",
        ),
        "score": Field(
            type=str, description="答案的评分：是或否"
        ),
        "identifies": Field(
            type=List[str],
            description="在答案中识别出的 PII 或其他敏感信息",
        ),
    },
)

# 计算第一个数据点的指标
print(metric(answer="John Doe 居住在斯普林菲尔德市主街 123 号。"))

💡 贡献

有兴趣参与贡献吗？请查看我们的贡献指南，了解更多详情。

资源

• 文档：链接
• 示例仓库：端到端示例仓库
• 博客文章：
  • RAG 流水线评估实用指南：第 1 部分：检索、第 2 部分：生成
  • 黄金数据集对 LLM 评估有多重要？ (链接)
  • 如何评估复杂的 GenAI 应用程序：一种细粒度的方法 (链接)
  • 如何充分利用 LLM 生产数据：模拟用户反馈 (链接)
  • 生成合成数据以测试 LLM 应用程序 (链接)
• Discord：加入我们的 LLM 开发者社区 Discord
• 联系创始人：电子邮件 或 预约聊天

许可证

本项目采用 Apache 2.0 许可证——详情请参阅 LICENSE 文件。

开放式分析

我们监控基本的匿名使用统计数据，以了解用户的偏好、为新功能提供参考，并识别可能需要改进的领域。 你可以在遥测代码中查看我们具体跟踪的内容。

要禁用使用情况跟踪，只需将 CONTINUOUS_EVAL_DO_NOT_TRACK 标志设置为 true。

continuous-eval 快速上手指南

continuous-eval 是一个专为 LLM 驱动应用设计的开源数据驱动评估工具。它支持模块化评估、丰富的指标库（涵盖 RAG、代码生成、Agent 等场景）以及概率性评估。

环境准备

• 系统要求：Python 3.8+
• 前置依赖：
  • 若需运行基于 LLM 的评估指标，需准备至少一个 LLM API Key（如 OpenAI, Anthropic 等）。
  • 请在项目根目录创建 .env 文件并配置密钥，参考示例文件 .env.example。

安装步骤

方式一：通过 PyPI 安装（推荐）

python3 -m pip install continuous-eval

国内加速建议：如遇下载缓慢，可使用清华或阿里镜像源：

python3 -m pip install continuous-eval -i https://pypi.tuna.tsinghua.edu.cn/simple

方式二：从源码安装

git clone https://github.com/relari-ai/continuous-eval.git && cd continuous-eval
poetry install --all-extras

基本使用

1. 运行单个指标

以下示例演示如何对单条数据运行 PrecisionRecallF1 检索指标：

from continuous_eval.metrics.retrieval import PrecisionRecallF1

datum = {
    "question": "What is the capital of France?",
    "retrieved_context": [
        "Paris is the capital of France and its largest city.",
        "Lyon is a major city in France.",
    ],
    "ground_truth_context": ["Paris is the capital of France."],
    "answer": "Paris",
    "ground_truths": ["Paris"],
}

metric = PrecisionRecallF1()

print(metric(**datum))

2. 运行数据集评估

使用 EvaluationRunner 对整个数据集进行评估，并运行预设测试（如召回率是否达标）：

from time import perf_counter

from continuous_eval.data_downloader import example_data_downloader
from continuous_eval.eval import EvaluationRunner, SingleModulePipeline
from continuous_eval.eval.tests import GreaterOrEqualThan
from continuous_eval.metrics.retrieval import (
    PrecisionRecallF1,
    RankedRetrievalMetrics,
)


def main():
    # 下载示例检索数据集
    dataset = example_data_downloader("retrieval")

    # 配置评估流水线（数据集、指标和测试条件）
    pipeline = SingleModulePipeline(
        dataset=dataset,
        eval=[
            PrecisionRecallF1().use(
                retrieved_context=dataset.retrieved_contexts,
                ground_truth_context=dataset.ground_truth_contexts,
            ),
            RankedRetrievalMetrics().use(
                retrieved_context=dataset.retrieved_contexts,
                ground_truth_context=dataset.ground_truth_contexts,
            ),
        ],
        tests=[
            GreaterOrEqualThan(
                test_name="Recall", metric_name="context_recall", min_value=0.8
            ),
        ],
    )

    # 启动评估管理器并运行
    tic = perf_counter()
    runner = EvaluationRunner(pipeline)
    eval_results = runner.evaluate()
    toc = perf_counter()
    print("Evaluation results:")
    print(eval_results.aggregate())
    print(f"Elapsed time: {toc - tic:.2f} seconds\n")

    print("Running tests...")
    test_results = runner.test(eval_results)
    print(test_results)


if __name__ == "__main__":
    # 必须使用 main 函数包裹以确保多进程正常工作
    main()

3. 自定义指标（LLM-as-a-Judge）

通过 CustomMetric 快速创建基于 LLM 的自定义评估逻辑，例如检查回答中是否包含敏感信息（PII）：

from continuous_eval.metrics.base.metric import Arg, Field
from continuous_eval.metrics.custom import CustomMetric
from typing import List

criteria = "Check that the generated answer does not contain PII or other sensitive information."
rubric = """Use the following rubric to assign a score to the answer based on its conciseness:
- Yes: The answer contains PII or other sensitive information.
- No: The answer does not contain PII or other sensitive information.
"""

metric = CustomMetric(
    name="PIICheck",
    criteria=criteria,
    rubric=rubric,
    arguments={"answer": Arg(type=str, description="The answer to evaluate.")},
    response_format={
        "reasoning": Field(
            type=str,
            description="The reasoning for the score given to the answer",
        ),
        "score": Field(
            type=str, description="The score of the answer: Yes or No"
        ),
        "identifies": Field(
            type=List[str],
            description="The PII or other sensitive information identified in the answer",
        ),
    },
)

# 计算第一条数据的指标
print(metric(answer="John Doe resides at 123 Main Street, Springfield."))

注意：涉及多进程评估时，请务必将执行代码包裹在 if __name__ == "__main__": 块中。