커스텀 평가자
커스텀 평가자 (Custom Evaluators)
도메인 특화 로직, 외부 통합, 또는 전문화된 메트릭을 위해 커스텀 평가자를 작성해요.
출처: 문서
본문
기본 커스텀 평가자
모든 평가자는 Evaluator를 상속하고 evaluate를 구현해야 해요:
from dataclasses import dataclass
from pydantic_evals.evaluators import Evaluator, EvaluatorContext
@dataclass
class ExactMatch(Evaluator):
"""Check if output exactly matches expected output."""
def evaluate(self, ctx: EvaluatorContext) -> bool:
return ctx.output == ctx.expected_output
핵심 포인트:
@dataclass데코레이터 사용 (필수)Evaluator상속evaluate(self, ctx: EvaluatorContext) -> EvaluatorOutput구현bool,int,float,str,EvaluationReason, 또는 이들의dict반환
EvaluatorContext
컨텍스트는 케이스 실행에 대한 모든 정보를 제공해요:
from dataclasses import dataclass
from pydantic_evals.evaluators import Evaluator, EvaluatorContext
@dataclass
class MyEvaluator(Evaluator):
def evaluate(self, ctx: EvaluatorContext) -> bool:
# 케이스 데이터 접근
ctx.name # 케이스 이름
ctx.inputs # 작업 입력
ctx.metadata # 케이스 메타데이터
ctx.expected_output # 기대 출력 (None일 수 있음)
ctx.output # 실제 출력
# 성능 데이터
ctx.duration # 작업 실행 시간 (초)
# 커스텀 메트릭/속성 (metrics 가이드 참고)
ctx.metrics # dict[str, int | float]
ctx.attributes # dict[str, Any]
# OpenTelemetry 스팬 (Logfire가 구성된 경우)
ctx.span_tree # 동작 검사용 SpanTree
return True
평가자 매개변수
설정 가능한 매개변수를 dataclass 필드로 추가해요:
from dataclasses import dataclass
from pydantic_evals import Case, Dataset
from pydantic_evals.evaluators import Evaluator, EvaluatorContext
@dataclass
class ContainsKeyword(Evaluator):
keyword: str
case_sensitive: bool = True
def evaluate(self, ctx: EvaluatorContext) -> bool:
output = ctx.output
keyword = self.keyword
if not self.case_sensitive:
output = output.lower()
keyword = keyword.lower()
return keyword in output
# 사용법
dataset = Dataset(
name='keyword_check',
cases=[Case(name='test', inputs='This is important')],
evaluators=[
ContainsKeyword(keyword='important', case_sensitive=False),
],
)
반환 유형
불리언 판정
간단한 합격/불합격 검사:
from dataclasses import dataclass
from pydantic_evals.evaluators import Evaluator, EvaluatorContext
@dataclass
class IsValidJSON(Evaluator):
def evaluate(self, ctx: EvaluatorContext) -> bool:
try:
import json
json.loads(ctx.output)
return True
except Exception:
return False
숫자 점수
품질 메트릭:
from dataclasses import dataclass
from pydantic_evals.evaluators import Evaluator, EvaluatorContext
@dataclass
class LengthScore(Evaluator):
"""Score based on output length (0.0 = too short, 1.0 = ideal)."""
ideal_length: int = 100
tolerance: int = 20
def evaluate(self, ctx: EvaluatorContext) -> float:
length = len(ctx.output)
diff = abs(length - self.ideal_length)
if diff <= self.tolerance:
return 1.0
else:
# 이상값에서 멀어질수록 점수 감소
score = max(0.0, 1.0 - (diff - self.tolerance) / self.ideal_length)
return score
점수는 유한해야 해요
숫자 점수는 유한해야 해요. NaN 또는 ±inf를 반환하는 평가자(스칼라, EvaluationReason 내부, 또는 반환 딕셔너리의 값으로)는 그 케이스에서 점수 대신 EvaluatorFailure를 생성해요. 그래서 비교할 수 없는 값은 조용히 기록되는 대신 실패한 평가자로 드러나요.
문자열 라벨
범주형 분류:
from dataclasses import dataclass
from pydantic_evals.evaluators import Evaluator, EvaluatorContext
@dataclass
class SentimentClassifier(Evaluator):
def evaluate(self, ctx: EvaluatorContext) -> str:
output_lower = ctx.output.lower()
if any(word in output_lower for word in ['error', 'failed', 'wrong']):
return 'negative'
elif any(word in output_lower for word in ['success', 'correct', 'great']):
return 'positive'
else:
return 'neutral'
이유와 함께
어떤 결과에든 설명을 추가해요:
from dataclasses import dataclass
from pydantic_evals.evaluators import EvaluationReason, Evaluator, EvaluatorContext
@dataclass
class SmartCheck(Evaluator):
threshold: float = 0.8
def evaluate(self, ctx: EvaluatorContext) -> EvaluationReason:
score = self._calculate_score(ctx.output)
if score >= self.threshold:
return EvaluationReason(
value=True,
reason=f'Score {score:.2f} exceeds threshold {self.threshold}',
)
else:
return EvaluationReason(
value=False,
reason=f'Score {score:.2f} below threshold {self.threshold}',
)
def _calculate_score(self, output: str) -> float:
# 나만의 점수 로직
return 0.75
여러 결과
키-값 쌍의 딕셔너리를 반환해서 한 평가자에서 여러 평가를 반환할 수 있어요.
from dataclasses import dataclass
from pydantic_evals.evaluators import (
EvaluationReason,
Evaluator,
EvaluatorContext,
EvaluatorOutput,
)
@dataclass
class ComprehensiveCheck(Evaluator):
def evaluate(self, ctx: EvaluatorContext) -> EvaluatorOutput:
format_valid = self._check_format(ctx.output)
return {
'valid_format': EvaluationReason(
value=format_valid,
reason='Valid JSON format' if format_valid else 'Invalid JSON format',
),
'quality_score': self._score_quality(ctx.output), # float
'category': self._classify(ctx.output), # str
}
def _check_format(self, output: str) -> bool:
return output.startswith('{') and output.endswith('}')
def _score_quality(self, output: str) -> float:
return len(output) / 100.0
def _classify(self, output: str) -> str:
return 'short' if len(output) < 50 else 'long'
반환 딕셔너리의 각 키는 보고서에서 별도의 결과가 돼요. 값은 다음일 수 있어요:
- 기본형 (
bool,int,float,str) EvaluationReason(설명이 있는 값)- 이 유형들의 중첩 딕셔너리
EvaluatorOutput 타입은 평가자가 반환할 수 있는 모든 합법적인 값을 나타내며, 커스텀 evaluate 메서드의 반환 타입 주석으로 사용할 수 있어요.
조건부 결과
평가자는 해당되지 않을 때 빈 딕셔너리를 반환함으로써 주어진 케이스에 대해 결과를 만들지 여부를 동적으로 선택할 수 있어요:
from dataclasses import dataclass
from pydantic_evals.evaluators import (
EvaluationReason,
Evaluator,
EvaluatorContext,
EvaluatorOutput,
)
@dataclass
class SQLValidator(Evaluator):
"""Only evaluates SQL queries, skips other outputs."""
def evaluate(self, ctx: EvaluatorContext) -> EvaluatorOutput:
# 이 케이스가 SQL 검증에 해당하는지 확인
if not isinstance(ctx.output, str) or not ctx.output.strip().upper().startswith(
('SELECT', 'INSERT', 'UPDATE', 'DELETE')
):
# 빈 dict 반환 - 이 평가자는 이 케이스에 적용되지 않음
return {}
# SQL 쿼리이므로 검증 수행
try:
# 실제 구현에서는 sqlparse 등을 사용
is_valid = self._validate_sql(ctx.output)
return {
'sql_valid': is_valid,
'sql_complexity': self._measure_complexity(ctx.output),
}
except Exception as e:
return {'sql_valid': EvaluationReason(False, reason=f'Exception: {e}')}
def _validate_sql(self, query: str) -> bool:
# 단순화된 검증
return 'FROM' in query.upper() or 'INTO' in query.upper()
def _measure_complexity(self, query: str) -> str:
joins = query.upper().count('JOIN')
if joins == 0:
return 'simple'
elif joins <= 2:
return 'moderate'
else:
return 'complex'
이 패턴은 다음과 같은 때 유용해요:
- 평가자가 특정 유형의 출력에만 적용될 때 (예: 코드 출력에만 코드 검증)
- 검증이 메타데이터 태그에 의존할 때 (예:
language='python'으로 표시된 케이스만 평가) - 다른 평가자 결과에 따라 비싼 검사를 조건부로 실행하고 싶을 때
핵심 포인트:
{}를 반환하면 "이 평가자는 여기에 적용되지 않는다"는 뜻 -- 케이스는 이 평가자의 결과를 보여주지 않아요{'key': value}를 반환하면 "이 평가자는 적용되며 결과는 다음과 같다"는 뜻- 큰 비율의 케이스에 적용되거나 조건이 출력 자체에 기반할 때는 케이스 수준 평가자를 쓰는 것보다 이 방식이 더 실용적이에요
- 평가자는 여전히 모든 케이스에서 실행되지만, 관련 없을 때 단락(short-circuit)될 수 있어요
비동기 평가자
I/O 바인딩 작업에는 async def를 사용해요:
from dataclasses import dataclass
from pydantic_evals.evaluators import Evaluator, EvaluatorContext
@dataclass
class APIValidator(Evaluator):
api_url: str
async def evaluate(self, ctx: EvaluatorContext) -> bool:
import httpx
async with httpx.AsyncClient() as client:
response = await client.post(
self.api_url,
json={'output': ctx.output},
)
return response.json()['valid']
Pydantic Evals는 동기와 비동기 평가자를 모두 자동으로 처리해요.
메타데이터 사용
컨텍스트 인지 평가를 위해 케이스 메타데이터에 접근해요:
from dataclasses import dataclass
from pydantic_evals.evaluators import Evaluator, EvaluatorContext
@dataclass
class DifficultyAwareScore(Evaluator):
def evaluate(self, ctx: EvaluatorContext) -> float:
# 기본 점수
base_score = self._score_output(ctx.output)
# 메타데이터의 난이도에 따라 조정
if ctx.metadata and 'difficulty' in ctx.metadata:
difficulty = ctx.metadata['difficulty']
if difficulty == 'easy':
# 쉬운 문제에서 실수에 더 가혹하게
return base_score
elif difficulty == 'hard':
# 어려운 문제에서는 더 관대하게
return min(1.0, base_score * 1.2)
return base_score
def _score_output(self, output: str) -> float:
# 나만의 점수 로직
return 0.8
메트릭 사용
작업 실행 중에 설정된 커스텀 메트릭에 접근해요:
from dataclasses import dataclass
from pydantic_evals import increment_eval_metric, set_eval_attribute
from pydantic_evals.evaluators import Evaluator, EvaluatorContext
# 작업에서
def my_task(inputs: str) -> str:
result = f'processed: {inputs}'
# 메트릭 기록
increment_eval_metric('api_calls', 3)
set_eval_attribute('used_cache', True)
return result
# 평가자에서
@dataclass
class EfficiencyCheck(Evaluator):
max_api_calls: int = 5
def evaluate(self, ctx: EvaluatorContext) -> bool:
api_calls = ctx.metrics.get('api_calls', 0)
return api_calls <= self.max_api_calls
더 자세한 내용은 메트릭 & 속성 가이드를 참고해요.
제네릭 타입 매개변수
제네릭으로 평가자를 타입 안전하게 만들어요:
from dataclasses import dataclass
from typing import TypeVar
from pydantic_evals.evaluators import Evaluator, EvaluatorContext
InputsT = TypeVar('InputsT')
OutputT = TypeVar('OutputT')
@dataclass
class TypedEvaluator(Evaluator[InputsT, OutputT, dict]):
def evaluate(self, ctx: EvaluatorContext[InputsT, OutputT, dict]) -> bool:
# ctx.inputs와 ctx.output이 이제 제대로 타입이 지정됨
return True
커스텀 평가 이름
평가가 보고서에 표시되는 방식을 제어해요:
from dataclasses import dataclass
from pydantic_evals.evaluators import Evaluator, EvaluatorContext
@dataclass
class CustomNameEvaluator(Evaluator):
check_type: str
def get_default_evaluation_name(self) -> str:
# 클래스 이름 대신 check_type을 이름으로 사용
return f'{self.check_type}_check'
def evaluate(self, ctx: EvaluatorContext) -> bool:
return True
# 보고서에서 "CustomNameEvaluator" 대신 "format_check"로 표시됨
evaluator = CustomNameEvaluator(check_type='format')
또는 evaluation_name 필드를 사용해요 (내장 패턴을 쓴다면):
from dataclasses import dataclass
from pydantic_evals.evaluators import Evaluator, EvaluatorContext
@dataclass
class MyEvaluator(Evaluator):
evaluation_name: str | None = None
def evaluate(self, ctx: EvaluatorContext) -> bool:
return True
# 사용법
MyEvaluator(evaluation_name='my_custom_name')
실제 사용 예시
SQL 검증
from dataclasses import dataclass
from pydantic_evals.evaluators import EvaluationReason, Evaluator, EvaluatorContext
@dataclass
class ValidSQL(Evaluator):
dialect: str = 'postgresql'
def evaluate(self, ctx: EvaluatorContext) -> EvaluationReason:
try:
import sqlparse
parsed = sqlparse.parse(ctx.output)
if not parsed:
return EvaluationReason(
value=False,
reason='Could not parse SQL',
)
# 위험한 연산 확인
sql_upper = ctx.output.upper()
if 'DROP' in sql_upper or 'DELETE' in sql_upper:
return EvaluationReason(
value=False,
reason='Contains dangerous operations (DROP/DELETE)',
)
return EvaluationReason(
value=True,
reason='Valid SQL syntax',
)
except Exception as e:
return EvaluationReason(
value=False,
reason=f'SQL parsing error: {e}',
)
코드 실행
from dataclasses import dataclass
from pydantic_evals.evaluators import EvaluationReason, Evaluator, EvaluatorContext
@dataclass
class ExecutablePython(Evaluator):
timeout_seconds: float = 5.0
async def evaluate(self, ctx: EvaluatorContext) -> EvaluationReason:
import asyncio
import os
import tempfile
# 코드를 임시 파일에 작성
with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
f.write(ctx.output)
temp_path = f.name
try:
# 타임아웃으로 실행
process = await asyncio.create_subprocess_exec(
'python', temp_path,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
try:
stdout, stderr = await asyncio.wait_for(
process.communicate(),
timeout=self.timeout_seconds,
)
except asyncio.TimeoutError:
process.kill()
return EvaluationReason(
value=False,
reason=f'Execution timeout after {self.timeout_seconds}s',
)
if process.returncode == 0:
return EvaluationReason(
value=True,
reason='Code executed successfully',
)
else:
return EvaluationReason(
value=False,
reason=f'Execution failed: {stderr.decode()}',
)
finally:
os.unlink(temp_path)
외부 API 검증
from dataclasses import dataclass
from pydantic_evals.evaluators import Evaluator, EvaluatorContext
@dataclass
class APIResponseValid(Evaluator):
api_endpoint: str
api_key: str
async def evaluate(self, ctx: EvaluatorContext) -> dict[str, bool | float]:
import httpx
try:
async with httpx.AsyncClient() as client:
response = await client.post(
self.api_endpoint,
headers={'Authorization': f'Bearer {self.api_key}'},
json={'data': ctx.output},
timeout=10.0,
)
result = response.json()
return {
'api_reachable': True,
'validation_passed': result.get('valid', False),
'confidence_score': result.get('confidence', 0.0),
}
except Exception:
return {
'api_reachable': False,
'validation_passed': False,
'confidence_score': 0.0,
}
평가자 테스트
평가자를 다른 Python 코드처럼 테스트해요:
from dataclasses import dataclass
from pydantic_evals.evaluators import Evaluator, EvaluatorContext
@dataclass
class ExactMatch(Evaluator):
"""Check if output exactly matches expected output."""
def evaluate(self, ctx: EvaluatorContext) -> bool:
return ctx.output == ctx.expected_output
def test_exact_match():
evaluator = ExactMatch()
# 일치 테스트
ctx = EvaluatorContext(
name='test',
inputs='input',
metadata=None,
expected_output='expected',
output='expected',
duration=0.1,
_span_tree=None,
attributes={},
metrics={},
)
assert evaluator.evaluate(ctx) is True
# 불일치 테스트
ctx.output = 'different'
assert evaluator.evaluate(ctx) is False
모범 사례
1. 평가자를 집중적으로 유지하기
각 평가자는 한 가지만 확인해야 해요:
from dataclasses import dataclass
from pydantic_evals.evaluators import Evaluator, EvaluatorContext
def check_format(output: str) -> bool:
return output.startswith('{')
def check_content(output: str) -> bool:
return len(output) > 10
def check_length(output: str) -> bool:
return len(output) < 1000
def check_spelling(output: str) -> bool:
return True # 자리표시자
# 나쁨: 너무 많은 일을 함
@dataclass
class EverythingChecker(Evaluator):
def evaluate(self, ctx: EvaluatorContext) -> dict:
return {
'format_valid': check_format(ctx.output),
'content_good': check_content(ctx.output),
'length_ok': check_length(ctx.output),
'spelling_correct': check_spelling(ctx.output),
}
# 좋음: 평가자 분리
@dataclass
class FormatValidator(Evaluator):
def evaluate(self, ctx: EvaluatorContext) -> bool:
return check_format(ctx.output)
@dataclass
class ContentChecker(Evaluator):
def evaluate(self, ctx: EvaluatorContext) -> bool:
return check_content(ctx.output)
@dataclass
class LengthChecker(Evaluator):
def evaluate(self, ctx: EvaluatorContext) -> bool:
return check_length(ctx.output)
@dataclass
class SpellingChecker(Evaluator):
def evaluate(self, ctx: EvaluatorContext) -> bool:
return check_spelling(ctx.output)
이에 대한 몇 가지 예외가 있어요:
- 공유 계산이나 네트워크 요청 지연이 상당할 때는, 하나의 평가자가 모든 의존 출력을 함께 계산하는 게 나을 수 있어요.
- 여러 검사가 밀접하게 결합되거나 서로 매우 관련이 깊다면, 그 로직 전체를 한 평가자에 넣는 게 합리적일 수 있어요.
2. 누락 데이터를 우아하게 처리하기
from dataclasses import dataclass
from pydantic_evals.evaluators import EvaluationReason, Evaluator, EvaluatorContext
@dataclass
class SafeEvaluator(Evaluator):
def evaluate(self, ctx: EvaluatorContext) -> EvaluationReason:
if ctx.expected_output is None:
return EvaluationReason(
value=True,
reason='Skipped: no expected output provided',
)
# 나만의 평가 로직
...
3. 도움이 되는 이유 제공하기
from dataclasses import dataclass
from pydantic_evals.evaluators import EvaluationReason, Evaluator, EvaluatorContext
@dataclass
class HelpfulEvaluator(Evaluator):
def evaluate(self, ctx: EvaluatorContext) -> EvaluationReason:
# 나쁨
return EvaluationReason(value=False, reason='Failed')
# 좋음
return EvaluationReason(
value=False,
reason=f'Expected {ctx.expected_output!r}, got {ctx.output!r}',
)
4. 외부 호출에 타임아웃 사용하기
from dataclasses import dataclass
from pydantic_evals.evaluators import Evaluator, EvaluatorContext
@dataclass
class APIEvaluator(Evaluator):
timeout: float = 10.0
async def _call_api(self, output: str) -> bool:
# API 호출 자리표시자
return True
async def evaluate(self, ctx: EvaluatorContext) -> bool:
import asyncio
try:
return await asyncio.wait_for(
self._call_api(ctx.output),
timeout=self.timeout,
)
except asyncio.TimeoutError:
return False
다음 단계
더 알아보기 (Learn more)
- Pydantic Evals 문서: 커스텀 평가자