재시도 전략

재시도 전략 (Retry Strategies)

자동 재시도 로직으로 작업과 평가자의 일시적 실패를 처리해요.

LLM 기반 시스템은 일시적 실패를 겪을 수 있어요:

• 속도 제한
• 네트워크 타임아웃
• 일시적 API 중단
• 컨텍스트 길이 오류

Pydantic Evals는 둘 다에 대해 재시도 구성을 지원해요:

• 작업 실행 - 평가되는 함수
• 평가자 실행 - 평가자 자체

출처: 문서

본문

기본 재시도 구성

Tenacity 매개변수를 사용해 evaluate() 또는 evaluate_sync()에 재시도 구성을 전달해요:

from tenacity import stop_after_attempt

from pydantic_evals import Case, Dataset


def my_function(inputs: str) -> str:
    return f'Result: {inputs}'


dataset = Dataset(name='basic_retry', cases=[Case(inputs='test')], evaluators=[])

report = dataset.evaluate_sync(
    task=my_function,
    retry_task={'stop': stop_after_attempt(3)},
    retry_evaluators={'stop': stop_after_attempt(2)},
)

재시도 구성 옵션

재시도 구성은 Tenacity를 사용하며 Pydantic AI의 RetryConfig와 같은 옵션을 지원해요:

from tenacity import stop_after_attempt, wait_exponential

from pydantic_evals import Case, Dataset


def my_function(inputs: str) -> str:
    return f'Result: {inputs}'


dataset = Dataset(name='retry_config', cases=[Case(inputs='test')], evaluators=[])

retry_config = {
    'stop': stop_after_attempt(3),  # 3번 시도 후 중단
    'wait': wait_exponential(multiplier=1, min=1, max=10),  # 지수 백오프: 1s, 2s, 4s, 8s (10s에서 상한)
    'reraise': True,  # 재시도 소진 후 원래 예외를 다시 던짐
}

dataset.evaluate_sync(
    task=my_function,
    retry_task=retry_config,
)

흔한 매개변수

재시도 구성은 tenacity retry 데코레이터의 어떤 매개변수든 받아요. 흔한 것들:

매개변수	유형	설명
stop	StopBaseT	중단 전략 (예: stop_after_attempt(3), stop_after_delay(60))
wait	WaitBaseT	대기 전략 (예: wait_exponential(), wait_fixed(2))
retry	RetryBaseT	재시도 조건 (예: retry_if_exception_type(TimeoutError))
reraise	bool	원래 예외를 다시 던질지 (기본: False)
before_sleep	Callable	재시도 사이 잠들기 전 콜백

사용 가능한 모든 옵션은 Tenacity 문서를 참고해요.

작업 재시도

작업 함수가 실패하면 재시도해요:

from tenacity import stop_after_attempt, wait_exponential

from pydantic_evals import Case, Dataset


async def call_llm(inputs: str) -> str:
    return f'LLM response to: {inputs}'


async def flaky_llm_task(inputs: str) -> str:
    """This might hit rate limits or timeout."""
    response = await call_llm(inputs)
    return response


dataset = Dataset(name='task_retry', cases=[Case(inputs='test')])

report = dataset.evaluate_sync(
    task=flaky_llm_task,
    retry_task={
        'stop': stop_after_attempt(5),  # 최대 5번 시도
        'wait': wait_exponential(multiplier=1, min=1, max=30),  # 지수 백오프, 30s에서 상한
        'reraise': True,
    },
)

작업 재시도가 언제 발동하나요?

작업이 예외를 던질 때 재시도가 발동돼요:

class RateLimitError(Exception):
    pass


class ValidationError(Exception):
    pass


async def call_api(inputs: str) -> str:
    return f'API response: {inputs}'


async def my_task(inputs: str) -> str:
    try:
        return await call_api(inputs)
    except RateLimitError:
        # 재시도를 발동시킴
        raise
    except ValidationError:
        # 이것도 재시도를 발동시킴
        raise

지수 백오프

wait_exponential()을 사용하면 지연이 기하급수적으로 증가해요:

Attempt 1: immediate
Attempt 2: ~1s delay (multiplier * 2^0)
Attempt 3: ~2s delay (multiplier * 2^1)
Attempt 4: ~4s delay (multiplier * 2^2)
Attempt 5: ~8s delay (multiplier * 2^3, capped at max)

실제 지연은 wait_exponential()에 전달한 multiplier, min, max 매개변수에 따라 달라져요.

평가자 재시도

평가자가 실패하면 재시도해요:

from tenacity import stop_after_attempt, wait_exponential

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


def my_task(inputs: str) -> str:
    return f'Result: {inputs}'


