pydantic-evals 案例生命周期钩子（CaseLifecycle）实战：Setup、Context 准备与 Teardown 的完整使用指南-育师

pydantic-evals 案例生命周期钩子（CaseLifecycle）实战：Setup、Context 准备与 Teardown 的完整使用指南

【免费下载链接】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-evals（即当前仓库pydantic_evals子包）提供的CaseLifecycle案例生命周期钩子机制。在基于 Dataset 的评测流水线中，每个测试案例（Case）依次经历“任务执行 → 评估器打分 → 结果汇总”，而CaseLifecycle允许你在任务执行前（setup）、任务完成后评估器运行前（prepare_context）以及评估器运行结束后（teardown）插入自定义逻辑。读完本文，你将掌握如何为单个案例注入一次性资源、在评估上下文中追加自定义指标与属性、以及按结果成败做差异化清理，并理解底层调用链与异常传播语义，可直接在真实评测项目中落地。

关联文档为 docs/api/pydantic_evals/lifecycle.md（API 自动生成页，主题源码见 pydantic_evals/pydantic_evals/lifecycle.py）。

一、为什么需要案例生命周期钩子

评测一个“随机函数”（例如 LLM 调用）通常需要为每个案例准备外部依赖、在评估前后做环境清理。如果把这些逻辑散落在 task 或 evaluator 内部，会造成职责混乱：task 应只负责生产输出，evaluator 应只负责打分。CaseLifecycle把“围绕单个案例的准备与收尾”独立成一个可复用的钩子类，其设计目标包括：

按案例隔离状态：每个案例在评估期间都会创建一个独立的实例，实例之间互不共享字段，天然避免并发评估时的状态串扰；
三个明确的插入点：任务前（setup）、任务后评估前（prepare_context）、评估完成后（teardown）；
失败也保证清理：setup或prepare_context抛出的异常会被捕获并记录为ReportCaseFailure，但teardown依然会被调用，确保资源不泄漏；
与观测体系集成：钩子内可以读取EvaluatorContext中的 span tree、metrics、attributes，也可以借助self.case访问案例元数据。

从源码结构看（pydantic_evals/pydantic_evals/init.py），CaseLifecycle与Case、Dataset、set_eval_attribute、increment_eval_metric一起作为包的顶层公共 API 导出，是 pydantic-evals 评测体系中一等公民的扩展点。

二、CaseLifecycle 核心 API 速览

CaseLifecycle是泛型类，签名如下（省略了默认值细节，完整实现见 lifecycle.py）：

from pydantic_evals.lifecycle import CaseLifecycle class CaseLifecycle(Generic[InputsT, OutputT, MetadataT]): def __init__(self, case: Case[InputsT, OutputT, MetadataT]) -> None: ... @property def case(self) -> Case[InputsT, OutputT, MetadataT]: ... async def setup(self) -> None: ... async def prepare_context(self, ctx: EvaluatorContext[InputsT, OutputT, MetadataT]) -> EvaluatorContext[InputsT, OutputT, MetadataT]: ... async def teardown( self, result: ReportCase[InputsT, OutputT, MetadataT] | ReportCaseFailure[InputsT, OutputT, MetadataT] | None, ) -> None: ...

关键设计点：

三个泛型参数InputsT、OutputT、MetadataT分别对应案例的输入、任务输出与元数据类型，默认均为Any，因此也可以直接写CaseLifecycle作为无约束版本（见测试 test_lifecycle_with_object_types）。
所有钩子默认是空操作（no-op）：你需要继承并覆盖需要的方法，其余方法保持默认即可。
self.case：在__init__中注入当前案例，钩子内部可通过self.case.metadata、self.case.name、self.case.inputs等访问案例信息。
__repr__已实现为TypeName(case=...)格式，便于调试输出（测试test_lifecycle_per_case_state中的assert 'StatefulLifecycle(case=' in repr(self)即验证了这一点）。

每个案例的执行流程

模块文档给出了单个案例的完整评估顺序（lifecycle.py）：

setup()—— 在任务执行之前调用；
任务运行（task(case.inputs)）；
prepare_context()—— 在任务完成后、评估器运行之前调用，可用来丰富 metrics / attributes；
评估器运行；
teardown()—— 在评估器全部完成后调用，收到完整结果（若运行被中断则为None）。

三、把生命周期钩子接入评测：lifecycle 参数详解

