온라인 평가

온라인 평가 (Online Evaluation)

온라인 평가를 사용하면 프로덕션(또는 스테이징) 함수에 평가자를 붙여서, 모든 호출(또는 샘플링된 일부)이 백그라운드에서 자동으로 평가되게 할 수 있어요. Dataset.evaluate()와 함께 쓰이던 동일한 Evaluator 클래스가 여기서도 동작해요. 차이점은 단지 연결 방식이에요.

출처: 문서

본문

온라인 평가는 언제 쓸까요?

온라인 평가는 다음을 원할 때 유용해요:

• 프로덕션 품질 모니터링: 루브릭에 대해 LLM 출력을 지속적으로 채점
• 회귀 포착: 배포에 걸친 에이전트 동작 저하 감지
• 평가 데이터 수집: 오프라인 분석을 위해 실제 트래픽에서 데이터셋 구축
• 비용 제어: 비싼 LLM judge를 트래픽의 일부에서만 샘플링하고, 저렴한 검사는 모든 것에 실행

배포 전 정제된 데이터셋으로 테스트하려면 Dataset.evaluate()를 사용하는 오프라인 평가를 대신 사용해요.

빠른 시작

evaluate() 데코레이터는 어떤 함수에든 평가자를 붙여요. 평가자는 호출자를 차단하지 않고 백그라운드에서 실행되며, 결과는 OpenTelemetry 이벤트로 내보내져요:

from dataclasses import dataclass

from pydantic_evals.evaluators import Evaluator, EvaluatorContext
from pydantic_evals.online import evaluate


@dataclass
class OutputNotEmpty(Evaluator):
    def evaluate(self, ctx: EvaluatorContext) -> bool:
        return bool(ctx.output)


@evaluate(OutputNotEmpty())
async def summarize(text: str) -> str:
    return f'Summary of: {text}'

애플리케이션 시작부의 다른 곳에서 OTel export(예: logfire.configure())를 연결해서 내보내진 gen_ai.evaluation.result 이벤트가 백엔드에 도달하게 해요. Pydantic Logfire를 사용할 때 이 이벤트들은 Live Evaluations 뷰에 표시되는데, 여기서 대상을 기준으로 결과를 탐색하고, 원래 트레이스로 드릴다운하고, 시간 창에 걸친 점수를 볼 수 있어요.