dataset = Dataset(
    name='evaluator_retry',
    cases=[Case(inputs='test')],
    evaluators=[
        # LLMJudge는 속도 제한에 걸릴 수 있음
        LLMJudge(rubric='Response is accurate'),
    ],
)

report = dataset.evaluate_sync(
    task=my_task,
    retry_evaluators={
        'stop': stop_after_attempt(3),
        'wait': wait_exponential(multiplier=1, min=0.5, max=10),
        'reraise': True,
    },
)

평가자 재시도가 언제 발동하나요?

평가자가 예외를 던질 때 재시도가 발동돼요:

from dataclasses import dataclass

from pydantic_evals.evaluators import Evaluator, EvaluatorContext


async def external_api_call(output: str) -> bool:
    return len(output) > 0


@dataclass
class APIEvaluator(Evaluator):
    async def evaluate(self, ctx: EvaluatorContext) -> bool:
        # 여기서 예외를 던지면 재시도 로직이 발동함
        result = await external_api_call(ctx.output)
        return result

평가자 실패

모든 재시도 후에도 평가자가 실패하면 EvaluatorFailure로 기록돼요:

from tenacity import stop_after_attempt

from pydantic_evals import Case, Dataset


def task(inputs: str) -> str:
    return f'Result: {inputs}'


dataset = Dataset(name='evaluator_failures', cases=[Case(inputs='test')], evaluators=[])

report = dataset.evaluate_sync(task, retry_evaluators={'stop': stop_after_attempt(3)})

# 평가자 실패 확인
for case in report.cases:
    if case.evaluator_failures:
        for failure in case.evaluator_failures:
            print(f'Evaluator {failure.name} failed: {failure.error_message}')
    #> (출력 없음 - 이 케이스에는 평가자 실패 없음)

보고서에서 평가자 실패를 봐요:

from pydantic_evals import Case, Dataset


def task(inputs: str) -> str:
    return f'Result: {inputs}'


dataset = Dataset(name='failure_report', cases=[Case(inputs='test')], evaluators=[])
report = dataset.evaluate_sync(task)

report.print(include_evaluator_failures=True)
"""
  Evaluation Summary:
         task
┏━━━━━━━━━━┳━━━━━━━━━━┓
┃ Case ID  ┃ Duration ┃
┡━━━━━━━━━━╇━━━━━━━━━━┩
│ Case 1   │     10ms │
├──────────┼──────────┤
│ Averages │     10ms │
└──────────┴──────────┘
"""
#>
#> ✅ case_0                       ━━━━━━━━━━━━━━━━━━━━━━━━━━━ 100% (0/0)

작업과 평가자 재시도 결합하기

둘을 독립적으로 구성할 수 있어요:

from tenacity import stop_after_attempt, wait_exponential

from pydantic_evals import Case, Dataset


def flaky_task(inputs: str) -> str:
    return f'Result: {inputs}'


dataset = Dataset(name='combined_retry', cases=[Case(inputs='test')], evaluators=[])

report = dataset.evaluate_sync(
    task=flaky_task,
    retry_task={
        'stop': stop_after_attempt(5),  # 작업을 최대 5번 재시도
        'wait': wait_exponential(multiplier=1, min=1, max=30),
        'reraise': True,
    },
    retry_evaluators={
        'stop': stop_after_attempt(3),  # 평가자를 최대 3번 재시도
        'wait': wait_exponential(multiplier=1, min=0.5, max=10),
        'reraise': True,
    },
)

실용 예시

속도 제한 처리

from tenacity import stop_after_attempt, wait_exponential

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


async def expensive_llm_call(inputs: str) -> str:
    return f'LLM response: {inputs}'


async def llm_task(inputs: str) -> str:
    """Task that might hit rate limits."""
    return await expensive_llm_call(inputs)


dataset = Dataset(
    name='rate_limit_retry',
    cases=[Case(inputs='test')],
    evaluators=[
        LLMJudge(rubric='Quality check'),  # 이것도 속도 제한에 걸릴 수 있음
    ],
)

# 속도 제한에 대한 넉넉한 재시도
report = dataset.evaluate_sync(
    task=llm_task,
    retry_task={
        'stop': stop_after_attempt(10),  # 속도 제한은 여러 번 재시도가 필요할 수 있음
        'wait': wait_exponential(multiplier=2, min=2, max=60),  # 2s에서 시작, 60s까지 지수 증가
        'reraise': True,
    },
    retry_evaluators={
        'stop': stop_after_attempt(5),
        'wait': wait_exponential(multiplier=2, min=2, max=30),
        'reraise': True,
    },
)

네트워크 타임아웃 처리

import httpx
from tenacity import stop_after_attempt, wait_exponential

from pydantic_evals import Case, Dataset


