스팬 기반 평가

스팬 기반 평가 (Span-Based Evaluation)

실행 중에 캡처된 OpenTelemetry 스팬을 분석해 AI 시스템의 동작을 평가해요.

출처: 문서

본문

logfire SDK가 필요해요

이 평가자들은 실행 중에 캡처된 OpenTelemetry 스팬 트리를 읽기 때문에, logfire SDK가 설치되고 구성되어 있어야 해요. Pydantic Logfire 계정은 필요 없어요. SDK가 스팬을 로컬에서 캡처해주니까요.

Terminal

pip install 'pydantic-evals[logfire]'

스팬 기반 평가를 사용하면 AI 시스템이 무엇을 만들어내는지뿐 아니라 어떻게 실행되는지도 평가할 수 있어요. 원하는 동작을 보장하기 위해 최종 출력뿐 아니라 실행 경로가 중요한 복잡한 에이전트에는 필수적이에요.

왜 스팬 기반 평가인가요?

전통적인 평가자는 작업 입력과 출력을 평가해요. 단순한 작업에서는 이것으로 충분할 수 있어요 -- 출력이 맞다면 작업은 성공한 거니까요. 하지만 복잡한 다단계 에이전트에서는 결과만큼 _과정_도 중요해요:

잘못된 방법으로 도달한 올바른 답 - 에이전트가 우연히 올바른 출력을 만들 수도 있어요 (예: 추측, 검색했어야 하는데 캐시 데이터 사용, 잘못된 툴을 호출했지만 운이 좋았던 경우)
필요한 동작의 검증 - 특정 툴이 호출됐는지, 특정 코드 경로가 실행됐는지, 특정 패턴을 따랐는지 확인해야 해요
성능과 효율성 - 에이전트가 불필요한 툴 호출, 무한 루프, 과도한 재시도 없이 효율적으로 답에 도달해야 해요
안전성과 규정 준수 - 위험한 연산이 시도되지 않았는지, 민감 데이터에 부적절하게 접근하지 않았는지, 가드레일을 우회하지 않았는지 확인하는 것이 중요해요

실제 시나리오

스팬 기반 평가는 특히 다음에 유용해요:

RAG 시스템 - 생성 전에 문서가 검색되고 재순위화됐는지 확인. 답변에 인용이 포함됐는지만 보는 게 아니라요.
멀티 에이전트 조정 - 오케스트레이터가 올바른 순서로 올바른 전문 에이전트에 위임했는지 확인
툴 호출 에이전트 - 특정 툴이 사용됐는지(또는 피해졌는지), 그리고 기대한 순서대로 사용됐는지 확인
디버깅과 회귀 테스트 - 출력은 올바르지만 내부 로직이 악화되는 동작 회귀를 잡아내기
프로덕션 정렬 - 평가 판정이 프로덕션에서 캡처된 것과 동일한 텔레메트리 데이터를 대상으로 동작하도록 해서, 평가 인사이트가 프로덕션 모니터링으로 직접 이어지게 하기

어떻게 동작하나요?

logfire를 구성하면(logfire.configure()), Pydantic Evals는 작업 실행 중에 생성된 모든 OpenTelemetry 스팬을 캡처해요. 그런 다음 다음에 대해 조건을 판정하는 평가자를 작성할 수 있어요:

어떤 툴이 호출됐는지 - HasMatchingSpan(query={'name_contains': 'search_tool'})
실행된 코드 경로 - 특정 함수가 실행됐는지, 특정 분기를 탔는지 확인
타이밍 특성 - 작업이 SLA 범위 안에 완료되는지 확인
오류 조건 - 재시도, 폴백, 또는 특정 실패 모드 감지
실행 구조 - 부모-자식 관계, 위임 패턴, 실행 순서 확인

이것은 근본적으로 다른 평가 패러다임을 만들어요. 입력-출력 관계뿐 아니라 동작 계약을 테스트하는 거예요.

기본 사용법

import logfire

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

# 스팬을 캡처하도록 logfire 구성
logfire.configure(send_to_logfire='if-token-present')

dataset = Dataset(
    name='span_basic',
    cases=[Case(inputs='test')],
    evaluators=[
        # 데이터베이스가 쿼리됐는지 확인
        HasMatchingSpan(
            query={'name_contains': 'database_query'},
            evaluation_name='used_database',
        ),
    ],
)

HasMatchingSpan 평가자

HasMatchingSpan 평가자는 어떤 스팬이 쿼리와 일치하는지 확인해요:

