尧图网页设计 
pydantic-ai 评估指南使用 Metrics、Attributes 与 Experiment Metadata 精细化追踪评估运行【免费下载链接】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 的 pydantic_evals 评估体系中默认的通过/失败判定往往不足以支撑复杂的质量分析。本指南讲解如何在任务执行期间记录自定义指标Metrics与属性Attributes如何在自定义评估器Evaluator中消费这些数据以及如何通过实验级**元数据Experiment Metadata**追踪整次评估运行的配置。读完本文你将掌握一套可复现、可对比、可定位问题的评估观测方案并能结合 Pydantic AI Agent 与 Logfire 实现自动化指标采集。概览三类可追踪的数据在执行评估任务时pydantic_evals 允许你在任务函数体内记录两类数据Metrics指标数值型数据int/float用于量化测量例如 API 调用次数、Token 消耗、耗时等Attributes属性任意类型数据用于记录定性信息例如使用的模型名、是否命中缓存、结构化配置等。这两类数据会出现在评估报告ReportCase中并可通过EvaluatorContext被评估器读取作为打分的依据。除此之外还有实验级元数据Experiment Metadata在调用Dataset.evaluate()时传入用于记录整次实验的配置信息。记录指标increment_eval_metric使用increment_eval_metric在任务执行期间累计数值。同名指标多次调用会累加而不是覆盖from dataclasses import dataclass from pydantic_evals.dataset import increment_eval_metric dataclass class APIResult: output: str usage: Usage dataclass class Usage: total_tokens: int def call_api(inputs: str) - APIResult: return APIResult(outputfResult: {inputs}, usageUsage(total_tokens100)) def my_task(inputs: str) - str: # 追踪 API 调用次数 increment_eval_metric(api_calls, 1) result call_api(inputs) # 追踪 Token 消耗 increment_eval_metric(tokens_used, result.usage.total_tokens) return result.output从源码实现看increment_eval_metric通过_task_run.CURRENT_TASK_RUN这个ContextVar获取当前任务运行的累加器再调用TaskRun.increment_metricpydantic_evals/_task_run.py完成累加。其中有一个值得注意的细节当当前值为 0 且累加结果仍为 0 时指标不会被创建。测试 tests/evals/test_dataset.py 中increment_eval_metric(phantom, 0)专门验证了这一点——零值指标不会出现在报告中避免无效数据污染。记录属性set_eval_attribute使用set_eval_attribute存储任意类型的数据。同名属性再次设置会覆盖旧值from pydantic_evals import set_eval_attribute def process(inputs: str) - str: return fProcessed: {inputs} def my_task(inputs: str) - str: # 记录使用了哪个模型 set_eval_attribute(model, gpt-5.2) # 记录功能开关状态 set_eval_attribute(used_cache, True) set_eval_attribute(retry_count, 2) # 记录结构化数据 set_eval_attribute(config, { temperature: 0.7, max_tokens: 100, }) return process(inputs)两个 API 都位于pydantic_evals.dataset模块顶层也可直接从pydantic_evals导入。底层实现基于ContextVar的上下文隔离机制pydantic_evals/_task_run.py因此并发场景下每个 case 的记录互不干扰——每个任务运行都有独立的TaskRun实例attributes与metrics两个字典。测试 tests/evals/test_online.py 验证了在异步装饰函数与同步装饰函数中调用这两个 API 都能正确传播到EvaluatorContext。在评估器中访问 Metrics 与 AttributesMetrics 和 Attributes 通过EvaluatorContext暴露给评估器分别对应ctx.metricsdict[str, int | float]与ctx.attributesdict[str, Any]from dataclasses import dataclass from pydantic_evals.evaluators import Evaluator, EvaluatorContext dataclass class EfficiencyChecker(Evaluator): max_api_calls: int 5 def evaluate(self, ctx: EvaluatorContext) - dict[str, bool]: # 读取指标 api_calls ctx.metrics.get(api_calls, 0) tokens_used ctx.metrics.get(tokens_used, 0) # 读取属性 used_cache ctx.attributes.get(used_cache, False) return { efficient_api_usage: api_calls self.max_api_calls, used_caching: used_cache, token_efficient: tokens_used 1000, }EvaluatorContext是评估器的唯一输入除metrics/attributes外还包含inputs、output、expected_output、metadatacase 级、duration以及span_tree任务执行的 OpenTelemetry 跨度树等字段。注意读取指标时建议使用.get(key, default)形式因为未记录的指标键在字典中不存在零值记录也不会被创建。在报告中查看评估完成后Metrics 与 Attributes 会随每个 case 出现在报告对象中from pydantic_evals import Case, Dataset def task(inputs: str) - str: return fResult: {inputs} dataset Dataset(namereport_viewing, cases[Case(inputstest)], evaluators[]) report dataset.evaluate_sync(task) for case in report.cases: print(f{case.name}:) # Case 1: print(f Metrics: {case.metrics}) # Metrics: {} print(f Attributes: {case.attributes}) # Attributes: {}在数据模型层面ReportCasepydantic_evals/reporting/init.py直接持有metrics: dict[str, float | int]与attributes: dict[str, Any]字段与EvaluatorContext中的对应字段共享同一来源——二者都来自任务执行期间TaskRun累积的数据。同时ReportCaseAggregate也会对指标做跨 case 的聚合统计。需要注意Metrics 与 Attributes 默认不会打印在控制台报告中需通过case.metrics/case.attributes编程访问或借助 Logfire 的可视化界面查看。自动指标Pydantic AI 与 Logfire 的集成当任务函数中使用 Pydantic AI Agent 且启用了 Logfire 时框架会从 OpenTelemetry 跨度树中自动提取一组标准指标import logfire from pydantic_ai import Agent logfire.configure(send_to_logfireif-token-present) agent Agent(openai:gpt-5.2) async def ai_task(inputs: str) - str: result await agent.run(inputs) return result.output # 自动追踪的指标 # - requests: LLM 调用次数 # - input_tokens: 输入 Token 总量 # - output_tokens: 输出 Token 总量 # - prompt_tokens: 提示词 Token如可用 # - completion_tokens: 补全 Token如可用 # - cost: 预估成本如使用 genai-prices这些自动指标可以在评估器中直接读取from dataclasses import dataclass from pydantic_evals.evaluators import Evaluator, EvaluatorContext dataclass class CostChecker(Evaluator): max_cost: float 0.01 # $0.01 def evaluate(self, ctx: EvaluatorContext) - bool: cost ctx.metrics.get(cost, 0.0) return cost self.max_cost自动提取的底层逻辑在 pydantic_evals/_task_run.py 的extract_span_tree_metrics函数中它遍历任务执行的 span 树识别带有gen_ai.request.model属性的节点从中提取gen_ai.operation.name chat的调用次数requests、operation.costcost以及所有gen_ai.usage.*前缀的用量属性input_tokens、output_tokens等。前提是安装了 Logfirepip install pydantic-evals[logfire]并配置了LOGFIRE_TOKEN环境变量详见 Logfire 集成指南。实战示例四种典型追踪场景API 调用与缓存命中追踪在带缓存的任务中区分缓存命中和真实 API 调用from dataclasses import dataclass from pydantic_evals import increment_eval_metric, set_eval_attribute from pydantic_evals.evaluators import Evaluator, EvaluatorContext def check_cache(inputs: str) - str | None: return None # 演示用无缓存命中 dataclass class APIResult: text: str usage: Usage dataclass class Usage: total_tokens: int async def call_api(inputs: str) - APIResult: return APIResult(textfResult: {inputs}, usageUsage(total_tokens100)) def save_to_cache(inputs: str, result: str) - None: pass # 保存到缓存 async def smart_task(inputs: str) - str: # 先尝试缓存 if cached : check_cache(inputs): set_eval_attribute(cache_hit, True) return cached set_eval_attribute(cache_hit, False) # 调用 API increment_eval_metric(api_calls, 1) result await call_api(inputs) increment_eval_metric(tokens, result.usage.total_tokens) # 缓存结果 save_to_cache(inputs, result.text) return result.text # 评估效率 dataclass class EfficiencyEvaluator(Evaluator): def evaluate(self, ctx: EvaluatorContext) - dict[str, bool | float]: api_calls ctx.metrics.get(api_calls, 0) cache_hit ctx.attributes.get(cache_hit, False) return { used_cache: cache_hit, made_api_call: api_calls 0, efficiency_score: 1.0 if cache_hit else 0.5, }Agent 工具调用追踪在 Pydantic AI Agent 的agent.tool装饰函数内记录指标与属性from dataclasses import dataclass from pydantic_ai import Agent, RunContext from pydantic_evals import increment_eval_metric, set_eval_attribute from pydantic_evals.evaluators import Evaluator, EvaluatorContext agent Agent(openai:gpt-5.2) def search(query: str) - str: return fSearch results for: {query} def call(endpoint: str) - str: return fAPI response from: {endpoint} agent.tool def search_database(ctx: RunContext, query: str) - str: increment_eval_metric(db_searches, 1) set_eval_attribute(last_query, query) return search(query) agent.tool def call_api(ctx: RunContext, endpoint: str) - str: increment_eval_metric(api_calls, 1) set_eval_attribute(last_endpoint, endpoint) return call(endpoint) # 评估工具使用情况 dataclass class ToolUsageEvaluator(Evaluator): def evaluate(self, ctx: EvaluatorContext) - dict[str, bool | int]: db_searches ctx.metrics.get(db_searches, 0) api_calls ctx.metrics.get(api_calls, 0) return { used_database: db_searches 0, used_api: api_calls 0, tool_call_count: db_searches api_calls, reasonable_tool_usage: (db_searches api_calls) 5, }性能追踪子操作耗时用time.perf_counter()测量任务内各子操作的耗时并记录为指标import time from dataclasses import dataclass from pydantic_evals import increment_eval_metric, set_eval_attribute from pydantic_evals.evaluators import Evaluator, EvaluatorContext async def retrieve_context(inputs: str) - list[str]: return [context1, context2] async def generate_response(context: list[str], inputs: str) - str: return fGenerated response for {inputs} async def monitored_task(inputs: str) - str: # 追踪子操作耗时 t0 time.perf_counter() context await retrieve_context(inputs) retrieve_time time.perf_counter() - t0 increment_eval_metric(retrieve_time, retrieve_time) t0 time.perf_counter() result await generate_response(context, inputs) generate_time time.perf_counter() - t0 increment_eval_metric(generate_time, generate_time) # 记录需要哪些操作 set_eval_attribute(needed_retrieval, len(context) 0) set_eval_attribute(context_chunks, len(context)) return result # 评估性能 dataclass class PerformanceEvaluator(Evaluator): max_retrieve_time: float 0.5 max_generate_time: float 2.0 def evaluate(self, ctx: EvaluatorContext) - dict[str, bool]: retrieve_time ctx.metrics.get(retrieve_time, 0.0) generate_time ctx.metrics.get(generate_time, 0.0) return { fast_retrieval: retrieve_time self.max_retrieve_time, fast_generation: generate_time self.max_generate_time, }质量追踪置信度与来源把 LLM 返回的置信度、引用来源等质量信号提取为属性供评估器计算综合评分from dataclasses import dataclass from pydantic_evals import set_eval_attribute from pydantic_evals.evaluators import Evaluator, EvaluatorContext async def llm_call(inputs: str) - dict: return {text: fResponse: {inputs}, confidence: 0.85, sources: [doc1, doc2]} async def quality_task(inputs: str) - str: result await llm_call(inputs) # 提取质量指标 confidence result.get(confidence, 0.0) sources_used result.get(sources, []) set_eval_attribute(confidence, confidence) set_eval_attribute(source_count, len(sources_used)) set_eval_attribute(sources, sources_used) return result[text] # 基于质量信号评估 dataclass class QualityEvaluator(Evaluator): min_confidence: float 0.7 def evaluate(self, ctx: EvaluatorContext) - dict[str, bool | float]: confidence ctx.attributes.get(confidence, 0.0) source_count ctx.attributes.get(source_count, 0) return { high_confidence: confidence self.min_confidence, used_sources: source_count 0, quality_score: confidence * (1.0 0.1 * source_count), }实验级元数据Experiment Metadata除 case 级数据外还可以在调用Dataset.evaluate()时传入实验级元数据metadata参数见 dataset.py记录整次评估运行的配置from pydantic_evals import Case, Dataset dataset Dataset( nameexperiment_metadata, cases[ Case( inputstest, metadata{difficulty: easy}, # Case 级元数据 ) ] ) async def task(inputs: str) - str: return fResult: {inputs} # 传入实验级元数据 async def main(): report await dataset.evaluate( task, metadata{ model: gpt-5.2, prompt_version: v2.1, temperature: 0.7, }, ) # 在报告中访问实验元数据 print(report.experiment_metadata) # {model: gpt-5.2, prompt_version: v2.1, temperature: 0.7}从源码看evaluate()会把metadata传入EvaluationReportreporting/init.py并同步设置到实验 span 的属性logfire.experiment.metadata中dataset.py同时报告评估器ReportEvaluatorContext也能读取到这份元数据。何时使用实验元数据实验元数据适合追踪适用于整个评估运行的配置信息模型配置模型名称、版本、参数提示词版本使用了哪个提示词模板基础设施部署环境、区域实验上下文开发者姓名、功能分支、提交哈希。在以下场景中尤其有价值跨时间比较多次评估运行追踪哪个配置产生了哪个结果依据历史数据复现评估结果。在报告中查看实验元数据会显示在打印报告report.render()的顶部from pydantic_evals import Case, Dataset dataset Dataset(namemetadata_report, cases[Case(inputshello, expected_outputHELLO)]) async def task(text: str) - str: return text.upper() async def main(): report await dataset.evaluate( task, metadata{model: gpt-5.2, version: v1.0}, ) print(report.render()) ╭─ Evaluation Summary: task ─╮ │ model: gpt-5.2 │ │ version: v1.0 │ ╰────────────────────────────╯ ┏━━━━━━━━━━┳━━━━━━━━━━┓ ┃ Case ID ┃ Duration ┃ ┡━━━━━━━━━━╇━━━━━━━━━━┩ │ Case 1 │ 10ms │ ├──────────┼──────────┤ │ Averages │ 10ms │ └──────────┴──────────┘ 在报告渲染层reporting/init.py实验元数据逐条显示为key: value行并且在进行基线baseline对比时会用/-标记新增或移除的配置项便于快速识别两次实验的配置差异。任务与实验元数据的同步单一事实来源实验元数据用于记录配置而不是配置任务。metadata字典不会自动改变任务的运行行为——你必须保证元数据中的值与任务实际使用的值一致。例如很容易出现元数据声称temperature: 0.7而任务实际使用temperature: 1.0的情况导致实验追踪错误、结果无法复现。为避免这一问题建议为配置建立单一事实来源single source of truth让任务与元数据共同引用它。下面给出几种推荐模式。模式一共享模块常量适用于简单场景用模块级常量统一管理from pydantic_ai import Agent from pydantic_evals import Case, Dataset # 模块常量作为单一事实来源 MODEL_NAME openai:gpt-5-mini TEMPERATURE 0.7 INSTRUCTIONS You are a helpful assistant. agent Agent(MODEL_NAME, model_settings{temperature: TEMPERATURE}, instructionsINSTRUCTIONS) async def task(inputs: str) - str: result await agent.run(inputs) return result.output async def main(): dataset Dataset(nameshared_constants, cases[Case(inputsWhat is the capital of France?)]) # 元数据引用同一组常量 await dataset.evaluate( task, metadata{ model: MODEL_NAME, temperature: TEMPERATURE, instructions: INSTRUCTIONS, }, )模式二配置对象推荐定义一次配置对象任务与元数据共用from dataclasses import asdict, dataclass from pydantic_ai import Agent from pydantic_evals import Case, Dataset dataclass class TaskConfig: 任务配置的单一事实来源。 包含所有希望出现在实验元数据中的变量。 model: str temperature: float max_tokens: int prompt_version: str # 只定义一次配置 config TaskConfig( modelopenai:gpt-5-mini, temperature0.7, max_tokens500, prompt_versionv2.1, ) # 在任务中使用配置 agent Agent( config.model, model_settings{temperature: config.temperature, max_tokens: config.max_tokens}, ) async def task(inputs: str) - str: 任务使用与元数据中记录相同的配置。 result await agent.run(inputs) return result.output # 用同一配置对象派生出元数据 async def main(): dataset Dataset(nameconfig_evaluation, cases[Case(inputsWhat is the capital of France?)]) report await dataset.evaluate( task, metadataasdict(config), # 保证与任务行为一致 ) print(report.experiment_metadata) { model: openai:gpt-5-mini, temperature: 0.7, max_tokens: 500, prompt_version: v2.1, } 如果全局配置对象不可行也可以在任务调用点创建TaskConfig实例并通过deps或类似机制传给 Agent但此时你仍需保证传给Dataset.evaluate的metadata值与任务实际使用的值始终一致。反模式重复配置务必避免以下常见错误——配置在多处重复定义极易失步from pydantic_ai import Agent from pydantic_evals import Case, Dataset # ❌ 错误配置在多处定义 agent Agent(openai:gpt-5-mini, model_settings{temperature: 0.7}) async def task(inputs: str) - str: result await agent.run(inputs) return result.output async def main(): dataset Dataset(nameanti_pattern, cases[Case(inputstest)]) # ❌ 错误手动输入元数据容易与任务失步 await dataset.evaluate( task, metadata{ model: openai:gpt-5-mini, # 重复定义可能与 Agent 定义不一致 temperature: 0.8, # ⚠️ 错误任务实际使用 0.7 }, )该反模式中元数据声称temperature: 0.8而任务实际使用0.7会导致实验追踪错误结果无法复现对比不同运行结果时产生困惑浪费大量时间排查为什么结果不同。Metrics vs Attributes vs Metadata差异速查特性MetricsAttributesCase MetadataExperiment Metadata设置位置任务执行期间任务执行期间Case 定义时evaluate()调用时类型int、float任意类型任意类型任意类型用途定量测量定性描述测试数据实验配置用于聚合统计上下文信息任务输入运行追踪可访问者评估器评估器任务与评估器仅报告作用域每个 case每个 case每个 case每次实验一个完整的区分示例from pydantic_evals import Case, Dataset, increment_eval_metric, set_eval_attribute # Case Metadata在 case 定义时设置执行前 case Case( inputsquestion, metadata{difficulty: hard, category: math}, # 每个 case 的元数据 ) dataset Dataset(namemetrics_demo, cases[case]) # Metrics Attributes在任务执行期间记录 async def task(inputs): # 这些是在执行期间为每个 case 记录的 increment_eval_metric(tokens, 100) set_eval_attribute(model, gpt-5.2) return fResult: {inputs} async def main(): # Experiment Metadata在评估调用时定义 await dataset.evaluate( task, metadata{ # 实验级元数据 prompt_version: v2.1, temperature: 0.7, }, )故障排查Metrics/attributes 没有出现请确认是在任务函数内部调用这两个 APIfrom pydantic_evals import increment_eval_metric def process(inputs: str) - str: return fProcessed: {inputs} # 错误在任务外调用 increment_eval_metric(count, 1) def bad_task(inputs): return process(inputs) # 正确在任务内调用 def good_task(inputs): increment_eval_metric(count, 1) return process(inputs)原理上increment_eval_metric/set_eval_attribute依赖CURRENT_TASK_RUNContextVar它只在run_task()上下文管理器pydantic_evals/_task_run.py内被设置在任务外调用时该 ContextVar 为None记录会被静默忽略。同理若调用发生在任务完成之后例如在评估器内也无法再追加记录。Metrics 没有累加检查是否误用了set_eval_attribute而不是increment_eval_metricfrom pydantic_evals import increment_eval_metric, set_eval_attribute # 错误这会覆盖而不是累加 set_eval_attribute(count, 1) set_eval_attribute(count, 1) # 仍是 1 # 正确累加 increment_eval_metric(count, 1) increment_eval_metric(count, 1) # 现在是 2Attributes 数据量太大存储摘要而非原始数据from pydantic_evals import set_eval_attribute giant_response_object {key str(i): value * 100 for i in range(1000)} # 错误存储巨型对象 set_eval_attribute(full_response, giant_response_object) # 正确存储摘要 set_eval_attribute(response_size_kb, len(str(giant_response_object)) / 1024) set_eval_attribute(response_keys, list(giant_response_object.keys())[:10]) # 前 10 个键大体积属性会拖慢报告序列化并污染日志建议只保留足以支撑评估判断的摘要信息。延伸阅读Case 生命周期钩子每个 case 的 setup、teardown 与上下文准备自定义评估器在评估器中使用 Metrics 与 AttributesLogfire 集成在 Logfire 中可视化查看 Metrics并发与性能优化优化评估运行性能。【免费下载链接】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），仅供参考 pydantic-ai 评估指南：使用 Metrics、Attributes 与 Experiment Metadata 精细化追踪评估运行
pydantic-ai 评估指南使用 Metrics、Attributes 与 Experiment Metadata 精细化追踪评估运行【免费下载链接】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 的 pydantic_evals 评估体系中默认的通过/失败判定往往不足以支撑复杂的质量分析。本指南讲解如何在任务执行期间记录自定义指标Metrics与属性Attributes如何在自定义评估器Evaluator中消费这些数据以及如何通过实验级**元数据Experiment Metadata**追踪整次评估运行的配置。读完本文你将掌握一套可复现、可对比、可定位问题的评估观测方案并能结合 Pydantic AI Agent 与 Logfire 实现自动化指标采集。概览三类可追踪的数据在执行评估任务时pydantic_evals 允许你在任务函数体内记录两类数据Metrics指标数值型数据int/float用于量化测量例如 API 调用次数、Token 消耗、耗时等Attributes属性任意类型数据用于记录定性信息例如使用的模型名、是否命中缓存、结构化配置等。这两类数据会出现在评估报告ReportCase中并可通过EvaluatorContext被评估器读取作为打分的依据。除此之外还有实验级元数据Experiment Metadata在调用Dataset.evaluate()时传入用于记录整次实验的配置信息。记录指标increment_eval_metric使用increment_eval_metric在任务执行期间累计数值。同名指标多次调用会累加而不是覆盖from dataclasses import dataclass from pydantic_evals.dataset import increment_eval_metric dataclass class APIResult: output: str usage: Usage dataclass class Usage: total_tokens: int def call_api(inputs: str) - APIResult: return APIResult(outputfResult: {inputs}, usageUsage(total_tokens100)) def my_task(inputs: str) - str: # 追踪 API 调用次数 increment_eval_metric(api_calls, 1) result call_api(inputs) # 追踪 Token 消耗 increment_eval_metric(tokens_used, result.usage.total_tokens) return result.output从源码实现看increment_eval_metric通过_task_run.CURRENT_TASK_RUN这个ContextVar获取当前任务运行的累加器再调用TaskRun.increment_metricpydantic_evals/_task_run.py完成累加。其中有一个值得注意的细节当当前值为 0 且累加结果仍为 0 时指标不会被创建。测试 tests/evals/test_dataset.py 中increment_eval_metric(phantom, 0)专门验证了这一点——零值指标不会出现在报告中避免无效数据污染。记录属性set_eval_attribute使用set_eval_attribute存储任意类型的数据。同名属性再次设置会覆盖旧值from pydantic_evals import set_eval_attribute def process(inputs: str) - str: return fProcessed: {inputs} def my_task(inputs: str) - str: # 记录使用了哪个模型 set_eval_attribute(model, gpt-5.2) # 记录功能开关状态 set_eval_attribute(used_cache, True) set_eval_attribute(retry_count, 2) # 记录结构化数据 set_eval_attribute(config, { temperature: 0.7, max_tokens: 100, }) return process(inputs)两个 API 都位于pydantic_evals.dataset模块顶层也可直接从pydantic_evals导入。底层实现基于ContextVar的上下文隔离机制pydantic_evals/_task_run.py因此并发场景下每个 case 的记录互不干扰——每个任务运行都有独立的TaskRun实例attributes与metrics两个字典。测试 tests/evals/test_online.py 验证了在异步装饰函数与同步装饰函数中调用这两个 API 都能正确传播到EvaluatorContext。在评估器中访问 Metrics 与 AttributesMetrics 和 Attributes 通过EvaluatorContext暴露给评估器分别对应ctx.metricsdict[str, int | float]与ctx.attributesdict[str, Any]from dataclasses import dataclass from pydantic_evals.evaluators import Evaluator, EvaluatorContext dataclass class EfficiencyChecker(Evaluator): max_api_calls: int 5 def evaluate(self, ctx: EvaluatorContext) - dict[str, bool]: # 读取指标 api_calls ctx.metrics.get(api_calls, 0) tokens_used ctx.metrics.get(tokens_used, 0) # 读取属性 used_cache ctx.attributes.get(used_cache, False) return { efficient_api_usage: api_calls self.max_api_calls, used_caching: used_cache, token_efficient: tokens_used 1000, }EvaluatorContext是评估器的唯一输入除metrics/attributes外还包含inputs、output、expected_output、metadatacase 级、duration以及span_tree任务执行的 OpenTelemetry 跨度树等字段。注意读取指标时建议使用.get(key, default)形式因为未记录的指标键在字典中不存在零值记录也不会被创建。在报告中查看评估完成后Metrics 与 Attributes 会随每个 case 出现在报告对象中from pydantic_evals import Case, Dataset def task(inputs: str) - str: return fResult: {inputs} dataset Dataset(namereport_viewing, cases[Case(inputstest)], evaluators[]) report dataset.evaluate_sync(task) for case in report.cases: print(f{case.name}:) # Case 1: print(f Metrics: {case.metrics}) # Metrics: {} print(f Attributes: {case.attributes}) # Attributes: {}在数据模型层面ReportCasepydantic_evals/reporting/init.py直接持有metrics: dict[str, float | int]与attributes: dict[str, Any]字段与EvaluatorContext中的对应字段共享同一来源——二者都来自任务执行期间TaskRun累积的数据。同时ReportCaseAggregate也会对指标做跨 case 的聚合统计。需要注意Metrics 与 Attributes 默认不会打印在控制台报告中需通过case.metrics/case.attributes编程访问或借助 Logfire 的可视化界面查看。自动指标Pydantic AI 与 Logfire 的集成当任务函数中使用 Pydantic AI Agent 且启用了 Logfire 时框架会从 OpenTelemetry 跨度树中自动提取一组标准指标import logfire from pydantic_ai import Agent logfire.configure(send_to_logfireif-token-present) agent Agent(openai:gpt-5.2) async def ai_task(inputs: str) - str: result await agent.run(inputs) return result.output # 自动追踪的指标 # - requests: LLM 调用次数 # - input_tokens: 输入 Token 总量 # - output_tokens: 输出 Token 总量 # - prompt_tokens: 提示词 Token如可用 # - completion_tokens: 补全 Token如可用 # - cost: 预估成本如使用 genai-prices这些自动指标可以在评估器中直接读取from dataclasses import dataclass from pydantic_evals.evaluators import Evaluator, EvaluatorContext dataclass class CostChecker(Evaluator): max_cost: float 0.01 # $0.01 def evaluate(self, ctx: EvaluatorContext) - bool: cost ctx.metrics.get(cost, 0.0) return cost self.max_cost自动提取的底层逻辑在 pydantic_evals/_task_run.py 的extract_span_tree_metrics函数中它遍历任务执行的 span 树识别带有gen_ai.request.model属性的节点从中提取gen_ai.operation.name chat的调用次数requests、operation.costcost以及所有gen_ai.usage.*前缀的用量属性input_tokens、output_tokens等。前提是安装了 Logfirepip install pydantic-evals[logfire]并配置了LOGFIRE_TOKEN环境变量详见 Logfire 集成指南。实战示例四种典型追踪场景API 调用与缓存命中追踪在带缓存的任务中区分缓存命中和真实 API 调用from dataclasses import dataclass from pydantic_evals import increment_eval_metric, set_eval_attribute from pydantic_evals.evaluators import Evaluator, EvaluatorContext def check_cache(inputs: str) - str | None: return None # 演示用无缓存命中 dataclass class APIResult: text: str usage: Usage dataclass class Usage: total_tokens: int async def call_api(inputs: str) - APIResult: return APIResult(textfResult: {inputs}, usageUsage(total_tokens100)) def save_to_cache(inputs: str, result: str) - None: pass # 保存到缓存 async def smart_task(inputs: str) - str: # 先尝试缓存 if cached : check_cache(inputs): set_eval_attribute(cache_hit, True) return cached set_eval_attribute(cache_hit, False) # 调用 API increment_eval_metric(api_calls, 1) result await call_api(inputs) increment_eval_metric(tokens, result.usage.total_tokens) # 缓存结果 save_to_cache(inputs, result.text) return result.text # 评估效率 dataclass class EfficiencyEvaluator(Evaluator): def evaluate(self, ctx: EvaluatorContext) - dict[str, bool | float]: api_calls ctx.metrics.get(api_calls, 0) cache_hit ctx.attributes.get(cache_hit, False) return { used_cache: cache_hit, made_api_call: api_calls 0, efficiency_score: 1.0 if cache_hit else 0.5, }Agent 工具调用追踪在 Pydantic AI Agent 的agent.tool装饰函数内记录指标与属性from dataclasses import dataclass from pydantic_ai import Agent, RunContext from pydantic_evals import increment_eval_metric, set_eval_attribute from pydantic_evals.evaluators import Evaluator, EvaluatorContext agent Agent(openai:gpt-5.2) def search(query: str) - str: return fSearch results for: {query} def call(endpoint: str) - str: return fAPI response from: {endpoint} agent.tool def search_database(ctx: RunContext, query: str) - str: increment_eval_metric(db_searches, 1) set_eval_attribute(last_query, query) return search(query) agent.tool def call_api(ctx: RunContext, endpoint: str) - str: increment_eval_metric(api_calls, 1) set_eval_attribute(last_endpoint, endpoint) return call(endpoint) # 评估工具使用情况 dataclass class ToolUsageEvaluator(Evaluator): def evaluate(self, ctx: EvaluatorContext) - dict[str, bool | int]: db_searches ctx.metrics.get(db_searches, 0) api_calls ctx.metrics.get(api_calls, 0) return { used_database: db_searches 0, used_api: api_calls 0, tool_call_count: db_searches api_calls, reasonable_tool_usage: (db_searches api_calls) 5, }性能追踪子操作耗时用time.perf_counter()测量任务内各子操作的耗时并记录为指标import time from dataclasses import dataclass from pydantic_evals import increment_eval_metric, set_eval_attribute from pydantic_evals.evaluators import Evaluator, EvaluatorContext async def retrieve_context(inputs: str) - list[str]: return [context1, context2] async def generate_response(context: list[str], inputs: str) - str: return fGenerated response for {inputs} async def monitored_task(inputs: str) - str: # 追踪子操作耗时 t0 time.perf_counter() context await retrieve_context(inputs) retrieve_time time.perf_counter() - t0 increment_eval_metric(retrieve_time, retrieve_time) t0 time.perf_counter() result await generate_response(context, inputs) generate_time time.perf_counter() - t0 increment_eval_metric(generate_time, generate_time) # 记录需要哪些操作 set_eval_attribute(needed_retrieval, len(context) 0) set_eval_attribute(context_chunks, len(context)) return result # 评估性能 dataclass class PerformanceEvaluator(Evaluator): max_retrieve_time: float 0.5 max_generate_time: float 2.0 def evaluate(self, ctx: EvaluatorContext) - dict[str, bool]: retrieve_time ctx.metrics.get(retrieve_time, 0.0) generate_time ctx.metrics.get(generate_time, 0.0) return { fast_retrieval: retrieve_time self.max_retrieve_time, fast_generation: generate_time self.max_generate_time, }质量追踪置信度与来源把 LLM 返回的置信度、引用来源等质量信号提取为属性供评估器计算综合评分from dataclasses import dataclass from pydantic_evals import set_eval_attribute from pydantic_evals.evaluators import Evaluator, EvaluatorContext async def llm_call(inputs: str) - dict: return {text: fResponse: {inputs}, confidence: 0.85, sources: [doc1, doc2]} async def quality_task(inputs: str) - str: result await llm_call(inputs) # 提取质量指标 confidence result.get(confidence, 0.0) sources_used result.get(sources, []) set_eval_attribute(confidence, confidence) set_eval_attribute(source_count, len(sources_used)) set_eval_attribute(sources, sources_used) return result[text] # 基于质量信号评估 dataclass class QualityEvaluator(Evaluator): min_confidence: float 0.7 def evaluate(self, ctx: EvaluatorContext) - dict[str, bool | float]: confidence ctx.attributes.get(confidence, 0.0) source_count ctx.attributes.get(source_count, 0) return { high_confidence: confidence self.min_confidence, used_sources: source_count 0, quality_score: confidence * (1.0 0.1 * source_count), }实验级元数据Experiment Metadata除 case 级数据外还可以在调用Dataset.evaluate()时传入实验级元数据metadata参数见 dataset.py记录整次评估运行的配置from pydantic_evals import Case, Dataset dataset Dataset( nameexperiment_metadata, cases[ Case( inputstest, metadata{difficulty: easy}, # Case 级元数据 ) ] ) async def task(inputs: str) - str: return fResult: {inputs} # 传入实验级元数据 async def main(): report await dataset.evaluate( task, metadata{ model: gpt-5.2, prompt_version: v2.1, temperature: 0.7, }, ) # 在报告中访问实验元数据 print(report.experiment_metadata) # {model: gpt-5.2, prompt_version: v2.1, temperature: 0.7}从源码看evaluate()会把metadata传入EvaluationReportreporting/init.py并同步设置到实验 span 的属性logfire.experiment.metadata中dataset.py同时报告评估器ReportEvaluatorContext也能读取到这份元数据。何时使用实验元数据实验元数据适合追踪适用于整个评估运行的配置信息模型配置模型名称、版本、参数提示词版本使用了哪个提示词模板基础设施部署环境、区域实验上下文开发者姓名、功能分支、提交哈希。在以下场景中尤其有价值跨时间比较多次评估运行追踪哪个配置产生了哪个结果依据历史数据复现评估结果。在报告中查看实验元数据会显示在打印报告report.render()的顶部from pydantic_evals import Case, Dataset dataset Dataset(namemetadata_report, cases[Case(inputshello, expected_outputHELLO)]) async def task(text: str) - str: return text.upper() async def main(): report await dataset.evaluate( task, metadata{model: gpt-5.2, version: v1.0}, ) print(report.render()) ╭─ Evaluation Summary: task ─╮ │ model: gpt-5.2 │ │ version: v1.0 │ ╰────────────────────────────╯ ┏━━━━━━━━━━┳━━━━━━━━━━┓ ┃ Case ID ┃ Duration ┃ ┡━━━━━━━━━━╇━━━━━━━━━━┩ │ Case 1 │ 10ms │ ├──────────┼──────────┤ │ Averages │ 10ms │ └──────────┴──────────┘ 在报告渲染层reporting/init.py实验元数据逐条显示为key: value行并且在进行基线baseline对比时会用/-标记新增或移除的配置项便于快速识别两次实验的配置差异。任务与实验元数据的同步单一事实来源实验元数据用于记录配置而不是配置任务。metadata字典不会自动改变任务的运行行为——你必须保证元数据中的值与任务实际使用的值一致。例如很容易出现元数据声称temperature: 0.7而任务实际使用temperature: 1.0的情况导致实验追踪错误、结果无法复现。为避免这一问题建议为配置建立单一事实来源single source of truth让任务与元数据共同引用它。下面给出几种推荐模式。模式一共享模块常量适用于简单场景用模块级常量统一管理from pydantic_ai import Agent from pydantic_evals import Case, Dataset # 模块常量作为单一事实来源 MODEL_NAME openai:gpt-5-mini TEMPERATURE 0.7 INSTRUCTIONS You are a helpful assistant. agent Agent(MODEL_NAME, model_settings{temperature: TEMPERATURE}, instructionsINSTRUCTIONS) async def task(inputs: str) - str: result await agent.run(inputs) return result.output async def main(): dataset Dataset(nameshared_constants, cases[Case(inputsWhat is the capital of France?)]) # 元数据引用同一组常量 await dataset.evaluate( task, metadata{ model: MODEL_NAME, temperature: TEMPERATURE, instructions: INSTRUCTIONS, }, )模式二配置对象推荐定义一次配置对象任务与元数据共用from dataclasses import asdict, dataclass from pydantic_ai import Agent from pydantic_evals import Case, Dataset dataclass class TaskConfig: 任务配置的单一事实来源。 包含所有希望出现在实验元数据中的变量。 model: str temperature: float max_tokens: int prompt_version: str # 只定义一次配置 config TaskConfig( modelopenai:gpt-5-mini, temperature0.7, max_tokens500, prompt_versionv2.1, ) # 在任务中使用配置 agent Agent( config.model, model_settings{temperature: config.temperature, max_tokens: config.max_tokens}, ) async def task(inputs: str) - str: 任务使用与元数据中记录相同的配置。 result await agent.run(inputs) return result.output # 用同一配置对象派生出元数据 async def main(): dataset Dataset(nameconfig_evaluation, cases[Case(inputsWhat is the capital of France?)]) report await dataset.evaluate( task, metadataasdict(config), # 保证与任务行为一致 ) print(report.experiment_metadata) { model: openai:gpt-5-mini, temperature: 0.7, max_tokens: 500, prompt_version: v2.1, } 如果全局配置对象不可行也可以在任务调用点创建TaskConfig实例并通过deps或类似机制传给 Agent但此时你仍需保证传给Dataset.evaluate的metadata值与任务实际使用的值始终一致。反模式重复配置务必避免以下常见错误——配置在多处重复定义极易失步from pydantic_ai import Agent from pydantic_evals import Case, Dataset # ❌ 错误配置在多处定义 agent Agent(openai:gpt-5-mini, model_settings{temperature: 0.7}) async def task(inputs: str) - str: result await agent.run(inputs) return result.output async def main(): dataset Dataset(nameanti_pattern, cases[Case(inputstest)]) # ❌ 错误手动输入元数据容易与任务失步 await dataset.evaluate( task, metadata{ model: openai:gpt-5-mini, # 重复定义可能与 Agent 定义不一致 temperature: 0.8, # ⚠️ 错误任务实际使用 0.7 }, )该反模式中元数据声称temperature: 0.8而任务实际使用0.7会导致实验追踪错误结果无法复现对比不同运行结果时产生困惑浪费大量时间排查为什么结果不同。Metrics vs Attributes vs Metadata差异速查特性MetricsAttributesCase MetadataExperiment Metadata设置位置任务执行期间任务执行期间Case 定义时evaluate()调用时类型int、float任意类型任意类型任意类型用途定量测量定性描述测试数据实验配置用于聚合统计上下文信息任务输入运行追踪可访问者评估器评估器任务与评估器仅报告作用域每个 case每个 case每个 case每次实验一个完整的区分示例from pydantic_evals import Case, Dataset, increment_eval_metric, set_eval_attribute # Case Metadata在 case 定义时设置执行前 case Case( inputsquestion, metadata{difficulty: hard, category: math}, # 每个 case 的元数据 ) dataset Dataset(namemetrics_demo, cases[case]) # Metrics Attributes在任务执行期间记录 async def task(inputs): # 这些是在执行期间为每个 case 记录的 increment_eval_metric(tokens, 100) set_eval_attribute(model, gpt-5.2) return fResult: {inputs} async def main(): # Experiment Metadata在评估调用时定义 await dataset.evaluate( task, metadata{ # 实验级元数据 prompt_version: v2.1, temperature: 0.7, }, )故障排查Metrics/attributes 没有出现请确认是在任务函数内部调用这两个 APIfrom pydantic_evals import increment_eval_metric def process(inputs: str) - str: return fProcessed: {inputs} # 错误在任务外调用 increment_eval_metric(count, 1) def bad_task(inputs): return process(inputs) # 正确在任务内调用 def good_task(inputs): increment_eval_metric(count, 1) return process(inputs)原理上increment_eval_metric/set_eval_attribute依赖CURRENT_TASK_RUNContextVar它只在run_task()上下文管理器pydantic_evals/_task_run.py内被设置在任务外调用时该 ContextVar 为None记录会被静默忽略。同理若调用发生在任务完成之后例如在评估器内也无法再追加记录。Metrics 没有累加检查是否误用了set_eval_attribute而不是increment_eval_metricfrom pydantic_evals import increment_eval_metric, set_eval_attribute # 错误这会覆盖而不是累加 set_eval_attribute(count, 1) set_eval_attribute(count, 1) # 仍是 1 # 正确累加 increment_eval_metric(count, 1) increment_eval_metric(count, 1) # 现在是 2Attributes 数据量太大存储摘要而非原始数据from pydantic_evals import set_eval_attribute giant_response_object {key str(i): value * 100 for i in range(1000)} # 错误存储巨型对象 set_eval_attribute(full_response, giant_response_object) # 正确存储摘要 set_eval_attribute(response_size_kb, len(str(giant_response_object)) / 1024) set_eval_attribute(response_keys, list(giant_response_object.keys())[:10]) # 前 10 个键大体积属性会拖慢报告序列化并污染日志建议只保留足以支撑评估判断的摘要信息。延伸阅读Case 生命周期钩子每个 case 的 setup、teardown 与上下文准备自定义评估器在评估器中使用 Metrics 与 AttributesLogfire 集成在 Logfire 中可视化查看 Metrics并发与性能优化优化评估运行性能。【免费下载链接】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），仅供参考