AI Agent

Pydantic Evals 실전 — 타입 안전 LLM 평가 데이터셋 구축과 프로덕션 회귀 탐지

cell-devlog 2026. 5. 29. 14:35
반응형

유닛 테스트는 결정론적입니다. 같은 입력 → 같은 출력. LLM 에이전트는 다릅니다. 같은 입력 → 확률적으로 다른 출력. 그래서 "테스트"가 아닌 "Eval"이 필요하고, Pydantic Evals는 그 Eval을 코드로 다루는 방법을 제시합니다.


핵심 요약 → Pydantic Evals = pydantic-evals 별도 패키지 (2026.05.21 최신 릴리즈, Production/Stable) → 핵심 구조: Dataset → Cases → Experiment → Evaluators → 3가지 평가자 레이어: 결정론적(정확 매치·정규식) + LLM-as-Judge + 스팬 기반 → 스팬 기반 평가(HasMatchingSpan): "최종 출력이 아닌 내부 동작을 평가" — 에이전트 Eval의 핵심 → YAML/JSON 데이터셋 직렬화 → 버전 관리, 비개발자 기여 가능 → Logfire 통합: logfire.instrument_pydantic_ai() 1줄 → Eval 결과 자동 대시보드 → CI/CD: evaluate_sync() 반환값 검사 → PR 머지 전 회귀 자동 차단


왜 LLM에는 유닛 테스트가 부족한가

# 유닛 테스트로 LLM 에이전트를 검사하면 어떻게 되나

def test_sentiment_agent():
    result = sentiment_agent.run_sync("이 제품 정말 별로예요")
    assert result.output.sentiment == "negative"  # ← 이건 작동함

    # 하지만 실제로 중요한 것들은 테스트 못함:
    # - "긍정도 아니고 중립도 아닌, 강하게 부정인가?"
    # - "신뢰도가 0.9 이상인가?"
    # - "내부적으로 DB 조회 툴을 호출했는가?"
    # - "응답이 2초 이내인가?"
    # - "새 모델 버전으로 교체했을 때 동일한 품질인가?"

# 이것들을 다루려면 Eval 프레임워크 필요

유닛 테스트와 달리 Eval은 신흥 기술입니다. 어떻게 정의해야 하는지 정확히 안다고 주장하는 사람이 있다면 무시해도 됩니다. Pydantic Evals는 지나치게 의견을 강요하지 않으면서 유연하고 유용하게 설계됐습니다.


1. 설치 + 핵심 구조

# 설치 (pydantic-ai와 별도 패키지)
pip install pydantic-evals
pip install pydantic-evals[logfire]   # Logfire 대시보드 포함

# Python 3.10+ 필수
# ── 핵심 구조 이해 ──
#
# Dataset ──────────── Cases (여러 테스트 케이스)
#    │
#    └── Experiments (실행마다 생성) ──── Case Results
#                                    └── Task (평가할 함수)
#                                    └── Evaluators (평가자들)

from pydantic_evals import Case, Dataset

# 가장 단순한 예시
dataset = Dataset(
    name="sentiment_eval",
    cases=[
        Case(
            name="강한 부정",
            inputs="이 제품 정말 별로예요. 환불할게요.",
            expected_output="negative",
        ),
        Case(
            name="약한 긍정",
            inputs="그냥 괜찮은 것 같아요",
            expected_output="positive",
        ),
        Case(
            name="명확한 긍정",
            inputs="완전 최고예요! 강력 추천합니다!",
            expected_output="positive",
        ),
    ]
)

# 평가할 태스크 함수
async def sentiment_task(inputs: str) -> str:
    result = await sentiment_agent.run(inputs)
    return result.output.sentiment   # "positive" | "negative" | "neutral"

# Eval 실행
report = await dataset.evaluate(sentiment_task)
print(report)

# 출력:
# sentiment_eval
#  강한 부정    ✅ pass
#  약한 긍정    ❌ fail (expected: positive, got: neutral)
#  명확한 긍정  ✅ pass
# Score: 66.7%

