공급망 수요 예측 시스템 AI API 아키텍처 마이그레이션 가이드

사례 연구: 부산의 한 전자상거래 팀

부산에 위치한 중견 전자상거래 플랫폼 'B사'는 매일 50만 개 이상의 SKU를 대상으로 재고 수요 예측을 수행하고 있었습니다. 이 팀은 기존 AI API 공급자를 사용하며 몇 가지 심각한 문제에 직면해 있었습니다. 네트워크 지연이 평균 420ms에 달해 실시간 재고 조정 알고리즘의 효과를 크게 저하시켰고, 월간 비용이 4,200달러에 달하면서도 서비스 가용성이 안정적이지 않았습니다. 특히 인기 제품 카테고리 급성장 시 API 응답 시간이 2초 이상으로 치솟는 문제가 반복되었습니다.

저는 이 마이그레이션 프로젝트의 기술 리드를 맡아 HolySheep AI로의 전환을 주도했습니다. 로컬 결제 지원으로 해외 신용카드 없이 즉시 시작할 수 있었고, 단일 API 키로 다중 모델을 통합할 수 있다는 점이 가장 큰 매력점이었습니다. 마이그레이션 후 30일 실측치로 네트워크 지연이 180ms로 개선되었고, 월간 비용은 680달러로 84% 절감이라는 놀라운 결과를 달성했습니다.

시스템 아키텍처 개요

기존 아키텍처의 문제점

기존 시스템은 단일 AI API 공급자에게 과도하게 의존하고 있었습니다. 이로 인해 발생하는 문제들은 다음과 같았습니다. 공급자 측 장애 시 전체 예측 시스템이 멈추는 단일 장애점 문제가 있었고, 모델별 최적화가 불가능하여 모든 요청에 동일한 모델을 사용해야 했습니다. 또한 비용 대비 성능 비율이 비효율적이었습니다.

마이그레이션 후 아키텍처

새로운 아키텍처는 HolySheep AI의 단일 엔드포인트를 중심으로 설계되었습니다. 다양한 모델을 상황에 따라 선택적으로 활용할 수 있으며, 각 모델의 강점을 최대한 살릴 수 있습니다. 요청 분류기에 따라 적절한 모델로 자동 라우팅되고, 폴백 메커니즘으로 가용성이 보장됩니다.

구체적인 마이그레이션 단계

1단계: Base URL 교체 및 SDK 설정

기존 SDK 설정에서 base_url만 교체하면 됩니다. HolySheep AI는 OpenAI 호환 API를 제공하므로 코드 변경이 최소화됩니다. 다음은 Python 기반 수요 예측 시스템의 SDK 설정 예시입니다.

import os
from openai import OpenAI

