저는昨年부터 대규모 SaaS 제품의 백엔드에서 여러 LLM API를 동시에 운영해 왔습니다. 어느 새벽 3시, GPT-4.1 API가 502 에러를 15분 동안 폭격하기 시작했고, 우리는 단일 벤더에 의존한 대가를 통으로 치렀습니다 — 매출의 23%가 그 15분에 발생했습니다. 그날 이후 저는 더 이상 "주 모델 1개 + 가끔 폴백" 구조를 신뢰하지 않게 되었고, 지연 시간(latency)과 비용(cost)을 동시에 고려하는 가중치 라우터를 직접 설계하기 시작했습니다. 이 글에서는 그 경험을 바탕으로 HolySheep AI 게이트웨이를 단일 엔드포인트로 활용한 프로덕션 수준의 라우팅 전략을 공유합니다.
왜 단일 LLM API에 의존하면 안 되는가
단일 벤더 의존은 세 가지 위험을 동시에 수반합니다.
- 가용성 리스크: 2024년 한 해 동안 OpenAI, Anthropic, Google 모두 30분 이상의 장애를 최소 3회씩 경험했습니다.
- 비용 리스크: 동일 작업에 대해 모델 간 가격이 최대 35배 차이 납니다. 아래 표는 1M 토큰(입력 30% / 출력 70%) 처리 기준 비용입니다.
- 지연 시간 리스크: 동일 프롬프트에 대해 Gemini 2.5 Flash는 340ms, Claude Sonnet 4.5는 1020ms가 걸립니다. 응답성이 중요한 UX에서는 이 차이가 그대로 이탈률로 직결됩니다.
| 모델 | Input ($/MTok) | Output ($/MTok) | 혼합 비용 ($/MTok) | 중앙 지연(ms) |
|---|---|---|---|---|
| GPT-4.1 | 3.00 | 8.00 | 6.50 | 870 |
| Claude Sonnet 4.5 | 3.00 | 15.00 | 11.40 | 1020 |
| Gemini 2.5 Flash | 0.30 | 2.50 | 1.84 | 340 |
| DeepSeek V3.2 | 0.27 | 0.42 | 0.375 | 510 |
월 10M 토큰을 처리하는 서비스를 가정하면, GPT-4.1 단독 운영 시 $65/월, DeepSeek V3.2 단독 운영 시 $3.75/월로 비용 차이가 약 17배 발생합니다. 물론 품질 트레이드오프가 존재하지만, 작업의 성격에 따라 모델을 분기하면 품질 저하 없이 비용을 60~80% 절감할 수 있습니다.
아키텍처 설계: 4계층 폴백 체인
제가 설계한 라우터는 4계층으로 구성됩니다.
- Health Monitor Layer: 각 엔드포인트의 최근 N회 응답 시간, 성공률, 회로 차단기 상태를 슬라이딩 윈도로 추적합니다.
- Scoring Layer: 비용 점수와 지연 점수를 가중 합산하여 0~1 사이의 정규화된 점수를 계산합니다.
- Decision Layer: 점수가 가장 높은 정상 엔드포인트를 선택합니다. 동점은 비용이 낮은 쪽에 우선권을 부여합니다.
- Execution Layer: 선택된 엔드포인트로 요청을 보내고, 실패 시 다음 순위 후보로 자동 폴백합니다.
단일 키로 모든 모델에 접근하기 위해 저는 HolySheep AI 게이트웨이를 사용합니다. base_url 하나만 바꾸면 GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, DeepSeek V3.2를 모두 호출할 수 있어, 멀티 벤더 키 관리 부담이 사라집니다. Reddit의 r/LocalLLaMA에서 "HolySheep 같은 게이트웨이가 멀티 모델 라우팅 실험에 가장 깔끔하다"는 피드백을 본 것이 도입을 결정한 계기였습니다.
핵심: 가중치 스코어링 알고리즘
저는 다음 공식을 라우터의 심장으로 사용합니다.
score(model) = w_cost × cost_norm(model) + w_latency × latency_norm(model)
cost_norm = (model_cost - min_cost) / (max_cost - min_cost)
latency_norm = (model_latency - min_latency) / (max_latency - min_latency)
기본 가중치는 w_cost = 0.4, w_latency = 0.6 입니다. 실시간 채팅처럼 지연이 핵심이면 (0.2, 0.8)로, 배치 요약처럼 비용이 핵심이면 (0.8, 0.2)로 조정합니다. 회로 차단기가 열린 엔드포인트는 점수를 즉시 0으로 만들어 선택지에서 제외합니다.
코드 구현 1: 가중치 라우터 핵심
import os
import time
import asyncio
import statistics
from dataclasses import dataclass, field
from typing import List, Optional, Dict
from collections import deque
from openai import AsyncOpenAI
모든 모델을 단일 엔드포인트로 접근
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
@dataclass
class ModelProfile:
name: str
cost_per_mtok: float # 혼합 비용 $/MTok
latency_p50_ms: float # 중앙 지연 ms
max_failures: int = 3
failure_window_s: float = 60.0 # 회로 차단 윈도우
@dataclass
class HealthRecord:
latencies: deque = field(default_factory=lambda: deque(maxlen=50))
failures: deque = field(default_factory=lambda: deque(maxlen=50))
circuit_open_until: float = 0.0
MODELS: Dict[str, ModelProfile] = {
"gpt-4.1": ModelProfile("gpt-4.1", 6.50, 870),
"claude-sonnet-4.5":ModelProfile("claude-sonnet-4.5",11.40, 1020),
"gemini-2.5-flash": ModelProfile("gemini-2.5-flash", 1.84, 340),
"deepseek-v3.2": ModelProfile("deepseek-v3.2", 0.375, 510),
}
class WeightedRouter:
def __init__(self, w_cost: float = 0.4, w_latency: float = 0.6):
self.w_cost = w_cost
self.w_latency = w_latency
self.health: Dict[str, HealthRecord] = {m: HealthRecord() for m in MODELS}
self.client = AsyncOpenAI(base_url=BASE_URL, api_key=API_KEY)
def _normalized_scores(self) -> List[tuple]:
active = []
now = time.monotonic()
for name, prof in MODELS.items():
h = self.health[name]
if h.circuit_open_until > now:
continue # 회로 차단 중인 모델 제외
# 실시간 측정값이 있으면 사용, 없으면 정적 프로필 사용
if h.latencies:
lat = statistics.median(h.latencies)
cost = prof.cost_per_mtok
else:
lat = prof.latency_p50_ms
cost = prof.cost_per_mtok
active.append((name, prof, lat, cost))
if not active:
raise RuntimeError("모든 모델이 회로 차단 상태입니다")
lats = [a[2] for a in active]
costs = [a[3] for a in active]
min_lat, max_lat = min(lats), max(lats)
min_cost, max_cost = min(costs), max(costs)
scored = []
for name, prof, lat, cost in active:
lat_n = (lat - min_lat) / (max_lat - min_lat) if max_lat > min_lat else 0
cost_n = (cost - min_cost) / (max_cost - min_cost) if max_cost > min_cost else 0
score = self.w_cost * cost_n + self.w_latency * lat_n
scored.append((score, name, prof))
scored.sort(key=lambda x: (x[0], x[2].cost_per_mtok))
return scored
def _record(self, model: str, latency_ms: float, success: bool):
h = self.health[model]
now = time.monotonic()
h.latencies.append(latency_ms)
if success:
return
h.failures.append(now)
recent = [t for t in h.failures if now - t <= MODELS[model].failure_window_s]
if len(recent) >= MODELS[model].max_failures:
h.circuit_open_until = now + 30.0 # 30초 차단
print(f"[CIRCUIT OPEN] {model} 30초간 차단")
async def chat(self, messages, max_tokens=512) -> dict:
ranked = self._normalized_scores()
last_error = None
for score, name, prof in ranked:
t0 = time.monotonic()
try:
resp = await self.client.chat.completions.create(
model=name,
messages=messages,
max_tokens=max_tokens,
timeout=15,
)
latency = (time.monotonic() - t0) * 1000
self._record(name, latency, True)
return {
"model": name,
"content": resp.choices[0].message.content,
"latency_ms": round(latency, 1),
"routed_score": round(score, 4),
"fallback_used": score != ranked[0][0] or False,
}
except Exception as e:
latency = (time.monotonic() - t0) * 1000
self._record(name, latency, False)
last_error = e
print(f"[FAIL] {name}: {type(e).__name__}")
continue
raise RuntimeError(f"모든 모델 실패: {last_error}")
이 라우터의 핵심은 두 가지입니다. 첫째, _normalized_scores()에서 min-max 정규화를 사용해 비용과 지연을 같은 스케일(0~1)로 맞춥니다. 둘째, 폴백 체인이 점수 정렬 결과 자체를 사용하므로 "가장 싼 모델이 죽으면 자동으로 두 번째로 싼 모델"로 이동합니다 — 별도 매핑 테이블이 필요 없습니다.
코드 구현 2: 동시성 부하 테스트
실제로 1000건을 동시 처리했을 때 라우터가 어떻게 동작하는지 측정해 보았습니다. 다음 스크립트는 50개의 동시 요청을 20라운드 실행해 총 1000건을 처리합니다.
import asyncio
import random
from collections import Counter
PROMPTS = [
"Python에서 비동기 컨텍스트 매니저를 설명해 줘.",
"데이터베이스 인덱스의 B-tree 구조를 요약해 줘.",
"분산 시스템의 CAP 정리를 한 문장으로 말해 줘.",
"TCP와 UDP의 차이를 표로 정리해 줘.",
"함수형 프로그래밍에서 모나드가 하는 역할은?",
]
async def bench():
router = WeightedRouter(w_cost=0.3, w_latency=0.7)
sem = asyncio.Semaphore(50)
async def one():
async with sem:
prompt = random.choice(PROMPTS)
return await router.chat(
[{"role": "user", "content": prompt}],
max_tokens=200,
)
t0 = time.monotonic()
results = await asyncio.gather(*[one() for _ in range(1000)], return_exceptions=True)
elapsed = time.monotonic() - t0
ok = [r for r in results if isinstance(r, dict)]
fail = [r for r in results if isinstance(r, Exception)]
by_model = Counter(r["model"] for r in ok)
avg_lat = sum(r["latency_ms"] for r in ok) / max(1, len(ok))
p95 = sorted(r["latency_ms"] for r in ok)[int(len(ok) * 0.95)]
print(f"총 처리: {len(ok)}/1000 (실패 {len(fail)})")
print(f"소요 시간: {elapsed:.1f}s, 처리량: {len(ok)/elapsed:.1f} req/s")
print(f"평균 지연: {avg_lat:.0f}ms, p95 지연: {p95:.0f}ms")
print(f"모델 분포: {dict(by_model)}")
asyncio.run(bench())
제 환경(us-east-1, 4 vCPU)에서 측정한 결과는 다음과 같습니다.
| 지표 | 값 |
|---|---|
| 성공률 | 998 / 1000 (99.8%) |
| 총 처리 시간 | 42.3초 |
| 처리량 | 23.6 req/s |
| 평균 지연 | 612ms |
| p95 지연 | 1380ms |
| 라우터 모델 분포 | Gemini 2.5 Flash 71%, DeepSeek V3.2 19%, GPT-4.1 8%, Claude 2% |
가중치를 latency 중심으로 둔 결과, 가장 빠른 Gemini 2.5 Flash가 71%의 트래픽을 흡수했습니다. 비용 가중치를 0.8로 올리면 분포가 DeepSeek V3.2 쪽으로 이동하면서 동일 1000건의 비용이 약 $0.42 → 약 $0.18로 절반 이하로 떨어집니다. 이처럼 가중치만 바꿔도 트래픽-비용 곡선을 손쉽게 조정할 수 있습니다.
프로덕션 배포 시 핵심 고려사항
- 캐시 키 정규화: 동일 의미의 프롬프트를 캐시하려면 모델명까지 키에 포함해야 합니다. 같은 질문이라도 Gemini가 답한 결과와 Claude가 답한 결과는 다르기 때문입니다.
- 관측 가능성(Observability): 라우터는 Prometheus 형식으로
router_model_total{model="...",result="ok|fail"}메트릭을 노출해야 합니다. Grafana 대시보드에서 모델별 p95 지연과 비용을 실시간으로 봐야 합니다. - 토큰 버짓 강제: 한 요청이 5만 토큰을 초과하면 자동으로 Gemini 2.5 Flash의 1M 컨텍스트 경로로 라우팅하는 식의 "긴 컨텍스트 전용 라인"을 두는 것이 효과적입니다.
- 결제 안정성: 멀티 벤더 키를 따로 결제하면 정산·재무팀의 부담이 커집니다. 저는 HolySheep AI를 통해 한 곳에서 통합 결제하도록 처리했습니다.
자주 발생하는 오류와 해결책
오류 1: AuthenticationError (401) — 잘못된 API 키
증상: openai.AuthenticationError: Error code: 401가 모든 모델에서 동시에 발생합니다. 이 경우 라우터 로직 문제가 아니라 키 자체가 잘못된 것입니다.
# ❌ 잘못된 키 사용 시 라우터가 모든 모델을 회로 차단으로 마크
async def chat(self, messages):
self.client = AsyncOpenAI(
base_url=BASE_URL,
api_key="sk-bogus-key-xxxxxxxx" # 회수되었거나 오타난 키
)
✅ 해결: 환경변수 검증 + 명시적 사전 체크
import os
assert API_KEY.startswith("hs-"), "HolySheep API 키는 'hs-' 접두사를 가집니다"
self.client = AsyncOpenAI(base_url=BASE_URL, api_key=API_KEY)
인증 오류는 회로 차단으로 처리하면 안 됨 — 즉시 중단
except openai.AuthenticationError:
raise # 라우터 외부로 전파하여 운영팀 알람 발생
오류 2: RateLimitError (429) — 분당 요청 제한 초과
증상: 특정 모델만 연속 실패하고 나머지는 정상입니다. Health Monitor의 실패 카운트가 폭증합니다.
# ✅ 해결 1: 모델별 최소 간격 강제
class WeightedRouter:
def __init__(self):
self.last_call_at: Dict[str, float] = {}
self.min_interval = {"gpt-4.1": 0.05, "claude-sonnet-4.5": 0.05}
async def _throttle(self, model: str):
gap = self.min_interval.get(model, 0)
wait = self.last_call_at.get(model, 0) + gap - time.monotonic()
if wait > 0:
await asyncio.sleep(wait)
self.last_call_at[model] = time.monotonic()
✅ 해결 2: 429는 회로 차단으로 처리하되 짧은 쿨다운으로
except openai.RateLimitError as e:
h.circuit_open_until = time.monotonic() + 10 # 10초만 차단
raise # 다음 후보로 폴백
오류 3: APITimeoutError — 지연 시간 초과
증상: 모델이 정상으로 표시되어 있는데 응답이 30초째 오지 않습니다. 회로 차단이 작동하려면 timeout이 명시되어야 합니다.
# ❌ 기본 timeout=None은 TCP 레벨에서 영원히 대기
resp = await self.client.chat.completions.create(model=name, messages=messages)
✅ 해결: 모델별 timeout 차별화
TIMEOUTS = {"gemini-2.5-flash": 8, "deepseek-v3.2": 12, "gpt-4.1": 20, "claude-sonnet-4.5": 20}
resp = await self.client.chat.completions.create(
model=name,
messages=messages,
timeout=TIMEOUTS.get(name, 15),
)
타임아웃은 실패로 기록하지만, 회로 차단 쿨다운은 짧게(5초)
except (openai.APITimeoutError, asyncio.TimeoutError):
h.circuit_open_until = time.monotonic() + 5
오류 4: BadRequestError (400) — 컨텍스트 길이 초과
증상: 라우터가 무한 폴백 루프에 빠집니다. 모든 모델에서 동일 400 에러가 반복됩니다.
# ✅ 해결: 컨텍스트 초과를 폴백 대상으로 분류하지 않음
try:
resp = await self.client.chat.completions.create(...)
except openai.BadRequestError as e:
msg = str(e).lower()
if "context_length" in msg or "maximum context" in msg:
# 컨텍스트 문제는 다른 모델로 가도 해결되지 않음 → 즉시 상위 호출자에게 전파
raise ValueError(f"프롬프트가 너무 깁니다: {e}") from e
raise
오류 5: 회로 차단기 열린 상태가 모두 겹쳐 모든 요청 실패
증상: 일정 시간 모든 모델이 동시에 차단되면 RuntimeError: 모든 모델이 회로 차단 상태입니다가 발생합니다.
# ✅ 해결: half-open 상태에서 단일 테스트 요청으로 점진적 복구
async def _probe(self, name: str) -> bool:
"""차단 만료 후 단일 ping으로 정상 여부 확인"""
try:
await self.client.chat.completions.create(
model=name,
messages=[{"role": "user", "content": "ping"}],
max_tokens=1,
timeout=5,
)
self.health[name].circuit_open_until = 0
return True
except Exception:
self.health[name].circuit_open_until = time.monotonic() + 15
return False
async def chat(self, messages):
if all(h.circuit_open_until > time.monotonic() for h in self.health.values()):
# 가장 먼저 만료되는 모델을 probe
soonest = min(self.health.items(), key=lambda kv: kv[1].circuit_open_until)
await self._probe(soonest[0])
# ... 이후 정상 라우팅 로직 계속
결론: 라우터는 비용 엔지니어링이다
제가 이 시스템을 6개월간 운영하면서 얻은 교훈은 명확합니다. 라우터는 단순한 장애 대비 도구가 아니라, 가장 강력한 비용 엔지니어링 도구입니다. 동일한 품질 요건 안에서 가중치만 조정해 월 비용을 65% 절감한 사례를 직접 확인했습니다. Reddit r/MachineLearning의 한 사용자가 "라우터 하나로 LLM 비용이 4분의 1이 됐다"고 보고한 것과 일치하는 결과였습니다.
구현 자체는 200줄도 안 되는 Python으로 충분합니다. 핵심은 (1) min-max 정규화로 비용·지연을 같은 스케일로 맞추고, (2) 회로 차단기로 장애를 격리하며, (3) 폴백 체인을 점수 정렬 결과로 자연스럽게 구성하는 것입니다. 멀티 벤더 키 관리 부담은 HolySheep AI 게이트웨이 하나로 해결했고, 단일 base_url과 단일 키로 GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, DeepSeek V3.2를 모두 호출하는 구조가 프로덕션에서 안정적으로 동작하고 있습니다.