저는 3년 전 이커머스 스타트업에서 AI 고객 서비스 시스템을 구축할 때 흥미로운 도전에 직면했습니다. 매출이 급증하면서 피크 타임에 ChatGPT API 응답이 8초 이상 지연되는 문제가 발생한 것이죠. 비용을 절감하면서도 응답 속도를 유지하는 방법을 찾다 보니, 여러 AI 모델을 통합하는 다중 모델聚合 게이트웨이가 필요하게 되었습니다. 오늘은 HolySheep AI를 활용하여 고성능 다중 모델聚合 게이트웨이를 구축하는 방법을 상세히 알려드리겠습니다.
다중 모델聚合 게이트웨이란?
다중 모델聚合 게이트웨이는 단일 API 엔드포인트를 통해 여러 AI 모델(GPT-5.5, Gemini, Claude 등)을 통합 관리하는 시스템입니다. 다음과 같은 핵심 이점을 제공합니다:
- 비용 최적화: Gemini 2.5 Flash는 $2.50/MTok, DeepSeek V3.2는 $0.42/MTok로 GPT-4.1($8/MTok) 대비 최대 95% 비용 절감 가능
- 장애 복원력: 특정 모델 장애 시 자동 Failover로 서비스 가용성 99.9% 확보
- 스마트 라우팅: 작업 유형에 따라 최적 모델 자동 선택
- 단일 키 관리: HolySheep AI의 단일 API 키로 모든 주요 모델 통합
실전 구축: 이커머스 AI 고객 서비스 시스템
실제 사용 사례로, 이커머스 플랫폼의 AI 고객 서비스를 구축해보겠습니다. 이 시스템은 다음과 같은 요구사항을 충족해야 합니다:
- 상품 문의 → Gemini 2.5 Flash (저비용, 빠른 응답)
- 복잡한 추천 → GPT-5.5 (고품질 분석)
- 긴 대화 컨텍스트 → Claude Sonnet 4.5 (200K 토큰 컨텍스트)
아키텍처 설계
┌─────────────────────────────────────────────────────────────┐
│ HolySheep AI Gateway │
│ https://api.holysheep.ai/v1 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Client │──▶│ Router │──▶│ Model │──▶│ Response │ │
│ │ Request │ │ (Task │ │ Pool │ │ Aggre- │ │
│ │ │ │ Analysis)│ │ │ │ gator │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
│ │ │
│ ┌─────────────┼─────────────┐ │
│ ▼ ▼ ▼ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Gemini │ │ GPT-5.5 │ │ Claude │ │
│ │ 2.5 │ │ │ │ Sonnet │ │
│ │ Flash │ │ │ │ 4.5 │ │
│ │ $2.50/M │ │ $8/MTok │ │ $15/MTok │ │
│ └──────────┘ └──────────┘ └──────────┘ │
└─────────────────────────────────────────────────────────────┘1단계: Python SDK 설치 및 기본 설정
# HolySheep AI Multi-Model Gateway SDK
Python 3.9+ 권장
pip install requests httpx asyncio openai anthropic google-generativeai
pip install holy-sheep-sdk # 실제 SDK (선택사항)
프로젝트 디렉토리 구조
mkdir -p multi_model_gateway/{core,models,utils}
cd multi_model_gateway
환경 변수 설정
export HOLYSHEEP_API_KEY="YOUR_HOLYSHEEP_API_KEY"
export HOLYSHEEP_BASE_URL="https://api.holysheep.ai/v1"2단계: HolySheep AI 기본 클라이언트 구현
# core/holy_sheep_client.py
"""
저는 HolySheep AI의 unified endpoint를 활용하여
여러 모델을 단일 인터페이스로 접근하는 클라이언트를 구현했습니다.
이 방식의 장점은 API 엔드포인트 변경 없이 모델 교체 가능하다는 점입니다.
"""
import os
import json
import time
from typing import Optional, Dict, Any, List
from dataclasses import dataclass
from enum import Enum
import requests
class ModelType(Enum):
GPT_55 = "gpt-5.5"
GEMINI_FLASH = "gemini-2.5-flash"
CLAUDE_SONNET = "claude-sonnet-4.5"
DEEPSEEK = "deepseek-v3.2"
@dataclass
class ModelConfig:
name: str
provider: str
cost_per_1m_tokens: float
max_tokens: int
avg_latency_ms: int
strengths: List[str]
MODEL_CONFIGS = {
ModelType.GPT_55: ModelConfig(
name="gpt-5.5",
provider="openai",
cost_per_1m_tokens=8.00, # $8/MTok
max_tokens=128000,
avg_latency_ms=1200,
strengths=["복잡한 추론", "코드 생성", "창작 작문"]
),
ModelType.GEMINI_FLASH: ModelConfig(
name="gemini-2.5-flash",
provider="google",
cost_per_1m_tokens=2.50, # $2.50/MTok
max_tokens=1000000,
avg_latency_ms=800,
strengths=["빠른 응답", "비용 효율", "장문 처리"]
),
ModelType.CLAUDE_SONNET: ModelConfig(
name="claude-sonnet-4.5",
provider="anthropic",
cost_per_1m_tokens=15.00, # $15/MTok
max_tokens=200000,
avg_latency_ms=1500,
strengths=["긴 컨텍스트", "안전한 출력", "명확한 설명"]
),
ModelType.DEEPSEEK: ModelConfig(
name="deepseek-v3.2",
provider="deepseek",
cost_per_1m_tokens=0.42, # $0.42/MTok
max_tokens=64000,
avg_latency_ms=900,
strengths=["초저비용", "수학 추론", "코드"]
),
}
class HolySheepClient:
"""HolySheep AI Unified Gateway Client"""
def __init__(self, api_key: str = None):
self.api_key = api_key or os.getenv("HOLYSHEEP_API_KEY")
if not self.api_key:
raise ValueError("HolySheep API key is required")
self.base_url = "https://api.holysheep.ai/v1"
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
})
# 실제 지연 시간 추적 (프로덕션 환경)
self.latency_tracker = {
ModelType.GPT_55: [],
ModelType.GEMINI_FLASH: [],
ModelType.CLAUDE_SONNET: [],
ModelType.DEEPSEEK: [],
}
def chat_completion(
self,
model: ModelType,
messages: List[Dict],
temperature: float = 0.7,
max_tokens: int = 2048
) -> Dict[str, Any]:
"""
HolySheep AI를 통한 채팅 완성 요청
모든 모델은 unified endpoint 하나로 접근
"""
start_time = time.time()
payload = {
"model": model.value,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
try:
response = self.session.post(
f"{self.base_url}/chat/completions",
json=payload,
timeout=30
)
response.raise_for_status()
result = response.json()
# 지연 시간 기록
latency_ms = (time.time() - start_time) * 1000
self._track_latency(model, latency_ms)
# 비용 계산
input_tokens = result.get("usage", {}).get("prompt_tokens", 0)
output_tokens = result.get("usage", {}).get("completion_tokens", 0)
cost = self._calculate_cost(model, input_tokens, output_tokens)
return {
"success": True,
"model": model.value,
"content": result["choices"][0]["message"]["content"],
"latency_ms": round(latency_ms, 2),
"tokens": {"input": input_tokens, "output": output_tokens},
"cost_usd": cost
}
except requests.exceptions.RequestException as e:
return {
"success": False,
"error": str(e),
"model": model.value
}
def _track_latency(self, model: ModelType, latency_ms: float):
"""지연 시간 추적 (최근 100개 샘플)"""
self.latency_tracker[model].append(latency_ms)
if len(self.latency_tracker[model]) > 100:
self.latency_tracker[model].pop(0)
def _calculate_cost(self, model: ModelType, input_tokens: int, output_tokens: int) -> float:
"""토큰 사용량 기반 비용 계산"""
config = MODEL_CONFIGS[model]
total_tokens = input_tokens + output_tokens
return round((total_tokens / 1_000_000) * config.cost_per_1m_tokens, 6)
def get_average_latency(self, model: ModelType) -> float:
"""평균 지연 시간 반환 (밀리초)"""
latencies = self.latency_tracker.get(model, [])
return round(sum(latencies) / len(latencies), 2) if latencies else 0
사용 예시
if __name__ == "__main__":
client = HolySheepClient()
# Gemini 2.5 Flash로 간단한 질문
response = client.chat_completion(
model=ModelType.GEMINI_FLASH,
messages=[
{"role": "system", "content": "당신은 친절한 고객 서비스 담당자입니다."},
{"role": "user", "content": "반품 정책이 어떻게 되나요?"}
]
)
print(f"Model: {response['model']}")
print(f"Latency: {response['latency_ms']}ms")
print(f"Cost: ${response['cost_usd']}")
print(f"Response: {response['content'][:100]}...")3단계: 스마트 라우팅 시스템 구현
# core/smart_router.py
"""
저는 이커머스 AI 고객 서비스를 구축하면서
작업 유형별로 최적 모델을 자동 선택하는 라우팅 로직을 구현했습니다.
이 시스템은 응답 시간, 비용, 품질을 자동으로 최적화합니다.
"""
import re
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass
from enum import Enum
class TaskType(Enum):
PRODUCT_INQUIRY = "product_inquiry" # 상품 문의
ORDER_STATUS = "order_status" # 주문 상태
RETURN_REFUND = "return_refund" # 반품/환불
PRODUCT_RECOMMENDATION = "recommendation" # 상품 추천
COMPLEX_ANALYSIS = "complex_analysis" # 복잡한 분석
CODE_GENERATION = "code_generation" # 코드 생성
GENERAL = "general" # 일반 대화
@dataclass
class RoutingRule:
task_type: TaskType
primary_model: ModelType
fallback_models: List[ModelType]
priority_keywords: List[str]
estimated_complexity: int # 1-10
class SmartRouter:
"""작업 유형 기반 스마트 모델 라우팅 시스템"""
def __init__(self, client: HolySheepClient):
self.client = client
self.rules = self._initialize_routing_rules()
def _initialize_routing_rules(self) -> List[RoutingRule]:
"""라우팅 규칙 초기화"""
return [
RoutingRule(
task_type=TaskType.PRODUCT_INQUIRY,
primary_model=ModelType.GEMINI_FLASH,
fallback_models=[ModelType.DEEPSEEK, ModelType.GPT_55],
priority_keywords=[
"상품", "가격", "사이즈", "색상", "재고",
"배송", "옵션", "규격", "사양"
],
estimated_complexity=3
),
RoutingRule(
task_type=TaskType.ORDER_STATUS,
primary_model=ModelType.GEMINI_FLASH,
fallback_models=[ModelType.DEEPSEEK],
priority_keywords=[
"주문", "배송", "도착", "追踪", "배달",
"발송", "상태", "예약", "결제"
],
estimated_complexity=2
),
RoutingRule(
task_type=TaskType.RETURN_REFUND,
primary_model=ModelType.GEMINI_FLASH,
fallback_models=[ModelType.CLAUDE_SONNET, ModelType.GPT_55],
priority_keywords=[
"반품", "환불", "취소", "교환", "리턴",
"환불금", "반품지", "회수"
],
estimated_complexity=4
),
RoutingRule(
task_type=TaskType.PRODUCT_RECOMMENDATION,
primary_model=ModelType.GPT_55,
fallback_models=[ModelType.GEMINI_FLASH, ModelType.CLAUDE_SONNET],
priority_keywords=[
"추천", "선호", "관심", "좋아하는", "스타일",
"메칭", "함께", "비슷한", "대안"
],
estimated_complexity=7
),
RoutingRule(
task_type=TaskType.COMPLEX_ANALYSIS,
primary_model=ModelType.GPT_55,
fallback_models=[ModelType.CLAUDE_SONNET],
priority_keywords=[
"분석", "비교", "평가", "장단점", "차이",
"이유", "원인", "결과", "영향"
],
estimated_complexity=9
),
RoutingRule(
task_type=TaskType.CODE_GENERATION,
primary_model=ModelType.DEEPSEEK,
fallback_models=[ModelType.GPT_55, ModelType.GEMINI_FLASH],
priority_keywords=[
"코드", "프로그래밍", "함수", "알고리즘",
"디버그", "리팩토링", "스키마", "API"
],
estimated_complexity=6
),
]
def classify_task(self, user_message: str) -> Tuple[TaskType, float]:
"""
사용자 메시지 기반으로 작업 유형 분류
반환값: (작업 유형, 신뢰도 점수)
"""
user_message_lower = user_message.lower()
best_match = TaskType.GENERAL
best_score = 0.0
for rule in self.rules:
score = 0
for keyword in rule.priority_keywords:
if keyword in user_message_lower:
score += 1
# 복잡도에 따른 가중치 조정
if score > 0:
final_score = score * (10 - rule.estimated_complexity) / 10
if final_score > best_score:
best_score = final_score
best_match = rule.task_type
return best_match, best_score
def select_model(self, task_type: TaskType) -> ModelType:
"""현재 시스템 상태 기반 최적 모델 선택"""
# 부하 분산 및 지연 시간 기반 선택
for rule in self.rules:
if rule.task_type == task_type:
# 기본적으로 primary 모델 선택
primary = rule.primary_model
# 지연 시간이 임계값 초과 시 faster 모델로 전환
avg_latency = self.client.get_average_latency(primary)
if avg_latency > 2000: # 2초 이상
for fallback in rule.fallback_models:
fallback_latency = self.client.get_average_latency(fallback)
if fallback_latency < avg_latency * 0.8:
return fallback
return primary
return ModelType.GEMINI_FLASH # 기본값
def execute_with_fallback(
self,
messages: List[Dict],
max_retries: int = 2
) -> Dict[str, Any]:
"""
실패 시 자동 Failover가 포함된 요청 실행
"""
task_type, confidence = self.classify_task(messages[-1]["content"])
primary_model = self.select_model(task_type)
# 라우팅 규칙 찾기
rule = next(
(r for r in self.rules if r.task_type == task_type),
None
)
fallback_chain = [primary_model]
if rule:
fallback_chain.extend(rule.fallback_models[:max_retries])
last_error = None
for model in fallback_chain[:max_retries + 1]:
response = self.client.chat_completion(
model=model,
messages=messages,
max_tokens=2048
)
if response["success"]:
return {
**response,
"task_type": task_type.value,
"confidence": round(confidence, 2),
"model_selection": "primary" if model == primary_model else "fallback"
}
last_error = response.get("error")
return {
"success": False,
"error": f"All models failed. Last error: {last_error}",
"task_type": task_type.value
}
통합 이커머스 AI 서비스
class EcommerceAIService:
"""이커머스 AI 고객 서비스 통합 시스템"""
def __init__(self, api_key: str):
self.client = HolySheepClient(api_key)
self.router = SmartRouter(self.client)
def chat(self, user_id: str, message: str, context: Dict = None) -> Dict:
"""통합 채팅 인터페이스"""
# 시스템 프롬프트 구성
system_prompt = self._build_system_prompt(context)
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": message}
]
# 스마트 라우팅 실행
result = self.router.execute_with_fallback(messages)
# 응답 로깅 (비용 추적용)
if result.get("success"):
self._log_interaction(user_id, result)
return result
def _build_system_prompt(self, context: Dict = None) -> str:
"""컨텍스트 기반 시스템 프롬프트 생성"""
base = """당신은 친절하고 전문적인 이커머스 고객 서비스 담당자입니다.
한국어로 답변하며, 간결하고 명확하게 정보를 제공합니다.
상품 문의, 주문 상태, 반품/환불 절차를 안내해주세요."""
if context and context.get("user_preferences"):
base += f"\n\n고객 선호사항: {context['user_preferences']}"
if context and context.get("recent_orders"):
base += f"\n\n최근 주문: {context['recent_orders']}"
return base
def _log_interaction(self, user_id: str, result: Dict):
"""대화 상호작용 로깅"""
print(f"[LOG] User: {user_id} | Model: {result['model']} | "
f"Cost: ${result.get('cost_usd', 0):.6f} | "
f"Latency: {result.get('latency_ms', 0)}ms | "
f"Task: {result.get('task_type')}")
사용 예시
if __name__ == "__main__":
service = EcommerceAIService(api_key="YOUR_HOLYSHEEP_API_KEY")
# 다양한 유형의 쿼리 테스트
queries = [
"상품 재고 확인해주세요",
"주문한 상품 언제 도착하나요?",
"이 상품과 비슷한 추천 해주세요",
"반품 신청하고 싶은데요"
]
for query in queries:
print(f"\n{'='*60}")
print(f"Query: {query}")
result = service.chat(user_id="user_123", message=query)
print(f"Result: {result.get('content', result.get('error'))[:200]}...")
print(f"Metadata: model={result.get('model')}, "
f"task={result.get('task_type')}, "
f"cost=${result.get('cost_usd', 0):.6f}")4단계: Node.js/JavaScript 구현
// models/multiModelClient.js
/**
* HolySheep AI Multi-Model Gateway Client for Node.js
* 비동기 API 호출 및 배치 처리 지원
*/
const https = require('https');
class HolySheepClient {
constructor(apiKey) {
this.apiKey = apiKey || process.env.HOLYSHEEP_API_KEY;
this.baseUrl = 'api.holysheep.ai';
this.basePath = '/v1';
if (!this.apiKey) {
throw new Error('HolySheep API key is required');
}
}
// HTTP 요청 헬퍼
async request(endpoint, payload, timeout = 30000) {
return new Promise((resolve, reject) => {
const startTime = Date.now();
const options = {
hostname: this.baseUrl,
path: ${this.basePath}${endpoint},
method: 'POST',
headers: {
'Authorization': Bearer ${this.apiKey},
'Content-Type': 'application/json'
},
timeout: timeout
};
const req = https.request(options, (res) => {
let data = '';
res.on('data', (chunk) => {
data += chunk;
});
res.on('end', () => {
const latencyMs = Date.now() - startTime;
try {
const result = JSON.parse(data);
// 토큰 사용량 및 비용 계산
const inputTokens = result.usage?.prompt_tokens || 0;
const outputTokens = result.usage?.completion_tokens || 0;
const cost = this.calculateCost(payload.model, inputTokens, outputTokens);
resolve({
success: true,
model: payload.model,
content: result.choices?.[0]?.message?.content || '',
latencyMs: latencyMs,
tokens: { input: inputTokens, output: outputTokens },
costUsd: cost
});
} catch (e) {
resolve({
success: false,
error: Parse error: ${e.message},
latencyMs: latencyMs
});
}
});
});
req.on('error', (e) => {
resolve({
success: false,
error: e.message
});
});
req.on('timeout', () => {
req.destroy();
resolve({
success: false,
error: 'Request timeout'
});
});
req.write(JSON.stringify(payload));
req.end();
});
}
calculateCost(model, inputTokens, outputTokens) {
const modelCosts = {
'gpt-5.5': 8.00, // $8/MTok
'gemini-2.5-flash': 2.50, // $2.50/MTok
'claude-sonnet-4.5': 15.00, // $15/MTok
'deepseek-v3.2': 0.42 // $0.42/MTok
};
const costPerM = modelCosts[model] || 8.00;
const totalTokens = inputTokens + outputTokens;
return parseFloat(((totalTokens / 1_000_000) * costPerM).toFixed(6));
}
// 채팅 완성
async chatCompletion(model, messages, options = {}) {
return this.request('/chat/completions', {
model: model,
messages: messages,
temperature: options.temperature || 0.7,
max_tokens: options.maxTokens || 2048
});
}
// 다중 모델 병렬 요청
async multiModelRequest(models, messages) {
const requests = models.map(model =>
this.chatCompletion(model, messages)
);
return Promise.allSettled(requests);
}
// 비용 최적화: cheapest model 먼저 시도
async costOptimizedRequest(messages, preferredModel = null) {
// 비용 순서: DeepSeek < Gemini Flash < GPT-5.5 < Claude
const modelPriority = ['deepseek-v3.2', 'gemini-2.5-flash', 'gpt-5.5', 'claude-sonnet-4.5'];
// 선호 모델이 있으면 먼저 시도
if (preferredModel && modelPriority.includes(preferredModel)) {
const idx = modelPriority.indexOf(preferredModel);
modelPriority.splice(idx, 1);
modelPriority.unshift(preferredModel);
}
for (const model of modelPriority) {
const result = await this.chatCompletion(model, messages);
if (result.success) {
return {
...result,
costOptimization: Used ${model} (cost rank: ${modelPriority.indexOf(model) + 1})
};
}
}
return { success: false, error: 'All models failed' };
}
}
// Express.js 기반 REST API 서버
const express = require('express');
const app = express();
app.use(express.json());
const client = new HolySheepClient(process.env.HOLYSHEEP_API_KEY);
// 기본 채팅 엔드포인트
app.post('/api/chat', async (req, res) => {
try {
const { model = 'gemini-2.5-flash', messages } = req.body;
if (!messages || !Array.isArray(messages)) {
return res.status(400).json({
error: 'messages array is required'
});
}
const result = await client.chatCompletion(model, messages);
res.json(result);
} catch (error) {
res.status(500).json({ error: error.message });
}
});
// 다중 모델 비교 엔드포인트
app.post('/api/compare', async (req, res) => {
try {
const { messages, models = ['gemini-2.5-flash', 'gpt-5.5'] } = req.body;
const results = await client.multiModelRequest(models, messages);
const comparison = results.map((result, idx) => ({
model: models[idx],
success: result.status === 'fulfilled',
latencyMs: result.value?.latencyMs || null,
costUsd: result.value?.costUsd || null,
preview: result.value?.content?.substring(0, 100) || result.reason?.message
}));
res.json({ comparison });
} catch (error) {
res.status(500).json({ error: error.message });
}
});
// 비용 최적화 엔드포인트
app.post('/api/chat/optimized', async (req, res) => {
try {
const { messages, preferredModel } = req.body;
const result = await client.costOptimizedRequest(messages, preferredModel);
res.json(result);
} catch (error) {
res.status(500).json({ error: error.message });
}
});
const PORT = process.env.PORT || 3000;
app.listen(PORT, () => {
console.log(HolySheep AI Gateway running on port ${PORT});
console.log(Base URL: https://api.holysheep.ai/v1);
});
// 실행 예시
async function example() {
const client = new HolySheepClient('YOUR_HOLYSHEEP_API_KEY');
const messages = [
{ role: 'system', content: '당신은 유용한 AI 어시스턴트입니다.' },
{ role: 'user', content: '한국의 유명한 관광지를 추천해주세요.' }
];
// 단일 모델 요청
console.log('Single Model Request:');
const singleResult = await client.chatCompletion('gemini-2.5-flash', messages);
console.log(Model: ${singleResult.model});
console.log(Latency: ${singleResult.latencyMs}ms);
console.log(Cost: $${singleResult.costUsd});
// 다중 모델 비교
console.log('\nMulti-Model Comparison:');
const multiResults = await client.multiModelRequest(
['gemini-2.5-flash', 'deepseek-v3.2'],
messages
);
multiResults.forEach((result, idx) => {
if (result.status === 'fulfilled') {
console.log(${['gemini-2.5-flash', 'deepseek-v3.2'][idx]}: ${result.value.latencyMs}ms, $${result.value.costUsd});
}
});
}
example().catch(console.error);
module.exports = { HolySheepClient };5단계: 비용 모니터링 대시보드
# utils/cost_monitor.py
"""
저는 HolySheep AI 게이트웨이 사용 시
비용을 세밀하게 모니터링하는 시스템을 구축했습니다.
30일간의 비용 추적 및 알림 기능을 포함합니다.
"""
import time
from datetime import datetime, timedelta
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Dict, List, Optional
import json
@dataclass
class UsageRecord:
timestamp: datetime
model: str
input_tokens: int
output_tokens: int
cost_usd: float
latency_ms: float
success: bool
@dataclass
class CostAlert:
threshold_usd: float
percentage: float # 임계값 대비 현재 사용량 %
class CostMonitor:
"""HolySheep AI 비용 모니터링 시스템"""
def __init__(self, daily_budget_usd: float = 100.0):
self.daily_budget = daily_budget_usd
self.monthly_budget = daily_budget_usd * 30
self.records: List[UsageRecord] = []
self.alerts: List[CostAlert] = []
# 모델별 비용 집계
self.model_costs = defaultdict(float)
self.model_tokens = defaultdict(int)
self.model_requests = defaultdict(int)
def record_usage(
self,
model: str,
input_tokens: int,
output_tokens: int,
cost_usd: float,
latency_ms: float,
success: bool = True
):
"""API 사용량 기록"""
record = UsageRecord(
timestamp=datetime.now(),
model=model,
input_tokens=input_tokens,
output_tokens=output_tokens,
cost_usd=cost_usd,
latency_ms=latency_ms,
success=success
)
self.records.append(record)
# 집계 업데이트
self.model_costs[model] += cost_usd
self.model_tokens[model] += input_tokens + output_tokens
self.model_requests[model] += 1
# 일일 예산 초과 체크
daily_cost = self.get_daily_cost()
if daily_cost > self.daily_budget * 0.9: # 90% 임계값
self._trigger_alert(daily_cost)
def get_daily_cost(self) -> float:
"""오늘 총 비용"""
today = datetime.now().date()
return sum(
r.cost_usd for r in self.records
if r.timestamp.date() == today and r.success
)
def get_monthly_cost(self) -> float:
"""이번 달 총 비용"""
first_of_month = datetime.now().replace(day=1, hour=0, minute=0, second=0)
return sum(
r.cost_usd for r in self.records
if r.timestamp >= first_of_month and r.success
)
def get_model_breakdown(self) -> Dict:
"""모델별 비용 상세"""
return {
model: {
"total_cost": round(cost, 6),
"total_tokens": tokens,
"request_count": self.model_requests[model],
"avg_cost_per_request": round(cost / max(self.model_requests[model], 1), 6),
"cost_percentage": round(cost / self.get_monthly_cost() * 100, 2)
}
for model, cost in self.model_costs.items()
}
def get_performance_stats(self) -> Dict:
"""성능 통계 (지연 시간 등)"""
stats = defaultdict(list)
for record in self.records:
if record.success:
stats[record.model].append(record.latency_ms)
return {
model: {
"avg_latency_ms": round(sum(lats) / len(lats), 2),
"min_latency_ms": round(min(lats), 2),
"max_latency_ms": round(max(lats), 2),
"p95_latency_ms": round(sorted(lats)[int(len(lats) * 0.95)], 2)
}
for model, lats in stats.items()
if lats
}
def generate_report(self) -> str:
"""비용 보고서 생성"""
daily = self.get_daily_cost()
monthly = self.get_monthly_cost()
report = f"""
{'='*60}
HolySheep AI 비용 보고서
생성일시: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
{'='*60}
📊 일일 비용
사용액: ${daily:.4f} / ${self.daily_budget:.2f}
사용률: {daily / self.daily_budget * 100:.1f}%
📊 월간 비용
사용액: ${monthly:.4f} / ${self.monthly_budget:.2f}
사용률: {monthly / self.monthly_budget * 100:.1f}%
📈 모델별 사용량
"""
for model, data in self.get_model_breakdown().items():
report += f"""
{model}:
-