HolySheep AI API 설정
client = OpenAI(
    api_key=os.environ.get("HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1"
)

def predict_demand(product_features, historical_data):
    """
    제품 특성 및 과거 데이터를 기반으로 수요 예측
    """
    prompt = f"""
    다음은 제품의 특성 및 최근 90일간 판매 데이터입니다.
    
    제품 특성:
    - 카테고리: {product_features['category']}
    - 가격대: {product_features['price_range']}
    - 계절성 지수: {product_features['seasonality_index']}
    
    최근 90일 일일 판매량:
    {historical_data}
    
    이 데이터를 분석하여 향후 30일간의 일일 수요 예측치를 제공해주세요.
    응답 형식: JSON (date, predicted_demand, confidence_interval 포함)
    """
    
    response = client.chat.completions.create(
        model="gpt-4.1",
        messages=[
            {"role": "system", "content": "당신은 공급망 최적화 전문가입니다."},
            {"role": "user", "content": prompt}
        ],
        temperature=0.3,
        max_tokens=2048
    )
    
    return response.choices[0].message.content

2단계: 카나리아 배포를 위한 모델 라우팅

카나리아 배포를 통해 기존 공급자와 HolySheep AI를 병렬로 운영하며 점진적으로 트래픽을 전환할 수 있습니다. 다음 코드는 트래픽 비율 기반 라우팅 로직을 보여줍니다.

import random
import os
from typing import Dict, List, Optional
from dataclasses import dataclass

@dataclass
class ModelConfig:
    name: str
    cost_per_million_tokens: float
    avg_latency_ms: float
    use_case: str

HolySheep AI에서 사용 가능한 모델 설정
MODEL_CONFIGS = {
    "demand_forecast_quick": ModelConfig(
        name="gemini-2.5-flash",
        cost_per_million_tokens=2.50,
        avg_latency_ms=150,
        use_case="빠른 초기 예측 및 이상치 탐지"
    ),
    "demand_forecast_detailed": ModelConfig(
        name="gpt-4.1",
        cost_per_million_tokens=8.00,
        avg_latency_ms=280,
        use_case="정밀한 장기 예측 및 시나리오 분석"
    ),
    "demand_forecast_complex": ModelConfig(
        name="claude-sonnet-4.5",
        cost_per_million_tokens=15.00,
        avg_latency_ms=350,
        use_case="복잡한 패턴 인식 및 공급업체 협상 지원"
    ),
    "demand_forecast_cost_efficient": ModelConfig(
        name="deepseek-v3.2",
        cost_per_million_tokens=0.42,
        avg_latency_ms=180,
        use_case="일상적 반복 작업 및 대량 처리"
    )
}

class CanaryRouter:
    """카나리아 배포를 위한 지능형 라우터"""
    
    def __init__(self, canary_percentage: float = 10.0):
        self.canary_percentage = canary_percentage
        self.client = OpenAI(
            api_key=os.environ.get("HOLYSHEEP_API_KEY"),
            base_url="https://api.holysheep.ai/v1"
        )
    
    def select_model(self, task_complexity: str, budget_priority: bool) -> str:
        """작업 특성 및 예산 우선순위에 따라 모델 선택"""
        
        if budget_priority:
            return "demand_forecast_cost_efficient"
        
        complexity_scores = {
            "low": 1,
            "medium": 2,
            "high": 3,
            "critical": 4
        }
        
        score = complexity_scores.get(task_complexity, 2)
        
        if score <= 2:
            return "demand_forecast_quick"
        elif score == 3:
            return random.choice(["demand_forecast_quick", "demand_forecast_detailed"])
        else:
            return "demand_forecast_detailed"
    
    def route_request(self, request_data: Dict) -> Dict:
        """요청을 적절한 모델로 라우팅"""
        
        # 카나리아 배포: 지정된 비율만큼 HolySheep AI 사용
        if random.random() * 100 < self.canary_percentage:
            model_key = self.select_model(
                request_data.get("complexity", "medium"),
                request_data.get("budget_first", False)
            )
            
            config = MODEL_CONFIGS[model_key]
            
            # HolySheep AI를 통한 요청 처리
            response = self._call_holysheep_api(config.name, request_data)
            
            return {
                "source": "holysheep",
                "model": config.name,
                "response": response,
                "estimated_cost": self._estimate_cost(response, config.cost_per_million_tokens)
            }
        else:
            # 기존 공급자 폴백
            return self._call_existing_provider(request_data)
    
    def _call_holysheep_api(self, model_name: str, data: Dict) -> str:
        """HolySheep AI API 호출"""
        
        response = self.client.chat.completions.create(
            model=model_name,
            messages=[
                {"role": "system", "content": "공급망 수요 예측 전문가"},
                {"role": "user", "content": str(data)}
            ],
            temperature=0.3,
            max_tokens=2048
        )
        
        return response.choices[0].message.content
    
    def _estimate_cost(self, response: str, cost_per_mtok: float) -> float:
        """토큰 기반 비용 추정"""
        # 대략적인 토큰 수 계산 (실제 사용 시 응답 헤더의 usage 정보 활용 권장)
        estimated_tokens = len(response) // 4
        return (estimated_tokens / 1_000_000) * cost_per_mtok

사용 예시
router = CanaryRouter(canary_percentage=10.0)

test_request = {
    "product_id": "SKU-12345",
    "complexity": "high",
    "budget_first": False,
    "data": {"sales_history": [120, 135, 142, 128, 156, 149]}
}

result = router.route_request(test_request)
print(f"라우팅 결과: {result['source']}, 모델: {result['model']}")

3단계: API 키 로테이션 및 보안 설정

프로덕션 환경에서는 정기적인 키 로테이션과 함께 환경 변수 관리 전략을 수립해야 합니다. HolySheep AI는 키 관리 대시보드를 통해 간편하게 새 키를 생성하고 기존 키를 폐기할 수 있습니다.

import os
import time
import hashlib
from functools import wraps
from typing import Callable, Any
from datetime import datetime, timedelta

class APIKeyManager:
    """HolySheep AI API 키 관리 및 로테이션"""
    
    def __init__(self):
        self.primary_key = os.environ.get("HOLYSHEEP_API_KEY")
        self.secondary_key = os.environ.get("HOLYSHEEP_API_KEY_BACKUP")
        self.key_version = int(os.environ.get("KEY_VERSION", "1"))
        self.last_rotation = datetime.now()
        self.rotation_interval_days = 30
    
    def should_rotate(self) -> bool:
        """키 로테이션 필요 여부 확인"""
        days_since_rotation = (datetime.now() - self.last_rotation).days
        return days_since_rotation >= self.rotation_interval_days
    
    def get_active_key(self) -> str:
        """현재 활성 키 반환"""
        return self.primary_key
    
    def verify_key_health(self, test_endpoint: str = "https://api.holysheep.ai/v1/models") -> bool:
        """API 키 유효성 검사"""
        import requests
        
        headers = {
            "Authorization": f"Bearer {self.get_active_key()}",
            "Content-Type": "application/json"
        }
        
        try:
            response = requests.get(test_endpoint, headers=headers, timeout=5)
            return response.status_code == 200
        except requests.RequestException:
            return False
    
    def rotate_keys(self) -> dict:
        """키 로테이션 실행 (HolySheep AI 대시보드에서 새 키 생성 후 실행)"""
        new_key = os.environ.get("HOLYSHEEP_API_KEY_NEW")
        
        if not new_key:
            raise ValueError("새 API 키가 환경 변수로 설정되지 않았습니다")
        
        # 키 교체
        self.secondary_key = self.primary_key
        self.primary_key = new_key
        self.key_version += 1
        self.last_rotation = datetime.now()
        
        # 환경 변수 업데이트
        os.environ["HOLYSHEEP_API_KEY"] = self.primary_key
        os.environ["HOLYSHEEP_API_KEY_BACKUP"] = self.secondary_key
        os.environ["KEY_VERSION"] = str(self.key_version)
        
        return {
            "status": "rotated",
            "version": self.key_version,
            "timestamp": self.last_rotation.isoformat()
        }

def with_key_health_check(func: Callable) -> Callable:
    """API 키 상태 확인 데코레이터"""
    @wraps(func)
    def wrapper(*args, **kwargs) -> Any:
        key_manager = APIKeyManager()
        
        # 키 상태 확인
        if not key_manager.verify_key_health():
            # 폴백 키로 전환
            if key_manager.secondary_key:
                os.environ["HOLYSHEEP_API_KEY"] = key_manager.secondary_key
        
        # 로테이션 필요 시 경고
        if key_manager.should_rotate():
            print(f"경고: API 키 로테이션 필요. "
                  f"마지막 로테이션: {key_manager.last_rotation}")
        
        return func(*args, **kwargs)
    
    return wrapper

사용 예시
@with_key_health_check
def run_demand_forecast():
    """수요 예측 배치 잡"""
    client = OpenAI(
        api_key=os.environ.get("HOLYSHEEP_API_KEY"),
        base_url="https://api.holysheep.ai/v1"
    )
    
    # 예측 로직 수행
    response = client.chat.completions.create(
        model="gpt-4.1",
        messages=[{"role": "user", "content": "오늘의 전체 SKU 수요 예측 실행"}]
    )
    
    return response.choices[0].message.content

스케줄러 설정 (매일 자정 실행)
*/0 * * * * /usr/bin/python3 /app/forecast.py >> /var/log/forecast.log 2>&1

비용 최적화 전략

모델별 최적 활용 가이드

HolySheep AI의 다양한 모델을 상황에 맞게 활용하면 비용을 극대화할 수 있습니다. 다음 표는 B사 팀이 실제 사용한 모델 배분 전략입니다.

• Gemini 2.5 Flash ($2.50/MTok): 실시간 이상치 탐지, 긴급 재주문 알림 — 일일 요청의 60%
• DeepSeek V3.2 ($0.42/MTok): 대량 데이터 전처리, 반복적 카테고리 분류 — 일일 요청의 25%
• GPT-4.1 ($8/MTok): 복잡한 수요 시나리오 분석, 신규 제품 출시 예측 — 일일 요청의 10%
• Claude Sonnet 4.5 ($15/MTok): 공급업체 협상 전략 수립, 고위험 재고 의사결정 — 일일 요청의 5%

이 전략을 적용한 결과 월간 비용이 4,200달러에서 680달러로 84% 절감되었습니다. 특히 DeepSeek V3.2의 도입이 가장 큰 비용 절감 효과(약 60% 감소)를 가져왔습니다.

성능 모니터링 및 최적화

마이그레이션 후 지속적인 모니터링을 위해 응답 시간, 토큰 사용량, 에러율 등의 지표를 추적해야 합니다. HolySheep AI 대시보드에서 실시간 사용량을 확인할 수 있으며, 커스텀 모니터링 대시보드도 구성할 수 있습니다.

import time
import logging
from collections import defaultdict
from datetime import datetime
import threading

class PerformanceMonitor:
    """성능 모니터링 및 로깅"""
    
    def __init__(self):
        self.metrics = defaultdict(list)
        self.lock = threading.Lock()
        self.logger = logging.getLogger("supply_chain_monitor")
        
    def record_request(self, model: str, latency_ms: float, tokens_used: int, 
                       success: bool, error_message: str = None):
        """요청 메트릭 기록"""
        
        with self.lock:
            self.metrics["requests"].append({
                "timestamp": datetime.now().isoformat(),
                "model": model,
                "latency_ms": latency_ms,
                "tokens": tokens_used,
                "success": success,
                "error": error_message
            })
            
            # 30일 이전 데이터 삭제
            cutoff = datetime.now().timestamp() - (30 * 24 * 60 * 60)
            self.metrics["requests"] = [
                m for m in self.metrics["requests"]
                if datetime.fromisoformat(m["timestamp"]).timestamp() > cutoff
            ]
    
    def get_average_latency(self, model: str = None, days: int = 7) -> float:
        """평균 응답 시간 조회"""
        
        with self.lock:
            cutoff = datetime.now().timestamp() - (days * 24 * 60 * 60)
            relevant = [
                m for m in self.metrics["requests"]
                if datetime.fromisoformat(m["timestamp"]).timestamp() > cutoff
                and m["success"]
                and (model is None or m["model"] == model)
            ]
            
            if not relevant:
                return 0.0
            
            return sum(m["latency_ms"] for m in relevant) / len(relevant)
    
    def get_cost_summary(self, days: int = 30) -> dict:
        """비용 요약 조회"""
        
        model_costs = {
            "gpt-4.1": 8.00,
            "claude-sonnet-4.5": 15.00,
            "gemini-2.5-flash": 2.50,
            "deepseek-v3.2": 0.42
        }
        
        with self.lock:
            cutoff = datetime.now().timestamp() - (days * 24 * 60 * 60)
            relevant = [
                m for m in self.metrics["requests"]
                if datetime.fromisoformat(m["timestamp"]).timestamp() > cutoff
            ]
            
            total_cost = 0.0
            model_breakdown = defaultdict(lambda: {"tokens": 0, "cost": 0.0})
            
            for m in relevant:
                model = m["model"]
                tokens = m["tokens"]
                cost_per_mtok = model_costs.get(model, 0)
                
                token_cost = (tokens / 1_000_000) * cost_per_mtok
                total_cost += token_cost
                
                model_breakdown[model]["tokens"] += tokens
                model_breakdown[model]["cost"] += token_cost
            
            return {
                "total_cost": total_cost,
                "period_days": days,
                "monthly_projected": (total_cost / days) * 30,
                "by_model": dict(model_breakdown)
            }
    
    def health_check(self) -> dict:
        """시스템 상태 확인"""
        
        with self.lock:
            recent = [
                m for m in self.metrics["requests"]
                if (datetime.now() - datetime.fromisoformat(m["timestamp"])).seconds < 300
            ]
            
            total_requests = len(recent)
            successful = sum(1 for m in recent if m["success"])
            failed = total_requests - successful
            
            avg_latency = (
                sum(m["latency_ms"] for m in recent if m["success"]) / successful
                if successful > 0 else 0
            )
            
            return {
                "status": "healthy" if failed / max(total_requests, 1) < 0.05 else "degraded",
                "requests_last_5min": total_requests,
                "success_rate": successful / max(total_requests, 1),
                "average_latency_ms": avg_latency,
                "active_models": len(set(m["model"] for m in recent))
            }

모니터링 인스턴스
monitor = PerformanceMonitor()

def monitored_api_call(client, model: str, prompt: str) -> str:
    """모니터링이 포함된 API 호출 래퍼"""
    
    start_time = time.time()
    tokens_used = 0
    success = False
    error = None
    response = None
    
    try:
        result = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            max_tokens=2048
        )
        
        response = result.choices[0].message.content
        tokens_used = result.usage.total_tokens if hasattr(result, 'usage') else 0
        success = True
        
    except Exception as e:
        error = str(e)
        monitor.logger.error(f"API 호출 실패: {error}")
    
    latency_ms = (time.time() - start_time) * 1000
    
    monitor.record_request(
        model=model,
        latency_ms=latency_ms,
        tokens_used=tokens_used,
        success=success,
        error_message=error
    )
    
    if not success:
        raise RuntimeError(f"API 호출 실패: {error}")
    
    return response