2. 내장 평가자 — 결정론적 검사

from pydantic_evals import Case, Dataset
from pydantic_evals.evaluators import (
    MatchesExpectedOutput,   # 정확 일치
    IsInstance,              # 타입 검사
    MaxDuration,             # 실행 시간 제한
    FuzzyMatchesOutput,      # 퍼지 매칭 (유사도 기반)
)

dataset = Dataset(
    name="code_review_eval",
    cases=[
        Case(
            name="ZeroDivisionError 탐지",
            inputs="def divide(a, b): return a / b",
            expected_output=CodeReview(
                has_bugs=True,
                severity="high",
                issues=["0으로 나누기 예외 처리 없음"],
                score=3,
            ),
        ),
    ],
    evaluators=[
        # 전체 케이스에 적용되는 평가자들
        IsInstance(type=CodeReview),        # 출력 타입 검사
        MaxDuration(seconds=5.0),           # 5초 이내 응답 강제
    ]
)

# 케이스별 평가자 — 각 케이스에 특화된 기준
dataset_with_case_evaluators = Dataset(
    name="qa_eval",
    cases=[
        Case(
            name="수도 질문",
            inputs="프랑스의 수도는?",
            expected_output="파리",
            evaluators=[
                MatchesExpectedOutput(),           # "파리" 정확 일치
            ]
        ),
        Case(
            name="개념 설명",
            inputs="머신러닝이 뭐야?",
            expected_output=None,                  # 정답 없음 — LLM Judge로 평가
            metadata={"topic": "ml_basics"},
        ),
    ]
)

3. LLM-as-Judge — 주관적 품질 평가

from pydantic_evals import Case, Dataset
from pydantic_evals.evaluators import LLMJudge

dataset = Dataset(
    name="customer_support_eval",
    cases=[
        Case(
            name="환불 요청 처리",
            inputs="3주 전에 산 제품이 고장났어요. 환불 받을 수 있나요?",
            expected_output=None,
            evaluators=[
                # 케이스별 특화 rubric
                LLMJudge(
                    rubric="응답이 환불 정책을 명확히 안내하고, 공감적 어조를 사용하며, 다음 단계를 구체적으로 제시했는가",
                    include_input=True,
                    model="anthropic:claude-sonnet-4-6",  # Judge 모델 지정
                )
            ]
        ),
        Case(
            name="기술 문의",
            inputs="앱이 계속 튕겨요",
            expected_output=None,
            evaluators=[
                LLMJudge(
                    rubric="응답이 구체적인 문제 해결 단계를 제공하고, 추가 정보 수집이 필요하면 명확한 질문을 했는가",
                    include_input=True,
                )
            ]
        ),
    ],
    # 전체 케이스에 적용되는 공통 Judge
    evaluators=[
        LLMJudge(
            rubric="응답이 전문적이고 정중한 어조를 유지했는가",
            model="openai:gpt-5.5",  # 더 강력한 Judge
        ),
        LLMJudge(
            rubric="응답에 개인정보나 민감한 데이터가 포함되어 있지 않은가",
            model="openai:gpt-5-mini",  # 단순 검사는 저렴한 모델로
        ),
        MaxDuration(seconds=10.0),
    ]
)

# Judge 모델 선택 전략
# - 단순 이진 검사 (있다/없다): gpt-5-mini ($저렴)
# - 중간 복잡도 품질 평가: claude-sonnet-4-6
# - 고난이도 전문 도메인: claude-opus-4-7 ($비쌈)

4. 스팬 기반 평가 — 에이전트 내부 동작 검사

스팬 기반 평가는 RAG 시스템에서 문서가 생성 전에 실제로 검색됐는지, 멀티 에이전트에서 오케스트레이터가 올바른 전문 에이전트에게 위임했는지 확인할 때 특히 유용합니다. 최종 출력이 맞더라도 내부 로직이 틀릴 수 있기 때문입니다.