from pydantic_evals.evaluators import HasMatchingSpan

HasMatchingSpan(
    query={'name_contains': 'test'},
    evaluation_name='span_check',
)

반환: bool - 쿼리와 일치하는 스팬이 하나라도 있으면 True

SpanQuery 레퍼런스

SpanQuery는 쿼리 조건을 가진 딕셔너리예요:

이름 조건

이름으로 스팬을 일치시켜요:

# 정확한 이름 일치
{'name_equals': 'search_database'}

# 부분 문자열 포함
{'name_contains': 'tool_call'}

# 정규식 패턴
{'name_matches_regex': r'llm_call_\d+'}

속성 조건

특정 속성을 가진 스팬을 일치시켜요:

# 특정 속성 값 보유
{'has_attributes': {'operation': 'search', 'status': 'success'}}

# 속성 키 보유 (값은 무관)
{'has_attribute_keys': ['user_id', 'request_id']}

상태 조건

상태로 스팬을 일치시켜요:

# 오류를 기록한 스팬
{'has_status': 'error'}

# 명시적으로 OK로 표시된 스팬 (참고: 성공 스팬은 보통 'unset'이지 'ok'가 아님)
{'has_status': 'ok'}

기간 조건

실행 시간에 따라 일치시켜요:

from datetime import timedelta

# 최소 기간
{'min_duration': 1.0}  # 초
{'min_duration': timedelta(seconds=1)}

# 최대 기간
{'max_duration': 5.0}  # 초
{'max_duration': timedelta(seconds=5)}

# 범위
{'min_duration': 0.5, 'max_duration': 2.0}

논리 연산자

조건을 결합해요:

# NOT
{'not_': {'name_contains': 'error'}}

# AND (모두 일치해야 함)
{'and_': [
    {'name_contains': 'tool'},
    {'max_duration': 1.0},
]}

# OR (하나라도 일치하면 됨)
{'or_': [
    {'name_equals': 'search'},
    {'name_equals': 'query'},
]}

자식/하위 항목 조건

스팬 간의 관계를 쿼리해요:

# 직접 자식 수
{'min_child_count': 1}
{'max_child_count': 5}

# 일부 자식이 쿼리와 일치
{'some_child_has': {'name_contains': 'retry'}}

# 모든 자식이 쿼리와 일치
{'all_children_have': {'max_duration': 0.5}}

# 어떤 자식도 쿼리와 일치하지 않음
{'no_child_has': {'has_status': 'error'}}

# 하위 항목 쿼리 (재귀적)
{'min_descendant_count': 5}
{'some_descendant_has': {'name_contains': 'api_call'}}

조상/깊이 조건

스팬 계층을 쿼리해요:

# 깊이 (루트 스팬은 깊이 0)
{'min_depth': 1}  # 루트 스팬이 아님
{'max_depth': 2}  # 최대 2단계 깊이

# 조상 쿼리
{'some_ancestor_has': {'name_equals': 'agent_run'}}
{'all_ancestors_have': {'max_duration': 10.0}}
{'no_ancestor_has': {'has_status': 'error'}}

재귀 중단

재귀 쿼리를 제어해요:

{
    'some_descendant_has': {'name_contains': 'expensive'},
    'stop_recursing_when': {'name_equals': 'boundary'},
}
# 'boundary'라는 스팬을 만날 때까지만 하위 항목 검색

실용 예시

툴 사용 검증

특정 툴이 호출됐는지 확인해요:

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

dataset = Dataset(
    name='tool_verification',
    cases=[Case(inputs='test')],
    evaluators=[
        # 반드시 search 툴 호출
        HasMatchingSpan(
            query={'name_contains': 'search_tool'},
            evaluation_name='used_search',
        ),

        # 위험한 툴은 반드시 호출하지 말 것
        HasMatchingSpan(
            query={'not_': {'name_contains': 'delete_database'}},
            evaluation_name='safe_execution',
        ),
    ],
)

여러 툴 확인

일련의 작업을 검증해요:

from pydantic_evals.evaluators import HasMatchingSpan

evaluators = [
    HasMatchingSpan(
        query={'name_contains': 'retrieve_context'},
        evaluation_name='retrieved_context',
    ),
    HasMatchingSpan(
        query={'name_contains': 'generate_response'},
        evaluation_name='generated_response',
    ),
    HasMatchingSpan(
        query={'and_': [
            {'name_contains': 'cite'},
            {'has_attribute_keys': ['source_id']},
        ]},
        evaluation_name='added_citations',
    ),
]

