企业官网建设流程全解析

Pydantic Evals 核心概念详解：Dataset、Case、Evaluator 与 EvaluationReport 的完整心智模型

【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai

本篇基于 pydantic-ai 仓库中的docs/evals/core-concepts.md与pydantic_evals源码，系统讲解 Pydantic Evals 的六大核心概念：Dataset（测试套件）、Case（测试场景）、Evaluator（评分器）、ReportEvaluator（实验级分析器）、Experiment（实验执行）与 EvaluationReport（评估报告）。读完后，你能够独立搭建一套类型安全、可序列化、可重复对比的 AI 系统评估流水线，并理解每次dataset.evaluate(task)调用背后从任务并发执行、逐用例评分到实验级聚合的完整调用链。

一、总览：定义、执行、结果三层分离

Pydantic Evals 围绕六个核心概念构建：

其中最关键的心智模型是三层分离：

• 定义（Definition）：Dataset及其Case、Evaluator、ReportEvaluator—— 描述"你想测什么"，与任务实现无关；
• 执行（Execution）：Experiment —— 把某个任务函数跑到这些测试上；
• 结果（Results）：EvaluationReport（逐用例结果 + 实验级分析）—— 描述"实验过程中发生了什么"。

这种分离正是后续"同一数据集跑多个任务版本做对比"能力的基础。

二、单元测试类比：快速建立心智模型

官方文档给出了一张与单元测试的对照表，这是理解整个框架最快的入口：

关键差异：AI 系统是概率性的，因此评估不再只是简单的通过/失败，而是可以有：

• 定量分数（0.0 到 1.0）；
• 定性标签（"good"、"acceptable"、"poor"）；
• 带解释性理由的通过/失败断言。

就像你可以多次对同一测试套件运行pytest，你也可以对同一数据集运行多次实验（multiple experiments）来比较不同实现或追踪时间变化——同一份 Dataset 定义可以反复复用。

三、Dataset：类型安全且可序列化的评估套件

3.1 基本定义

Dataset是一组测试用例和评估器的集合，定义了一个完整的评估套件：

from pydantic_evals import Case, Dataset from pydantic_evals.evaluators import IsInstance dataset = Dataset( name='my_eval_suite', cases=[ Case(inputs='test input', expected_output='test output'), ], evaluators=[ IsInstance(type_name='str'), ], )

从源码 Dataset 类定义 看，它有四个字段：name、cases、evaluators（应用于所有用例）和report_evaluators（对整份报告做实验级分析）。构造时会做一项重要校验：用例名不允许重复，出现重名会直接抛出ValueError: Duplicate case name: ...——因为用例名是报告展示与结果过滤的标识。

3.2 三大关键特性

1. 类型安全（Type-safe）：Dataset泛型化于InputsT、OutputT、MetadataT三个类型变量（见 dataset.py）。这意味着你可以写Dataset[MyInput, str, MyMeta]获得输入/输出/元数据的完整类型推导；
2. 可序列化（Serializable）：支持保存/加载为 YAML 或 JSON 文件（下一节详述）；
3. 可评估（Evaluable）：可以针对任何输入/输出类型匹配的任务函数运行。

3.3 保存与加载：YAML / JSON + JSON Schema

Dataset提供了to_file/from_file/from_text/from_dict等方法（见 dataset.py），要点如下：

• 格式推断：fmt参数可省略，会从文件扩展名推断（.yaml/.yml→ YAML，.json→ JSON），推断失败则要求显式传fmt；
• JSON Schema 自动导出：to_file默认按模板./{stem}_schema.json（源码常量DEFAULT_SCHEMA_PATH_TEMPLATE，dataset.py）写出对应的 JSON Schema，供 YAML 编辑器做校验与补全；YAML 文件头部还会写入# yaml-language-server: $schema=...行；
• 自定义评估器反序列化：from_file接受custom_evaluator_types/custom_report_evaluator_types参数，把自定义评估器类注册进加载器注册表——序列化时评估器以规格（spec）形式存储，加载时按注册表还原为真实对象。

