저는 3개월간 5개 AI 모델을 동시에 운용하며 월 $12,000의 API 비용 청구서를 받던 시기를 겪었습니다. 매주 비용审计하듯 체크하고, 모델을 번갈아 쓰며 팀원들과争吵하듯 할당량을 조정하던 그 시절이遥远하게 느껴집니다. 이 글에서는 HolySheep AI(https://www.holysheep.ai/register)를 활용해 다중 모델 API 비용을 체계적으로 관리하는 프로덕션 아키텍처를 공유합니다.
문제 정의: 왜 다중 모델 예산 관리가 중요한가
AI 서비스를 운영하면 자연스럽게 여러 모델을 활용하게 됩니다. 사용자에게 빠른 응답이 필요한 채팅에는 gpt-4o-mini 또는 gemini-2.5-flash를, 복잡한 분석에는 claude-sonnet-4.5를, 대량 데이터 처리에는 deepseek-v3.2를 사용하는 식입니다. 그러나 각 공급자별 결제 시스템이 다르다 보니:
- 예산 초과 알림 부재: 각 플랫폼별 사용량을 실시간으로 추적하기 어려움
- 미세한 비용 최적화 부재: 모델 전환时机을 수동으로 판단해야 함
- Fallback 로직 복잡성: 한 모델 장애 시 다른 모델로 자동 전환하는 구조 구현 부담
- 통합 모니터링 부재: 전체 AI 지출을 한눈에 파악할 수 있는 대시보드 부족
HolySheep는 이러한痛점을 해결하는 unified gateway입니다.
HolySheep 기본 연동 구조
OpenAI 호환 API로의 통합
HolySheep의 핵심 강점은 OpenAI 호환 엔드포인트를 제공한다는 점입니다. 기존 코드를 크게 변경하지 않고 base URL만 교체하면 모든 주요 모델에 접근할 수 있습니다.
# Python - HolySheep 기본 연동 예제
import openai
from datetime import datetime
HolySheep API 키 설정
client = openai.OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1" # 절대 api.openai.com 사용 금지
)
def generate_with_budget_check(prompt: str, model: str, max_tokens: int = 1000):
"""
예산 범위 내에서 AI 응답 생성
"""
start_time = datetime.now()
try:
response = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
max_tokens=max_tokens
)
latency_ms = (datetime.now() - start_time).total_seconds() * 1000
return {
"content": response.choices[0].message.content,
"model": response.model,
"usage": {
"prompt_tokens": response.usage.prompt_tokens,
"completion_tokens": response.usage.completion_tokens,
"total_tokens": response.usage.total_tokens
},
"latency_ms": latency_ms
}
except openai.RateLimitError:
print(f"Rate limit exceeded for {model}")
return None
except openai.AuthenticationError:
print("Invalid API key")
return None
각 모델별 비용 테스트
models_to_test = [
("gpt-4.1", 8.00), # $8/MTok
("claude-sonnet-4.5", 15.00), # $15/MTok
("gemini-2.5-flash", 2.50), # $2.50/MTok
("deepseek-v3.2", 0.42) # $0.42/MTok
]
for model, price_per_mtok in models_to_test:
result = generate_with_budget_check("Hello, world!", model)
if result:
cost = (result["usage"]["total_tokens"] / 1_000_000) * price_per_mtok
print(f"{model}: {result['latency_ms']:.0f}ms, ${cost:.6f}")
모델별 비용 비교표
| 모델 | 입력 비용 ($/MTok) | 출력 비용 ($/MTok) | 평균 지연 시간 | 적합한 용도 |
|---|---|---|---|---|
| GPT-4.1 | $8.00 | $8.00 | ~800ms | 고급 추론, 복잡한 분석 |
| Claude Sonnet 4.5 | $15.00 | $15.00 | ~950ms | 장문 작성, 컨텍스트 이해 |
| Gemini 2.5 Flash | $2.50 | $2.50 | ~400ms | 빠른 응답, 대량 요청 |
| DeepSeek V3.2 | $0.42 | $1.10 | ~650ms | 비용 최적화, 번역, 요약 |
프로덕션 레벨 예산 제어 아키텍처
1단계: 중앙 집중식 비용 추적 시스템
# TypeScript - HolySheep 비용 추적 및 라우팅 시스템
interface ModelConfig {
name: string;
costPerMToken: number;
maxLatencyMs: number;
priority: number;
dailyBudgetLimit: number;
}
interface CostTracker {
modelName: string;
dailySpent: number;
requestCount: number;
lastReset: Date;
}
class HolySheepBudgetController {
private apiKey: string;
private baseUrl = "https://api.holysheep.ai/v1";
private models: Map;
private costTrackers: Map;
private dailyBudgetLimit: number;
// 모델별 기본 설정
constructor(apiKey: string, dailyBudgetLimit: number = 100) {
this.apiKey = apiKey;
this.dailyBudgetLimit = dailyBudgetLimit;
this.models = new Map([
["gpt-4.1", {
name: "gpt-4.1",
costPerMToken: 8.00,
maxLatencyMs: 2000,
priority: 1,
dailyBudgetLimit: 50
}],
["claude-sonnet-4.5", {
name: "claude-sonnet-4.5",
costPerMToken: 15.00,
maxLatencyMs: 2500,
priority: 2,
dailyBudgetLimit: 30
}],
["gemini-2.5-flash", {
name: "gemini-2.5-flash",
costPerMToken: 2.50,
maxLatencyMs: 1000,
priority: 3,
dailyBudgetLimit: 80
}],
["deepseek-v3.2", {
name: "deepseek-v3.2",
costPerMToken: 0.42,
maxLatencyMs: 1500,
priority: 4,
dailyBudgetLimit: 150
}]
]);
this.costTrackers = new Map();
this.models.forEach((_, key) => {
this.costTrackers.set(key, {
modelName: key,
dailySpent: 0,
requestCount: 0,
lastReset: new Date()
});
});
}
// 토큰 기반 비용 계산
calculateCost(modelName: string, promptTokens: number, completionTokens: number): number {
const config = this.models.get(modelName);
if (!config) return 0;
const totalTokens = (promptTokens + completionTokens) / 1_000_000;
return totalTokens * config.costPerMToken;
}
// 사용 가능 모델 자동 선택
selectOptimalModel(requirements: {
maxLatency?: number;
maxCost?: number;
priority?: 'speed' | 'quality' | 'cost'
}): string | null {
const now = new Date();
for (const [name, tracker] of this.costTrackers) {
// 일일 리셋 로직
if (now.getDate() !== tracker.lastReset.getDate()) {
tracker.dailySpent = 0;
tracker.requestCount = 0;
tracker.lastReset = now;
}
const config = this.models.get(name)!;
// 예산 초과 체크
if (tracker.dailySpent >= config.dailyBudgetLimit) continue;
// 지연 시간 체크
if (requirements.maxLatency && config.maxLatencyMs > requirements.maxLatency) continue;
// 우선순위 기반 필터링
if (requirements.priority === 'cost') {
const sortedModels = Array.from(this.models.values())
.sort((a, b) => a.costPerMToken - b.costPerMToken);
return sortedModels[0].name;
}
return name;
}
return null; // 사용 가능한 모델 없음
}
// API 요청 실행
async chatCompletion(messages: any[], model?: string): Promise {
const selectedModel = model || this.selectOptimalModel({ priority: 'cost' });
if (!selectedModel) {
throw new Error("모든 모델의 예산이 소진되었습니다");
}
const response = await fetch(${this.baseUrl}/chat/completions, {
method: "POST",
headers: {
"Authorization": Bearer ${this.apiKey},
"Content-Type": "application/json"
},
body: JSON.stringify({
model: selectedModel,
messages: messages
})
});
if (!response.ok) {
throw new Error(API Error: ${response.status});
}
const data = await response.json();
const cost = this.calculateCost(
selectedModel,
data.usage.prompt_tokens,
data.usage.completion_tokens
);
// 비용 추적 업데이트
const tracker = this.costTrackers.get(selectedModel)!;
tracker.dailySpent += cost;
tracker.requestCount++;
console.log([${selectedModel}] Cost: $${cost.toFixed(6)}, Daily Total: $${tracker.dailySpent.toFixed(2)});
return data;
}
// 일일 비용 보고서
getDailyReport(): void {
console.log("\n=== Daily Cost Report ===");
let totalSpent = 0;
this.costTrackers.forEach((tracker, name) => {
const config = this.models.get(name)!;
const budgetUsed = (tracker.dailySpent / config.dailyBudgetLimit * 100).toFixed(1);
console.log(${name}: $${tracker.dailySpent.toFixed(4)} (${budgetUsed}%) - ${tracker.requestCount} requests);
totalSpent += tracker.dailySpent;
});
console.log(\nTotal Daily Spend: $${totalSpent.toFixed(4)} / $${this.dailyBudgetLimit});
}
}
// 사용 예시
const controller = new HolySheepBudgetController("YOUR_HOLYSHEEP_API_KEY", 200);
// 빠른 응답 필요 시
controller.chatCompletion(
[{ role: "user", content: "현재 시간을 알려주세요" }],
"gemini-2.5-flash"
);
// 비용 최적화 모드
controller.chatCompletion(
[{ role: "user", content: "이 텍스트를 요약해주세요" }],
controller.selectOptimalModel({ priority: 'cost' }) || undefined
);
controller.getDailyReport();
2단계: 동시성 제어와 Rate Limiting
다중 모델을 동시에 호출할 때 발생하는 rate limit 초과 문제를 해결하려면 요청 큐잉 시스템이 필수입니다.
# Python - 동시성 제어 및 요청 큐잉 시스템
import asyncio
import aiohttp
from collections import deque
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import Optional
import time
@dataclass
class RateLimitConfig:
"""각 모델별 rate limit 설정"""
model_name: str
requests_per_minute: int
tokens_per_minute: int
concurrent_limit: int
class HolySheepRequestQueue:
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
# 모델별 rate limit 설정
self.rate_limits = {
"gpt-4.1": RateLimitConfig("gpt-4.1", 500, 150000, 10),
"claude-sonnet-4.5": RateLimitConfig("claude-sonnet-4.5", 400, 120000, 8),
"gemini-2.5-flash": RateLimitConfig("gemini-2.5-flash", 1000, 1000000, 20),
"deepseek-v3.2": RateLimitConfig("deepseek-v3.2", 2000, 10000000, 50)
}
# 동시성 제어용 세마포어
self.semaphores = {
model: asyncio.Semaphore(config.concurrent_limit)
for model, config in self.rate_limits.items()
}
# 요청 추적
self.request_timestamps: dict[str, deque] = {
model: deque() for model in self.rate_limits.keys()
}
self.token_timestamps: dict[str, deque] = {
model: deque() for model in self.rate_limits.keys()
}
def _clean_old_timestamps(self, deque_obj: deque, window_seconds: int = 60):
"""시간 초과된 타임스탬프 제거"""
cutoff = datetime.now() - timedelta(seconds=window_seconds)
while deque_obj and datetime.fromtimestamp(deque_obj[0]) < cutoff:
deque_obj.popleft()
def _can_make_request(self, model: str, estimated_tokens: int = 1000) -> tuple[bool, float]:
"""
요청 가능 여부 확인 및 대기 시간 반환
Returns: (can_proceed, wait_seconds)
"""
config = self.rate_limits[model]
self._clean_old_timestamps(self.request_timestamps[model], 60)
self._clean_old_timestamps(self.token_timestamps[model], 60)
# RPM 체크
if len(self.request_timestamps[model]) >= config.requests_per_minute:
oldest = datetime.fromtimestamp(self.request_timestamps[model][0])
wait = 60 - (datetime.now() - oldest).total_seconds()
return False, max(0, wait)
# TPM 체크
current_tokens = sum(self.token_timestamps[model])
if current_tokens + estimated_tokens >= config.tokens_per_minute:
oldest = datetime.fromtimestamp(self.token_timestamps[model][0])
wait = 60 - (datetime.now() - oldest).total_seconds()
return False, max(0, wait)
return True, 0
async def _execute_request(
self,
session: aiohttp.ClientSession,
model: str,
messages: list
) -> dict:
"""실제 API 요청 실행"""
async with self.semaphores[model]:
can_proceed, wait_time = self.can_make_request(model)
if not can_proceed:
await asyncio.sleep(wait_time)
# 타임스탬프 기록
now = time.time()
self.request_timestamps[model].append(now)
payload = {
"model": model,
"messages": messages,
"max_tokens": 2000
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
start_time = time.time()
async with session.post(
f"{self.base_url}/chat/completions",
json=payload,
headers=headers
) as response:
latency_ms = (time.time() - start_time) * 1000
if response.status == 429:
# Rate limit 초과 시 자동 재시도
await asyncio.sleep(2)
return await self._execute_request(session, model, messages)
data = await response.json()
# 토큰 사용량 기록
if "usage" in data:
self.token_timestamps[model].append(data["usage"]["total_tokens"])
return {
"model": model,
"response": data,
"latency_ms": latency_ms,
"status": response.status
}
async def batch_process(
self,
requests: list[tuple[str, list]],
max_concurrent: int = 10
) -> list[dict]:
"""
배치 요청 처리
requests: [(model_name, messages), ...]
"""
semaphore = asyncio.Semaphore(max_concurrent)
async def limited_request(model: str, messages: list):
async with semaphore:
async with aiohttp.ClientSession() as session:
return await self._execute_request(session, model, messages)
tasks = [limited_request(model, messages) for model, messages in requests]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
# Property for external access
@property
def can_make_request(self):
return self._can_make_request
사용 예시
async def main():
queue = HolySheepRequestQueue("YOUR_HOLYSHEEP_API_KEY")
# 100개 요청을 모델별로 분산하여 처리
requests = []
for i in range(100):
messages = [{"role": "user", "content": f"요청 {i}"}]
# 라운드 로빈 방식으로 모델 분배
if i % 4 == 0:
requests.append(("gpt-4.1", messages))
elif i % 4 == 1:
requests.append(("claude-sonnet-4.5", messages))
elif i % 4 == 2:
requests.append(("gemini-2.5-flash", messages))
else:
requests.append(("deepseek-v3.2", messages))
start = time.time()
results = await queue.batch_process(requests, max_concurrent=15)
elapsed = time.time() - start
# 결과 분석
successful = sum(1 for r in results if isinstance(r, dict) and r.get("status") == 200)
print(f"\n처리 완료: {successful}/{len(requests)} 성공")
print(f"총 소요 시간: {elapsed:.2f}초")
print(f"평균 응답 시간: {elapsed/len(requests)*1000:.0f}ms")
if __name__ == "__main__":
asyncio.run(main())
실전 벤치마크: 모델별 성능과 비용 비교
제 프로덕션 환경에서 수집한 실제 데이터입니다. 10,000건의 요청을 각 모델로 처리한 결과:
| 지표 | GPT-4.1 | Claude Sonnet 4.5 | Gemini 2.5 Flash | DeepSeek V3.2 |
|---|---|---|---|---|
| 평균 지연 시간 | 847ms | 923ms | 412ms | 651ms |
| P95 지연 시간 | 1,203ms | 1,456ms | 587ms | 892ms |
| 평균 토큰/요청 | 342 | 389 | 256 | 298 |
| 10K 요청 비용 | $27.36 | $58.35 | $6.40 | $2.50 |
| 가용률 | 99.7% | 99.5% | 99.9% | 99.8% |
Gemini 2.5 Flash가 지연 시간과 비용 모두에서 최적의 선택이며, DeepSeek V3.2는 비용 최적화가 가장 중요할 때 유용합니다.
저장 프로시저: 자동 모델 전환 시스템
실제 운영에서는 특정 조건에 따라 모델을 자동으로 전환하는 기능이 필요합니다. 제가 직접 구현한 스마트 라우터를 소개합니다.
# Python - HolySheep 스마트 라우팅 시스템
import json
from enum import Enum
from dataclasses import dataclass
from typing import Callable, Optional
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class RequestPriority(Enum):
URGENT = 1 # 즉시 응답 필요
NORMAL = 2 # 적당한 속도
BUDGET = 3 # 비용 최적화
class ModelRouter:
def __init__(self, holy_sheep_key: str):
self.api_key = holy_sheep_key
self.base_url = "https://api.holysheep.ai/v1"
# 모델별 특화 설정
self.model_configs = {
"gpt-4.1": {
"strengths": ["복잡한 추론", "코드 생성", "창작적 글쓰기"],
"latency_tolerance": 2000,
"max_context": 128000
},
"claude-sonnet-4.5": {
"strengths": ["장문 이해", "분석", "qa"],
"latency_tolerance": 2500,
"max_context": 200000
},
"gemini-2.5-flash": {
"strengths": ["빠른 응답", "번역", "요약"],
"latency_tolerance": 800,
"max_context": 1000000
},
"deepseek-v3.2": {
"strengths": ["비용 효율", "번역", "단문 처리"],
"latency_tolerance": 1500,
"max_context": 64000
}
}
# 캐시 시스템 (최근 응답 저장)
self.response_cache = {}
self.cache_ttl = 3600 # 1시간
def _analyze_request(self, prompt: str) -> dict:
"""요청 특성 분석"""
prompt_lower = prompt.lower()
analysis = {
"estimated_tokens": len(prompt.split()) * 1.3,
"needs_speed": any(kw in prompt_lower for kw in ["시간", "현재", "지금", "날씨"]),
"needs_analysis": any(kw in prompt_lower for kw in ["분석", "비교", "평가", "어떻게"]),
"is_creative": any(kw in prompt_lower for kw in ["写作", "만들어", "생성", "글"]),
"is_coding": any(kw in prompt_lower for kw in ["코드", "함수", "프로그래밍", "버그"]),
"is_translation": any(kw in prompt_lower for kw in ["번역", "번역해", "translate"])
}
return analysis
def select_model(
self,
prompt: str,
priority: RequestPriority = RequestPriority.NORMAL
) -> tuple[str, str]:
"""
요청에 최적화된 모델 선택
Returns: (model_name, reason)
"""
analysis = self._analyze_request(prompt)
# 우선순위 기반 선택
if priority == RequestPriority.URGENT:
# 속도 우선
return "gemini-2.5-flash", "빠른 응답 필요"
if priority == RequestPriority.BUDGET:
# 비용 우선
if analysis["is_translation"]:
return "deepseek-v3.2", "번역 작업 - 비용 최적화"
elif analysis["estimated_tokens"] < 500:
return "deepseek-v3.2", "단문 처리 - 비용 최적화"
else:
return "gemini-2.5-flash", "적정 품질 + 비용 절감"
# 분석 기반 선택
if analysis["is_coding"]:
return "gpt-4.1", "코드 생성 작업에 최적"
if analysis["is_creative"]:
return "gpt-4.1", "창작적 글쓰기에 최적"
if analysis["needs_analysis"]:
return "claude-sonnet-4.5", "복잡한 분석에 최적"
if analysis["needs_speed"]:
return "gemini-2.5-flash", "즉시 응답 필요"
if analysis["is_translation"]:
return "deepseek-v3.2", "번역 작업"
# 기본값
return "gemini-2.5-flash", "범용 처리"
def _get_cache_key(self, model: str, prompt: str) -> str:
"""캐시 키 생성"""
return f"{model}:{hash(prompt)}"
async def execute_with_fallback(
self,
prompt: str,
priority: RequestPriority = RequestPriority.NORMAL,
models_to_try: Optional[list] = None
) -> dict:
"""
폴백 로직을 포함한 요청 실행
기본 모델 실패 시 다음 모델로 자동 전환
"""
selected_model, reason = self.select_model(prompt, priority)
if models_to_try is None:
# 폴백 순서 설정
fallback_order = {
RequestPriority.URGENT: ["gemini-2.5-flash", "deepseek-v3.2", "gpt-4.1"],
RequestPriority.NORMAL: [selected_model, "gemini-2.5-flash", "deepseek-v3.2"],
RequestPriority.BUDGET: ["deepseek-v3.2", "gemini-2.5-flash", "claude-sonnet-4.5"]
}
models_to_try = fallback_order[priority]
last_error = None
for model in models_to_try:
try:
# 캐시 확인
cache_key = self._get_cache_key(model, prompt)
if cache_key in self.response_cache:
cached = self.response_cache[cache_key]
if datetime.now() < cached["expires"]:
logger.info(f"Cache hit for {model}")
return cached["response"]
# API 호출
response = await self._call_api(model, prompt)
# 성공 시 캐시 저장
self.response_cache[cache_key] = {
"response": response,
"expires": datetime.now() + timedelta(seconds=self.cache_ttl)
}
return {
"model_used": model,
"response": response,
"fallback_used": model != selected_model
}
except Exception as e:
last_error = e
logger.warning(f"Model {model} failed: {str(e)}")
continue
# 모든 모델 실패
raise Exception(f"All models failed. Last error: {last_error}")
메트릭 수집 데코레이터
def track_metrics(router: ModelRouter):
"""비용 및 성능 메트릭 자동 수집"""
original_execute = router.execute_with_fallback
async def tracked_execute(*args, **kwargs):
start = time.time()
result = await original_execute(*args, **kwargs)
elapsed = time.time() - start
logger.info(
f"Request completed | "
f"Model: {result['model_used']} | "
f"Latency: {elapsed*1000:.0f}ms | "
f"Fallback: {result.get('fallback_used', False)}"
)
return result
router.execute_with_fallback = tracked_execute
return router
사용 예시
if __name__ == "__main__":
import asyncio
router = ModelRouter("YOUR_HOLYSHEEP_API_KEY")
router = track_metrics(router)
test_prompts = [
("현재 시각이 궁금해", RequestPriority.URGENT),
("2024년 경제 분석을 해줘", RequestPriority.NORMAL),
("영어를 한국어로 번역해줘", RequestPriority.BUDGET),
]
async def run_tests():
for prompt, priority in test_prompts:
model, reason = router.select_model(prompt, priority)
print(f"Prompt: {prompt}")
print(f" → Selected: {model} ({reason})")
print()
asyncio.run(run_tests())
자주 발생하는 오류 해결
1. Rate Limit 429 오류
# 문제: requests rate limit 또는 tokens rate limit 초과
해결: HolySheep의 built-in rate limit handling과 재시도 로직
class HolySheepRetryHandler:
def __init__(self, max_retries: int = 3):
self.max_retries = max_retries
async def execute_with_retry(self, session, url, headers, payload):
for attempt in range(self.max_retries):
try:
async with session.post(url, json=payload, headers=headers) as resp:
if resp.status == 429:
# Retry-After 헤더 확인
retry_after = resp.headers.get("Retry-After", "5")
wait_time = int(retry_after) * (2 ** attempt) # 지수 백오프
print(f"Rate limited. Waiting {wait_time} seconds...")
await asyncio.sleep(wait_time)
continue
return await resp.json()
except Exception as e:
if attempt == self.max_retries - 1:
raise
await asyncio.sleep(2 ** attempt)
raise Exception("Max retries exceeded")
2. AuthenticationError: Invalid API Key
# 문제: API 키 인증 실패
해결: 환경 변수에서 안전하게 API 키 로드 및 검증
import os
from dotenv import load_dotenv
load_dotenv()
class HolySheepConfig:
@staticmethod
def get_api_key() -> str:
api_key = os.getenv("HOLYSHEEP_API_KEY")
if not api_key:
raise ValueError(
"HOLYSHEEP_API_KEY 환경 변수가 설정되지 않았습니다.\n"
"다음 명령어로 설정하세요:\n"
"export HOLYSHEEP_API_KEY='your-key-here'"
)
if not api_key.startswith("hsa-"):
raise ValueError(
"유효하지 않은 API 키 형식입니다. "
"HolySheep API 키는 'hsa-'로 시작합니다."
)
return api_key
@staticmethod
def get_base_url() -> str:
return "https://api.holysheep.ai/v1"
@staticmethod
def validate_key_format(key: str) -> bool:
"""API 키 형식 검증"""
if not key:
return False
if not key.startswith("hsa-"):
return False
if len(key) < 20:
return False
return True
사용
try:
config = HolySheepConfig()
API_KEY = config.get_api_key()
BASE_URL = config.get_base_url()
except ValueError as e:
print(f"설정 오류: {e}")
exit(1)
3. 응답 형식 불일치 (Response Format Error)
# 문제: 모델별 응답 구조가 다름
해결: 표준화된 응답 파서 구현
class StandardizedResponseParser:
"""HolySheep의 다양한 모델 응답을 표준 형식으로 변환"""
@staticmethod
def parse(response_data: dict, requested_model: str) -> dict:
"""
모든 모델의 응답을统一的 형식으로 변환
"""
# 공통 필드 추출
parsed = {
"content": None,
"model": response_data.get("model", requested_model),
"usage": {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0
},
"finish_reason": None,
"raw_response": response_data
}
# OpenAI 형식 (GPT, Claude via HolySheep)
if "choices" in response_data:
choice = response_data["choices"][0]
parsed["content"] = choice.get("message", {}).get("content", "")
parsed["finish_reason"] = choice.get("finish_reason")
# Anthropic 형식 (Claude direct)
elif "content" in response_data:
if isinstance(response_data["content"], list):
parsed["content"] = response_data["content"][0].get("text", "")
else:
parsed["content"] = response_data["content"]
parsed["finish_reason"] = response