메트릭 & 속성
메트릭 & 속성 (Metrics & Attributes)
작업 실행 중에 커스텀 메트릭과 속성을 추적해서 더 풍부한 평가 인사이트를 얻어요.
평가 작업을 실행하는 동안 다음을 기록할 수 있어요:
- 메트릭 - 정량적 측정을 위한 숫자 값 (int/float)
- 속성 - 정성적 정보를 위한 어떤 데이터든
이것들은 평가 보고서에 나타나며 평가자가 평가에 사용할 수 있어요.
출처: 문서
본문
메트릭 기록하기
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)
# 사용된 토큰 추적
increment_eval_metric('tokens_used', result.usage.total_tokens)
return result.output
속성 기록하기
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)
평가자에서 접근하기
메트릭과 속성은 EvaluatorContext에서 사용할 수 있어요:
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,
}
보고서에서 보기
메트릭과 속성은 리포트 데이터에 나타나요:
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: {}
출력된 보고서에서도 표시할 수 있어요:
from pydantic_evals import Case, Dataset
def task(inputs: str) -> str:
return f'Result: {inputs}'
dataset = Dataset(name='report_printing', cases=[Case(inputs='test')], evaluators=[])
report = dataset.evaluate_sync(task)
# 메트릭과 속성은 사용 가능하지만 기본적으로는 표시되지 않음
# 프로그래밍 방식으로 또는 Logfire를 통해 접근
for case in report.cases:
print(f'\nCase: {case.name}')
"""
Case: Case 1
"""
print(f'Metrics: {case.metrics}')
#> Metrics: {}
print(f'Attributes: {case.attributes}')
#> Attributes: {}
자동 메트릭
Pydantic AI와 Logfire를 사용할 때 일부 메트릭은 자동으로 추적돼요:
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: 총 입력 토큰
# - output_tokens: 총 출력 토큰
# - prompt_tokens: 프롬프트 토큰 (사용 가능한 경우)
# - completion_tokens: 완료 토큰 (사용 가능한 경우)
# - 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
실용 예시
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,
}
툴 사용 추적
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,
}
성능 추적
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,
}
품질 추적
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),
}
실험 수준 메타데이터
케이스 수준 메타데이터 외에도, evaluate()를 호출할 때 실험 수준 메타데이터를 전달할 수 있어요:
from pydantic_evals import Case, Dataset
dataset = Dataset(
name='experiment_metadata',
cases=[
Case(
inputs='test',
metadata={'difficulty': 'easy'}, # 케이스 수준 메타데이터
)
]
)
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}
실험 메타데이터는 언제 쓸까요?
실험 메타데이터는 전체 평가 실행에 적용되는 구성을 추적하는 데 유용해요:
- 모델 구성: 모델 이름, 버전, 매개변수
- 프롬프트 버전 관리: 어떤 프롬프트 템플릿을 사용했는지
- 인프라: 배포 환경, 리전
- 실험 컨텍스트: 개발자 이름, 기능 브랜치, 커밋 해시
이 메타데이터는 특히 다음 때 가치가 있어요:
- 시간이 지나면서 여러 평가 실행을 비교
- 어떤 구성이 어떤 결과를 만들었는지 추적
- 과거 데이터에서 평가 결과를 재현
보고서에서 보기
실험 메타데이터는 출력된 보고서 상단에 나타나요:
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 │
└──────────┴──────────┘
"""
작업과 실험 메타데이터의 동기화
실험 메타데이터는 작업을 _구성_하기 위한 것이 아니라 구성을 _기록_하기 위한 것이에요. 메타데이터 딕셔너리는 작업의 동작을 자동으로 구성하지 않아요. 메타데이터 딕셔너리의 값이 작업이 실제로 사용하는 값과 일치하도록 직접 보장해야 해요. 예를 들어 메타데이터에 temperature: 0.7을 주장하면서 작업은 실제로 temperature: 1.0을 사용하면 잘못된 실험 추적과 재현 불가능한 결과가 생기기 쉬워요.
이 문제를 피하려면 작업과 메타데이터가 모두 참조하는 구성의 단일 진실 소스(single source of truth)를 세우는 걸 권장해요. 아래에 이 동기화를 달성하기 위한 몇 가지 권장 패턴이 있어요.
패턴 1: 공유 모듈 상수
간단한 경우에는 모듈 수준 상수를 사용해요:
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,
},
)
패턴 2: 구성 객체 (권장)
구성을 한 번 정의하고 모든 곳에서 사용해요:
from dataclasses import asdict, dataclass
from pydantic_ai import Agent
from pydantic_evals import Case, Dataset
@dataclass
class TaskConfig:
"""Single source of truth for task configuration.
Includes all variables you'd like to see in experiment metadata.
"""
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',
)
# 작업에서 config 사용
agent = Agent(
config.model,
model_settings={'temperature': config.temperature, 'max_tokens': config.max_tokens},
)
async def task(inputs: str) -> str:
"""Task uses the same config that's recorded in metadata."""
result = await agent.run(inputs)
return result.output
# 같은 config에서 파생된 메타데이터로 평가
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 등으로 에이전트에 전달할 수도 있어요. 하지만 그 경우에도 값이 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', # 중복! 에이전트 정의와 어긋날 수 있음
'temperature': 0.8, # ⚠️ 틀림! 작업은 실제로 0.7을 사용
},
)
이 안티 패턴에서는 메타데이터가 temperature: 0.8을 주장하지만 작업은 0.7을 사용해요. 이로 인해:
- 잘못된 실험 추적
- 결과 재현 불가능
- 실행 비교 시 혼란
- "결과가 왜 다른지" 디버깅에 시간 낭비
메트릭 vs 속성 vs 메타데이터
차이를 이해해요:
| 기능 | 메트릭 | 속성 | 케이스 메타데이터 | 실험 메타데이터 |
|---|---|---|---|---|
| 설정 위치 | 작업 실행 | 작업 실행 | 케이스 정의 | evaluate() 호출 |
| 유형 | int, float | Any | Any | Any |
| 목적 | 정량적 | 정성적 | 테스트 데이터 | 실험 구성 |
| 용도 | 집계 | 컨텍스트 | 작업 입력 | 실행 추적 |
| 사용 가능한 곳 | 평가자 | 평가자 | 작업 & 평가자 | 보고서만 |
| 범위 | 케이스별 | 케이스별 | 케이스별 | 실험별 |
from pydantic_evals import Case, Dataset, increment_eval_metric, set_eval_attribute
# 케이스 메타데이터: 케이스에서 정의 (실행 전)
case = Case(
inputs='question',
metadata={'difficulty': 'hard', 'category': 'math'}, # 케이스별 메타데이터
)
dataset = Dataset(name='metrics_demo', cases=[case])
# 메트릭 & 속성: 실행 중 기록
async def task(inputs):
# 각 케이스에 대해 실행 중에 기록됨
increment_eval_metric('tokens', 100)
set_eval_attribute('model', 'gpt-5.2')
return f'Result: {inputs}'
async def main():
# 실험 메타데이터: 평가 시점에 정의
await dataset.evaluate(
task,
metadata={ # 실험 수준 메타데이터
'prompt_version': 'v2.1',
'temperature': 0.7,
},
)
문제 해결
"메트릭/속성이 나타나지 않음"
작업 내부에서 함수를 호출하고 있는지 확인해요:
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)
"메트릭이 증가하지 않음"
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
"속성에 데이터가 너무 많음"
원시 데이터가 아니라 요약을 저장해요:
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개 키
다음 단계
- 케이스 생명주기 훅 - 케이스별 설정, 정리, 컨텍스트 준비
- 커스텀 평가자 - 평가자에서 메트릭/속성 사용
- Logfire 통합 - Logfire에서 메트릭 보기
- 동시성 & 성능 - 평가 성능 최적화
더 알아보기 (Learn more)
- Pydantic Evals 문서: 메트릭 & 속성