이 글은 누구를 위한 것인가
- 여러 팀이 하나의 LLM API 키를 공유해서 비용 추적이 안 되는 팀
- Claude와 GPT-4를 상황에 따라 선택하는 라우팅 로직이 필요한 팀
- LLM API 장애 시 자동 폴백이 필요한 팀
들어가며
AI 게이트웨이는 모든 LLM 호출이 거치는 중앙 프록시다. 여기서 인증, 비용 추적, 레이트 리밋, 모델 라우팅, 폴백을 처리한다. 팀이 커질수록 필수가 된다.
이 글은 bluefoxdev.kr의 LLM 인프라 설계 를 참고하여 작성했습니다.
1. AI 게이트웨이 아키텍처
[AI 게이트웨이 처리 흐름]
클라이언트 요청
↓
인증 (팀 API 키 검증)
↓
레이트 리밋 확인 (팀별/전체)
↓
비용 예산 확인 (팀별 월 한도)
↓
모델 라우팅 (요청 특성에 따라 모델 선택)
↓
LLM API 호출 (Claude/GPT-4/Gemini)
↓ (실패 시)
폴백 모델 시도
↓
응답 처리 (로깅, 비용 기록)
↓
클라이언트에 반환
[모델 라우팅 전략]
빠른 분류/단순 작업 → Claude Haiku (저렴)
복잡한 추론 → Claude Opus (정확)
코드 생성 → Claude Sonnet (균형)
이미지 분석 → Claude Opus 또는 GPT-4V
비용 초과 시 → 더 저렴한 모델로 강제 라우팅
2. LLM 게이트웨이 구현
from fastapi import FastAPI, HTTPException, Header, Depends
from fastapi.middleware.cors import CORSMiddleware
import anthropic
import httpx
import redis.asyncio as redis
from datetime import datetime
app = FastAPI(title="AI Gateway")
redis_client = redis.from_url("redis://localhost:6379")
class TeamConfig:
def __init__(self, team_id: str):
self.team_id = team_id
self.monthly_budget_usd = 100.0
self.rate_limit_rpm = 60 # requests per minute
self.allowed_models = ["claude-haiku-4-5-20251001", "claude-sonnet-4-6"]
async def verify_team_api_key(x_team_api_key: str = Header(...)) -> str:
"""팀 API 키 검증"""
team_id = await redis_client.get(f"api_key:{x_team_api_key}")
if not team_id:
raise HTTPException(status_code=401, detail="Invalid API key")
return team_id.decode()
async def check_rate_limit(team_id: str) -> bool:
"""레이트 리밋 확인"""
key = f"rate_limit:{team_id}:{datetime.now().strftime('%Y%m%d%H%M')}"
count = await redis_client.incr(key)
await redis_client.expire(key, 60)
config = await get_team_config(team_id)
if count > config.rate_limit_rpm:
raise HTTPException(status_code=429, detail="Rate limit exceeded")
return True
async def check_budget(team_id: str, estimated_cost: float) -> bool:
"""예산 확인"""
spent_key = f"budget:{team_id}:{datetime.now().strftime('%Y%m')}"
current_spent = float(await redis_client.get(spent_key) or 0)
config = await get_team_config(team_id)
if current_spent + estimated_cost > config.monthly_budget_usd:
raise HTTPException(status_code=402, detail="Monthly budget exceeded")
return True
@app.post("/v1/messages")
async def proxy_messages(
request: dict,
team_id: str = Depends(verify_team_api_key),
):
"""Claude API 프록시"""
await check_rate_limit(team_id)
# 모델 라우팅
requested_model = request.get("model", "claude-sonnet-4-6")
config = await get_team_config(team_id)
if requested_model not in config.allowed_models:
requested_model = config.allowed_models[0] # 기본 모델 사용
# 비용 추정
input_tokens = estimate_tokens(request.get("messages", []))
estimated_cost = calculate_cost(requested_model, input_tokens, request.get("max_tokens", 1000))
await check_budget(team_id, estimated_cost)
# LLM 호출 (폴백 포함)
models_to_try = [requested_model] + get_fallback_models(requested_model)
for model in models_to_try:
try:
anthropic_client = anthropic.Anthropic(api_key=get_api_key_for_model(model))
response = anthropic_client.messages.create(
model=model,
max_tokens=request.get("max_tokens", 1000),
messages=request.get("messages", []),
system=request.get("system"),
)
# 비용 기록
actual_cost = calculate_actual_cost(
model,
response.usage.input_tokens,
response.usage.output_tokens,
)
spent_key = f"budget:{team_id}:{datetime.now().strftime('%Y%m')}"
await redis_client.incrbyfloat(spent_key, actual_cost)
await redis_client.expire(spent_key, 86400 * 35)
# 사용량 로깅
await log_usage({
"team_id": team_id,
"model": model,
"input_tokens": response.usage.input_tokens,
"output_tokens": response.usage.output_tokens,
"cost_usd": actual_cost,
"timestamp": datetime.utcnow().isoformat(),
})
return response.model_dump()
except anthropic.APIError as e:
if model == models_to_try[-1]:
raise HTTPException(status_code=503, detail=f"All models failed: {e}")
continue # 다음 폴백 모델 시도
def get_fallback_models(primary_model: str) -> list[str]:
"""폴백 모델 목록"""
fallbacks = {
"claude-opus-4-7": ["claude-sonnet-4-6", "claude-haiku-4-5-20251001"],
"claude-sonnet-4-6": ["claude-haiku-4-5-20251001"],
"claude-haiku-4-5-20251001": [],
}
return fallbacks.get(primary_model, [])
마무리
AI 게이트웨이는 팀이 3개 이상이 되면 반드시 필요하다. 처음에는 간단한 레이트 리밋과 로깅부터 시작하고, 점진적으로 모델 라우팅과 폴백을 추가하라. LiteLLM, PortKey 같은 오픈소스 솔루션을 먼저 검토하고, 요구사항이 특수하면 직접 구현하는 것이 더 효율적이다.