상태 확인 스케줄러 (매 5분마다)
curl -X POST http://localhost:8080/health-check

자주 발생하는 오류와 해결책

1. API 키 인증 실패 오류

증상: 401 Unauthorized 또는 "Invalid API key" 에러 메시지가 반환됩니다.

원인: 환경 변수에 설정된 API 키가 유효하지 않거나, 키 앞에 불필요한 공백이나 따옴표가 포함된 경우입니다.

해결 코드:

import os

잘못된 설정 예시
HOLYSHEEP_API_KEY = "sk-xxxxx"  # 따옴표가 포함됨
HOLYSHEEP_API_KEY = sk-xxxxx    # 따옴표 없음, 공백 가능성

올바른 설정
os.environ["HOLYSHEEP_API_KEY"] = os.environ.get("HOLYSHEEP_API_KEY", "").strip()

키 검증
if not os.environ.get("HOLYSHEEP_API_KEY"):
    raise ValueError("HOLYSHEEP_API_KEY 환경 변수가 설정되지 않았습니다")

키 포맷 검증
api_key = os.environ["HOLYSHEEP_API_KEY"]
if not api_key.startswith("hs_"):
    print(f"경고: API 키 형식이 다를 수 있습니다. HolySheep 대시보드에서 확인하세요.")

