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.mdAPI 自动生成页主题源码见 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.pyCaseLifecycle与Case、Dataset、set_eval_attribute、increment_eval_metric一起作为包的顶层公共 API 导出是 pydantic-evals 评测体系中一等公民的扩展点。二、CaseLifecycle 核心 API 速览CaseLifecycle是泛型类签名如下省略了默认值细节完整实现见 lifecycle.pyfrom 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.pysetup()—— 在任务执行之前调用任务运行task(case.inputs)prepare_context()—— 在任务完成后、评估器运行之前调用可用来丰富 metrics / attributes评估器运行teardown()—— 在评估器全部完成后调用收到完整结果若运行被中断则为None。三、把生命周期钩子接入评测lifecycle 参数详解CaseLifecycle本身并不直接运行而是通过Dataset.evaluate()或Dataset.evaluate_sync()的lifecycle关键字参数注入。参数类型有三种形式见 dataset.pylifecycle: ( 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.pyfrom 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(namelifecycle_demo, cases[Case(nametest, inputshello)]) report dataset.evaluate_sync(lambda inputs: inputs.upper(), lifecycleEnrichMetrics) 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(fsetup:{self.case.name}) async def teardown(self, result): events.append(fteardown:{self.case.name}:{type(result).__name__ if result is not None else NoneType}) await example_dataset.evaluate(task, max_concurrency1, lifecycleTrackingLifecycle) assert events snapshot([setup:case1, teardown:case1:ReportCase, setup:case2, teardown:case2:ReportCase])这个快照同时验证了两件事setup 一定先于任务且teardown 一定在评估完成后执行即使使用max_concurrency1串行执行顺序也严格为 setup → 任务/评估 → teardown → 下一个案例。4.2 prepare_context()评估前的上下文增强prepare_context(ctx)在任务完成后、评估器运行前被调用。它的典型用途是从任务输出、span tree 或外部状态中派生额外指标和属性注入EvaluatorContext从而让评估器看到更丰富的信息。EvaluatorContext是一个kw_onlydataclassevaluators/context.py主要字段包括字段含义name案例名称inputs传给任务的输入metadata案例元数据可能为Noneexpected_output期望输出可能为Noneoutput任务的实际输出duration任务运行耗时秒metricsdict[str, int \| float]可在任务代码中通过increment_eval_metric()累加也可在钩子中直接修改attributesdict[str, Any]可在任务代码中通过set_eval_attribute()设置也可在钩子中直接修改span_treeproperty任务执行期间记录的 OpenTelemetry span 树含计时与自定义 span若未安装 opentelemetry 或使用了不兼容的 TracerProvider访问时会抛出SpanTreeRecordingErrorset_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.pyReportCaseFailure任务执行期间抛出了异常包含error_message、error_stacktrace、trace/span id 等reporting/init.pyNone运行在没有报告对象的情况下结束例如被取消。文档建议利用这个差异做条件化清理例如失败时保留资源便于现场排查成功时直接释放。测试 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_concurrency1, lifecycleTeardownTracker) 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_teardownsetup抛出RuntimeError(setup failed)后报告中出现ReportCaseFailure且teardown仍被调用result为ReportCaseFailureerror_message包含setup failedtest_lifecycle_teardown_exception_propagatesteardown抛出RuntimeError(teardown exploded)后dataset.evaluate(...)以ExceptionGroupunhandled 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_config123) await example_dataset.evaluate(task, lifecyclelifecycle)这正是测试 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(fcase {self.case.name} failed, keeping artifacts) await self._close_client() except Exception: # teardown 的异常会向上传播务必自行消化 pass async def main() - None: dataset Datasetstr, str, dict, Case(namefail, inputsboom, 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_concurrency4, # 案例并发执行各实例状态互不干扰 lifecyclepartial(ServiceLifecycle, base_urlhttp://localhost:8080), ) print(fpassed{len(report.cases)} failed{len(report.failures)}) asyncio.run(main())该示例综合演示了三个钩子、partial注入配置、按结果成败的差异化 teardown以及“teardown 异常自行处理”的防御性写法。八、与并发和 repeat 的组合注意点并发评估evaluate默认并发执行所有案例可用max_concurrency限制并发度None表示不限。由于生命周期实例按案例创建即使高并发下状态也互不干扰但如果你在钩子里共享了模块级或类级可变对象仍需要自行保证线程/任务安全。多轮重复repeat 1repeat会让每个案例运行多次并做聚合dataset.py。从调用链看每次运行都会被包装成独立的任务条目因此每次运行都会获得一个全新的生命周期实例setup/teardown也会随之执行多次符合“每轮评估独立准备与清理”的预期。同步/异步入口evaluate_sync只是evaluate的同步包装钩子始终以 async 方法定义两种入口下语义完全一致。九、源码导航与延伸阅读钩子类完整实现pydantic_evals/pydantic_evals/lifecycle.py生命周期参数定义与调用链_run_task_and_evaluatorspydantic_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.pypydantic-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 各司其职评测代码更易维护、更可复用。【免费下载链接】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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考