성능 판정

작업이 지연 요구사항을 충족하는지 확인해요:

from pydantic_evals.evaluators import HasMatchingSpan

evaluators = [
    # 데이터베이스 쿼리는 빨라야 함
    HasMatchingSpan(
        query={'and_': [
            {'name_contains': 'database'},
            {'max_duration': 0.1},  # 최대 100ms
        ]},
        evaluation_name='fast_db_queries',
    ),

    # 전체가 빨리 완료되어야 함
    HasMatchingSpan(
        query={'and_': [
            {'name_equals': 'task_execution'},
            {'max_duration': 2.0},
        ]},
        evaluation_name='within_sla',
    ),
]

오류 탐지

스팬 상태를 사용해 오류 조건을 확인해요:

from pydantic_evals.evaluators import HasMatchingSpan

evaluators = [
    # 트레이스 어딘가에서 오류 발생
    HasMatchingSpan(
        query={'has_status': 'error'},
        evaluation_name='had_errors',
    ),

    # 오류 없음: HasMatchingSpan은 *어떤* 스팬이 일치하면 통과하므로,
    # 루트 스팬에 쿼리를 고정하고 그것과 모든 하위 항목을 확인
    HasMatchingSpan(
        query={
            'name_equals': 'task_execution',
            'not_': {'has_status': 'error'},
            'no_descendant_has': {'has_status': 'error'},
        },
        evaluation_name='no_errors',
    ),

    # 재시도가 발생함
    HasMatchingSpan(
        query={'name_contains': 'retry'},
        evaluation_name='had_retries',
    ),

    # 폴백이 사용됨
    HasMatchingSpan(
        query={'name_contains': 'fallback_model'},
        evaluation_name='used_fallback',
    ),
]

복잡한 동작 검사

정교한 동작 패턴을 검증해요:

from pydantic_evals.evaluators import HasMatchingSpan

evaluators = [
    # 에이전트가 하위 에이전트에 위임함
    HasMatchingSpan(
        query={'and_': [
            {'name_contains': 'agent'},
            {'some_child_has': {'name_contains': 'delegate'}},
        ]},
        evaluation_name='used_delegation',
    ),

    # 재시도와 함께 여러 LLM 호출
    HasMatchingSpan(
        query={'and_': [
            {'name_contains': 'llm_call'},
            {'some_descendant_has': {'name_contains': 'retry'}},
            {'min_descendant_count': 3},
        ]},
        evaluation_name='retry_pattern',
    ),
]

SpanTree로 커스텀 평가자

더 복잡한 스팬 분석을 위해 커스텀 평가자를 작성해요:

from dataclasses import dataclass

from pydantic_evals.evaluators import Evaluator, EvaluatorContext


@dataclass
class CustomSpanCheck(Evaluator):
    def evaluate(self, ctx: EvaluatorContext) -> dict[str, bool | int]:
        span_tree = ctx.span_tree

        # 특정 스팬 찾기
        llm_spans = span_tree.find(lambda node: 'llm' in node.name)
        tool_spans = span_tree.find(lambda node: 'tool' in node.name)

        # 메트릭 계산
        total_llm_time = sum(
            span.duration.total_seconds() for span in llm_spans
        )

        return {
            'used_llm': len(llm_spans) > 0,
            'used_tools': len(tool_spans) > 0,
            'tool_count': len(tool_spans),
            'llm_fast': total_llm_time < 2.0,
        }

SpanTree API

SpanTree는 스팬 분석 메서드를 제공해요:

from pydantic_evals.otel import SpanTree


# 예시 API (컨텍스트의 span_tree 필요)
def example_api(span_tree: SpanTree) -> None:
    span_tree.find(lambda n: True)  # 일치하는 모든 노드 찾기
    span_tree.any({'name_contains': 'test'})  # 일치하는 스팬이 있는지 확인
    span_tree.all({'name_contains': 'test'})  # 모든 스팬이 일치하는지 확인
    span_tree.count({'name_contains': 'test'})  # 일치하는 스팬 개수

    # 반복
    for node in span_tree:
        print(node.name, node.duration, node.attributes)

SpanNode 속성

각 SpanNode는 다음을 가져요:

from pydantic_evals.otel import SpanNode