2. Rate Limit 초과 에러

증상: 429 Too Many Requests 에러가 발생하며 요청이 거부됩니다.

원인: 짧은 시간内に了大量의 요청을 전송하여 HolySheep AI의 Rate Limit에 도달했습니다. 대량 배치 처리 시 자주 발생합니다.

해결 코드:

import time
import asyncio
from typing import List, Callable, Any
from ratelimit import limits, sleep_and_retry

class RateLimitedClient:
    """Rate Limit을 고려한 HolySheep API 클라이언트"""
    
    def __init__(self, requests_per_minute: int = 60):
        self.rpm = requests_per_minute
        self.client = OpenAI(
            api_key=os.environ.get("HOLYSHEEP_API_KEY"),
            base_url="https://api.holysheep.ai/v1"
        )
    
    @sleep_and_retry
    @limits(calls=60, period=60)
    def _rate_limited_call(self, model: str, prompt: str) -> dict:
        """Rate Limit 적용된 API 호출"""
        return self.client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            max_tokens=2048
        )
    
    def batch_process(self, items: List[dict], model: str = "gemini-2.5-flash") -> List[dict]:
        """배치 처리 with 자동 재시도 및 Rate Limit 처리"""
        
        results = []
        max_retries = 3
        
        for idx, item in enumerate(items):
            for attempt in range(max_retries):
                try:
                    response = self._rate_limited_call(model, str(item))
                    results.append({
                        "index": idx,
                        "result": response.choices[0].message.content,
                        "status": "success"
                    })
                    break
                    
                except Exception as e:
                    if "429" in str(e) or "rate limit" in str(e).lower():
                        wait_time = (attempt + 1) * 2  # 지수 백오프
                        print(f"Rate Limit 도달. {wait_time}초 후 재시도 ({attempt + 1}/{max_retries})")
                        time.sleep(wait_time)
                    else:
                        results.append({
                            "index": idx,
                            "error": str(e),
                            "status": "failed"
                        })
                        break
            else:
                results.append({
                    "index": idx,
                    "error": "Max retries exceeded",
                    "status": "failed"
                })
        
        return results