async def api_task(inputs: str) -> str:
    """Task that calls external API which might timeout."""
    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.post('https://api.example.com', json={'input': inputs})
        return response.text


dataset = Dataset(name='timeout_retry', cases=[Case(inputs='test')], evaluators=[])

# 네트워크 문제에 대한 빠른 재시도
report = dataset.evaluate_sync(
    task=api_task,
    retry_task={
        'stop': stop_after_attempt(4),  # 빠른 재시도 몇 번
        'wait': wait_exponential(multiplier=0.5, min=0.5, max=5),  # 빠른 재시도, 5s에서 상한
        'reraise': True,
    },
)

컨텍스트 길이 처리

from tenacity import stop_after_attempt

from pydantic_evals import Case, Dataset


class ContextLengthError(Exception):
    pass


async def llm_call(inputs: str, max_tokens: int = 8000) -> str:
    return f'LLM response: {inputs[:100]}'


async def smart_llm_task(inputs: str) -> str:
    """Task that might exceed context length."""
    try:
        return await llm_call(inputs, max_tokens=8000)
    except ContextLengthError:
        # 더 짧은 컨텍스트로 재시도
        truncated_inputs = inputs[:4000]
        return await llm_call(truncated_inputs, max_tokens=4000)


dataset = Dataset(name='context_length', cases=[Case(inputs='test')], evaluators=[])

# 컨텍스트 길이 오류는 재시도하지 않음 (작업에서 처리)
report = dataset.evaluate_sync(
    task=smart_llm_task,
    retry_task={'stop': stop_after_attempt(1)},  # 재시도 없음, 우리가 처리함
)

재시도 vs 오류 처리

**재시도를 쓸 때: **

• 일시적 실패 (속도 제한, 타임아웃)
• 네트워크 문제
• 일시적 서비스 중단
• 복구 가능한 오류

오류 처리를 쓸 때:

• 검증 오류
• 논리 오류
• 영구 실패
• 예상된 오류 조건

class RateLimitError(Exception):
    pass


async def llm_call(inputs: str) -> str:
    return f'LLM response: {inputs}'


def is_valid(result: str) -> bool:
    return len(result) > 0


async def smart_task(inputs: str) -> str:
    """Handle expected errors, let retries handle transient failures."""
    try:
        result = await llm_call(inputs)

        # 출력 검증 (검증 오류는 재시도하지 않음)
        if not is_valid(result):
            return 'ERROR: Invalid output format'

        return result

    except RateLimitError:
        # 재시도 로직이 처리하게 함
        raise

    except ValueError as e:
        # 재시도하지 않음 - 영구 오류
        return f'ERROR: {e}'

문제 해결

"재시도 후에도 계속 실패함"

재시도 횟수를 늘리거나 오류가 재시도 가능한지 확인해요:

import logging

from tenacity import stop_after_attempt

from pydantic_evals import Case, Dataset


def task(inputs: str) -> str:
    return f'Result: {inputs}'


# 무엇이 실패하는지 보려면 로깅 추가
logging.basicConfig(level=logging.DEBUG)

dataset = Dataset(name='troubleshooting', cases=[Case(inputs='test')], evaluators=[])

# Tenacity가 재시도 시도를 로그
report = dataset.evaluate_sync(task, retry_task={'stop': stop_after_attempt(5)})

"평가가 너무 오래 걸림"

재시도 횟수나 대기 시간을 줄여요:

from tenacity import stop_after_attempt, wait_exponential

# 더 빠른 재시도
retry_config = {
    'stop': stop_after_attempt(3),  # 시도 줄임
    'wait': wait_exponential(multiplier=0.1, min=0.1, max=2),  # 빠른 재시도, 2s에서 상한
    'reraise': True,
}

"재시도에도 속도 제한에 걸림"

지연을 늘리거나 max_concurrency를 사용해요:

from tenacity import stop_after_attempt, wait_exponential

from pydantic_evals import Case, Dataset


def task(inputs: str) -> str:
    return f'Result: {inputs}'


dataset = Dataset(name='rate_limit_config', cases=[Case(inputs='test')], evaluators=[])

# 더 긴 지연
retry_config = {
    'stop': stop_after_attempt(5),
    'wait': wait_exponential(multiplier=5, min=5, max=60),  # 5s에서 시작, 60s까지 지수 증가
    'reraise': True,
}

# 동시성도 줄임
report = dataset.evaluate_sync(
    task=task,
    retry_task=retry_config,
    max_concurrency=2,  # 동시 작업 2개만
)

다음 단계

• 동시성 & 성능 - 평가 성능 최적화
• Logfire 통합 - Logfire에서 재시도 보기

더 알아보기 (Learn more)

• Pydantic Evals 문서: 재시도 전략