from pydantic_evals import Case, Dataset
from pydantic_evals.evaluators import HasMatchingSpan

# ── RAG 에이전트 Eval: 문서를 실제로 검색했는가? ──
rag_eval = Dataset(
    name="rag_agent_eval",
    cases=[
        Case(
            name="기술 문서 질문",
            inputs="FastAPI에서 의존성 주입 어떻게 쓰나요?",
            expected_output=None,
        ),
    ],
    evaluators=[
        # 벡터 검색 툴이 호출됐는가?
        HasMatchingSpan(
            query={"name_contains": "vector_search"},
            evaluation_name="retrieved_docs",
        ),
        # Reranker가 실행됐는가?
        HasMatchingSpan(
            query={"name_contains": "rerank"},
            evaluation_name="reranked",
        ),
        # 검색 → 생성 순서가 맞는가?
        HasMatchingSpan(
            query={"and_": [
                {"name_contains": "vector_search"},
                {"name_contains": "generate"},   # 검색 후에 생성 있어야 함
            ]},
            evaluation_name="search_before_generate",
        ),
        # 응답 생성이 2초 이내
        HasMatchingSpan(
            query={
                "name_contains": "generate",
                "max_duration": 2.0,
            },
            evaluation_name="fast_generation",
        ),
        # 에러 없이 완료
        HasMatchingSpan(
            query={"not_": {"has_attributes": {"error": True}}},
            evaluation_name="no_errors",
        ),
    ]
)


# ── 멀티 에이전트 Eval: 올바른 서브에이전트에게 위임했는가? ──
multi_agent_eval = Dataset(
    name="orchestrator_eval",
    cases=[
        Case(
            name="재무 질문 → finance_agent 위임",
            inputs="2분기 영업이익 분석해줘",
            evaluators=[
                HasMatchingSpan(
                    query={"name_contains": "finance_agent"},
                    evaluation_name="delegated_to_finance",
                ),
            ]
        ),
        Case(
            name="코드 질문 → code_agent 위임",
            inputs="이 Python 코드 최적화해줘",
            evaluators=[
                HasMatchingSpan(
                    query={"name_contains": "code_agent"},
                    evaluation_name="delegated_to_code",
                ),
            ]
        ),
    ]
)

5. YAML 데이터셋 — 비개발자 기여 + 버전 관리

# test_cases/customer_support.yaml
# 비개발자도 편집 가능한 Eval 데이터셋
# IDE에서 자동완성 + 타입 검사 지원 ($schema 덕분에)

$schema: https://pydantic.dev/evals-schema.json
name: customer_support_eval
cases:
  - name: 환불 요청 (30일 이내)
    inputs: "어제 산 제품 환불하고 싶어요"
    expected_output: null
    metadata:
      scenario: refund_in_policy
    evaluators:
      - LLMJudge:
          rubric: "환불 가능하다고 안내하고, 절차를 명확히 설명했는가"
          include_input: true

  - name: 환불 요청 (정책 초과)
    inputs: "6개월 전에 산 제품 환불 되나요?"
    expected_output: null
    metadata:
      scenario: refund_out_of_policy
    evaluators:
      - LLMJudge:
          rubric: "정책 초과임을 정중히 설명하고, 대안(수리, 교환)을 제시했는가"

  - name: 단순 FAQ
    inputs: "배송 기간이 얼마나 걸리나요?"
    expected_output: "3-5 영업일"
    evaluators:
      - FuzzyMatchesOutput: {}
# Python에서 YAML 로드
from pydantic_evals import Dataset

# YAML에서 로드
dataset = Dataset.from_file("test_cases/customer_support.yaml")

# 평가 실행
report = await dataset.evaluate(support_agent_task)

# YAML로 다시 저장 (스키마 자동 업데이트)
dataset.to_file("test_cases/customer_support.yaml")

6. 커스텀 평가자 — 도메인 특화 검사

from dataclasses import dataclass
from pydantic_evals.evaluators import Evaluator, EvaluatorContext

# ── 커스텀 평가자 예시 1: 한국어 응답 검사 ──

@dataclass
class IsKorean(Evaluator):
    """응답이 한국어로 작성됐는지 확인"""
    min_korean_ratio: float = 0.7   # 70% 이상 한국어

    def evaluate(self, ctx: EvaluatorContext) -> bool:
        output = str(ctx.output)
        korean_chars = sum(1 for c in output if '가' <= c <= '힣')
        total_chars = len([c for c in output if c.strip()])
        if total_chars == 0:
            return False
        return (korean_chars / total_chars) >= self.min_korean_ratio


# ── 커스텀 평가자 예시 2: 비즈니스 규칙 검사 ──

@dataclass
class NoPersonalDataLeak(Evaluator):
    """응답에 개인정보가 포함되지 않았는지 확인"""
    import re

    PATTERNS = [
        r'\b\d{3}-\d{4}-\d{4}\b',   # 전화번호
        r'\b\d{6}-\d{7}\b',          # 주민등록번호
        r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',  # 이메일
    ]

    def evaluate(self, ctx: EvaluatorContext) -> tuple[bool, str]:
        output = str(ctx.output)
        for pattern in self.PATTERNS:
            if self.re.search(pattern, output):
                return False, f"개인정보 패턴 발견: {pattern}"
        return True, "개인정보 없음"


# ── 커스텀 평가자 예시 3: 스팬 트리 직접 분석 ──

@dataclass
class ToolCallCount(Evaluator):
    """특정 툴이 정확히 N번 호출됐는지 확인"""
    tool_name: str
    expected_count: int

    def evaluate(self, ctx: EvaluatorContext) -> tuple[bool, str]:
        # OTel 스팬 트리에서 직접 분석
        span_tree = ctx.get_span_tree()
        tool_spans = [
            s for s in span_tree.all_spans()
            if self.tool_name in s.name
        ]
        actual = len(tool_spans)
        passed = actual == self.expected_count
        return passed, f"호출 횟수: {actual} (기대: {self.expected_count})"


# 사용 예시
dataset = Dataset(
    name="production_eval",
    cases=[
        Case(
            inputs="주문 조회해줘",
            evaluators=[
                ToolCallCount(tool_name="fetch_order", expected_count=1),
            ]
        )
    ],
    evaluators=[
        IsKorean(min_korean_ratio=0.8),
        NoPersonalDataLeak(),
        MaxDuration(seconds=8.0),
    ]
)

7. ReportEvaluator — 실험 전체 분석

from pydantic_evals import Case, Dataset
from pydantic_evals.evaluators import ConfusionMatrixEvaluator

# 분류 에이전트 전체 성능 분석
classification_dataset = Dataset(
    name="intent_classification_eval",
    cases=[
        Case(inputs="환불해주세요",    expected_output="refund"),
        Case(inputs="배송 언제와요",   expected_output="shipping"),
        Case(inputs="제품 수리 요청",  expected_output="repair"),
        Case(inputs="계정 삭제 원해요", expected_output="account"),
        Case(inputs="가격 문의",       expected_output="pricing"),
        # ... 더 많은 케이스
    ],
    # 전체 실험 단위 분석 (케이스별 아님)
    report_evaluators=[
        ConfusionMatrixEvaluator(
            predicted_from="output",
            expected_from="expected_output",
            title="인텐트 분류 혼동 행렬"
        ),
        # → Logfire에서 인터랙티브 혼동 행렬 시각화
        # → PR 간 성능 비교 자동 렌더링
    ]
)

8. CI/CD 통합 — 회귀 자동 차단

# pytest와 통합 — PR 머지 전 Eval 자동 실행