# 예시 속성 (컨텍스트의 node 필요)
def example_properties(node: SpanNode) -> None:
    _ = node.name  # 스팬 이름
    _ = node.duration  # timedelta
    _ = node.attributes  # dict[str, AttributeValue]
    _ = node.start_timestamp  # datetime
    _ = node.end_timestamp  # datetime
    _ = node.status  # 'unset' | 'ok' | 'error'
    _ = node.children  # list[SpanNode]
    _ = node.descendants  # list[SpanNode] (재귀적)
    _ = node.ancestors  # list[SpanNode]
    _ = node.parent  # SpanNode | None

스팬 쿼리 디버깅

Logfire에서 스팬 보기

데이터를 Logfire로 보내고 있다면 웹 UI에서 모든 스팬을 볼 수 있어 트레이스 구조를 이해할 수 있어요.

스팬 트리 출력

from dataclasses import dataclass

from pydantic_evals.evaluators import Evaluator, EvaluatorContext


@dataclass
class DebugSpans(Evaluator):
    def evaluate(self, ctx: EvaluatorContext) -> bool:
        for node in ctx.span_tree:
            print(f"{'  ' * len(node.ancestors)}{node.name} ({node.duration})")
        return True

쿼리 테스트

쿼리를 점진적으로 테스트해요:

from pydantic_evals.evaluators import HasMatchingSpan

# 간단하게 시작
query = {'name_contains': 'tool'}

# 조건을 점진적으로 추가
query = {'and_': [
    {'name_contains': 'tool'},
    {'max_duration': 1.0},
]}

# 평가자에서 테스트
HasMatchingSpan(query=query, evaluation_name='test')

사용 사례

RAG 시스템 검증

검색-증강 생성 워크플로를 검증해요:

from pydantic_evals.evaluators import HasMatchingSpan

evaluators = [
    # 문서 검색됨
    HasMatchingSpan(
        query={'name_contains': 'vector_search'},
        evaluation_name='retrieved_docs',
    ),

    # 재순위화된 결과
    HasMatchingSpan(
        query={'name_contains': 'rerank'},
        evaluation_name='reranked_results',
    ),

    # 컨텍스트로 생성됨
    HasMatchingSpan(
        query={'and_': [
            {'name_contains': 'generate'},
            {'has_attribute_keys': ['context_ids']},
        ]},
        evaluation_name='used_context',
    ),
]

멀티 에이전트 시스템

에이전트 조정을 검증해요:

from pydantic_evals.evaluators import HasMatchingSpan

evaluators = [
    # 마스터 에이전트 실행
    HasMatchingSpan(
        query={'name_equals': 'master_agent'},
        evaluation_name='master_ran',
    ),

    # 전문가에게 위임
    HasMatchingSpan(
        query={'and_': [
            {'name_contains': 'specialist_agent'},
            {'some_ancestor_has': {'name_equals': 'master_agent'}},
        ]},
        evaluation_name='delegated_correctly',
    ),

    # 순환 위임 없음
    HasMatchingSpan(
        query={'not_': {'and_': [
            {'name_contains': 'agent'},
            {'some_descendant_has': {'name_contains': 'agent'}},
            {'some_ancestor_has': {'name_contains': 'agent'}},
        ]}},
        evaluation_name='no_circular_delegation',
    ),
]

툴 사용 패턴

지능적인 툴 선택을 검증해요:

from pydantic_evals.evaluators import HasMatchingSpan

evaluators = [
    # 답변하기 전에 검색 사용
    HasMatchingSpan(
        query={'and_': [
            {'name_contains': 'search'},
            {'some_ancestor_has': {'name_contains': 'answer'}},
        ]},
        evaluation_name='searched_before_answering',
    ),

    # 툴 호출 제한 (루프 없음)
    HasMatchingSpan(
        query={'and_': [
            {'name_contains': 'tool'},
            {'max_child_count': 5},
        ]},
        evaluation_name='reasonable_tool_usage',
    ),
]

모범 사례

간단하게 시작해요: 기본 이름 쿼리로 시작하고 필요에 따라 복잡성을 추가해요
설명적인 이름을 사용해요: 애플리케이션 코드에서 스팬 이름을 잘 지어요
쿼리를 테스트해요: 전체 평가를 실행하기 전에 쿼리가 동작하는지 확인해요
다른 평가자와 결합해요: 출력 검증과 함께 스팬 검사를 사용해요
기대치를 문서화해요: 특정 스팬이 있어야/없어야 하는 이유를 주석으로 남겨요

다음 단계

Logfire 통합 - 스팬 캡처를 위한 Logfire 설정
커스텀 평가자 - 고급 스팬 분석 작성
내장 평가자 - 기타 평가자 유형

더 알아보기 (Learn more)

Pydantic Evals 문서: 스팬 기반 평가