CaseLifecycle本身并不直接运行，而是通过Dataset.evaluate()或Dataset.evaluate_sync()的lifecycle关键字参数注入。参数类型有三种形式（见 dataset.py）：

lifecycle: ( type[CaseLifecycle[InputsT, OutputT, MetadataT]] | Callable[[Case[InputsT, OutputT, MetadataT]], CaseLifecycle[InputsT, OutputT, MetadataT]] | None ) = None

也就是说，你可以传：

一个 CaseLifecycle 子类：框架会为每个案例自动实例化（lifecycle(case)）；
一个可调用对象（如functools.partial）：框架会用它构造每个案例的实例，便于给钩子注入额外配置；
None（默认）：不启用任何钩子。

evaluate是异步入口，evaluate_sync是其同步包装（内部通过run_until_complete调用evaluate），两者都接受相同的lifecycle参数（dataset.py）。

最小可用示例

来自模块文档的官方示例（lifecycle.py）：

from pydantic_evals import Case, Dataset from pydantic_evals.evaluators.context import EvaluatorContext from pydantic_evals.lifecycle import CaseLifecycle class EnrichMetrics(CaseLifecycle): async def prepare_context(self, ctx: EvaluatorContext) -> EvaluatorContext: ctx.metrics['custom_metric'] = 42 return ctx dataset = Dataset(name='lifecycle_demo', cases=[Case(name='test', inputs='hello')]) report = dataset.evaluate_sync(lambda inputs: inputs.upper(), lifecycle=EnrichMetrics) print(report.cases[0].metrics['custom_metric']) #> 42

注意prepare_context的返回值语义：它接收EvaluatorContext，返回（可能被修改的）EvaluatorContext，这个返回值会被传给后续的评估器。

四、三个钩子的实战语义

4.1 setup()：任务执行前的资源准备

setup()在任务执行前调用，官方文档建议用它做每案例级别的资源准备，例如创建测试数据库、启动临时服务等。案例元数据可通过self.case.metadata访问，这样不同案例可以用自己的 metadata 决定初始化参数。

测试 test_lifecycle_setup_and_teardown 用事件列表验证了顺序：

class TrackingLifecycle(CaseLifecycle[TaskInput, TaskOutput, TaskMetadata]): async def setup(self) -> None: events.append(f'setup:{self.case.name}') async def teardown(self, result): events.append(f'teardown:{self.case.name}:{type(result).__name__ if result is not None else "NoneType"}') await example_dataset.evaluate(task, max_concurrency=1, lifecycle=TrackingLifecycle) assert events == snapshot(['setup:case1', 'teardown:case1:ReportCase', 'setup:case2', 'teardown:case2:ReportCase'])

这个快照同时验证了两件事：setup 一定先于任务，且teardown 一定在评估完成后执行（即使使用max_concurrency=1串行执行，顺序也严格为 setup → 任务/评估 → teardown → 下一个案例）。

4.2 prepare_context()：评估前的上下文增强

prepare_context(ctx)在任务完成后、评估器运行前被调用。它的典型用途是从任务输出、span tree 或外部状态中派生额外指标和属性，注入EvaluatorContext，从而让评估器看到更丰富的信息。

EvaluatorContext是一个kw_onlydataclass（evaluators/context.py），主要字段包括：

字段	含义
`name`	案例名称
`inputs`	传给任务的输入
`metadata`	案例元数据（可能为`None`）
`expected_output`	期望输出（可能为`None`）
`output`	任务的实际输出
`duration`	任务运行耗时（秒）
`metrics`	`dict[str, int \\| float]`，可在任务代码中通过`increment_eval_metric()`累加，也可在钩子中直接修改
`attributes`	`dict[str, Any]`，可在任务代码中通过`set_eval_attribute()`设置，也可在钩子中直接修改
`span_tree`（property）	任务执行期间记录的 OpenTelemetry span 树，含计时与自定义 span；若未安装 opentelemetry 或使用了不兼容的 TracerProvider，访问时会抛出`SpanTreeRecordingError`

set_eval_attribute与increment_eval_metric定义在 dataset.py，通过_task_run.CURRENT_TASK_RUN这个 ContextVar 找到当前任务运行并写入累加器，钩子与任务代码共享同一份数据。

官方文档的prepare_context示例同时展示了基于案例输入计算指标的用法：