사용 예시
batch_items = [{"sku": f"ITEM-{i}", "quantity": i * 10} for i in range(100)]
client = RateLimitedClient(requests_per_minute=60)
results = client.batch_process(batch_items)

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import socket

class ResilientClient:
    """네트워크 오류에 강한 HolySheep AI 클라이언트"""
    
    def __init__(self, timeout: int = 30, max_retries: int = 3):
        self.timeout = timeout
        
        self.session = requests.Session()
        
        # 재시도 전략 설정
        retry_strategy = Retry(
            total=max_retries,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["POST", "GET"]
        )
        
        adapter = HTTPAdapter(max_retries=retry_strategy)
        self.session.mount("https://", adapter)
        self.session.mount("http://", adapter)
    
    def create_completion(self, model: str, prompt: str) -> dict:
        """타임아웃 및 재시도가 적용된 API 호출"""
        
        # Base URL 검증
        base_url = "https://api.holysheep.ai/v1"
        
        try:
            response = self.session.post(
                f"{base_url}/chat/completions",
                headers={
                    "Authorization": f"Bearer {os.environ.get('HOLYSHEEP_API_KEY')}",
                    "Content-Type": "application/json"
                },
                json={
                    "model": model,
                    "messages": [{"role": "user", "content": prompt}],
                    "max_tokens": 2048
                },
                timeout=self.timeout
            )
            
            response.raise_for_status()
            return response.json()
            
        except requests.Timeout:
            # 타임아웃 발생 시 폴백 모델 시도
            print(f"타임아웃 발생. 폴백 모델로 재시도...")
            return self._fallback_call(model, prompt)
            
        except requests.ConnectionError as e:
            # 연결 오류 시 DNS 및 네트워크 확인
            print(f"연결 오류 발생: {e}")
            self._diagnose_network_issue()
            raise
    
    def _fallback_call(self, original_model: str, prompt: str) -> dict:
        """폴백 모델로 재시도"""
        
        # 비용 효율적인 폴백 모델 우선 선택
        fallback_models = ["gemini-2.5-flash", "deepseek-v3.2"]
        
        for model in fallback_models:
            try:
                print(f"폴백 모델 {model} 시도...")
                return self.create_completion(model, prompt)
            except Exception:
                continue
        
        raise RuntimeError("모든 폴백 모델 실패")
    
    def _diagnose_network_issue(self):
        """네트워크 문제 진단"""
        
        # DNS 확인
        try:
            ip = socket.gethostbyname("api.holysheep.ai")
            print(f"DNS 확인됨: api.holysheep.ai -> {ip}")
        except socket.gaierror:
            print("DNS 확인 실패: HolySheheep AI 도메인 해결 불가")
        
        # 포트 확인
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(5)
            result = sock.connect_ex(("api.holysheep.ai", 443))
            sock.close()
            
            if result == 0:
                print("포트 443 연결 성공")
            else:
                print(f"포트 443 연결 실패: {result}")
        except Exception as e:
            print(f"포트 연결 테스트 실패: {e}")