这一机制使得"评估套件"可以像测试文件一样进入版本控制、被人工评审，甚至由 LLM 辅助生成。

3.4 Dataset 级 vs Case 级评估器

评估器可以定义在两个层级：

• Dataset 级：应用于数据集中的所有用例；
• Case 级：仅应用于特定用例。

from pydantic_evals import Case, Dataset from pydantic_evals.evaluators import EqualsExpected, IsInstance dataset = Dataset( name='case_level_evaluators', cases=[ Case( name='special_case', inputs='test', expected_output='TEST', evaluators=[ # 这个评估器只为该用例运行 EqualsExpected(), ], ), ], evaluators=[ # 这个评估器为所有用例运行 IsInstance(type_name='str'), ], )

运行时的合并逻辑在_run_task_and_evaluators中：evaluators = case.evaluators + dataset_evaluators，即 Case 级在前、Dataset 级在后，两者全部并发执行。此外Dataset还提供了两个便捷方法：

• add_case(...)：增量添加用例（同样校验重名，dataset.py）；
• add_evaluator(evaluator, specific_case=None)：不传specific_case时加入 Dataset 级，传入用例名则只挂到对应用例上，找不到该用例抛ValueError（dataset.py）。

四、Experiment：一次实验发生了什么

Experiment就是把任务函数执行于数据集中所有用例的过程，它是静态定义（Dataset）与结果（EvaluationReport）之间的桥梁。

4.1 启动实验

通过evaluate()（异步）或evaluate_sync()（同步封装）运行：

from pydantic_evals import Case, Dataset # 定义数据集（静态定义） dataset = Dataset( name='uppercase_experiment', cases=[ Case(inputs='hello', expected_output='HELLO'), Case(inputs='world', expected_output='WORLD'), ], ) # 定义任务 def uppercase_task(text: str) -> str: return text.upper() # 运行实验（执行） report = dataset.evaluate_sync(uppercase_task)

对照Dataset.evaluate签名，完整参数为：

4.2 执行流水线（源码级拆解）

文档描述了五步流程，对照源码可以逐层印证：

1. Setup：evaluate打开一个logfire_span('evaluate {name}')，携带task_name、dataset_name、n_cases等属性，并用anyio.Semaphore(max_concurrency)建立并发门（dataset.py）；
2. Execution：所有用例经task_group_gather并发分发，每个用例在_run_task中执行——在execute {task}span 下调用task(case.inputs)，用time.perf_counter计时（优先从 span 计算时长），同步函数通过to_thread.run_sync放到线程池以免阻塞事件循环；同时捕获该次执行的 OpenTelemetry span 子树（SpanTree）；
3. Case Evaluation：每个用例的评估器列表（case.evaluators + dataset_evaluators）同样并发执行，输出先经_group_evaluator_outputs_by_type按值类型分流为assertions（bool）、scores（int/float）、labels（str）三个字典，重名评估器会自动加_2、_3后缀去重；
4. Report Evaluation：若配置了report_evaluators，它们在全量结果上运行（_run_report_evaluators），产出的分析（混淆矩阵、PR 曲线、标量、表格）追加到report.analyses；单个 report evaluator 抛错不会中断实验，而是记入report.report_evaluator_failures；
5. Reporting：所有结果聚合为EvaluationReport，包含逐用例结果与实验级分析；trace_id/span_id从实验 span 的 context 中取出（32/16 位十六进制字符串）。

任务抛异常的用例不会被丢弃：异常被捕获后转为ReportCaseFailure（含error_message与完整error_stacktrace），归入report.failures，与成功用例分列展示——这保证了"跑挂的用例"在报告中显性可见，而不是静默消失。

4.3 一个数据集，多个实验

Pydantic Evals 的关键能力：同一数据集可以针对不同任务实现反复运行：

from pydantic_evals import Case, Dataset from pydantic_evals.evaluators import EqualsExpected dataset = Dataset( name='comparison_test', cases=[ Case(inputs='hello', expected_output='HELLO'), ], evaluators=[EqualsExpected()], ) # 原始实现 def task_v1(text: str) -> str: return text.upper() # 改进实现（加了感叹号） def task_v2(text: str) -> str: return text.upper() + '!' # 对比结果 report_v1 = dataset.evaluate_sync(task_v1) report_v2 = dataset.evaluate_sync(task_v2) avg_v1 = report_v1.averages() avg_v2 = report_v2.averages() print(f'V1 pass rate: {avg_v1.assertions if avg_v1 and avg_v1.assertions else 0}') #> V1 pass rate: 1.0 print(f'V2 pass rate: {avg_v2.assertions if avg_v2 and avg_v2.assertions else 0}') #> V2 pass rate: 0

report.averages()返回ReportCaseAggregate，其中assertions是所有用例断言的通过率（reporting/init.py）。这种模式支持：

• 对比实现：跨版本比较；
• 追踪表现：随时间观测指标漂移；
• A/B 测试：不同方法互验；
• 部署前验证：变更先过评估再上线。

五、Case：单个测试场景的四个组成部分

Case表示一个带特定输入与可选期望输出的测试场景（dataset.py）：

from pydantic_evals import Case from pydantic_evals.evaluators import EqualsExpected case = Case( name='test_uppercase', # 可选，但报告展示强烈建议提供 inputs='hello world', # 必填：传给任务的输入 expected_output='HELLO WORLD', # 可选：期望输出 metadata={'category': 'basic'}, # 可选：任意元数据 evaluators=[EqualsExpected()], # 可选：该用例专属评估器 )

5.1 Inputs（输入）

传给被评估任务的输入，可以是任意类型：

from pydantic import BaseModel from pydantic_evals import Case class MyInputModel(BaseModel): field1: str # 简单类型 Case(inputs='hello') Case(inputs=42) # 复杂类型 Case(inputs={'query': 'What is AI?', 'max_tokens': 100}) Case(inputs=MyInputModel(field1='value'))

5.2 Expected Output（期望输出）

期望结果，供EqualsExpected等评估器使用：

from pydantic_evals import Case Case( inputs='2 + 2', expected_output='4', )

注意None的语义："未提供"与"期望为 None"共用同一表示。若未提供expected_output，依赖它的评估器会跳过该用例——EqualsExpected.evaluate在源码中直接if ctx.expected_output is None: return {}（common.py），返回空字典即不产生任何断言。因此若要断言任务恰好返回None，应使用Equals(value=None)而不是expected_output=None。

5.3 Metadata（元数据）

任意数据，评估器可通过EvaluatorContext.metadata访问：

from pydantic_evals import Case Case( inputs='question', metadata={ 'difficulty': 'hard', 'category': 'math', 'source': 'exam_2024', }, )

元数据用途：分析时过滤用例、为评估器提供上下文、组织测试套件（例如按难度/来源做分组统计）。

5.4 用例专属 Evaluators

用例可以挂载只为自己运行的评估器。这对构建综合性评估套件特别有力：不同用例有不同的验收标准。官方文档的表述很精辟——如果你能写出一套对所有用例都完美适用的评估标准，那它本就该并入 agent 的 instructions；而用例级的LLMJudge评估器尤其适合为每个场景单独描述"好"长什么样，从而快速构建可维护的 golden dataset（详见 Case-specific evaluators）。

六、Evaluator：三类返回值与 EvaluatorContext

Evaluator对任务输出进行评估，返回一个或多个分数（score）、标签（label）或断言（assertion），每一项还可附带字符串形式的解释理由。

6.1 三种返回类型

自定义评估器只需继承Evaluator并实现evaluate(self, ctx: EvaluatorContext)（evaluator.py），且必须以@dataclass装饰（序列化注册表的校验逻辑会检查__dataclass_fields__，见 dataset.py）：

from dataclasses import dataclass from pydantic_evals.evaluators import Evaluator, EvaluatorContext @dataclass class ExactMatch(Evaluator): def evaluate(self, ctx: EvaluatorContext) -> bool: return ctx.output == ctx.expected_output # 断言 @dataclass class Confidence(Evaluator): def evaluate(self, ctx: EvaluatorContext) -> float: # 分析输出并返回置信度分数 return 0.95 # 分数 @dataclass class Classifier(Evaluator): def evaluate(self, ctx: EvaluatorContext) -> str: if 'error' in ctx.output.lower(): return 'error' # 标签 return 'success'

评估器还可以返回EvaluationReason实例，或"标签 → 值"的字典以一次产出多项结果。内置评估器一览（导出自 evaluators/init.py）：

• 通用：Equals、EqualsExpected、Contains、IsInstance、MaxDuration、LLMJudge、GEval、HasMatchingSpan；
• Agent 轨迹：ToolCorrectness、TrajectoryMatch、TrajectoryOrder、ArgumentCorrectness、MaxToolCalls、MaxModelRequests；
• 实验级（ReportEvaluator）：ConfusionMatrixEvaluator、PrecisionRecallEvaluator、ROCAUCEvaluator、KolmogorovSmirnovEvaluator。

完整内置参考见 Native Evaluators。

6.2 EvaluatorContext：评估器的唯一输入

所有评估器都会收到一个EvaluatorContext实例，字段如下：

其中attributes与metrics可以在任务执行期间通过两个全局辅助函数动态写入（dataset.py）：

from pydantic_evals import set_eval_attribute, increment_eval_metric # 任务内部调用： set_eval_attribute('retries_used', 2) # 写入 attributes increment_eval_metric('tokens', 1520) # 累加 metrics

这让评估器不仅能看"输出对不对"，还能拿到"过程中花了多少 token、走了哪些 span"。

6.3 多结果返回与 EvaluationReason

返回字典 = 一次评估产出多项结果（每项再按 bool/数值/字符串归类）：

from dataclasses import dataclass from pydantic_evals.evaluators import Evaluator, EvaluatorContext @dataclass class MultiCheck(Evaluator): def evaluate(self, ctx: EvaluatorContext) -> dict[str, bool | float | str]: return { 'is_valid': isinstance(ctx.output, str), # 断言 'length': len(ctx.output), # 指标 'category': 'long' if len(ctx.output) > 100 else 'short', # 标签 }

用EvaluationReason附带解释：

from dataclasses import dataclass from pydantic_evals.evaluators import EvaluationReason, Evaluator, EvaluatorContext @dataclass class SmartCheck(Evaluator): def evaluate(self, ctx: EvaluatorContext) -> EvaluationReason: if ctx.output == ctx.expected_output: return EvaluationReason( value=True, reason='Exact match with expected output', ) return EvaluationReason( value=False, reason=f'Expected {ctx.expected_output!r}, got {ctx.output!r}', )

理由会在报告渲染时通过include_reasons=True参数显示出来——对 LLM 生成的失败原因做归因排查时非常有用。自定义返回类型的更多细节见 custom evaluator return types。

七、EvaluationReport：实验结果的完整结构

EvaluationReport是实验的产物，包含任务在数据集中所有用例上执行并跑完全部评估器后的全部数据：

from pydantic_evals import Case, Dataset from pydantic_evals.evaluators import EqualsExpected dataset = Dataset( name='report_example', cases=[Case(inputs='hello', expected_output='HELLO')], evaluators=[EqualsExpected()], ) def my_task(text: str) -> str: return text.upper() # 运行实验 report = dataset.evaluate_sync(my_task) # 打印到控制台 report.print() """ Evaluation Summary: my_task ┏━━━━━━━━━━┳━━━━━━━━━━━━┳━━━━━━━━━━┓ ┃ Case ID ┃ Assertions ┃ Duration ┃ ┡━━━━━━━━━━╇━━━━━━━━━━━━╇━━━━━━━━━━┩ │ Case 1 │ ✔ │ 10ms │ ├──────────┼────────────┼──────────┤ │ Averages │ 100.0% ✔ │ 10ms │ └──────────┴────────────┴──────────┘ """ # 以编程方式访问数据 for case in report.cases: print(f'{case.name}: {case.scores}') #> Case 1: {}

7.1 报告结构

对照EvaluationReport定义：

• name：实验名称；
• cases：成功用例的评估结果列表（ReportCase）；
• failures：执行失败的用例列表（ReportCaseFailure，含错误信息与堆栈）；
• analyses：实验级分析列表，来自 report evaluators（混淆矩阵、PR 曲线、标量、表格）；
• report_evaluator_failures：report evaluator 抛错记录；
• experiment_metadata：传给evaluate(metadata=...)的实验元数据，渲染时显示为 "Evaluation Summary" 面板；
• trace_id/span_id：OpenTelemetry 追踪标识（可选）。

7.2 ReportCase：单用例结果

每个成功用例结果（ReportCase）包含四类信息：

用例数据：name、inputs、metadata（可选）、expected_output（可选）、output。

评估结果：

• scores：评估器产出的数值分数字典；
• labels：评估器产出的分类标签字典；
• assertions：评估器产出的通过/失败断言字典。

性能数据：

• task_duration：任务执行耗时；
• total_duration：含评估器执行的总耗时（源码注释明确includes evaluator execution time，且repeat/teardown 之后会再校准一次）。

附加数据与追踪：metrics、attributes、trace_id（可选）、span_id（可选）；多轮实验下还有source_case_name作为分组键。

错误：evaluator_failures——评估器自身执行失败（而非任务失败）的记录列表。

7.3 聚合与渲染

• report.averages()：跨用例聚合出ReportCaseAggregate——分数/指标取均值、标签取分布占比、断言取通过率、耗时取平均；多轮实验（repeat > 1）下先按case_groups()逐组聚合再取组间平均；
• report.render(...)/report.print(...)：渲染参数多达二十余个，包括include_input、include_output、include_durations、include_errors、include_averages、include_reasons、各列的格式化配置（score_configs、metric_configs、duration_config）等；
• 基线对比：print/render都接受baseline参数，传入另一份报告即渲染出逐列 diff 表（旧值 → 新值，显著差异按 score/metric 语义染绿/染红），是"V1 vs V2"对比的现成工具；
• console_table()直接返回rich渲染对象，便于嵌入自己的 Console。

八、数据模型关系：一张图理清依赖

静态定义

• 一个Dataset包含：
  • 多个Case（带输入与期望输出的测试场景）
  • 多个Evaluator（给单个输出打分的逻辑）
  • 多个Report Evaluator（分析整组实验结果的逻辑）

执行（Experiment）

调用dataset.evaluate(task)时，一个Experiment被触发：

• Task函数在整个Dataset的所有Case上执行；
• 所有Evaluator（Dataset 级与 Case 级）按适用范围对每个输出运行；
• 最终产出一份EvaluationReport。

结果

• EvaluationReport包含：
  • 每个Case的结果（输入、输出、分数、断言、标签）
  • report evaluator 产出的实验级Analyses（混淆矩阵、PR 曲线、标量、表格）
  • 汇总统计（平均值、通过率）
  • 性能数据（耗时）
  • 追踪信息（OpenTelemetry span）

关键基数关系

• 一个 Dataset → 多个 Experiment：同一数据集可以针对不同任务实现运行，或重复运行以追踪变化；
• 一个 Experiment → 一份 Report：每次dataset.evaluate(...)调用产出一份报告；
• 一个 Experiment → 多个 Case 结果：报告覆盖数据集中每个用例的结果。

九、动手实践与延伸阅读

仓库自带的评估示例位于 examples/pydantic_ai_examples/evals/，覆盖数据集生成（example_01_generate_dataset.py）、自定义评估器（example_02_add_custom_evaluators.py）、单元测试式评估（example_03_unit_testing.py）与模型对比（example_04_compare_models.py），其中agent.py展示了如何把pydantic-ai的 Agent 作为 task 直接塞进dataset.evaluate，是本文概念到实战的最短路径。

继续深入建议按以下顺序阅读：

• Evaluators Overview—— 何时使用不同类型的评估器，以及用例专属评估器；
• Native Evaluators—— 内置评估器完整参考；
• Custom Evaluators—— 编写自己的评估逻辑与返回类型；
• Report Evaluators—— 实验级分析（混淆矩阵、PR 曲线等）；
• Dataset Management—— 数据集的保存、加载与生成。

【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai