케이스 생명주기 훅

케이스 생명주기 훅 (Case Lifecycle Hooks)

CaseLifecycle를 사용해 평가 중 케이스별 설정, 컨텍스트 준비, 정리(teardown)를 제어해요.

CaseLifecycle는 케이스 평가의 각 단계에서 훅을 제공해요. Dataset.evaluate](https://pydantic.dev/docs/ai/api/pydantic_evals/dataset/#pydantic_evals.dataset.Dataset.evaluate)에 생명주기 클래스(인스턴스가 아니라)를 전달하면, 각 케이스마다 새 인스턴스가 만들어져요. 그래서 인스턴스 속성이 자연스럽게 케이스별 상태를 담아요.

출처: 문서

본문

평가 흐름

각 케이스는 다음 흐름을 따라요:

  1. setup() -- 작업 실행 전에 호출
  2. 작업 실행
  3. prepare_context() -- 작업 후, 평가자 전에 호출
  4. 평가자 실행
  5. teardown() -- 평가자 완료 후, 또는 케이스가 중단되면 정리 중에 호출

케이스별 설정과 정리

각 케이스가 고유한 환경이 필요할 때 setup()과 teardown()을 사용해요 -- 예를 들어 데이터베이스를 만들거나, 서비스를 시작하거나, 케이스 메타데이터로 구동되는 픽스처를 준비하는 경우예요. 각 케이스마다 새 생명주기 인스턴스가 만들어지므로 인스턴스 속성은 자연스럽게 케이스 범위로 한정돼요:

from pydantic_evals import Case, Dataset
from pydantic_evals.evaluators.context import EvaluatorContext
from pydantic_evals.lifecycle import CaseLifecycle
from pydantic_evals.reporting import ReportCase, ReportCaseFailure


class SetupFromMetadata(CaseLifecycle[str, str, dict]):
    async def setup(self) -> None:
        prefix = (self.case.metadata or {}).get('prefix', '')
        self.prefix = prefix

    async def prepare_context(
        self, ctx: EvaluatorContext[str, str, dict]
    ) -> EvaluatorContext[str, str, dict]:
        ctx.metrics['prefix_length'] = len(self.prefix)
        return ctx

    async def teardown(
        self,
        result: ReportCase[str, str, dict] | ReportCaseFailure[str, str, dict] | None,
    ) -> None:
        pass  # 여기서 리소스 정리


dataset = Dataset(
    name='setup_teardown',
    cases=[
        Case(name='no_prefix', inputs='hello', metadata={'prefix': ''}),
        Case(name='with_prefix', inputs='hello', metadata={'prefix': 'PREFIX:'}),
    ]
)

report = dataset.evaluate_sync(lambda inputs: inputs.upper(), lifecycle=SetupFromMetadata)

metrics = {c.name: c.metrics for c in report.cases}
print(metrics['no_prefix']['prefix_length'])
#> 0
print(metrics['with_prefix']['prefix_length'])
#> 7

케이스 메타데이터가 커스텀 Case 서브클래스나 직렬화 없이 케이스별 동작을 이끌어요.

조건부 정리

teardown() 훅은 전체 결과를 받으므로, 성공한 케이스와 실패한 케이스의 정리를 다르게 할 수 있어요. 예를 들어 실패한 케이스의 환경을 수동 검사용으로 유지할 수 있어요. 케이스가 보고서 결과를 만들기 전에 평가가 중단되면 result는 None이에요. 정리가 결과에 의존할 때는 그 경우를 처리해요:

from pydantic_evals import Case, Dataset
from pydantic_evals.lifecycle import CaseLifecycle
from pydantic_evals.reporting import ReportCase, ReportCaseFailure

cleaned_up: list[str] = []


class ConditionalCleanup(CaseLifecycle[str, str, dict]):
    async def setup(self) -> None:
        self.resource_id = self.case.name

    async def teardown(
        self,
        result: ReportCase[str, str, dict] | ReportCaseFailure[str, str, dict] | None,
    ) -> None:
        keep_on_failure = (self.case.metadata or {}).get('keep_on_failure', False)
        if result is None:
            # 비정상 종료
            cleaned_up.append(self.resource_id)
        elif isinstance(result, ReportCaseFailure) and keep_on_failure:
            # 케이스 실패
            pass  # 검사용 리소스 유지
        else:
            # 케이스 성공
            cleaned_up.append(self.resource_id)


dataset = Dataset(
    name='conditional_cleanup',
    cases=[
        Case(name='success_case', inputs='hello', metadata={'keep_on_failure': True}),
        Case(name='failure_case', inputs='fail', metadata={'keep_on_failure': True}),
    ]
)


def task(inputs: str) -> str:
    if inputs == 'fail':
        raise ValueError('intentional failure')
    return inputs.upper()


report = dataset.evaluate_sync(task, max_concurrency=1, lifecycle=ConditionalCleanup)

print(cleaned_up)
#> ['success_case']

평가자 컨텍스트 준비

prepare_context() 훅은 작업이 완료된 후, 평가자가 컨텍스트를 보기 전에 실행돼요. 작업 출력, 스팬 트리, 또는 다른 상태에 기반해 메트릭이나 속성을 추가하는 데 사용할 수 있어요 -- 예를 들어 계측된 스팬에서 메트릭을 도출하거나(툴 호출 횟수, API 지연 같은), setup()에서 준비한 외부 리소스에서 값을 계산하는 경우요:

from dataclasses import dataclass

from pydantic_evals import Case, Dataset
from pydantic_evals.evaluators import Evaluator, EvaluatorContext
from pydantic_evals.lifecycle import CaseLifecycle


class EnrichMetrics(CaseLifecycle):
    async def prepare_context(self, ctx: EvaluatorContext) -> EvaluatorContext:
        ctx.metrics['output_length'] = len(str(ctx.output))
        return ctx


@dataclass
class CheckLength(Evaluator):
    max_length: int = 50

    def evaluate(self, ctx: EvaluatorContext) -> bool:
        return ctx.metrics.get('output_length', 0) <= self.max_length


dataset = Dataset(
    name='context_enrichment',
    cases=[Case(name='short', inputs='hi'), Case(name='long', inputs='hello world')],
    evaluators=[CheckLength()],
)

report = dataset.evaluate_sync(lambda inputs: inputs.upper(), lifecycle=EnrichMetrics)

for case in report.cases:
    print(f'{case.name}: output_length={case.metrics["output_length"]}')
    #> short: output_length=2
    #> long: output_length=11

타입 매개변수

CaseLifecycle는 Case와 같은 세 타입 매개변수 InputsT, OutputT, MetadataT에 대해 제네릭이에요. 셋 다 기본값이 Any라서, 훅이 타입별 접근을 필요로 하지 않을 때는 생략할 수 있어요:

from pydantic_evals import Case, Dataset
from pydantic_evals.evaluators.context import EvaluatorContext
from pydantic_evals.lifecycle import CaseLifecycle


# 어떤 데이터셋이든 동작 -- 타입 매개변수 불필요
class GenericMetricEnricher(CaseLifecycle):
    async def prepare_context(self, ctx: EvaluatorContext) -> EvaluatorContext:
        ctx.metrics['custom'] = 42
        return ctx


dataset = Dataset(name='generic_lifecycle', cases=[Case(inputs='test')])
report = dataset.evaluate_sync(lambda inputs: inputs, lifecycle=GenericMetricEnricher)

print(report.cases[0].metrics['custom'])
#> 42

다음 단계

더 알아보기 (Learn more)