사용 예시
resilient_client = ResilientClient(timeout=30, max_retries=3)
result = resilient_client.create_completion("gpt-4.1", "수요 예측 실행")

import json
import re
from typing import Any, Optional

class ResponseParser:
    """다양한 응답 형식을 처리하는 파서"""
    
    @staticmethod
    def parse_json_response(response_text: str) -> Optional[dict]:
        """JSON 응답 파싱 (다양한 형식 처리)"""
        
        # 1순위: 완전한 JSON 객체 파싱 시도
        try:
            return json.loads(response_text)
        except json.JSONDecodeError:
            pass
        
        # 2순위: Markdown 코드 블록 내 JSON 추출
        code_block_pattern = r'``(?:json)?\s*([\s\S]*?)\s*``'
        matches = re.findall(code_block_pattern, response_text)
        
        for match in matches:
            try:
                return json.loads(match.strip())
            except json.JSONDecodeError:
                continue
        
        # 3순위: 첫 번째 { 및 } 사이 텍스트 추출
        json_pattern = r'\{[\s\S]*\}'
        match = re.search(json_pattern, response_text)
        
        if match:
            try:
                return json.loads(match.group())
            except json.JSONDecodeError:
                pass
        
        return None
    
    @staticmethod
    def validate_demand_forecast(data: dict) -> dict:
        """수요 예측 응답 검증 및 정규화"""
        
        required_fields = ["date", "predicted_demand", "confidence_interval"]
        
        # 필드 검증
        missing = [f for f in required_fields if f not in data]
        if missing:
            raise ValueError(f"필수 필드 누락: {missing}")
        
        # 타입 검증 및 정규화
        validated = {
            "date": data["date"],
            "predicted_demand": float(data["predicted_demand"]),
            "confidence_interval": {
                "low": float(data["confidence_interval"].get("low", 0)),
                "high": float(data["confidence_interval"].get("high", 0))
            }
        }
        
        return validated
    
    @staticmethod
    def extract_and_validate(response: Any, expected_format: str = "json") -> Any:
        """응답 형식 자동 감지 및 검증"""
        
        if isinstance(response, dict):
            response_text = json.dumps(response)
        else:
            response_text = str(response)
        
        if expected_format == "json":
            parsed = ResponseParser.parse_json_response(response_text)
            if parsed:
                return ResponseParser.validate_demand_forecast(parsed)
            else:
                raise ValueError(f"JSON 파싱 실패: {response_text[:200]}")
        
        return response_text

사용 예시
parser = ResponseParser()

테스트
test_responses = [
    '{"date": "2024-01-15", "predicted_demand": 150, "confidence_interval": {"low": 120, "high": 180}}',
    '``json\n{"date": "2024-01-15", "predicted_demand": 150, "confidence_interval": {"low": 120, "high": 180}}\n``',
    '날짜는 2024-01-15이고, 예측 수요량은 150개입니다. 신뢰 구간은 120~180입니다.'
]

for resp in test_responses:
    try:
        result = parser.extract_and_validate(resp, "json")
        print(f"파싱 성공: {result}")
    except Exception as e:
        print(f"파싱 실패: {e}")