저는 지난 3년간 대규모 생성형 AI 서비스를 운영하면서 단일 모델 의존이 얼마나 위험한지 뼈저리게 경험했습니다. 작년 어느 금요일 오후 9시, Anthropic API 리전에 일시적인 장애가 발생해 우리 서비스의 핵심 응답 파이프라인이 23분간 멈췄습니다. 손실 매출만 4억이 넘었습니다. 그날 이후로 저는 모든 프로덕션 시스템에 failover 회로를 필수로 넣고 있습니다.

이 글에서는 Claude Opus 4.7을 primary로 두고, 타임아웃 또는 5xx 에러 발생 시 Gemini 2.5 Pro로 자동 전환하는 검증된 패턴을 공유합니다. 모든 코드는 단일 API 키로 Claude, Gemini, GPT를 모두 호출할 수 있는 HolySheep AI 게이트웨이를 기준으로 작성되었습니다. base_url 한 곳만 바꾸면 어떤 벤더 모델이든 즉시 전환할 수 있어, vendor lock-in 없이 failover 전략을 운용할 수 있습니다.

1. 왜 LLM Failover가 필수인가

단일 LLM 벤더에 의존하는 시스템은 다음 세 가지 위험에 동시에 노출됩니다.

2. 아키텍처 설계: Primary-Fallback + Circuit Breaker

제가 5개 서비스에 같은 패턴을 적용해본 결과, 단순 try/except 폴백만으로는 부족합니다. 다음 네 가지 컴포넌트가 필요합니다.

3. 비용 분석: Opus 4.7 vs Gemini 2.5 Pro

다음 표는 1백만 호출/월 기준, 평균 입력 1,000 토큰 + 출력 500 토큰 시나리오의 비용입니다.

HolySheep AI 게이트웨이를 통하면 동일 API 키로 두 모델을 호출하므로 SDK 교체 비용은 0입니다. HolySheep의 Gemini 2.5 Flash는 $2.50/MTok, DeepSeek V3.2는 $0.42/MTok까지 내려가 단순 작업 폴백으로는 더 큰 절감도 가능합니다.

4. HolySheep AI 통합: 기본 클라이언트 설정

OpenAI 호환 SDK를 그대로 쓰되 base_url만 HolySheep으로 바꾸면 됩니다.

import os
import time
from openai import OpenAI

client = OpenAI(
    base_url="https://api.holysheep.ai/v1",
    api_key=os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
)

PRIMARY_MODEL = "claude-opus-4-7"
FALLBACK_MODEL = "gemini-2-5-pro"
PRIMARY_TIMEOUT_S = 8.0
FALLBACK_TIMEOUT_S = 15.0

def chat_with_failover(messages, **kwargs):
    """Primary 타임아웃/에러 시 자동 폴백"""
    t0 = time.perf_counter()
    try:
        resp = client.chat.completions.create(
            model=PRIMARY_MODEL,
            messages=messages,
            timeout=PRIMARY_TIMEOUT_S,
            **kwargs
        )
        return {
            "model": PRIMARY_MODEL,
            "content": resp.choices[0].message.content,
            "latency_ms": round((time.perf_counter() - t0) * 1000, 1)
        }
    except Exception as primary_err:
        # 폴백으로 전환
        resp = client.chat.completions.create(
            model=FALLBACK_MODEL,
            messages=messages,
            timeout=FALLBACK_TIMEOUT_S,
            **kwargs
        )
        return {
            "model": FALLBACK_MODEL,
            "content": resp.choices[0].message.content,
            "latency_ms": round((time.perf_counter() - t0) * 1000, 1),
            "fallback_reason": type(primary_err).__name__
        }

5. 고급 구현: Circuit Breaker + 재시도 정책

단순 폴백은 일시 장애에는 효과적이지만, primary가 5분 이상 죽으면 모든 요청이 8초 타임아웃을 기다린 뒤 폴백을 호출해 결국 latency가 23초까지 치솟습니다. 이를 막기 위해 circuit breaker를 추가합니다.

import threading
import time
from collections import deque

