企业官网建设流程全解析

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(output=f'Result: {inputs}', usage=Usage(total_tokens=100)) 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_metric（pydantic_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 f'Processed: {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 与 Attributes

Metrics 和 Attributes 通过EvaluatorContext暴露给评估器，分别对应ctx.metrics（dict[str, int | float]）与ctx.attributes（dict[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、metadata（case 级）、duration以及span_tree（任务执行的 OpenTelemetry 跨度树）等字段。注意：读取指标时建议使用.get(key, default)形式，因为未记录的指标键在字典中不存在（零值记录也不会被创建）。

在报告中查看

评估完成后，Metrics 与 Attributes 会随每个 case 出现在报告对象中：

from pydantic_evals import Case, Dataset def task(inputs: str) -> str: return f'Result: {inputs}' dataset = Dataset(name='report_viewing', cases=[Case(inputs='test')], 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: {}

在数据模型层面，ReportCase（pydantic_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_logfire='if-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.cost（cost）以及所有gen_ai.usage.*前缀的用量属性（input_tokens、output_tokens等）。前提是安装了 Logfire（pip 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(text=f'Result: {inputs}', usage=Usage(total_tokens=100)) 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 f'Search results for: {query}' def call(endpoint: str) -> str: return f'API 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 f'Generated 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': f'Response: {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( name='experiment_metadata', cases=[ Case( inputs='test', metadata={'difficulty': 'easy'}, # Case 级元数据 ) ] ) async def task(inputs: str) -> str: return f'Result: {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传入EvaluationReport（reporting/init.py），并同步设置到实验 span 的属性logfire.experiment.metadata中（dataset.py），同时报告评估器（ReportEvaluatorContext）也能读取到这份元数据。

何时使用实验元数据

实验元数据适合追踪适用于整个评估运行的配置信息：

• 模型配置：模型名称、版本、参数；
• 提示词版本：使用了哪个提示词模板；
• 基础设施：部署环境、区域；
• 实验上下文：开发者姓名、功能分支、提交哈希。

在以下场景中尤其有价值：

• 跨时间比较多次评估运行；
• 追踪"哪个配置产生了哪个结果"；
• 依据历史数据复现评估结果。

在报告中查看

实验元数据会显示在打印报告（report.render()）的顶部：

from pydantic_evals import Case, Dataset dataset = Dataset(name='metadata_report', cases=[Case(inputs='hello', expected_output='HELLO')]) 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}, instructions=INSTRUCTIONS) async def task(inputs: str) -> str: result = await agent.run(inputs) return result.output async def main(): dataset = Dataset(name='shared_constants', cases=[Case(inputs='What 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( model='openai:gpt-5-mini', temperature=0.7, max_tokens=500, prompt_version='v2.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(name='config_evaluation', cases=[Case(inputs='What is the capital of France?')]) report = await dataset.evaluate( task, metadata=asdict(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(name='anti_pattern', cases=[Case(inputs='test')]) # ❌ 错误：手动输入元数据，容易与任务失步 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：差异速查

一个完整的区分示例：

from pydantic_evals import Case, Dataset, increment_eval_metric, set_eval_attribute # Case Metadata：在 case 定义时设置（执行前） case = Case( inputs='question', metadata={'difficulty': 'hard', 'category': 'math'}, # 每个 case 的元数据 ) dataset = Dataset(name='metrics_demo', cases=[case]) # Metrics & Attributes：在任务执行期间记录 async def task(inputs): # 这些是在执行期间为每个 case 记录的 increment_eval_metric('tokens', 100) set_eval_attribute('model', 'gpt-5.2') return f'Result: {inputs}' async def main(): # Experiment Metadata：在评估调用时定义 await dataset.evaluate( task, metadata={ # 实验级元数据 'prompt_version': 'v2.1', 'temperature': 0.7, }, )

故障排查

"Metrics/attributes 没有出现"

请确认是在任务函数内部调用这两个 API：

from pydantic_evals import increment_eval_metric def process(inputs: str) -> str: return f'Processed: {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_metric：

from 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) # 现在是 2

"Attributes 数据量太大"

存储摘要而非原始数据：

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 与 Attributes；
• Logfire 集成：在 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