반응형
1편에서 기본 호출까지 했습니다. 2편은 OpenRouter를 단순 프록시가 아닌 인텔리전트 라우터로 쓰는 방법입니다. 프로바이더가 죽어도 앱이 살아있고, 같은 모델을 가장 저렴한 프로바이더로 자동 라우팅합니다.
[2편 핵심 요약]
→ 기본 로드밸런싱: 가격 역제곱 가중치 + 30초 이내 장애 프로바이더 회피
→ 폴백 라우팅: models 배열에 여러 모델 → 순서대로 시도 (실패 시 자동 전환)
→ provider.order: 특정 프로바이더 우선순위 지정 (Anthropic → Bedrock → Vertex)
→ provider.sort: price / throughput / latency 중 하나로 정렬
→ provider.ignore: 특정 프로바이더 제외 (데이터 리전 제한 등)
→ max_price: 최대 토큰 가격 필터 (이 이상이면 라우팅 거부)
→ require_parameters: function_calling, streaming 등 필수 기능 있는 프로바이더만
→ BYOK: 자체 API 키 연결 → DeepSeek 시간대 할인 등 프로바이더 할인 직접 적용
OpenRouter 기본 라우팅 전략
코드 한 줄 없이 자동으로 작동합니다.
[기본 로드밸런싱 알고리즘]
Step 1: 30초 내 장애 발생 프로바이더 제외
Step 2: 안정적인 프로바이더 중 가격 역제곱 가중치 정렬
→ Provider A: $1/M → 가중치 = 1/1² = 1.0
→ Provider B: $2/M → 가중치 = 1/2² = 0.25
→ Provider C: $3/M → 가중치 = 1/3² = 0.11
→ A가 C보다 9배 먼저 선택됨
Step 3: 선택된 프로바이더 실패 시 다음 순서로 폴백
실제 예시 (Claude Sonnet):
→ Anthropic 직접: $3.00/M
→ AWS Bedrock: $3.00/M (동일 가격)
→ GCP Vertex AI: $3.00/M (동일 가격)
→ 동일 가격이면 업타임으로 가중치 결정
→ 하나 죽으면 자동으로 다른 프로바이더로 전환
실전 1 — 모델 폴백 설정
from openai import OpenAI
import os
client = OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.environ["OPENROUTER_API_KEY"],
)
# ── 기본 폴백 설정 ────────────────────────────────────
# models 배열: 앞에서부터 시도, 실패하면 다음 모델로
response = client.chat.completions.create(
model="anthropic/claude-opus-4-7", # 주 모델
extra_body={
"models": [
"anthropic/claude-opus-4-7", # 1순위
"anthropic/claude-sonnet-4-6", # 2순위 (Opus 장애 시)
"openai/gpt-5.4", # 3순위
"google/gemini-3-flash", # 4순위 (마지막 보루)
]
},
messages=[{"role": "user", "content": "안녕"}]
)
# 실제 어떤 모델이 응답했는지 확인
print(f"응답 모델: {response.model}")
# → "anthropic/claude-opus-4-7" 또는 폴백된 모델
# ── 폴백 트리거 조건 제어 ─────────────────────────────
response = client.chat.completions.create(
model="anthropic/claude-opus-4-7",
extra_body={
"models": [
"anthropic/claude-opus-4-7",
"anthropic/claude-sonnet-4-6",
],
# 어떤 에러에서 폴백할지 설정
# 기본값: 모든 에러 (rate limit, 500, 모더레이션 등)
# "on": ["rate_limit", "server_error"] → 특정 에러만
# "fallback": False → 폴백 비활성화
},
messages=[{"role": "user", "content": "복잡한 추론 문제"}]
)
# ── 폴백 + 정렬 조합 ─────────────────────────────────
response = client.chat.completions.create(
model="anthropic/claude-sonnet-4-6",
extra_body={
"models": [
"anthropic/claude-sonnet-4-6",
"openai/gpt-5.4",
"google/gemini-3-flash",
],
"provider": {
# 폴백 간 모델 경계 제거 — 전체 엔드포인트를 가격순 정렬
# partition: "none" → 저렴한 엔드포인트가 먼저 시도됨
"sort": {
"by": "price",
"partition": "none" # 모델 경계 없이 전역 정렬
}
}
},
messages=[{"role": "user", "content": "분류 태스크"}]
)
실전 2 — 프로바이더 직접 제어
# ── 특정 프로바이더 우선순위 지정 ─────────────────────
# Claude를 항상 Bedrock 통해서 쓰고 싶을 때
response = client.chat.completions.create(
model="anthropic/claude-sonnet-4-6",
extra_body={
"provider": {
"order": [
"Amazon Bedrock", # 1순위
"Anthropic", # 2순위 (Bedrock 장애 시)
]
}
},
messages=[{"role": "user", "content": "안녕"}]
)
# ── 특정 프로바이더 제외 ─────────────────────────────
# 데이터 리전 제한 (EU 외 라우팅 금지 등)
response = client.chat.completions.create(
model="anthropic/claude-sonnet-4-6",
extra_body={
"provider": {
"ignore": [
"Amazon Bedrock", # 특정 리전 사용 안 할 때
"Google Vertex", # GCP 계정 없을 때
]
}
},
messages=[{"role": "user", "content": "안녕"}]
)
# ── 단일 프로바이더 고정 ──────────────────────────────
# 폴백 없이 Anthropic만 쓰고 싶을 때
response = client.chat.completions.create(
model="anthropic/claude-sonnet-4-6",
extra_body={
"provider": {
"order": ["Anthropic"],
"allow_fallbacks": False # 폴백 완전 비활성화
}
},
messages=[{"role": "user", "content": "안녕"}]
)
# ── sort로 최적화 목표 설정 ───────────────────────────
# 가장 저렴한 프로바이더 (== :floor variant와 동일)
response = client.chat.completions.create(
model="anthropic/claude-sonnet-4-6",
extra_body={
"provider": {"sort": "price"}
},
messages=[{"role": "user", "content": "대량 처리 태스크"}]
)
# 가장 빠른 프로바이더 (== :nitro variant와 동일)
response = client.chat.completions.create(
model="anthropic/claude-sonnet-4-6",
extra_body={
"provider": {"sort": "throughput"} # 또는 "latency"
},
messages=[{"role": "user", "content": "실시간 채팅"}]
)
실전 3 — 가격 필터와 필수 기능 필터
# ── 최대 가격 필터 ────────────────────────────────────
# 이 가격 이상이면 라우팅 거부
response = client.chat.completions.create(
model="anthropic/claude-sonnet-4-6",
extra_body={
"provider": {
"max_price": {
"prompt": 3, # $3/M input 이하만
"completion": 15 # $15/M output 이하만
}
}
},
messages=[{"role": "user", "content": "안녕"}]
)
# ── 필수 기능 필터 ────────────────────────────────────
# 툴 콜링, 스트리밍 등 특정 기능 지원하는 프로바이더만
# Function Calling 필수
response = client.chat.completions.create(
model="anthropic/claude-sonnet-4-6",
extra_body={
"provider": {
"require_parameters": True # 요청한 파라미터 모두 지원하는 프로바이더만
}
},
tools=[{
"type": "function",
"function": {
"name": "get_weather",
"description": "날씨 조회",
"parameters": {
"type": "object",
"properties": {"location": {"type": "string"}},
"required": ["location"]
}
}
}],
messages=[{"role": "user", "content": "서울 날씨 알려줘"}]
)
# ── 양자화 레벨 필터 ──────────────────────────────────
# 오픈소스 모델의 품질 vs 비용 트레이드오프
response = client.chat.completions.create(
model="meta-llama/llama-3.3-70b-instruct",
extra_body={
"provider": {
"quantizations": ["fp16", "bf16"] # 고품질 양자화만
# 옵션: "fp32", "fp16", "bf16", "fp8", "int8", "int4", "int2"
}
},
messages=[{"role": "user", "content": "복잡한 추론"}]
)
실전 4 — BYOK(자체 API 키) 설정
[BYOK 설정 위치]
OpenRouter 콘솔 → Settings → Integrations → Add Provider Key
지원 프로바이더:
→ Anthropic (Claude)
→ OpenAI (GPT)
→ Google (Gemini)
→ AWS Bedrock
→ Azure OpenAI
→ DeepSeek
→ Mistral AI
→ Cohere ... 등
BYOK 비용 구조:
→ 월 100만 요청 이하: 무료
→ 100만 요청 초과: 표준 비용의 5%
→ 프로바이더 요금은 자체 계정에 직접 청구
# BYOK 설정 후 특정 프로바이더 키 사용 강제
response = client.chat.completions.create(
model="anthropic/claude-sonnet-4-6",
extra_body={
"provider": {
"order": ["Anthropic"], # 내 Anthropic 키 사용
"allow_fallbacks": False
}
},
messages=[{"role": "user", "content": "안녕"}]
)
# DeepSeek BYOK + 시간대 할인 활용
# DeepSeek: 오프피크(UTC 16:30~00:30) 75% 할인
# → BYOK 연결하면 OpenRouter가 직접 할인 요금으로 청구
response = client.chat.completions.create(
model="deepseek/deepseek-chat",
extra_body={
"provider": {
"order": ["DeepSeek"], # 내 DeepSeek 키 사용 → 할인 직접 적용
}
},
messages=[{"role": "user", "content": "분석 태스크"}]
)
[BYOK가 유리한 케이스]
→ 이미 특정 프로바이더 계약·크레딧 보유
→ 레이트 리밋을 직접 관리하고 싶을 때
→ DeepSeek 오프피크 75% 할인 활용
→ AWS Bedrock 기업 계약 할인 적용
→ 프로바이더별 데이터 처리 정책 직접 제어
[BYOK가 필요 없는 케이스]
→ 프로바이더 계정 없는 경우
→ OpenRouter 크레딧으로 간단히 시작
→ 월 트래픽이 많지 않은 경우
실전 5 — 응답에서 라우팅 정보 추출
import httpx
import os
import json
def chat_with_routing_info(
messages: list[dict],
model: str = "anthropic/claude-sonnet-4-6",
fallback_models: list[str] = None
) -> dict:
"""라우팅 정보를 포함한 응답 반환"""
body = {
"model": model,
"messages": messages,
}
if fallback_models:
body["models"] = [model] + fallback_models
response = httpx.post(
"https://openrouter.ai/api/v1/chat/completions",
headers={
"Authorization": f"Bearer {os.environ['OPENROUTER_API_KEY']}",
"Content-Type": "application/json",
},
json=body,
timeout=60
)
data = response.json()
return {
# 실제 응답한 모델 (폴백 발생 시 원본 모델과 다름)
"model_used": data.get("model"),
"content": data["choices"][0]["message"]["content"],
# 토큰 사용량
"input_tokens": data.get("usage", {}).get("prompt_tokens", 0),
"output_tokens": data.get("usage", {}).get("completion_tokens", 0),
# 비용 (달러)
"cost_usd": data.get("usage", {}).get("cost", 0),
# 생성 ID (추후 조회용)
"generation_id": data.get("id"),
}
# 사용
result = chat_with_routing_info(
messages=[{"role": "user", "content": "안녕"}],
model="anthropic/claude-opus-4-7",
fallback_models=["anthropic/claude-sonnet-4-6", "openai/gpt-5.4"]
)
print(f"사용 모델: {result['model_used']}")
print(f"비용: ${result['cost_usd']:.6f}")
print(f"응답: {result['content']}")
# 폴백 여부 감지
requested_model = "anthropic/claude-opus-4-7"
if result["model_used"] != requested_model:
print(f"⚠️ 폴백 발생: {requested_model} → {result['model_used']}")
# Generation ID로 상세 정보 조회
import httpx
def get_generation_details(generation_id: str) -> dict:
"""응답 후 상세 라우팅 정보 조회"""
response = httpx.get(
f"https://openrouter.ai/api/v1/generation?id={generation_id}",
headers={"Authorization": f"Bearer {os.environ['OPENROUTER_API_KEY']}"}
)
data = response.json()["data"]
return {
"model": data["model"],
"provider": data.get("provider"), # 실제 프로바이더
"latency_ms": data.get("latency"), # 레이턴시 (ms)
"tokens_prompt": data.get("tokens_prompt"),
"tokens_completion": data.get("tokens_completion"),
"total_cost": data.get("total_cost"),
"finish_reason": data.get("finish_reason"),
}
details = get_generation_details(result["generation_id"])
print(f"프로바이더: {details['provider']}")
print(f"레이턴시: {details['latency_ms']}ms")
실전 6 — 프로덕션 비용 최적화 패턴
from openai import OpenAI
import os
from functools import lru_cache
client = OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.environ["OPENROUTER_API_KEY"],
)
# ── 패턴 1: 캐스케이딩 품질 업스케일 ────────────────
# 싼 모델로 먼저 시도 → 품질 검증 → 실패 시 더 좋은 모델
def cascading_request(prompt: str) -> str:
"""비용 최적화된 캐스케이딩 요청"""
# 1단계: 저렴한 모델로 시도
response = client.chat.completions.create(
model="google/gemini-3-flash", # $0.10/M — 매우 저렴
extra_body={
"provider": {"sort": "price"}
},
messages=[
{"role": "system", "content": "간단한 태스크만 처리. 복잡하면 'ESCALATE' 출력."},
{"role": "user", "content": prompt}
],
max_tokens=500
)
result = response.choices[0].message.content
# 에스컬레이션 필요 여부 확인
if "ESCALATE" not in result:
return result # 저렴한 모델로 해결
# 2단계: 더 강력한 모델로 재시도
print("📈 Sonnet으로 업스케일...")
response = client.chat.completions.create(
model="anthropic/claude-sonnet-4-6",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
# ── 패턴 2: 배치 + 저렴한 모델 조합 ─────────────────
# 급하지 않은 대량 처리: 무료 모델 or :floor variant
def batch_process(prompts: list[str]) -> list[str]:
"""대량 처리: 가장 저렴한 모델로"""
results = []
for prompt in prompts:
response = client.chat.completions.create(
model="meta-llama/llama-3.3-70b-instruct:free", # 무료
messages=[{"role": "user", "content": prompt}]
)
results.append(response.choices[0].message.content)
return results
# ── 패턴 3: 태스크 유형별 모델 매핑 ─────────────────
TASK_MODEL_MAP = {
# 분류·감정분석·단순 쿼리 → 무료 or 초저가
"classify": "google/gemini-3.1-flash-lite:free",
"summarize": "deepseek/deepseek-chat", # $0.27/M
"translate": "mistralai/mistral-small", # $0.10/M
# 코딩·추론 → 중급 모델
"code": "anthropic/claude-sonnet-4-6", # $3/M
"reasoning": "deepseek/deepseek-r1", # 추론 특화
# 에이전트·복잡한 태스크 → 최상위 모델
"agent": "anthropic/claude-opus-4-7", # $5/M
"vision": "anthropic/claude-opus-4-7", # 고해상도 이미지
}
def smart_request(prompt: str, task_type: str = "summarize") -> str:
model = TASK_MODEL_MAP.get(task_type, "anthropic/claude-sonnet-4-6")
response = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
실전 7 — 비용 모니터링 자동화
import httpx
from datetime import datetime, timedelta
def get_usage_report(days: int = 7) -> dict:
"""
최근 N일 사용량 리포트
OpenRouter Generations API 활용
"""
end_date = datetime.now()
start_date = end_date - timedelta(days=days)
response = httpx.get(
"https://openrouter.ai/api/v1/generations",
headers={"Authorization": f"Bearer {os.environ['OPENROUTER_API_KEY']}"},
params={
"date_start": start_date.isoformat(),
"date_end": end_date.isoformat(),
}
)
generations = response.json().get("data", [])
# 모델별 비용 집계
model_costs = {}
total_cost = 0
for gen in generations:
model = gen.get("model", "unknown")
cost = gen.get("total_cost", 0)
tokens = gen.get("tokens_prompt", 0) + gen.get("tokens_completion", 0)
if model not in model_costs:
model_costs[model] = {"cost": 0, "tokens": 0, "requests": 0}
model_costs[model]["cost"] += cost
model_costs[model]["tokens"] += tokens
model_costs[model]["requests"] += 1
total_cost += cost
# 비용 순 정렬
sorted_models = sorted(
model_costs.items(),
key=lambda x: x[1]["cost"],
reverse=True
)
print(f"\n{'='*50}")
print(f"최근 {days}일 사용량 리포트")
print(f"{'='*50}")
print(f"총 비용: ${total_cost:.4f}")
print(f"총 요청: {len(generations):,}")
print(f"\n[모델별 비용 Top 5]")
for model, data in sorted_models[:5]:
pct = (data["cost"] / total_cost * 100) if total_cost > 0 else 0
print(f" {model[:40]}")
print(f" 비용: ${data['cost']:.4f} ({pct:.1f}%)")
print(f" 요청: {data['requests']:,} / 토큰: {data['tokens']:,}")
return {"total_cost": total_cost, "by_model": dict(sorted_models)}
# 실행
report = get_usage_report(days=7)
마무리
✅ 2편에서 한 것
→ 기본 로드밸런싱 알고리즘 이해 (가격 역제곱 가중치)
→ 모델 폴백: models 배열로 순서 지정
→ 프로바이더 제어: order / ignore / allow_fallbacks
→ 정렬 최적화: price / throughput / latency
→ 가격 필터: max_price
→ 기능 필터: require_parameters / quantizations
→ BYOK: 자체 API 키 연결 + DeepSeek 시간대 할인
→ 응답에서 실제 사용 모델·프로바이더 확인
→ 비용 최적화 패턴: 캐스케이딩, 태스크별 모델 매핑
❌ 3편에서 다룰 것
→ Python/TypeScript SDK 완전 통합
→ LangChain · LangGraph에 OpenRouter 붙이기
→ 스트리밍 실전
→ 멀티모달 (이미지·PDF) 처리
→ 구조화 출력 (JSON mode)
관련 글
반응형