10장: 에러 처리, 폴백, 관측 가능성

import asyncio
import random
from typing import TypeVar, Callable
 
T = TypeVar("T")
 
async def retry_with_exponential_backoff(
    func: Callable[..., T],
    *args,
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    jitter: bool = True,
    retryable_errors: tuple = (Exception,),
    **kwargs,
) -> T:
    """지수 백오프 + 지터를 적용한 재시도"""
    last_error = None
 
    for attempt in range(max_retries):
        try:
            return await func(*args, **kwargs)
        except retryable_errors as e:
            last_error = e
 
            if attempt == max_retries - 1:
                raise
 
            delay = min(base_delay * (2 ** attempt), max_delay)
            if jitter:
                delay = delay * (0.5 + random.random())
 
            print(f"재시도 {attempt + 1}/{max_retries}, "
                  f"{delay:.1f}초 대기 (에러: {e})")
            await asyncio.sleep(delay)
 
    raise last_error

from langchain_openai import ChatOpenAI
 
model = ChatOpenAI(model="gpt-4o")
 
# 재시도 설정
chain_with_retry = (prompt | model | parser).with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True,
    retry_if_exception_type=(
        ConnectionError,
        TimeoutError,
    ),
)

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
 
# 1차: 주 모델
primary = ChatOpenAI(model="gpt-4o", request_timeout=10)
 
# 2차: 같은 공급자의 저비용 모델
secondary = ChatOpenAI(model="gpt-4o-mini", request_timeout=15)
 
# 3차: 다른 공급자
tertiary = ChatAnthropic(
    model="claude-3-5-haiku-20241022",
    timeout=20,
)
 
# 폴백 체인
robust_chain = (
    prompt
    | primary.with_fallbacks([secondary, tertiary])
    | parser
)

async def query_with_strategy_fallback(query: str) -> str:
    """전략 수준의 폴백"""
    strategies = [
        ("RAG + GPT-4o", rag_gpt4_chain),
        ("RAG + GPT-4o-mini", rag_gpt4mini_chain),
        ("직접 응답 (RAG 없이)", direct_chain),
        ("사전 정의된 응답", cached_response),
    ]
 
    for strategy_name, chain in strategies:
        try:
            result = await chain.ainvoke({"query": query})
            print(f"성공 전략: {strategy_name}")
            return result
        except Exception as e:
            print(f"전략 실패 [{strategy_name}]: {e}")
            continue
 
    return "죄송합니다, 현재 서비스를 이용할 수 없습니다. 잠시 후 다시 시도해주세요."

import asyncio
 
async def call_with_timeout(chain, input_data, timeout: float = 30.0):
    """타임아웃이 적용된 체인 호출"""
    try:
        async with asyncio.timeout(timeout):
            return await chain.ainvoke(input_data)
    except asyncio.TimeoutError:
        # 타임아웃 시 폴백
        return await fallback_chain.ainvoke(input_data)
 
# LangChain 모델 수준 타임아웃
model = ChatOpenAI(
    model="gpt-4o",
    request_timeout=10,  # HTTP 요청 타임아웃
    max_retries=2,       # API 클라이언트 수준 재시도
)

import time
from enum import Enum
 
class CircuitState(Enum):
    CLOSED = "closed"      # 정상 상태
    OPEN = "open"          # 차단 상태
    HALF_OPEN = "half_open"  # 시험 상태
 
class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: float = 60.0,
        half_open_max_calls: int = 3,
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_max_calls = half_open_max_calls
 
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = 0
        self.half_open_calls = 0
 
    async def call(self, func, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                self.half_open_calls = 0
            else:
                raise CircuitOpenError(
                    f"서킷 브레이커 OPEN 상태. "
                    f"{self.recovery_timeout}초 후 재시도됩니다."
                )
 
        try:
            result = await func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise
 
    def _on_success(self):
        self.failure_count = 0
        if self.state == CircuitState.HALF_OPEN:
            self.half_open_calls += 1
            if self.half_open_calls >= self.half_open_max_calls:
                self.state = CircuitState.CLOSED
 
    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN
 
# 사용
breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60)
 
try:
    result = await breaker.call(model.ainvoke, prompt)
except CircuitOpenError:
    result = await fallback_model.ainvoke(prompt)

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanExporter
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
    OTLPSpanExporter,
)
from opentelemetry.sdk.resources import Resource
 
# 리소스 정의
resource = Resource.create({
    "service.name": "ai-orchestration-service",
    "service.version": "1.0.0",
    "deployment.environment": "production",
})
 
# 트레이서 설정
provider = TracerProvider(resource=resource)
exporter = OTLPSpanExporter(endpoint="http://otel-collector:4317")
provider.add_span_processor(BatchSpanExporter(exporter))
trace.set_tracer_provider(provider)
 
tracer = trace.get_tracer("ai.orchestration")

from opentelemetry import trace
 
tracer = trace.get_tracer("ai.orchestration")
 
