바이브코더를 위한 프로덕션 생존 가이드

import logging
import time
from functools import wraps

# 기본 로깅 설정
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s | %(levelname)s | %(message)s'
)
logger = logging.getLogger(__name__)

def log_request(func):
    """API 요청 로깅 데코레이터"""
    @wraps(func)
    async def wrapper(*args, **kwargs):
        start_time = time.time()
        request_id = generate_request_id()

        try:
            result = await func(*args, **kwargs)
            elapsed = time.time() - start_time

            logger.info(f"[{request_id}] {func.__name__} | "
                       f"status=200 | duration={elapsed:.3f}s")
            return result

        except Exception as e:
            elapsed = time.time() - start_time
            logger.error(f"[{request_id}] {func.__name__} | "
                        f"status=500 | duration={elapsed:.3f}s | "
                        f"error={type(e).__name__}: {str(e)}")
            raise

    return wrapper

# 사용 예시
@log_request
async def call_llm_api(prompt: str):
    response = await openai_client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

from pydantic_settings import BaseSettings
from pydantic import field_validator

class Settings(BaseSettings):
    """필수 환경변수 - 하나라도 없으면 앱 시작 불가"""

    # API Keys
    OPENAI_API_KEY: str
    ANTHROPIC_API_KEY: str

    # Database
    DATABASE_URL: str

    # Optional with defaults
    MAX_TOKENS: int = 4000
    TIMEOUT_SECONDS: int = 30

    @field_validator('OPENAI_API_KEY', 'ANTHROPIC_API_KEY')
    @classmethod
    def validate_api_key(cls, v: str, info) -> str:
        if not v or v.startswith('sk-xxx'):
            raise ValueError(f"{info.field_name} is not set or is a placeholder")
        return v

    @field_validator('DATABASE_URL')
    @classmethod
    def validate_db_url(cls, v: str) -> str:
        if 'localhost' in v and not v.startswith('postgresql://'):
            raise ValueError("Production DATABASE_URL should not use localhost")
        return v

    class Config:
        env_file = ".env"

# 앱 시작 시 검증 - 실패하면 서버가 뜨지 않음
try:
    settings = Settings()
    print("Environment validated successfully")
except Exception as e:
    print(f"FATAL: Environment validation failed - {e}")
    exit(1)

import httpx
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type
)

# 타임아웃 설정된 HTTP 클라이언트
http_client = httpx.AsyncClient(
    timeout=httpx.Timeout(
        connect=5.0,    # 연결 타임아웃: 5초
        read=30.0,      # 읽기 타임아웃: 30초
        write=10.0,     # 쓰기 타임아웃: 10초
        pool=5.0        # 커넥션 풀 타임아웃: 5초
    )
)

# 재시도 데코레이터 (지수 백오프)
@retry(
    stop=stop_after_attempt(3),                    # 최대 3회
    wait=wait_exponential(multiplier=1, max=10),   # 1s → 2s → 4s
    retry=retry_if_exception_type((
        httpx.TimeoutException,
        httpx.NetworkError,
    )),
    reraise=True
)
async def call_external_api(url: str, payload: dict) -> dict:
    """타임아웃 + 재시도가 적용된 외부 API 호출"""
    response = await http_client.post(url, json=payload)

    # 4xx 에러는 재시도하지 않음 (클라이언트 잘못)
    if 400 <= response.status_code < 500:
        raise ValueError(f"Client error: {response.status_code}")

    response.raise_for_status()
    return response.json()

# 폴백 패턴
async def call_with_fallback(prompt: str) -> str:
    """메인 실패 시 폴백으로 전환"""
    try:
        return await call_openai(prompt)
    except Exception as e:
        logger.warning(f"OpenAI failed, falling back to Claude: {e}")
        try:
            return await call_anthropic(prompt)
        except Exception as e2:
            logger.error(f"All LLM providers failed: {e2}")
            return "죄송합니다. 일시적인 오류가 발생했습니다. 잠시 후 다시 시도해주세요."

from datetime import datetime, timedelta
from collections import defaultdict
import hashlib

class RateLimiter:
    """간단한 인메모리 Rate Limiter"""

    def __init__(self, max_requests: int = 100, window_seconds: int = 60):
        self.max_requests = max_requests
        self.window = timedelta(seconds=window_seconds)
        self.requests = defaultdict(list)

    def is_allowed(self, user_id: str) -> bool:
        now = datetime.now()
        cutoff = now - self.window

        # 윈도우 밖의 요청 제거
        self.requests[user_id] = [
            t for t in self.requests[user_id] if t > cutoff
        ]

        if len(self.requests[user_id]) >= self.max_requests:
            return False

        self.requests[user_id].append(now)
        return True

class CostGuard:
    """비용 가드레일"""

    def __init__(self, daily_limit: float = 100.0):
        self.daily_limit = daily_limit
        self.daily_cost = 0.0
        self.last_reset = datetime.now().date()

    def check_and_add(self, estimated_cost: float) -> bool:
        today = datetime.now().date()

        # 날짜 바뀌면 리셋
        if today > self.last_reset:
            self.daily_cost = 0.0
            self.last_reset = today

        # 한도 초과 체크
        if self.daily_cost + estimated_cost > self.daily_limit:
            logger.warning(f"Daily cost limit reached: ${self.daily_cost:.2f}")
            return False

        self.daily_cost += estimated_cost

        # 80% 도달 시 경고
        if self.daily_cost > self.daily_limit * 0.8:
            logger.warning(f"Cost warning: 80% of daily limit used (${self.daily_cost:.2f})")

        return True