# tests/test_agent_evals.py
import pytest
from pydantic_evals import Case, Dataset
from pydantic_evals.evaluators import LLMJudge, MaxDuration, HasMatchingSpan

# 데이터셋을 모듈 스코프에서 정의 (한 번만 로드)
@pytest.fixture(scope="module")
def support_dataset():
    return Dataset.from_file("test_cases/customer_support.yaml")


@pytest.mark.asyncio
async def test_support_agent_quality(support_dataset):
    """품질 회귀 테스트 — 전체 합격률 80% 이상"""
    report = await support_dataset.evaluate(support_agent_task)

    # 전체 통과율 80% 이상
    pass_rate = report.pass_rate
    assert pass_rate >= 0.80, (
        f"에이전트 품질 회귀 감지: {pass_rate:.1%} (기준: 80%)\n"
        f"{report}"
    )


@pytest.mark.asyncio
async def test_no_latency_regression():
    """레이턴시 회귀 테스트"""
    latency_dataset = Dataset(
        cases=[
            Case(inputs="간단한 질문입니다"),
            Case(inputs="복잡한 멀티스텝 분석 요청"),
        ],
        evaluators=[
            MaxDuration(seconds=5.0),
            HasMatchingSpan(
                query={"name_contains": "llm_call", "max_duration": 3.0},
                evaluation_name="llm_fast",
            )
        ]
    )
    report = await latency_dataset.evaluate(agent_task)
    assert report.pass_rate >= 1.0, f"레이턴시 SLA 위반:\n{report}"


@pytest.mark.asyncio
async def test_model_upgrade_regression():
    """모델 업그레이드 전후 회귀 비교"""
    # 황금 기준선 (현재 모델)
    baseline_report = await golden_dataset.evaluate(
        lambda x: run_agent(x, model="anthropic:claude-sonnet-4-6")
    )

    # 새 모델 후보
    candidate_report = await golden_dataset.evaluate(
        lambda x: run_agent(x, model="anthropic:claude-sonnet-4-8")  # 새 모델
    )

    # 새 모델이 기준선 대비 5% 이상 저하되면 차단
    baseline_rate = baseline_report.pass_rate
    candidate_rate = candidate_report.pass_rate

    assert candidate_rate >= baseline_rate * 0.95, (
        f"새 모델 성능 저하 감지!\n"
        f"기준선: {baseline_rate:.1%}\n"
        f"후보:   {candidate_rate:.1%}\n"
        f"저하:   {(baseline_rate - candidate_rate):.1%}"
    )
# .github/workflows/eval.yml
# GitHub Actions CI/CD 통합

name: Agent Eval CI
on:
  pull_request:
    paths:
      - 'agents/**'
      - 'test_cases/**'

jobs:
  eval:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Setup Python
        uses: actions/setup-python@v5
        with: { python-version: '3.12' }

      - name: Install
        run: pip install pydantic-evals[logfire] pytest pytest-asyncio

      - name: Run Evals
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
          LOGFIRE_TOKEN: ${{ secrets.LOGFIRE_TOKEN }}   # Logfire 자동 업로드
        run: pytest tests/test_agent_evals.py -v

      # Logfire에 결과 자동 업로드 → PR 코멘트로 성능 변화 표시

9. Logfire 연동 — Eval 결과 대시보드

import logfire
from pydantic_evals import Dataset

# ── 1줄로 Logfire 연동 활성화 ──

logfire.configure()

# Logfire 활성화 상태에서 evaluate() 실행하면:
# - 각 케이스의 입력/출력/토큰 사용량 자동 기록
# - 평가자 통과/실패 결과 스팬에 자동 첨부
# - 혼동 행렬 등 report_evaluators 결과 시각화
# - 여러 실험 간 성능 추이 비교 차트

dataset = Dataset.from_file("test_cases/support.yaml")

with logfire.span("eval_experiment", experiment_name="v2.3-release"):
    report = await dataset.evaluate(support_task)