class EnrichMetrics(CaseLifecycle[TaskInput, TaskOutput, TaskMetadata]): async def prepare_context(self, ctx: EvaluatorContext) -> EvaluatorContext: ctx.metrics['custom_metric'] = 42 ctx.metrics['input_length'] = len(self.case.inputs.query) return ctx

对应的测试 test_lifecycle_prepare_context 断言每个案例的case.metrics['custom_metric'] == 42且'input_length' in case.metrics。

评估器确实能看见增强后的上下文：测试 test_lifecycle_evaluator_sees_enriched_context 在prepare_context里设置ctx.metrics['enriched'] = 1，然后让一个自定义Evaluator检查ctx.metrics.get('enriched') == 1，最终断言report.cases[0].assertions['CheckMetric'].value is True——证明钩子修改的上下文会原样流向评估器。

4.3 teardown()：评估完成后的差异化清理

teardown(result)在评估器全部完成后调用，result参数有三种可能：

ReportCase：评估成功，包含输出、指标、属性、分数、标签、断言、任务耗时、总耗时（含评估器执行时间）、trace/span id 等完整信息（reporting/init.py）；
ReportCaseFailure：任务执行期间抛出了异常，包含error_message、error_stacktrace、trace/span id 等（reporting/init.py）；
None：运行在没有报告对象的情况下结束，例如被取消。

文档建议利用这个差异做条件化清理：例如失败时保留资源便于现场排查，成功时直接释放。

测试 test_lifecycle_teardown_on_task_failure 展示了失败路径：

async def task(inputs: str) -> str: if inputs == 'fail': raise ValueError('boom') return inputs.upper() report = await dataset.evaluate(task, max_concurrency=1, lifecycle=TeardownTracker) assert len(report.cases) == 1 # 成功的案例 assert len(report.failures) == 1 # 失败的案例 assert len(teardown_results) == 2 # teardown 两个都执行了 assert result_types == {'ReportCase', 'ReportCaseFailure'}

即无论任务成败，teardown 都会被调用，且能通过result的类型区分成败。

五、异常语义：谁会被捕获、谁会向上传播

CaseLifecycle的异常处理语义是钩子设计中最容易被忽视、也最关键的部分，文档明确如下：

setup()或prepare_context()抛出的异常：会被捕获，并记录为一个ReportCaseFailure（异常类型与消息会进入error_message），之后teardown()仍然会被调用，让你有机会做清理；
teardown()抛出的异常：会向上传播给调用者，可能导致整个评估运行中止。如果不想让 teardown 的异常搞崩评测，就应该在teardown()实现内部自行 try/except 处理。

底层实现在 dataset.py 的_run_task_and_evaluators中：setup/prepare_context位于try块内，异常会被except Exception捕获并构造ReportCaseFailure；而teardown位于finally块中，其异常故意不捕获，直接向调用方传播（源码注释明确说明了这一设计意图）。

对应的测试证据

test_lifecycle_setup_failure_produces_case_failure_and_calls_teardown：setup抛出RuntimeError('setup failed')后，报告中出现ReportCaseFailure，且teardown仍被调用，result为ReportCaseFailure，error_message包含'setup failed'；
test_lifecycle_teardown_exception_propagates：teardown抛出RuntimeError('teardown exploded')后，dataset.evaluate(...)以ExceptionGroup（'unhandled errors in a TaskGroup'）的形式向上抛出。

六、进阶模式：按案例隔离状态与可配置生命周期

6.1 每个案例独立实例，状态天然隔离

Dataset.evaluate中每个案例都会执行lc = lifecycle(case)实例化，因此钩子实例是按案例创建的。测试 test_lifecycle_per_case_state 验证了这一点：实例字段setup_called在setup中置真，随后prepare_context断言它已被调用，并在不同案例上计算出不同的case_name_length指标——证明状态不会跨案例泄漏，也证明setup一定先于prepare_context执行。

6.2 用 functools.partial 注入配置

由于lifecycle参数接受任意“接收 Case 返回 CaseLifecycle”的可调用对象，你可以用functools.partial给生命周期构造函数传递额外配置：

from functools import partial class ConfigurableLifecycle(CaseLifecycle[TaskInput, TaskOutput, TaskMetadata]): def __init__(self, case: Case[TaskInput, TaskOutput, TaskMetadata], my_config: int) -> None: super().__init__(case) self.my_config = my_config lifecycle = partial(ConfigurableLifecycle, my_config=123) await example_dataset.evaluate(task, lifecycle=lifecycle)