async def rag_query(query: str) -> str:
    with tracer.start_as_current_span("rag_query") as span:
        span.set_attribute("query", query)
 
        # 검색 단계
        with tracer.start_as_current_span("retrieval") as retrieval_span:
            docs = await retriever.aretrieve(query)
            retrieval_span.set_attribute("num_docs", len(docs))
            retrieval_span.set_attribute(
                "avg_score",
                sum(d.score for d in docs) / len(docs) if docs else 0,
            )
 
        # 생성 단계
        with tracer.start_as_current_span("generation") as gen_span:
            response = await model.ainvoke(prompt.format(
                context=docs, question=query
            ))
            gen_span.set_attribute(
                "tokens.input", response.usage_metadata["input_tokens"]
            )
            gen_span.set_attribute(
                "tokens.output", response.usage_metadata["output_tokens"]
            )
 
        span.set_attribute("response_length", len(str(response)))
        return str(response)

from opentelemetry import metrics
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import (
    OTLPMetricExporter,
)
 
# 메트릭 설정
metric_reader = PeriodicExportingMetricReader(
    OTLPMetricExporter(endpoint="http://otel-collector:4317"),
    export_interval_millis=10000,
)
meter_provider = MeterProvider(metric_readers=[metric_reader])
metrics.set_meter_provider(meter_provider)
 
meter = metrics.get_meter("ai.orchestration")
 
# 메트릭 정의
request_counter = meter.create_counter(
    "ai.requests.total",
    description="총 AI 요청 수",
)
token_counter = meter.create_counter(
    "ai.tokens.total",
    description="총 토큰 사용량",
)
latency_histogram = meter.create_histogram(
    "ai.request.duration",
    description="요청 처리 시간 (ms)",
    unit="ms",
)
error_counter = meter.create_counter(
    "ai.errors.total",
    description="에러 발생 수",
)
 
# 메트릭 기록
async def tracked_invoke(chain, input_data, model_name: str):
    start = time.perf_counter()
    try:
        result = await chain.ainvoke(input_data)
        request_counter.add(1, {"model": model_name, "status": "success"})
        if hasattr(result, "usage_metadata"):
            token_counter.add(
                result.usage_metadata["total_tokens"],
                {"model": model_name},
            )
        return result
    except Exception as e:
        error_counter.add(1, {"model": model_name, "error_type": type(e).__name__})
        raise
    finally:
        duration = (time.perf_counter() - start) * 1000
        latency_histogram.record(duration, {"model": model_name})

from dataclasses import dataclass, field
from datetime import datetime
 
# 2026년 기준 대략적 가격 (실제 가격은 공급자 확인)
MODEL_PRICING = {
    "gpt-4o": {"input": 2.50, "output": 10.00},       # per 1M tokens
    "gpt-4o-mini": {"input": 0.15, "output": 0.60},
    "claude-3-5-sonnet": {"input": 3.00, "output": 15.00},
    "claude-3-5-haiku": {"input": 0.25, "output": 1.25},
}
 
@dataclass
class CostTracker:
    daily_budget: float = 100.0  # USD
    daily_spent: float = 0.0
    records: list = field(default_factory=list)
 
    def record_usage(
        self,
        model: str,
        input_tokens: int,
        output_tokens: int,
    ) -> float:
        pricing = MODEL_PRICING.get(model, {"input": 0, "output": 0})
        cost = (
            (input_tokens / 1_000_000) * pricing["input"]
            + (output_tokens / 1_000_000) * pricing["output"]
        )
        self.daily_spent += cost
        self.records.append({
            "timestamp": datetime.now().isoformat(),
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cost_usd": cost,
        })
        return cost
 
    def check_budget(self) -> bool:
        """예산 초과 여부 확인"""
        return self.daily_spent < self.daily_budget
 
    def get_summary(self) -> dict:
        return {
            "daily_budget": self.daily_budget,
            "daily_spent": round(self.daily_spent, 4),
            "remaining": round(self.daily_budget - self.daily_spent, 4),
            "usage_percent": round(
                (self.daily_spent / self.daily_budget) * 100, 1
            ),
            "total_requests": len(self.records),
        }

class AlertManager:
    def __init__(self, thresholds: dict):
        self.thresholds = thresholds
 
    async def check_and_alert(self, metrics: dict):
        alerts = []
 
        if metrics["error_rate"] > self.thresholds["error_rate"]:
            alerts.append({
                "severity": "critical",
                "message": f"에러율 {metrics['error_rate']:.1%}이 "
                          f"임계값 {self.thresholds['error_rate']:.1%}을 초과했습니다.",
            })
 
        if metrics["p95_latency"] > self.thresholds["p95_latency"]:
            alerts.append({
                "severity": "warning",
                "message": f"P95 지연 시간 {metrics['p95_latency']:.1f}ms이 "
                          f"임계값을 초과했습니다.",
            })
 
        if metrics["daily_cost"] > self.thresholds["daily_budget"] * 0.8:
            alerts.append({
                "severity": "warning",
                "message": f"일일 비용이 예산의 "
                          f"{(metrics['daily_cost']/self.thresholds['daily_budget'])*100:.0f}%에 도달했습니다.",
            })
 
        for alert in alerts:
            await self.send_alert(alert)