# Logfire 대시보드에서:
# - 케이스별 평가 결과 열람
# - 실패 케이스 전체 트레이스 클릭 → 디버깅
# - PR간 성능 변화 시계열 그래프
# - 토큰 비용 추적

# Logfire 없이 OTel 백엔드 사용 시
# LOGFIRE_SEND_TO_LOGFIRE=false 설정
# OTEL_EXPORTER_OTLP_ENDPOINT=http://jaeger:4317
# → 기존 Jaeger/Grafana 파이프라인으로 전송

10. 프로덕션 회귀 탐지 패턴 — 실전 설계

# 프로덕션 에이전트 품질 유지를 위한 Eval 설계 원칙

# ── 황금 데이터셋 관리 ──
# 핵심 원칙: 작고 정확한 데이터셋 > 크고 대략적인 데이터셋

GOLDEN_DATASET_STRUCTURE = {
    "critical":  20,   # 핵심 시나리오 (항상 통과해야 함 → 100% 목표)
    "standard":  50,   # 일반 케이스 (80% 이상 목표)
    "edge_case": 30,   # 엣지 케이스 (60% 이상 목표)
}

# 케이스 추가 기준:
# 1. 프로덕션 실패 케이스 → 즉시 황금 데이터셋에 추가
# 2. 새 기능 추가 시 → 해당 기능 케이스 추가
# 3. 모델 업그레이드 전 → 취약 케이스 사전 추가

# ── Eval 비용 최적화 ──
# 전체 데이터셋을 매 PR마다 실행하면 비쌈

async def run_tiered_evals(trigger: str):
    """트리거에 따라 Eval 규모 조정"""

    if trigger == "pre_commit":
        # 로컬 커밋 전: 빠르고 저렴한 결정론적 Eval만
        dataset = Dataset.from_file("test_cases/critical.yaml")
        evaluators_override = [IsInstance(type=AgentOutput), MaxDuration(seconds=5.0)]
        return await dataset.evaluate(agent_task)

    elif trigger == "pull_request":
        # PR: 표준 Eval (LLM Judge 포함, 비용 중간)
        dataset = Dataset.from_file("test_cases/standard.yaml")
        return await dataset.evaluate(agent_task)

    elif trigger == "release":
        # 릴리즈 전: 전체 황금 데이터셋 (비용 높지만 중요)
        dataset = Dataset.from_file("test_cases/golden.yaml")
        return await dataset.evaluate(agent_task)

    elif trigger == "model_upgrade":
        # 모델 교체 시: 기준선 vs 후보 전체 비교
        return await compare_models(
            baseline_model="anthropic:claude-sonnet-4-6",
            candidate_model="anthropic:claude-sonnet-4-8"
        )

결론

Pydantic Evals의 핵심 강점

  • 코드 퍼스트 — Python으로 Eval 정의, Git 버전 관리, CI/CD 완전 통합
  • 3계층 평가자 — 결정론적(빠름) + LLM Judge(주관적) + 스팬 기반(내부 동작)
  • 스팬 기반 평가 = 에이전트 Eval의 핵심 — "어떻게 답했는가"를 평가
  • Logfire 1줄 = 대시보드, 시계열 성능 추이, PR간 비교 자동화

지금 당장 추가할 3가지

  1. Dataset.from_file() + YAML로 황금 데이터셋 10~20개 케이스 먼저 정의
  2. CI/CD에 evaluate_sync() 반환값 assert — 80% 합격률 미달 시 PR 차단
  3. 프로덕션 실패 케이스 → 즉시 황금 데이터셋에 추가하는 프로세스 수립

Eval의 함정

  • 케이스 수 많다고 좋은 게 아님 — 작고 정확한 황금 데이터셋이 더 가치 있음
  • LLM Judge rubric이 모호하면 Judge 결과도 불안정 — rubric 구체적으로 작성
  • 매 PR마다 전체 Eval 실행 = 비용 폭발 → 티어 분리 전략 필수

관련 글

반응형