这正是测试 test_lifecycle_via_partial 所覆盖的用法，适用于需要为不同实验传入不同配置（如数据库连接、服务地址）的场景。

七、完整实战：一个带外部资源管理的评测例子

综合以上内容，一个典型的实战写法如下（结合文档示例与测试语义）：

import asyncio from functools import partial from pydantic_evals import Case, Dataset from pydantic_evals.evaluators.context import EvaluatorContext from pydantic_evals.lifecycle import CaseLifecycle class ServiceLifecycle(CaseLifecycle[str, str, dict]): """为每个案例准备并清理一个外部资源。""" def __init__(self, case: Case[str, str, dict], base_url: str) -> None: super().__init__(case) self.base_url = base_url self.client = None async def setup(self) -> None: # 每个案例独立初始化资源，metadata 可携带差异化配置 self.client = await self._create_client(self.base_url, **self.case.metadata or {}) async def prepare_context(self, ctx: EvaluatorContext) -> EvaluatorContext: # 从任务输出与外部状态派生指标 ctx.metrics['output_length'] = len(ctx.output) ctx.attributes['client_ready'] = self.client is not None return ctx async def teardown(self, result) -> None: try: if result is None or isinstance(result, __import__('pydantic_evals').ReportCaseFailure): # 失败时保留现场用于排查，这里仅记录 print(f'case {self.case.name} failed, keeping artifacts') await self._close_client() except Exception: # teardown 的异常会向上传播，务必自行消化 pass async def main() -> None: dataset = Datasetstr, str, dict, Case(name='fail', inputs='boom', metadata={'region': 'eu-west'}), ], ) async def task(inputs: str) -> str: if inputs == 'boom': raise RuntimeError('task failed') return inputs.upper() report = await dataset.evaluate( task, max_concurrency=4, # 案例并发执行，各实例状态互不干扰 lifecycle=partial(ServiceLifecycle, base_url='http://localhost:8080'), ) print(f'passed={len(report.cases)} failed={len(report.failures)}') asyncio.run(main())

该示例综合演示了三个钩子、partial注入配置、按结果成败的差异化 teardown，以及“teardown 异常自行处理”的防御性写法。

八、与并发和 repeat 的组合注意点

并发评估：evaluate默认并发执行所有案例，可用max_concurrency限制并发度（None表示不限）。由于生命周期实例按案例创建，即使高并发下状态也互不干扰；但如果你在钩子里共享了模块级或类级可变对象，仍需要自行保证线程/任务安全。
多轮重复（repeat > 1）：repeat会让每个案例运行多次并做聚合（dataset.py）。从调用链看，每次运行都会被包装成独立的任务条目，因此每次运行都会获得一个全新的生命周期实例，setup/teardown也会随之执行多次，符合“每轮评估独立准备与清理”的预期。
同步/异步入口：evaluate_sync只是evaluate的同步包装，钩子始终以 async 方法定义，两种入口下语义完全一致。

九、源码导航与延伸阅读

钩子类完整实现：pydantic_evals/pydantic_evals/lifecycle.py
生命周期参数定义与调用链（_run_task_and_evaluators）：pydantic_evals/pydantic_evals/dataset.py
评估上下文对象：EvaluatorContext见 pydantic_evals/pydantic_evals/evaluators/context.py
报告对象：ReportCase/ReportCaseFailure见 pydantic_evals/pydantic_evals/reporting/init.py
顶层导出（CaseLifecycle等）：pydantic_evals/pydantic_evals/init.py
生命周期测试套件（覆盖顺序、失败路径、状态隔离、partial 注入等全部语义）：tests/evals/test_dataset.py
pydantic-evals 在线评测与报告相关 API 文档：docs/api/pydantic_evals/online.md、docs/api/pydantic_evals/reporting.md；评测整体概念见 docs/evals/core-concepts.md

十、小结

CaseLifecycle为 pydantic-evals 的按案例评测提供了三个干净、可组合的插入点：setup负责前置资源准备，prepare_context负责在评估前丰富上下文（指标与属性），teardown负责无论成败都执行的收尾清理。其核心语义——每案例独立实例、失败也保证 teardown、teardown 异常向上传播——均由源码（dataset.py）与完整测试套件（tests/evals/test_dataset.py）双重背书。在构建真实 LLM 评测流水线时，把资源准备与清理收敛进CaseLifecycle，能让 task、evaluator 各司其职，评测代码更易维护、更可复用。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考