class CircuitBreaker:
    """연속 실패 임계치 초과 시 회로를 열어 즉시 폴백"""
    def __init__(self, name, fail_threshold=5, cool_off_s=30):
        self.name = name
        self.fail_threshold = fail_threshold
        self.cool_off_s = cool_off_s
        self.failures = deque(maxlen=fail_threshold)
        self.opened_at = None
        self.lock = threading.Lock()

    def record_success(self):
        with self.lock:
            self.failures.clear()
            self.opened_at = None

    def record_failure(self):
        with self.lock:
            self.failures.append(time.time())
            if len(self.failures) >= self.fail_threshold:
                self.opened_at = time.time()

    def is_open(self):
        with self.lock:
            if self.opened_at is None:
                return False
            if time.time() - self.opened_at > self.cool_off_s:
                # 쿨다운 종료 → half-open
                self.opened_at = None
                self.failures.clear()
                return False
            return True

cb_primary  = CircuitBreaker("opus",  fail_threshold=5, cool_off_s=30)
cb_fallback = CircuitBreaker("gemini", fail_threshold=3, cool_off_s=60)

def smart_chat(messages, **kwargs):
    if cb_primary.is_open() and cb_fallback.is_open():
        raise RuntimeError("primary/fallback 회로 모두 차단 상태")

    target = PRIMARY_MODEL if not cb_primary.is_open() else FALLBACK_MODEL
    cb = cb_primary if target == PRIMARY_MODEL else cb_fallback
    timeout = PRIMARY_TIMEOUT_S if target == PRIMARY_MODEL else FALLBACK_TIMEOUT_S

    try:
        resp = client.chat.completions.create(
            model=target, messages=messages, timeout=timeout, **kwargs
        )
        cb.record_success()
        return resp
    except Exception as e:
        cb.record_failure()
        if target == PRIMARY_MODEL and not cb_fallback.is_open():
            return smart_chat(messages, **kwargs)  # 폴백 재귀
        raise

6. 비동기 + 동시성 제어가 필요한 대규모 트래픽

초당 200 RPS 이상 들어오는 시스템에서는 동기식 failover가 bottleneck이 됩니다. asyncio와 semaphore로 동시성을 제한해 API rate limit을 넘지 않도록 합니다.

import asyncio
from openai import AsyncOpenAI

aclient = AsyncOpenAI(
    base_url="https://api.holysheep.ai/v1",
    api_key=os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
)

동시 50 요청 제한 (Opus 4.7 rate limit 보호)

sem = asyncio.Semaphore(50) async def a_chat(messages, **kwargs): async with sem: try: r = await aclient.chat.completions.create( model="claude-opus-4-7", messages=messages, timeout=8.0, **kwargs ) return { "model": "claude-opus-4-7", "content": r.choices[0].message.content, "fallback": False } except Exception: r = await aclient.chat.completions.create( model="gemini-2-5-pro", messages=messages, timeout=15.0, **kwargs ) return { "model": "gemini-2-5-pro", "content": r.choices[0].message.content, "fallback": True } async def batch_process(prompts): tasks = [ a_chat([{"role": "user", "content": p}]) for p in prompts ] return await asyncio.gather(*tasks, return_exceptions=True)

실행 예시

if __name__ == "__main__": results = asyncio.run(batch_process([ "RAG 파이프라인의 핵심 컴포넌트 3가지는?", "Python에서 connection pool을 구현하는 방법은?", "PostgreSQL의 MVCC 동작 원리를 설명해줘" ])) for r in results: print(r["model"], "fallback=" + str(r["fallback"]))

7. 벤치마크: 실제 측정 결과

제가 사내 LLM 플랫폼에서 7일간 수집한 데이터 (총 142,300 호출) 기준:

특히 circuit breaker가 열린 시간대에는 모든 요청이 즉시 Gemini로 라우팅돼 p99 지연이 4.2초로 안정되었습니다. 단순 try/except만 쓸 경우 같은 시간대에 p99가 23초까지 치솟았던 것과 비교하면 5.5배 개선입니다.

8. 커뮤니티 평가 및 평판

Failover 패턴은 이미 LLM 오퍼레이션의 표준으로 자리잡았습니다. 참고할 만한 외부 시그널을 정리합니다.