각 데코레이션된 호출은 OTel GenAI 평가 semconv를 따라 평가자 결과마다 하나의 gen_ai.evaluation.result OTel 이벤트를 내보내요. 이것은 오프라인 평가가 logfire.span을 통해 OTel 스팬을 내보내는 방식과 대칭이에요. 프로세스에 어떤 OTel SDK든 구성되어 있으면(logfire.configure()](https://pydantic.dev/docs/ai/integrations/logfire/#using-logfire), OTel SDK 직접, 또는 벤더 계측을 통해) 이벤트는 백엔드로 흘러가요. 그렇지 않으면 내보내기는 저렴한 no-op이에요.

결과를 Python 코드에서 추가로 처리하려면 -- 알림, 맞춤 집계, 메모리 내 테스트 캡처, 또는 비-OTel 대상용으로 -- sink를 등록해요. Sink는 OTel 이벤트 내보내기 외에도 실행돼요.

모듈 수준 configure()와 evaluate() 함수는 전역 OnlineEvalConfig에 위임해요. 여러 구성이나 격리된 설정이 필요하면 직접 구성 인스턴스를 만들 수 있어요(아래 OnlineEvalConfig 참고).

대상 (Target)

각 데코레이션된 함수(또는 에이전트)는 target으로 태그된 결과를 내보내요. target은 다운스트림 sink와 대시보드에서 결과를 그룹짓는 이름이에요. 기본적으로 target은 데코레이션된 함수의 __name__이지만, target=...으로 덮어쓸 수 있어요:

from dataclasses import dataclass

from pydantic_evals.evaluators import Evaluator, EvaluatorContext
from pydantic_evals.online import evaluate


@dataclass
class OutputNotEmpty(Evaluator):
    def evaluate(self, ctx: EvaluatorContext) -> bool:
        return bool(ctx.output)


# 기본: target='summarize' (함수 이름)
@evaluate(OutputNotEmpty())
async def summarize(text: str) -> str: ...


# 덮어쓰기: 친숙한 이름 사용
@evaluate(OutputNotEmpty(), target='customer_support')
async def run_agent(prompt: str) -> str: ...

target 이름은 모든 submit() 호출에서 평문 str로 sink에 제공돼요. 단일 sink 인스턴스는 여러 데코레이션된 함수나 에이전트를 처리해요.

에이전트 기능의 경우 target 이름은 에이전트 자체의 name 속성에서 가져와요(Agent 통합 참고). 에이전트 여부로 분류하거나 라우팅하려면 구성에 메타데이터를 추가해요(예: metadata={'kind': 'agent'}).

핵심 개념

OnlineEvaluator

평가자마다 서로 다른 설정이 필요해요. 저렴한 휴리스틱은 트래픽의 100%에서 실행할 수 있고, 비싼 LLM judge는 1%에서 실행할 수 있어요. OnlineEvaluator는 Evaluator를 평가자별 구성으로 감싸요:

from dataclasses import dataclass

from pydantic_evals.evaluators import Evaluator, EvaluatorContext, LLMJudge
from pydantic_evals.online import OnlineEvaluator


@dataclass
class IsHelpful(Evaluator):
    def evaluate(self, ctx: EvaluatorContext) -> bool:
        return len(str(ctx.output)) > 10


# 저렴한 평가자: 모든 요청에서 실행
always_check = OnlineEvaluator(evaluator=IsHelpful(), sample_rate=1.0)

# 비싼 평가자: 요청의 1%에서 실행, 동시성 제한
rare_check = OnlineEvaluator(
    evaluator=LLMJudge(rubric='Is the response helpful?'),
    sample_rate=0.01,
    max_concurrency=5,
)

evaluate() 데코레이터에 맨 Evaluator를 전달하면 구성의 기본 샘플 비율을 가진 OnlineEvaluator로 자동 감싸져요.

OnlineEvalConfig

OnlineEvalConfig는 교차 평가자 기본값(샘플 비율, 메타데이터, 선택적 추가 sink, OTel 내보내기 토글)을 담아요. 전역 기본 인스턴스가 있고, 다른 구성마다 커스텀 인스턴스를 만들 수 있어요:

import asyncio
from collections.abc import Sequence
from dataclasses import dataclass

from pydantic_evals.evaluators import (
    EvaluationResult,
    Evaluator,
    EvaluatorContext,
    EvaluatorFailure,
)
from pydantic_evals.online import OnlineEvalConfig, wait_for_evaluations


@dataclass
class IsNonEmpty(Evaluator):
    def evaluate(self, ctx: EvaluatorContext) -> bool:
        return bool(ctx.output)


results_log: list[str] = []


async def log_sink(
    results: Sequence[EvaluationResult],
    failures: Sequence[EvaluatorFailure],
    context: EvaluatorContext,
) -> None:
    for r in results:
        results_log.append(f'{r.name}={r.value}')


my_eval = OnlineEvalConfig(
    default_sink=log_sink,
    default_sample_rate=1.0,
    metadata={'service': 'my-app'},
)


@my_eval.evaluate(IsNonEmpty())
async def my_function(query: str) -> str:
    return f'Answer to: {query}'


async def main():
    result = await my_function('What is 2+2?')
    print(result)
    #> Answer to: What is 2+2?
    await wait_for_evaluations()
    print(results_log)
    #> ['IsNonEmpty=True']


asyncio.run(main())

Sink

OTel 이벤트 내보내기는 온라인 평가의 기본 관측 가능성 표면이에요(기본 OTel 이벤트 내보내기 참고). Sink는 Python 코드에서의 추가 처리를 위한 것이에요 -- 메모리 내 테스트 캡처, 알림, 비-OTel 대상으로의 fan-out, 또는 맞춤 집계. EvaluationSink가 프로토콜이며, 단일 구성에 여러 sink를 등록할 수 있어요.

내장 CallbackSink는 결과, 실패, 컨텍스트를 받는 모든 콜러블(동기 또는 비동기)을 감싸요. sink가 기대되는 곳에 맨 콜러블을 전달할 수도 있어요 -- CallbackSink로 자동 감싸져요.

커스텀 sink를 위해 EvaluationSink 프로토콜을 구현해요. 각 submit() 호출은 주어진 함수 호출에 대해 실행된 하나 이상의 평가자로부터의 결과, 실패, 컨텍스트, 스팬 참조, target을 묶은 SinkPayload를 받아요:

from pydantic_evals.online import SinkPayload


class PrintSink:
    """Prints evaluation results to stdout."""

    async def submit(self, payload: SinkPayload) -> None:
        for r in payload.results:
            version = f' ({r.evaluator_version})' if r.evaluator_version else ''
            print(f'  [{payload.target}] {r.name}{version}: {r.value}')
        for f in payload.failures:
            version = f' ({f.evaluator_version})' if f.evaluator_version else ''
            print(f'  [{payload.target}] FAILED {f.name}{version}: {f.error_message}')

payload.results와 payload.failures는 단일 함수 호출의 하나 이상의 평가자를 다룰 수 있어요. 여러 평가자가 sink를 공유하면 그 결과가 단일 submit() 호출로 배치돼요. 각 결과는 자체 귀속 정보(name, EvaluationResult와 EvaluatorFailure의 evaluator_version, 그리고 원본 spec)를 갖춰서 sink가 다운스트림에서 분리할 수 있어요. 평가자 버전 관리 참고. payload.target은 평가 중인 함수 또는 에이전트를 식별해요(Target 참고).

기본 OTel 이벤트 내보내기

디스패치된 각 평가자는 EvaluationResult 또는 EvaluatorFailure마다 하나의 gen_ai.evaluation.result OTel 로그 이벤트를 무조건 내보내요 -- sink 등록이 필요 없어요. 이벤트는 그것을 만든 스팬에 부모로 붙어서 트레이스에서 원래 함수 호출 아래에 중첩되어 나타나요. 프로세스에 OTel SDK가 구성되지 않았으면 내보내기는 저렴한 no-op이에요.

각 이벤트는 event.name = 'gen_ai.evaluation.result'와 짧은 사람이 읽을 수 있는 본문(예: evaluation: accuracy=0.87, 또는 evaluation: accuracy failed: <error>)을 가져요. 내보내기는 OpenTelemetry GenAI 평가 semconv를 따르며 다음 속성을 가져요:

• gen_ai.evaluation.name -- evaluate()가 스칼라를 반환하면 평가자 클래스 이름, {'accuracy': ..., 'score': ...}를 반환하면 매핑 키. 출처: EvaluationResult.name / EvaluatorFailure.name.
• gen_ai.evaluation.score.value -- bool(True→1.0, False→0.0) 및 숫자 반환에 대해 채워짐. str 반환에는 생략됨.
• gen_ai.evaluation.score.label -- bool(True→'pass', False→'fail') 및 str 반환에 대해 채워짐(라벨로 직접 사용). 숫자 반환에는 생략됨.
• gen_ai.evaluation.explanation -- 성공 시 EvaluationResult.reason, 실패 시 EvaluatorFailure.error_message. 없으면 생략됨. 커스텀 평가자 내부에서 EvaluationResult를 만들 때 reason=...로 설정해요.
• error.type (실패 이벤트만) -- 잡힌 예외로부터 실패가 만들어졌을 때 예외 클래스 이름(예: 'ValueError'). 그것 없이 만들어진 EvaluatorFailure 인스턴스는 'pydantic_evals.EvaluatorFailure'로 대체됨. 성공 평가에는 없음. 출처: EvaluatorFailure.error_type.
• gen_ai.evaluation.target -- @evaluate(target=...) 또는 에이전트 name. Target 참고.
• gen_ai.evaluation.evaluator.version -- Evaluator.evaluator_version 클래스 속성. 클래스가 설정하지 않으면 생략됨. 평가자 버전 관리 참고.
• gen_ai.evaluation.evaluator.source -- JSON 직렬화된 EvaluatorSpec으로 평가자 클래스와 생성자 인자를 식별해요. 다운스트림 쿼리가 name만으로 의존하지 않고 평가자 정체성으로 그룹지을 수 있게 해줘요(서로 다른 두 LLMJudge(rubric=...) 인스턴스는 이름을 공유하지만 source는 달라요).

OTel baggage 항목(있으면)도 구성의 include_baggage로 설정 가능한 속성으로 각 이벤트에 붙어요. 위의 gen_ai.*와 error.type 속성은 baggage와 충돌 시 항상 우선해요.

예를 들어 위의 OutputNotEmpty 평가자를 @evaluate(OutputNotEmpty(), target='customer_support')로 장식하고 주어진 호출에서 True를 반환하면 다음으로 이벤트를 하나 내보내요:

• gen_ai.evaluation.name = 'OutputNotEmpty'
• gen_ai.evaluation.score.value = 1.0
• gen_ai.evaluation.score.label = 'pass'
• gen_ai.evaluation.target = 'customer_support'
• gen_ai.evaluation.evaluator.source = '{"name":"OutputNotEmpty","arguments":null}'

생성자 인자가 있는 평가자는 그 인자를 source에 렌더링해요. 예를 들어 LLMJudge(rubric='Is the response helpful?')는 gen_ai.evaluation.evaluator.source = '{"name":"LLMJudge","arguments":["Is the response helpful?"]}'를 내보내서, 루브릭이 다른 두 LLMJudge 인스턴스가 다운스트림에서도 구분 가능하게 해줘요.

gen_ai.evaluation.evaluator.* 아래의 속성은 pydantic-evals 확장이에요. 현재 OTel GenAI semconv에는 없고, 미래 semconv 추가와 맞추기 위해 이름이 바뀔 수 있어요.

기본 내보내기를 비활성화하려면(예: 커스텀 sink에만 assert하려는 테스트 하네스에서) 구성에 emit_otel_events=False를 설정해요:

from pydantic_evals.online import OnlineEvalConfig

config = OnlineEvalConfig(emit_otel_events=False)

평가자 버전 관리

Evaluator 서브클래스에서 get_evaluator_version을 재정의하면 그것이 내보내는 모든 결과에 버전 문자열을 찍어요. 이는 내보내진 이벤트에서 gen_ai.evaluation.evaluator.version으로, 각 EvaluationResult와 EvaluatorFailure에서 evaluator_version으로 표면화돼요. 이렇게 하면 추세선과 대시보드가 히스토리 행을 삭제하지 않고 퇴역한 평가자 버전이 만든 결과를 걸러낼 수 있어요. LLM judge의 프롬프트를 바꾸거나 이전 점수를 무효화하는 방식으로 휴리스틱을 재작업할 때 유용해요:

from dataclasses import dataclass

from pydantic_evals.evaluators import Evaluator, EvaluatorContext


@dataclass
class ToneCheck(Evaluator):
    def evaluate(self, ctx: EvaluatorContext) -> str:
        return 'neutral'

    def get_evaluator_version(self) -> str | None:
        return 'v2'  # 프롬프트 재작성 후 올림

버전은 평가자가 만드는 모든 결과에 적용돼요(그래서 평가자 클래스 하나는 버전 하나에 매핑되고, 평가자가 명명된 결과의 매핑을 반환해도 동일해요).

샘플링

품질 모니터링과 비용의 균형을 맞추기 위해 평가자별 샘플 비율로 평가 빈도를 제어해요.

참고

샘플링은 데코레이션된 함수가 실행되기 전에 결정돼요. 주어진 호출에 대해 샘플링된 평가자가 없으면 함수는 추가 계측 오버헤드 없이 실행돼요(logfire 스팬이나 스팬 트리 캡처가 없어요).

정적 샘플 비율

0.0과 1.0 사이의 sample_rate는 각 호출을 평가할 확률을 설정해요:

from dataclasses import dataclass

from pydantic_evals.evaluators import Evaluator, EvaluatorContext
from pydantic_evals.online import OnlineEvaluator


@dataclass
class QuickCheck(Evaluator):
    def evaluate(self, ctx: EvaluatorContext) -> bool:
        return bool(ctx.output)


# 모든 요청에서 실행
always = OnlineEvaluator(evaluator=QuickCheck(), sample_rate=1.0)

# 요청의 10%에서 실행
sometimes = OnlineEvaluator(evaluator=QuickCheck(), sample_rate=0.1)

# 절대 실행 안 함 (사실상 비활성화)
never = OnlineEvaluator(evaluator=QuickCheck(), sample_rate=0.0)

동적 샘플 비율

콜러블을 전달해 런타임 구성 가능 또는 입력 의존적인 샘플링을 가능하게 해요. 콜러블은 평가자 인스턴스, 함수 입력, 구성 메타데이터, 호출별 임의 시드를 담은 SamplingContext를 받고 float(확률) 또는 bool(항상/절대)을 반환해요:

from dataclasses import dataclass

from pydantic_evals.evaluators import Evaluator, EvaluatorContext
from pydantic_evals.online import OnlineEvaluator, SamplingContext


def get_current_rate(ctx: SamplingContext) -> float:
    return 0.5


@dataclass
class QuickCheck(Evaluator):
    def evaluate(self, ctx: EvaluatorContext) -> bool:
        return bool(ctx.output)


dynamic = OnlineEvaluator(evaluator=QuickCheck(), sample_rate=get_current_rate)

이것은 기능 플래그, 관리 변수, 또는 구성 시스템과의 통합을 가능하게 해요. 예를 들어 get_current_rate를 런타임에 원격 구성 서비스(예: Logfire 관리 변수)를 읽는 함수로 교체해서, 애플리케이션을 재배포하지 않고 확률을 바꿀 수 있어요.

SamplingContext를 사용해 함수 입력에 기반해 샘플링 결정을 내릴 수도 있어요:

from dataclasses import dataclass

from pydantic_evals.evaluators import Evaluator, EvaluatorContext
from pydantic_evals.online import OnlineEvaluator, SamplingContext


def sample_long_inputs(ctx: SamplingContext) -> bool:
    """Only evaluate calls with long input text."""
    return len(str(ctx.inputs.get('text', ''))) > 100


@dataclass
class QualityCheck(Evaluator):
    def evaluate(self, ctx: EvaluatorContext) -> bool:
        return len(str(ctx.output)) > 10


expensive = OnlineEvaluator(evaluator=QualityCheck(), sample_rate=sample_long_inputs)

상관 샘플링

기본적으로 각 평가자는 독립적으로 샘플링해요. 각각 10%인 평가자 셋이 있으면 호출의 약 27%(1 − 0.9³)가 평가 오버헤드를 겪어요. 같은 10%의 호출이 모든 평가자를 실행하기를 선호한다면 sampling_mode='correlated'를 설정해요:

from dataclasses import dataclass

from pydantic_evals.evaluators import Evaluator, EvaluatorContext
from pydantic_evals.online import OnlineEvalConfig, OnlineEvaluator


@dataclass
class CheckA(Evaluator):
    def evaluate(self, ctx: EvaluatorContext) -> bool:
        return True


@dataclass
class CheckB(Evaluator):
    def evaluate(self, ctx: EvaluatorContext) -> bool:
        return True


config = OnlineEvalConfig(
    default_sink=lambda results, failures, ctx: None,
    sampling_mode='correlated',
)

# 둘 다 같은 ~10%의 호출에서 실행
check_a = OnlineEvaluator(evaluator=CheckA(), sample_rate=0.1)
check_b = OnlineEvaluator(evaluator=CheckB(), sample_rate=0.1)

상관 모드에서는 단일 임의 call_seed(0.0과 1.0 사이 균등 분포)가 함수 호출당 생성되어 모든 평가자에 공유돼요. 평가자는 call_seed < sample_rate일 때 실행되므로, 낮은 비율 평가자의 호출은 항상 높은 비율 평가자의 부분 집합이 되고, 총 오버헤드 확률은 누적되지 않고 최대 비율과 같아져요.

call_seed는 모드와 무관하게 자신만의 상관 로직을 구현하고 싶은 커스텀 sample_rate 콜러블을 위해 SamplingContext에서도 사용할 수 있어요.

평가 비활성화

disable_evaluation()을 사용해 스코프 안에서 모든 온라인 평가를 억제해요. 테스트에서 유용할 수 있어요:

import asyncio
from collections.abc import Sequence
from dataclasses import dataclass

from pydantic_evals.evaluators import (
    EvaluationResult,
    Evaluator,
    EvaluatorContext,
    EvaluatorFailure,
)
from pydantic_evals.online import (
    OnlineEvalConfig,
    disable_evaluation,
    wait_for_evaluations,
)


@dataclass
class OutputCheck(Evaluator):
    def evaluate(self, ctx: EvaluatorContext) -> bool:
        return bool(ctx.output)


results_log: list[str] = []


async def log_sink(
    results: Sequence[EvaluationResult],
    failures: Sequence[EvaluatorFailure],
    context: EvaluatorContext,
) -> None:
    for r in results:
        results_log.append(f'{r.name}={r.value}')


config = OnlineEvalConfig(default_sink=log_sink)


@config.evaluate(OutputCheck())
async def my_function(x: int) -> int:
    return x * 2


async def main():
    # 이 블록 안에서는 평가자가 억제됨
    with disable_evaluation():
        result = await my_function(21)
        print(result)
        #> 42

    await wait_for_evaluations()
    print(f'evaluations run: {len(results_log)}')
    #> evaluations run: 0

    # 블록 밖에서 평가자 재개
    await my_function(21)
    await wait_for_evaluations()
    print(f'evaluations run: {len(results_log)}')
    #> evaluations run: 1


asyncio.run(main())

조건부 평가

비용 제어를 위해 단일 커스텀 평가자 안에서 값비싼 평가 로직을 조건부로 실행할 수 있어요. 실행된 검사에 대해서만 키를 포함한 매핑을 반환해요. 수행하고 싶지 않은 검사는 결과에서 그냥 생략할 수 있어요:

import asyncio
from collections.abc import Sequence
from dataclasses import dataclass

from pydantic_evals.evaluators import (
    EvaluationResult,
    Evaluator,
    EvaluatorContext,
    EvaluatorFailure,
)
from pydantic_evals.online import (
    OnlineEvalConfig,
    wait_for_evaluations,
)

results_log: list[str] = []


async def log_sink(
    results: Sequence[EvaluationResult],
    failures: Sequence[EvaluatorFailure],
    context: EvaluatorContext,
) -> None:
    for r in results:
        results_log.append(f'{r.name}={r.value}')


@dataclass
class ConditionalAnalysis(Evaluator):
    """Runs a cheap check on every call, and an expensive check only on long outputs."""

    def evaluate(self, ctx: EvaluatorContext) -> dict[str, float | bool]:
        output = str(ctx.output)
        results: dict[str, float | bool] = {
            'has_content': len(output) > 0,
        }
        # 비싼 분석은 긴 출력에서만 실행
        if len(output) > 20:
            # 다음 줄이 비싸다고 가정..
            results['detail_score'] = len(output) / 100.0
        return results


config = OnlineEvalConfig(default_sink=log_sink)


@config.evaluate(ConditionalAnalysis())
async def generate(prompt: str) -> str:
    return f'Response to: {prompt}'


async def main():
    await generate('hi')  # 짧은 출력 -- 저렴한 검사만 실행
    await wait_for_evaluations()
    print(results_log)
    #> ['has_content=True']

    results_log.clear()
    await generate('tell me a long story about dragons')  # 긴 출력 -- 두 검사 모두 실행
    await wait_for_evaluations()
    print(sorted(results_log))
    #> ['detail_score=0.47', 'has_content=True']


asyncio.run(main())

이 패턴은 저렴한 검사와 비싼 검사를 한 평가자에 결합해, 조건이 충족되지 않을 때 불필요한 작업을 피하게 해줘요.

동기 함수 지원

evaluate() 데코레이터는 비동기와 동기 함수 모두에서 동작해요:

import asyncio
from collections.abc import Sequence
from dataclasses import dataclass

from pydantic_evals.evaluators import (
    EvaluationResult,
    Evaluator,
    EvaluatorContext,
    EvaluatorFailure,
)
from pydantic_evals.online import OnlineEvalConfig, wait_for_evaluations

results_log: list[str] = []


async def log_sink(
    results: Sequence[EvaluationResult],
    failures: Sequence[EvaluatorFailure],
    context: EvaluatorContext,
) -> None:
    for r in results:
        results_log.append(f'{r.name}={r.value}')


@dataclass
class OutputCheck(Evaluator):
    def evaluate(self, ctx: EvaluatorContext) -> bool:
        return bool(ctx.output)


config = OnlineEvalConfig(default_sink=log_sink)


@config.evaluate(OutputCheck())
def process(text: str) -> str:
    return text.upper()


async def main():
    # 동기 데코레이션 함수는 비동기 컨텍스트에서도 동작
    result = process('hello')
    print(result)
    #> HELLO

    await wait_for_evaluations()
    print(results_log)
    #> ['OutputCheck=True']


asyncio.run(main())

동기 데코레이션 함수는 동기와 비동기 컨텍스트 모두에서 동작해요. 실행 중인 이벤트 루프가 있으면 평가자는 그 루프의 백그라운드 작업으로 디스패치돼요. 그렇지 않으면 자체 이벤트 루프를 가진 백그라운드 스레드가 생성돼요.

평가자별 Sink 덮어쓰기

개별 평가자는 구성의 기본 sink를 덮어쓸 수 있어요. 서로 다른 평가자가 다른 대상으로 결과를 보내야 할 때 유용해요:

import asyncio
from collections.abc import Sequence
from dataclasses import dataclass

from pydantic_evals.evaluators import (
    EvaluationResult,
    Evaluator,
    EvaluatorContext,
    EvaluatorFailure,
)
from pydantic_evals.online import (
    OnlineEvalConfig,
    OnlineEvaluator,
    wait_for_evaluations,
)

default_log: list[str] = []
special_log: list[str] = []


async def default_sink(
    results: Sequence[EvaluationResult],
    failures: Sequence[EvaluatorFailure],
    context: EvaluatorContext,
) -> None:
    for r in results:
        default_log.append(r.name)


async def special_sink(
    results: Sequence[EvaluationResult],
    failures: Sequence[EvaluatorFailure],
    context: EvaluatorContext,
) -> None:
    for r in results:
        special_log.append(r.name)


@dataclass
class FastCheck(Evaluator):
    def evaluate(self, ctx: EvaluatorContext) -> bool:
        return True


@dataclass
class ImportantCheck(Evaluator):
    def evaluate(self, ctx: EvaluatorContext) -> bool:
        return True


config = OnlineEvalConfig(default_sink=default_sink)


@config.evaluate(
    FastCheck(),  # 기본 sink 사용
    OnlineEvaluator(evaluator=ImportantCheck(), sink=special_sink),  # 특수 sink 사용
)
async def my_function(x: int) -> int:
    return x


async def main():
    await my_function(42)
    await wait_for_evaluations()

    print(f'default: {default_log}')
    #> default: ['FastCheck']
    print(f'special: {special_log}')
    #> special: ['ImportantCheck']


asyncio.run(main())

저장된 데이터에서 평가자 재실행

온라인 평가의 핵심 기능은 원래 함수를 재실행하지 않고 평가자를 재실행하는 거예요. 업데이트된 루브릭으로 과거 데이터를 평가하거나 기존 트레이스에 추가 평가자를 실행할 때 유용해요.

run_evaluators

run_evaluators()는 EvaluatorContext에 대해 평가자 목록을 실행하고 결과를 반환해요:

import asyncio
from dataclasses import dataclass

from pydantic_evals.evaluators import Evaluator, EvaluatorContext
from pydantic_evals.online import run_evaluators
from pydantic_evals.otel.span_tree import SpanTree


@dataclass
class LengthCheck(Evaluator):
    min_length: int = 10

    def evaluate(self, ctx: EvaluatorContext) -> bool:
        return len(str(ctx.output)) >= self.min_length


@dataclass
class HasKeyword(Evaluator):
    keyword: str = 'hello'

    def evaluate(self, ctx: EvaluatorContext) -> bool:
        return self.keyword in str(ctx.output).lower()


async def main():
    # 컨텍스트를 수동으로 구축 (실제로는 저장된 데이터에서 얻음)
    # 보통 EvaluatorContext는 수동으로 구성되지 않음 --
    # @evaluate 데코레이터나 OnlineEvaluation 기능이 자동으로 만들거나,
    # EvaluatorContextSource에서 만듦 (아래 참고).
    ctx = EvaluatorContext(
        name='example',
        inputs={'query': 'greet the user'},
        output='Hello! How can I help you today?',
        expected_output=None,
        metadata=None,
        duration=0.5,
        _span_tree=SpanTree(),
        attributes={},
        metrics={},
    )

    results, failures = await run_evaluators(
        [LengthCheck(min_length=10), HasKeyword(keyword='hello')],
        ctx,
    )

    for r in results:
        print(f'{r.name}: {r.value}')
        #> LengthCheck: True
        #> HasKeyword: True
    print(f'failures: {len(failures)}')
    #> failures: 0


asyncio.run(main())

EvaluatorContextSource 프로토콜

외부 저장소(예: Pydantic Logfire)에서 컨텍스트 데이터를 가져오려면 EvaluatorContextSource 프로토콜을 구현해요. 그것은 저장된 데이터에서 EvaluatorContext 객체를 반환하는 fetch()와 fetch_many() 메서드를 정의해요:

import asyncio
from collections.abc import Sequence

from pydantic_evals.evaluators import EvaluatorContext
from pydantic_evals.online import SpanReference
from pydantic_evals.otel.span_tree import SpanTree


class MyContextSource:
    """Example source that fetches context from a hypothetical store."""

    def __init__(self, store: dict[str, EvaluatorContext]) -> None:
        self._store = store

    async def fetch(self, span: SpanReference) -> EvaluatorContext:
        return self._store[span.span_id]

    async def fetch_many(self, spans: Sequence[SpanReference]) -> list[EvaluatorContext]:
        return [self._store[s.span_id] for s in spans]


def _make_context(
    *,
    inputs: object = None,
    output: object = None,
    metadata: object = None,
    duration: float = 0.0,
) -> EvaluatorContext:
    # 보통 EvaluatorContext는 수동으로 구성되지 않음 --
    # @evaluate 데코레이터나 OnlineEvaluation 기능이 자동으로 만듦.
    return EvaluatorContext(
        name=None,
        inputs=inputs,
        output=output,
        expected_output=None,
        metadata=metadata,
        duration=duration,
        _span_tree=SpanTree(),
        attributes={},
        metrics={},
    )


async def main():
    source = MyContextSource({
        'span_abc': _make_context(
            inputs={'query': 'What is AI?'},
            output='AI is artificial intelligence.',
            metadata={'model': 'gpt-4o'},
            duration=1.2,
        ),
        'span_def': _make_context(
            inputs={'query': 'What is ML?'},
            output='ML is machine learning.',
            metadata={'model': 'gpt-4o'},
            duration=0.8,
        ),
    })

    # 단일 컨텍스트 가져오기
    ctx = await source.fetch(SpanReference(trace_id='t1', span_id='span_abc'))
    print(f'inputs: {ctx.inputs}')
    #> inputs: {'query': 'What is AI?'}
    print(f'output: {ctx.output}')
    #> output: AI is artificial intelligence.

    # 배치로 여러 컨텍스트 가져오기
    spans = [
        SpanReference(trace_id='t1', span_id='span_abc'),
        SpanReference(trace_id='t1', span_id='span_def'),
    ]
    contexts = await source.fetch_many(spans)
    print(f'batch size: {len(contexts)}')
    #> batch size: 2


asyncio.run(main())

EvaluatorContext 직렬화

위 같은 저장소를 채우려면 EvaluatorContext를 JSON으로 직렬화(그리고 다시 읽기)해야 해요. EvaluatorContext는 Pydantic 직렬화 가능한 dataclass이므로 TypeAdapter가 양방향을 처리해요. 컨텍스트가 담는 구체적인 inputs, output, metadata 타입에 바인딩해서 그 필드가 충실히 재구성되게 해요:

from pydantic import TypeAdapter

from pydantic_evals.evaluators import EvaluatorContext
from pydantic_evals.otel.span_tree import SpanTree

context_adapter = TypeAdapter(EvaluatorContext[dict[str, str], str, dict[str, str]])

ctx = EvaluatorContext[dict[str, str], str, dict[str, str]](
    name='span_abc',
    inputs={'query': 'What is AI?'},
    output='AI is artificial intelligence.',
    expected_output=None,
    metadata={'model': 'gpt-4o'},
    duration=1.2,
    _span_tree=SpanTree(),
    attributes={},
    metrics={},
)

json_bytes = context_adapter.dump_json(ctx)
restored = context_adapter.validate_json(json_bytes)
print(restored.output)
#> AI is artificial intelligence.

타입 매개변수 바인딩

맨 TypeAdapter(EvaluatorContext)는 inputs, output, metadata, expected_output의 Any 기본값으로 동작해요. 그것은 구체적인 Python 타입을 재구성하지 않으므로, 비-기본형 값은 평문 dict/list로 왕복되고, JSON 직렬화할 수 없는 값은 dump 시점에 PydanticSerializationError를 일으켜요. 충실한 왕복을 위해 위처럼 구체적인 타입 매개변수를 전달해요.

동시성 제어

각 OnlineEvaluator는 max_concurrency 한도(기본: 10)를 가져요. 한도에 도달하면 해당 평가자의 새 평가 요청은 버려져요(대기열에 쌓이지 않아요). 이렇게 하면 비싼 평가자가 무제한 리소스를 소비하지 못하게 막아요:

from dataclasses import dataclass

from pydantic_evals.evaluators import Evaluator, EvaluatorContext
from pydantic_evals.online import OnlineEvaluator


@dataclass
class ExpensiveCheck(Evaluator):
    async def evaluate(self, ctx: EvaluatorContext) -> bool:
        # 이게 LLM에 느린 호출을 한다고 상상해보세요
        return True


# 최대 3개의 동시 평가 허용
limited = OnlineEvaluator(
    evaluator=ExpensiveCheck(),
    sample_rate=0.1,
    max_concurrency=3,
)

버려진 평가에 반응하려면 OnlineEvaluator에 on_max_concurrency를 설정하거나 OnlineEvalConfig에 기본값으로 설정해요. 콜백은 평가됐을 EvaluatorContext를 받으며 동기 또는 비동기일 수 있어요:

import warnings
from dataclasses import dataclass

from pydantic_evals.evaluators import Evaluator, EvaluatorContext
from pydantic_evals.online import OnlineEvalConfig, OnlineEvaluator


@dataclass
class ExpensiveCheck(Evaluator):
    async def evaluate(self, ctx: EvaluatorContext) -> bool:
        return True


def warn_on_drop(ctx: EvaluatorContext) -> None:
    warnings.warn('Evaluation dropped due to max concurrency', stacklevel=1)


# 평가자별 핸들러
limited = OnlineEvaluator(
    evaluator=ExpensiveCheck(),
    max_concurrency=3,
    on_max_concurrency=warn_on_drop,
)

# 또는 구성의 모든 평가자에 대한 전역 기본값 설정
config = OnlineEvalConfig(on_max_concurrency=warn_on_drop)

참고

평가자별과 구성 수준 on_max_concurrency 모두 설정되지 않으면 버려진 평가는 조용히 무시돼요.

오류 처리

두 가지 유형의 오류 처리가 있어요:

• on_sampling_error: sample_rate 콜러블이 예외를 던질 때 동기적으로 호출돼요. 예외와 Evaluator를 받아요. 동기여야 해요(비동기 아님). 설정되면 평가자를 건너뜁니다. 설정되지 않으면 예외는 호출자에게 전파돼요.
• on_error: sink 또는 on_max_concurrency 콜백에서 예외가 발생할 때 호출돼요. 예외, EvaluatorContext, Evaluator, 그리고 OnErrorLocation 문자열을 받아요. 동기 또는 비동기일 수 있어요. 설정되지 않으면 예외는 조용히 억제돼요. 'sink' 위치는 넓어요 -- 커스텀 sink 실패와 더 드문 기본 OTel 이벤트 내보내기 실패를 모두 다루기 때문에, 위치에 따라 분기하는 핸들러는 'sink'를 "결과 전달이 잘못됨"으로 취급해야 해요.

이것들은 전역 기본값으로 OnlineEvalConfig에 설정하거나, 평가자별로 덮어쓰도록 OnlineEvaluator에 설정해요:

from dataclasses import dataclass

from pydantic_evals.evaluators import Evaluator, EvaluatorContext
from pydantic_evals.online import OnErrorLocation, OnlineEvalConfig, OnlineEvaluator


def log_errors(
    exc: Exception,
    ctx: EvaluatorContext,
    evaluator: Evaluator,
    location: OnErrorLocation,
) -> None:
    print(f'[{location}] {type(exc).__name__}: {exc}')


@dataclass
class MyCheck(Evaluator):
    def evaluate(self, ctx: EvaluatorContext) -> bool:
        return True


# 전역 기본값 -- 이 구성의 모든 평가자에 적용
config = OnlineEvalConfig(
    default_sink=lambda results, failures, context: None,
    on_error=log_errors,
)

# 평가자별 덮어쓰기
custom = OnlineEvaluator(evaluator=MyCheck(), on_error=log_errors)

핵심 동작:

• 평가자 예외는 EvaluatorFailure 객체로 변환되어 sink에 전달됨으로 처리돼요 -- on_error를 거치지 않아요.
• 한 평가자의 오류는 형제들에 영향을 주지 않아요 -- 각 평가자는 격리된 오류 처리를 가진 자체 작업에서 실행돼요.
• 한 sink의 오류는 다른 sink에 영향을 주지 않아요 -- 각 sink 제출은 개별적으로 감싸져요.
• on_error 자체가 예외를 던지면 형제 평가자를 보호하기 위해 예외가 조용히 억제돼요.
• on_error가 설정되지 않으면 예외는 조용히 억제돼요 -- 이것이 안전한 기본값이에요.

실패한 호출 평가

기본적으로 데코레이션된 함수나 래핑된 에이전트 실행이 예외를 던지면 평가자는 디스패치되지 않아요 -- 성공한 결과만 평가자에게 도달해요. 예외는 평소대로 호출자에게 전파돼요.

실패 모드를 채점하려면(예: 예외 유형 분류, 툴 오류 개수, 회귀 알림) OnlineEvaluator에 run_on_errors=True를 설정해 평가자를 opt-in해요. 호출이 예외를 던지면 그 평가자들은 EvaluatorContext.output으로 예외와 함께 디스패치돼요. 디스패치 후에도 예외는 여전히 전파돼요:

from dataclasses import dataclass

from pydantic_evals.evaluators import Evaluator, EvaluatorContext
from pydantic_evals.online import OnlineEvaluator, evaluate


@dataclass
class CategorizeError(Evaluator):
    def evaluate(self, ctx: EvaluatorContext) -> str:
        # 실패한 호출에서 ctx.output은 던져진 예외
        if isinstance(ctx.output, Exception):
            return type(ctx.output).__name__
        return 'ok'


@evaluate(OnlineEvaluator(evaluator=CategorizeError(), run_on_errors=True))
async def my_function(x: int) -> int:
    if x < 0:
        raise ValueError('negative input')
    return x * 2

호출을 위해 샘플링됐지만 run_on_errors=True가 없는 평가자는 오류 경로에서 건너뛰어져서, 저렴한 성공 전용 검사가 같은 데코레이터 안에서 전용 오류 분류기와 나란히 있을 수 있어요. 이 플래그는 OnlineEvaluation 에이전트 기능에서도 인정돼요.

에이전트 통합

OnlineEvaluation 기능은 온라인 평가를 Pydantic AI 에이전트에 가져와요. 함수를 장식하는 대신 에이전트에 기능을 추가해요. @evaluate 데코레이터처럼 평가자는 백그라운드에서 디스패치되고 결과는 기본적으로 OTel 이벤트로 내보내져요 -- sink 등록이 필요 없어요:

from dataclasses import dataclass

from pydantic_ai import Agent
from pydantic_evals.evaluators import Evaluator, EvaluatorContext
from pydantic_evals.online_capability import OnlineEvaluation


@dataclass
class OutputNotEmpty(Evaluator):
    def evaluate(self, ctx: EvaluatorContext) -> bool:
        return bool(ctx.output)


agent = Agent(
    'openai:gpt-5.2',
    name='assistant',
    capabilities=[OnlineEvaluation(evaluators=[OutputNotEmpty()])],
)

내보내진 각 이벤트에 쓰이는 target 이름은 에이전트 자체의 name 속성이에요. 그래서 agent = Agent(..., name='assistant')의 이벤트는 gen_ai.evaluation.target = 'assistant' 아래에 놓여요. 에이전트에 이름이 없으면 target은 리터럴 문자열 'agent'로 대체돼요.

각 에이전트 실행이 끝난 후 그 기능은:

1. sample_rate 구성에 따라 평가자 샘플링
2. 실행 결과에서 EvaluatorContext 구축(출력, 프롬프트, 토큰 사용량, 기간, 스팬 트리) -- context.name은 에이전트 실행의 run_id로 채워짐
3. 백그라운드에서 평가자를 비동기 디스패치
4. 평가자가 끝날 때까지 기다리지 않고 호출자에게 제어권 반환

추가 sink를 붙이거나 샘플링 기본값을 덮어쓰려면 @evaluate 데코레이터와 마찬가지로 OnlineEvalConfig를 전달해요: OnlineEvaluation(evaluators=[...], config=OnlineEvalConfig(default_sample_rate=0.1)).

이 기능은 @evaluate() 데코레이터와 같은 모든 기능을 지원해요: 샘플링, 평가자별 sink, 동시성 제어, 오류 처리. config 매개변수는 선택이며 기본적으로 전역 DEFAULT_CONFIG로 설정돼요.

참고

OnlineEvaluation은 실행이 최종 결과에 도달할 때 agent.run(), agent.run_stream(), agent.iter()를 감싸요. 스트리밍 실행의 경우 평가자는 최종 결과를 사용할 수 있고 주변 컨텍스트 관리자가 종료된 후에만 디스패치돼요. 같은 지연 디스패치 동작이 agent.iter() 실행을 완료까지 몰아갈 때도 적용되는데, 이것이 일반적으로 선호되는 스트리밍 API예요.

API 레퍼런스

pydantic_evals.online 모듈의 완전한 API는 API 레퍼런스에 문서화되어 있어요.

다음 단계

• 커스텀 평가자 -- 도메인용 평가자 작성
• 내장 평가자 -- 바로 쓸 수 있는 평가자 사용
• Logfire의 Live Evaluations -- Logfire 웹 UI에서 온라인 평가 결과 탐색·필터·추세
• Logfire 통합 -- Logfire에서 평가 결과 시각화
• 빠른 시작 -- Dataset.evaluate()로 오프라인 평가

더 알아보기 (Learn more)

• Pydantic Evals 문서: 온라인 평가