PROD · ERROR HANDLING

에러 핸들링

🔄 재시도 전략 ⚡ Circuit Breaker 🔀 Fallback 체인

LLM API 오류, 도구 실패, 타임아웃은 피할 수 없습니다. 지수 백오프 재시도, Circuit Breaker, Fallback 체인으로 탄력적인 Agent 시스템을 구축합니다.

에러 분류 체계

에러 유형예시재시도 가능처리 방법
일시적 오류429 Rate Limit, 503 Service Unavailable✅ 재시도지수 백오프
영구적 오류400 Bad Request, 401 Unauthorized❌ 재시도 무의미즉시 실패 처리
타임아웃네트워크 타임아웃, LLM 응답 지연✅ 재시도더 짧은 타임아웃 + 재시도
LLM 품질 오류잘못된 JSON, 빈 응답, 환각✅ 재시도프롬프트 수정 후 재시도
도구 실행 오류외부 API 오류, DB 연결 실패✅ 재시도Fallback 도구 사용

지수 백오프 재시도

pythonretry.py — 타입별 재시도 전략
import asyncio, random
from anthropic import RateLimitError, APIStatusError, APIConnectionError
from functools import wraps
from typing import TypeVar, Callable

T = TypeVar("T")

RETRYABLE_ERRORS = (RateLimitError, APIConnectionError)
PERMANENT_ERRORS = (APIStatusError,)

async def with_retry(
    fn: Callable,
    *args,
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    **kwargs
):
    """지수 백오프 + 지터를 적용한 재시도"""
    last_error = None

    for attempt in range(max_retries + 1):
        try:
            return await fn(*args, **kwargs)

        except RateLimitError as e:
            # Rate limit: Retry-After 헤더 우선 사용
            retry_after = float(e.response.headers.get("retry-after", base_delay))
            delay = min(retry_after, max_delay)
            last_error = e

        except APIConnectionError as e:
            # 연결 오류: 지수 백오프
            delay = min(base_delay * (2 ** attempt), max_delay)
            delay += random.uniform(0, delay * 0.1)  # 지터
            last_error = e

        except APIStatusError as e:
            if e.status_code in (400, 401, 403):
                raise  # 영구 오류 → 즉시 실패
            delay = min(base_delay * (2 ** attempt), max_delay)
            last_error = e

        if attempt < max_retries:
            print(f"[Retry] {attempt+1}/{max_retries} — {delay:.1f}초 후 재시도: {last_error}")
            await asyncio.sleep(delay)

    raise last_error

# LangChain LLM에 적용
from langchain_anthropic import ChatAnthropic
llm = ChatAnthropic(
    model="claude-opus-4-7",
    max_retries=3,           # LangChain 내장 재시도
    timeout=60
)

Circuit Breaker 패턴

pythoncircuit_breaker.py — 연쇄 장애 방지
import time
from enum import Enum

class State(Enum):
    CLOSED   = "closed"    # 정상 — 요청 통과
    OPEN     = "open"      # 차단 — 즉시 실패
    HALF_OPEN = "half_open"  # 탐색 — 일부 요청만

class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        reset_timeout: float = 60.0,
        half_open_max_calls: int = 3
    ):
        self.failure_threshold   = failure_threshold
        self.reset_timeout       = reset_timeout
        self.half_open_max_calls = half_open_max_calls
        self.state               = State.CLOSED
        self.failure_count       = 0
        self.last_failure_time   = 0
        self.half_open_calls     = 0

    async def call(self, fn, *args, **kwargs):
        if self.state == State.OPEN:
            if time.time() - self.last_failure_time > self.reset_timeout:
                self.state = State.HALF_OPEN
                self.half_open_calls = 0
            else:
                raise RuntimeError("Circuit OPEN — 서비스 일시 중단")

        if self.state == State.HALF_OPEN:
            self.half_open_calls += 1
            if self.half_open_calls > self.half_open_max_calls:
                raise RuntimeError("Circuit HALF_OPEN — 제한 초과")

        try:
            result = await fn(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

    def _on_success(self):
        self.failure_count = 0
        self.state = State.CLOSED

    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = State.OPEN
            print(f"[CB] Circuit OPEN — {self.failure_count}회 실패")

# 서비스별 Circuit Breaker 등록
breakers = {
    "anthropic_api": CircuitBreaker(failure_threshold=5,  reset_timeout=60),
    "search_api":    CircuitBreaker(failure_threshold=3,  reset_timeout=30),
    "db":            CircuitBreaker(failure_threshold=10, reset_timeout=120),
}

Fallback 체인

pythonfallback.py — 단계별 Fallback 체인
from langchain_anthropic import ChatAnthropic

# LangChain .with_fallbacks() 내장 지원
primary_llm  = ChatAnthropic(model="claude-opus-4-7")
fallback_llm = ChatAnthropic(model="claude-haiku-4-5-20251001")

llm_with_fallback = primary_llm.with_fallbacks(
    [fallback_llm],
    exceptions_to_handle=(Exception,)
)

# 다단계 Fallback (검색 → 캐시 → 기본 응답)
async def search_with_fallback(query: str) -> str:
    strategies = [
        ("tavily",   lambda q: tavily_search(q)),
        ("duckduckgo", lambda q: ddg_search(q)),
        ("cache",     lambda q: cache.get(q)),
        ("default",   lambda q: "검색 서비스를 일시적으로 사용할 수 없습니다."),
    ]

    for name, strategy in strategies:
        try:
            result = await strategy(query)
            if result:
                print(f"[Fallback] 성공: {name}")
                return result
        except Exception as e:
            print(f"[Fallback] {name} 실패: {e}")
            continue

    return "검색 결과를 가져올 수 없습니다."
↑ 목차