class IdempotencyGuard:
    """중복 요청 방지"""

    def __init__(self, ttl_seconds: int = 300):
        self.cache = {}  # 실제로는 Redis 사용 권장
        self.ttl = timedelta(seconds=ttl_seconds)

    def get_key(self, user_id: str, request_data: dict) -> str:
        data_str = f"{user_id}:{sorted(request_data.items())}"
        return hashlib.sha256(data_str.encode()).hexdigest()

    def check_duplicate(self, user_id: str, request_data: dict) -> tuple[bool, any]:
        key = self.get_key(user_id, request_data)
        now = datetime.now()

        if key in self.cache:
            cached_time, cached_result = self.cache[key]
            if now - cached_time < self.ttl:
                logger.info(f"Duplicate request detected, returning cached result")
                return True, cached_result

        return False, None

    def store_result(self, user_id: str, request_data: dict, result: any):
        key = self.get_key(user_id, request_data)
        self.cache[key] = (datetime.now(), result)

# 사용 예시
rate_limiter = RateLimiter(max_requests=100, window_seconds=60)
cost_guard = CostGuard(daily_limit=50.0)
idempotency = IdempotencyGuard()

async def handle_request(user_id: str, request_data: dict):
    # 1. Rate Limit 체크
    if not rate_limiter.is_allowed(user_id):
        return {"error": "Too many requests. Please wait."}, 429

    # 2. 중복 요청 체크
    is_duplicate, cached = idempotency.check_duplicate(user_id, request_data)
    if is_duplicate:
        return cached, 200

    # 3. 비용 체크
    estimated_cost = estimate_cost(request_data)
    if not cost_guard.check_and_add(estimated_cost):
        return {"error": "Daily limit exceeded. Try again tomorrow."}, 503

    # 4. 실제 처리
    result = await process_request(request_data)

    # 5. 결과 캐싱
    idempotency.store_result(user_id, request_data, result)

    return result, 200

import tiktoken

class TokenGovernor:
    """토큰 사용량 관리"""

    def __init__(
        self,
        max_input_tokens: int = 4000,
        max_output_tokens: int = 1000,
        max_history_messages: int = 10
    ):
        self.max_input = max_input_tokens
        self.max_output = max_output_tokens
        self.max_history = max_history_messages
        self.encoder = tiktoken.encoding_for_model("gpt-4")

    def count_tokens(self, text: str) -> int:
        return len(self.encoder.encode(text))

    def validate_input(self, prompt: str) -> tuple[bool, str]:
        """입력 검증 - API 호출 전에 체크"""
        token_count = self.count_tokens(prompt)

        if token_count > self.max_input:
            return False, f"입력이 너무 깁니다. ({token_count} tokens > {self.max_input} limit)"

        return True, ""

    def trim_history(self, messages: list[dict]) -> list[dict]:
        """대화 히스토리 자르기 - 최근 N개만 유지"""
        if len(messages) <= self.max_history:
            return messages

        # 시스템 메시지는 항상 유지
        system_msgs = [m for m in messages if m.get("role") == "system"]
        other_msgs = [m for m in messages if m.get("role") != "system"]

        # 최근 메시지만 유지
        trimmed = other_msgs[-(self.max_history - len(system_msgs)):]

        return system_msgs + trimmed

    def summarize_if_needed(self, messages: list[dict]) -> list[dict]:
        """토큰 초과 시 이전 대화 요약"""
        total_tokens = sum(self.count_tokens(m.get("content", "")) for m in messages)

        if total_tokens <= self.max_input:
            return messages

        # 시스템 + 최근 2개 메시지 보존
        system_msgs = [m for m in messages if m.get("role") == "system"]
        recent = [m for m in messages if m.get("role") != "system"][-2:]
        old_msgs = [m for m in messages if m.get("role") != "system"][:-2]

        if not old_msgs:
            return messages

        # 이전 대화 요약
        old_content = "\n".join(m.get("content", "") for m in old_msgs)
        summary = f"[이전 대화 요약: {old_content[:500]}...]"

        summary_msg = {"role": "system", "content": summary}

        return system_msgs + [summary_msg] + recent

# 사용 예시
governor = TokenGovernor(
    max_input_tokens=4000,
    max_output_tokens=1000,
    max_history_messages=10
)

async def chat(user_input: str, history: list[dict]) -> str:
    # 1. 입력 검증
    is_valid, error_msg = governor.validate_input(user_input)
    if not is_valid:
        return error_msg

    # 2. 히스토리 정리
    history = governor.trim_history(history)
    history = governor.summarize_if_needed(history)

    # 3. 새 메시지 추가
    history.append({"role": "user", "content": user_input})

    # 4. API 호출
    response = await openai_client.chat.completions.create(
        model="gpt-4",
        messages=history,
        max_tokens=governor.max_output
    )

    return response.choices[0].message.content