프로덕션 환경에서 AI Agent를 운영할 때 가장 큰 고민 중 하나는 비용입니다. 하루에 수십만 요청이 들어오는 시스템에서 모델 선택 하나가 월 수백만 원의 비용 차이를 만듭니다. 이번 튜토리얼에서는 HolySheep AI 게이트웨이를 활용한 토큰 예산 관리와 동적 모델 전환 전략을 실전 코드와 함께 다룹니다.
실제 오류 시나리오로 시작하기
지난달, 제 팀은凌晨 3시에 급하게 호출을 받았습니다. AI Agent 서비스의 일일 비용이 평소의 3배로 급증한 것입니다. 로그를 확인해보니 원인은 명확했습니다:
"""
오류 로그 분석 결과
"""
문제 상황: Claude API 비용 초과 경보
2024-12-15 03:17:42 UTC
Error: 429 Rate Limit Exceeded
Cost: $847.32 (일일 예상 비용의 340%)
근본 원인: 단순 라우팅만 사용
모든 요청 → Claude Sonnet 4.5 ($15/MTok)
복잡도 불문 모든 쿼리에高价 모델 사용
요청 유형별 분포 (로그 분석)
requests = {
"간단한 질문応答": 65, # 예: "오늘 날씨 알려줘"
"중간 난이도 변환": 25, # 예: "JSON을 YAML로 변환"
"복잡한 분석 작업": 10 # 예: "코드 리뷰 + 최적화 제안"
}
비용 비교 (일 10만 요청 기준)
costs_per_10k = {
"Claude Sonnet 4.5 (단일 사용)": "$150/일",
"Gemini 2.5 Flash (단일 사용)": "$25/일",
"스마트 라우팅 (본 가이드 방법)": "$47/일"
}
print("비용 최적화 효과: 69% 절감 가능")이 경험이 이번 튜토리얼을 쓰게 된 계기입니다. HolySheep AI의 단일 API 키로 여러 모델을 통합 관리하면서 비용을 효과적으로 제어하는 방법을 공유합니다.
토큰 예산 관리 시스템 설계
1. Budget Manager 클래스 구현
비용 통제의 핵심은 토큰 사용량을 실시간으로 추적하고 한도를 초과하기 전에 조취를 취하는 것입니다. 다음 BudgetManager 클래스는 일간, 월간 예산을 설정하고 각 요청마다 잔여 예산을 확인합니다.
"""
HolySheep AI 토큰 예산 관리 시스템
Author: HolySheep AI Technical Team
"""
import time
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Optional, List, Dict
from enum import Enum
import threading
class BudgetPeriod(Enum):
DAILY = "daily"
MONTHLY = "monthly"
@dataclass
class TokenUsage:
prompt_tokens: int
completion_tokens: int
total_tokens: int
cost_cents: float
model: str
timestamp: datetime = field(default_factory=datetime.now)
@dataclass
class BudgetConfig:
daily_limit_cents: float = 10000 # 일일 $100
monthly_limit_cents: float = 200000 # 월 $2000
warning_threshold: float = 0.8 # 80% 사용 시 경고
class BudgetManager:
"""토큰 예산을 관리하고 모델 전환을 제어하는 클래스"""
def __init__(self, config: BudgetConfig):
self.config = config
self.daily_usage_cents = 0.0
self.monthly_usage_cents = 0.0
self.usage_history: List[TokenUsage] = []
self._lock = threading.Lock()
self._last_reset = datetime.now().date()
# 모델별 비용 테이블 (HolySheep AI 공식 가격)
self.model_costs = {
"gpt-4.1": 8.0, # $8/MTok
"gpt-4.1-mini": 0.6, # $0.60/MTok
"claude-sonnet-4": 15.0, # $15/MTok
"claude-haiku-4": 1.5, # $1.50/MTok
"gemini-2.5-flash": 2.5, # $2.50/MTok
"gemini-2.5-flash-8b": 0.30, # $0.30/MTok
"deepseek-v3.2": 0.42, # $0.42/MTok
}
def _check_and_reset_daily(self):
"""일일 리셋 체크"""
current_date = datetime.now().date()
if current_date > self._last_reset:
self.daily_usage_cents = 0.0
self._last_reset = current_date
def record_usage(self, usage: TokenUsage) -> bool:
"""토큰 사용량 기록 및 예산 확인"""
with self._lock:
self._check_and_reset_daily()
# 비용 계산
total_cost = (usage.total_tokens / 1_000_000) * self.model_costs.get(
usage.model, 8.0
)
cost_cents = total_cost * 100
# 예산 초과 체크
new_daily = self.daily_usage_cents + cost_cents
new_monthly = self.monthly_usage_cents + cost_cents
if new_daily > self.config.daily_limit_cents:
raise BudgetExceededError(
f"일일 예산 초과! 현재: ${self.daily_usage_cents/100:.2f}, "
f"한도: ${self.config.daily_limit_cents/100:.2f}"
)
if new_monthly > self.config.monthly_limit_cents:
raise BudgetExceededError(
f"월간 예산 초과! 현재: ${self.monthly_usage_cents/100:.2f}, "
f"한도: ${self.config.monthly_limit_cents/100:.2f}"
)
# 사용량 업데이트
self.daily_usage_cents = new_daily
self.monthly_usage_cents = new_monthly
self.usage_history.append(usage)
return True
def get_remaining_budget(self, period: BudgetPeriod) -> Dict:
"""잔여 예산 조회"""
with self._lock:
self._check_and_reset_daily()
if period == BudgetPeriod.DAILY:
used = self.daily_usage_cents
limit = self.config.daily_limit_cents
else:
used = self.monthly_usage_cents
limit = self.config.monthly_limit_cents
return {
"used_cents": used,
"remaining_cents": limit - used,
"limit_cents": limit,
"usage_percent": (used / limit) * 100 if limit > 0 else 0,
"is_warning": (used / limit) >= self.config.warning_threshold
}
def select_model_by_budget(self, task_complexity: str) -> str:
"""예산 상태에 따른 모델 선택"""
daily = self.get_remaining_budget(BudgetPeriod.DAILY)
# 긴급 상황: 일일 예산의 90% 이상 사용
if daily["usage_percent"] >= 90:
return "deepseek-v3.2" # 가장 저렴한 모델
# 경고 상황: 80% 이상 사용
if daily["usage_percent"] >= 80:
if task_complexity == "simple":
return "gemini-2.5-flash-8b"
return "gemini-2.5-flash"
# 정상 상황: 복잡도에 따른 선택
complexity_models = {
"simple": "gpt-4.1-mini",
"medium": "gemini-2.5-flash",
"complex": "claude-sonnet-4"
}
return complexity_models.get(task_complexity, "gemini-2.5-flash")
class BudgetExceededError(Exception):
"""예산 초과 예외"""
pass
사용 예시
if __name__ == "__main__":
config = BudgetConfig(
daily_limit_cents=5000, # 일일 $50
monthly_limit_cents=100000, # 월 $1000
)
manager = BudgetManager(config)
# 예산 상태 확인
status = manager.get_remaining_budget(BudgetPeriod.DAILY)
print(f"일일 예산 상태: ${status['remaining_cents']/100:.2f} 남음")
print(f"사용률: {status['usage_percent']:.1f}%")
# 모델 선택 테스트
selected = manager.select_model_by_budget("complex")
print(f"선택된 모델: {selected}")2. HolySheep AI API 통합
이제 BudgetManager와 HolySheep AI API를 연동하는 AgentClient를 구현합니다. HolySheep AI는 단일 API 키로 다양한 모델에 접근할 수 있어 모델 전환이 매우 유연합니다.
"""
HolySheep AI API 통합 Agent Client
base_url: https://api.holysheep.ai/v1
"""
import os
import json
from typing import Optional, Dict, Any, List
from openai import OpenAI
from dataclasses import dataclass
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class AgentResponse:
content: str
model_used: str
tokens_used: int
cost_cents: float
latency_ms: int
success: bool
error_message: Optional[str] = None
class HolySheepAgent:
"""HolySheep AI 게이트웨이 기반 AI Agent"""
BASE_URL = "https://api.holysheep.ai/v1"
# 모델별 지연 시간 참고값 (실제 측정치)
MODEL_LATENCY = {
"gpt-4.1": {"avg_ms": 1200, "p95_ms": 2500},
"gpt-4.1-mini": {"avg_ms": 400, "p95_ms": 800},
"claude-sonnet-4": {"avg_ms": 1500, "p95_ms": 3000},
"claude-haiku-4": {"avg_ms": 500, "p95_ms": 1000},
"gemini-2.5-flash": {"avg_ms": 600, "p95_ms": 1200},
"gemini-2.5-flash-8b": {"avg_ms": 200, "p95_ms": 400},
"deepseek-v3.2": {"avg_ms": 350, "p95_ms": 700},
}
# 모델별 비용 ($/MTok)
MODEL_COSTS = {
"gpt-4.1": 8.0,
"gpt-4.1-mini": 0.6,
"claude-sonnet-4": 15.0,
"claude-haiku-4": 1.5,
"gemini-2.5-flash": 2.5,
"gemini-2.5-flash-8b": 0.30,
"deepseek-v3.2": 0.42,
}
def __init__(self, api_key: str, budget_manager: BudgetManager):
self.client = OpenAI(
api_key=api_key,
base_url=self.BASE_URL,
timeout=60.0,
max_retries=2
)
self.budget_manager = budget_manager
self.request_count = 0
self.success_count = 0
def _estimate_task_complexity(self, prompt: str, system_prompt: str = "") -> str:
"""태스크 복잡도 추정"""
combined = f"{system_prompt} {prompt}"
word_count = len(combined.split())
# 복잡도 판단 기준
complexity_indicators = {
"simple": ["告诉我", "what is", "今天的日期", "translate to", "단어 뜻"],
"complex": ["analyze", "리뷰해줘", "optimize", "compare", "evaluate"]
}
# 키워드 기반 판단
for indicator in complexity_indicators.get("complex", []):
if indicator.lower() in combined.lower():
return "complex"
# 토큰 수 기반 판단
if word_count > 500:
return "complex"
elif word_count > 150:
return "medium"
else:
return "simple"
def _calculate_cost(self, model: str, usage: Dict[str, int]) -> float:
"""토큰 사용량 기반 비용 계산"""
total_tokens = usage.get("total_tokens", 0)
cost_per_token = self.MODEL_COSTS.get(model, 8.0) / 1_000_000
return cost_per_token * total_tokens * 100 # cents 단위
def _make_request(
self,
model: str,
messages: List[Dict],
temperature: float = 0.7,
max_tokens: Optional[int] = None
) -> AgentResponse:
"""HolySheep AI API 요청 실행"""
import time
start_time = time.time()
try:
response = self.client.chat.completions.create(
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens
)
latency_ms = int((time.time() - start_time) * 1000)
# 토큰 사용량 추출
usage = {
"prompt_tokens": response.usage.prompt_tokens,
"completion_tokens": response.usage.completion_tokens,
"total_tokens": response.usage.total_tokens
}
# 비용 기록
cost_cents = self._calculate_cost(model, usage)
# BudgetManager에 사용량 기록
self.budget_manager.record_usage(TokenUsage(
prompt_tokens=usage["prompt_tokens"],
completion_tokens=usage["completion_tokens"],
total_tokens=usage["total_tokens"],
cost_cents=cost_cents,
model=model
))
self.success_count += 1
return AgentResponse(
content=response.choices[0].message.content,
model_used=model,
tokens_used=usage["total_tokens"],
cost_cents=cost_cents,
latency_ms=latency_ms,
success=True
)
except Exception as e:
latency_ms = int((time.time() - start_time) * 1000)
logger.error(f"API 요청 실패: {str(e)}")
return AgentResponse(
content="",
model_used=model,
tokens_used=0,
cost_cents=0,
latency_ms=latency_ms,
success=False,
error_message=str(e)
)
def chat(
self,
prompt: str,
system_prompt: str = "당신은 유용한 AI 어시스턴트입니다.",
auto_select_model: bool = True,
preferred_model: Optional[str] = None,
max_latency_ms: Optional[int] = None
) -> AgentResponse:
"""AI Agent 채팅 실행 (자동 모델 선택 지원)"""
self.request_count += 1
# 모델 선택
if auto_select_model:
complexity = self._estimate_task_complexity(prompt, system_prompt)
model = self.budget_manager.select_model_by_budget(complexity)
# 지연 시간 제약이 있는 경우
if max_latency_ms:
available_models = [
m for m, lat in self.MODEL_LATENCY.items()
if lat["p95_ms"] <= max_latency_ms
]
if available_models:
model = min(available_models,
key=lambda m: self.MODEL_COSTS.get(m, 999))
else:
model = preferred_model or "gemini-2.5-flash"
logger.info(f"[요청 #{self.request_count}] 모델: {model}, "
f"예산 상태: {self.budget_manager.get_remaining_budget(BudgetPeriod.DAILY)['usage_percent']:.1f}%")
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.append({"role": "user", "content": prompt})
return self._make_request(model, messages)
def batch_chat(
self,
requests: List[Dict[str, str]],
system_prompt: str = "",
parallel: bool = True
) -> List[AgentResponse]:
"""배치 처리로 여러 요청 동시 실행"""
responses = []
if parallel:
from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=5) as executor:
futures = [
executor.submit(self.chat, req["prompt"], system_prompt)
for req in requests
]
responses = [f.result() for f in futures]
else:
for req in requests:
response = self.chat(req["prompt"], system_prompt)
responses.append(response)
return responses
def get_statistics(self) -> Dict[str, Any]:
"""사용 통계 조회"""
daily = self.budget_manager.get_remaining_budget(BudgetPeriod.DAILY)
return {
"total_requests": self.request_count,
"successful_requests": self.success_count,
"success_rate": (self.success_count / self.request_count * 100)
if self.request_count > 0 else 0,
"daily_budget": {
"used_cents": daily["used_cents"],
"remaining_cents": daily["remaining_cents"],
"usage_percent": daily["usage_percent"]
}
}
===== HolySheep AI API 키 설정 =====
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # HolySheep AI 대시보드에서 발급
===== 사용 예시 =====
if __name__ == "__main__":
# 예산 관리자 초기화
config = BudgetConfig(
daily_limit_cents=10000, # 일일 $100
monthly_limit_cents=200000 # 월 $2000
)
budget_manager = BudgetManager(config)
# Agent 초기화
agent = HolySheepAgent(HOLYSHEEP_API_KEY, budget_manager)
# 다양한 태스크 테스트
test_tasks = [
{"prompt": "한국의 수도는?", "expected_complexity": "simple"},
{"prompt": "다음 JSON을 YAML로 변환해주세요: {\"name\": \"test\"}",
"expected_complexity": "medium"},
{"prompt": "이 코드를 리뷰하고 개선점을 제안해주세요. "
"특히 성능과 보안 측면에서 분석해주세요.",
"expected_complexity": "complex"},
]
print("=" * 60)
print("HolySheep AI 비용 최적화 테스트")
print("=" * 60)
for i, task in enumerate(test_tasks, 1):
response = agent.chat(task["prompt"], max_latency_ms=2000)
print(f"\n[요청 {i}] 복잡도: {task['expected_complexity']}")
print(f"모델: {response.model_used}")
print(f"토큰: {response.tokens_used}")
print(f"비용: ${response.cost_cents/100:.4f}")
print(f"지연: {response.latency_ms}ms")
print(f"성공: {response.success}")
# 최종 통계
stats = agent.get_statistics()
print("\n" + "=" * 60)
print("전체 통계")
print("=" * 60)
print(f"총 요청: {stats['total_requests']}")
print(f"성공률: {stats['success_rate']:.1f}%")
print(f"일일 예산 사용: {stats['daily_budget']['usage_percent']:.1f}%")
print(f"잔여 예산: ${stats['daily_budget']['remaining_cents']/100:.2f}")동적 모델 전환 전략
1. 요청 특성 기반 자동 라우팅
성능과 비용 사이의 최적점을 찾기 위해 요청의 특성을 분석하여 적절한 모델을 선택합니다. 다음 표는 주요 모델들의 특성을 정리한 것입니다:
- DeepSeek V3.2 ($0.42/MTok): 단순 질문, 반복 작업, 대량 배치 처리
- Gemini 2.5 Flash-8B ($0.30/MTok): 빠른 응답 필요, 실시간 인터랙션
- Gemini 2.5 Flash ($2.50/MTok): 일반적인 코딩, 번역, 요약
- GPT-4.1 Mini ($0.60/MTok): 한국어 중심의 일반 작업
- Claude Haiku 4 ($1.50/MTok): 빠른 분석, 컨텍스트 이해
- Claude Sonnet 4 ($15/MTok): 복잡한 분석, 코드 리뷰, 장문 생성
- GPT-4.1 ($8/MTok): 최고 품질의 복잡한 작업
"""
동적 모델 라우팅 시스템
요청 특성 분석 → 최적 모델 자동 선택
"""
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass
import re
@dataclass
class ModelCapability:
name: str
cost_per_mtok: float
avg_latency_ms: int
strengths: List[str]
weaknesses: List[str]
max_tokens: int
context_window: int
class SmartRouter:
"""요청 특성 분석 기반 스마트 라우터"""
# HolySheep AI에서 사용 가능한 모델 정의
MODELS = {
"deepseek-v3.2": ModelCapability(
name="deepseek-v3.2",
cost_per_mtok=0.42,
avg_latency_ms=350,
strengths=["코딩", "수학", "저렴한 비용", "빠른 처리"],
weaknesses=["창작 작문"],
max_tokens=64000,
context_window=128000
),
"gemini-2.5-flash-8b": ModelCapability(
name="gemini-2.5-flash-8b",
cost_per_mtok=0.30,
avg_latency_ms=200,
strengths=["초고속 응답", "实时 인터랙션", "가장 저렴"],
weaknesses=["복잡한 reasoning"],
max_tokens=64000,
context_window=1000000
),
"gemini-2.5-flash": ModelCapability(
name="gemini-2.5-flash",
cost_per_mtok=2.50,
avg_latency_ms=600,
strengths=["균형잡힌 성능", "긴 컨텍스트", "다국어"],
weaknesses=["복잡한 코드"],
max_tokens=64000,
context_window=1000000
),
"claude-haiku-4": ModelCapability(
name="claude-haiku-4",
cost_per_mtok=1.50,
avg_latency_ms=500,
strengths=["빠른 분석", "컨텍스트 이해", "한국어"],
weaknesses=["장문 생성"],
max_tokens=4096,
context_window=200000
),
"claude-sonnet-4": ModelCapability(
name="claude-sonnet-4",
cost_per_mtok=15.0,
avg_latency_ms=1500,
strengths=["복잡한 분석", "코드 리뷰", "창작"],
weaknesses=["느린 응답", "고가"],
max_tokens=8192,
context_window=200000
),
}
# 태스크 분류 키워드
TASK_PATTERNS = {
"code_generation": [
"코드를 작성", "함수를 만들어", "program", "implement", "代码"
],
"code_review": [
"리뷰해줘", "검토", "review", "check the code", "代码审查"
],
"translation": [
"번역", "translate", "변환", "convert", "翻译"
],
"summarization": [
"요약", "summary", "요약해줘", "핵심 정리", "摘要"
],
"analysis": [
"분석해줘", "analyze", "비교해줘", "evaluate", "分析"
],
"simple_qa": [
"뭐야", "무엇", "who is", "what is", "어디야", "who"
],
"creative": [
"글을 써줘", "시에", "소설", "write a", "creative", "创作"
],
}
# 복잡도 점수 키워드
COMPLEXITY_KEYWORDS = {
"high": ["분석해줘", "비교", "평가", "리뷰", "optimize", "analyze",
"批判的", "evaluates", "complex", "advantage", "disadvantage"],
"low": ["뭐", "무엇", "who", "what", "where", "언제", "날짜", "simple"]
}
def __init__(self, budget_threshold_percent: float = 80.0):
self.budget_threshold = budget_threshold_percent
def classify_task(self, prompt: str, system_prompt: str = "") -> Tuple[str, float]:
"""
요청 분류 및 복잡도 점수 계산
Returns: (task_type, complexity_score)
"""
combined = f"{system_prompt} {prompt}".lower()
# 태스크 유형 분류
task_scores = {}
for task_type, patterns in self.TASK_PATTERNS.items():
score = sum(1 for p in patterns if p.lower() in combined)
task_scores[task_type] = score
# 가장 높은 점수의 태스크 선택
if max(task_scores.values()) > 0:
task_type = max(task_scores, key=task_scores.get)
else:
task_type = "general"
# 복잡도 점수 계산 (0.0 ~ 1.0)
complexity_score = 0.0
# 키워드 기반 복잡도
for keyword in self.COMPLEXITY_KEYWORDS["high"]:
if keyword.lower() in combined:
complexity_score += 0.2
for keyword in self.COMPLEXITY_KEYWORDS["low"]:
if keyword.lower() in combined:
complexity_score -= 0.15
# 토큰 수 기반 복잡도
word_count = len(combined.split())
if word_count > 300:
complexity_score += 0.3
elif word_count > 100:
complexity_score += 0.1
# 코드/마크다운 presence
if "``" in prompt or "``" in system_prompt:
complexity_score += 0.2
# 요청 수 제한
complexity_score = max(0.0, min(1.0, complexity_score))
return task_type, complexity_score
def route(
self,
prompt: str,
system_prompt: str = "",
budget_usage_percent: float = 0.0,
required_latency_ms: Optional[int] = None,
require_high_quality: bool = False
) -> str:
"""
요청 특성 및 예산 상태 기반 모델 선택
"""
task_type, complexity = self.classify_task(prompt, system_prompt)
# ===== 예산 긴급 상황: 가장 저렴한 모델 강제 사용 =====
if budget_usage_percent >= 90:
return "gemini-2.5-flash-8b"
# ===== 품질 요구가 높은 경우 =====
if require_high_quality or complexity >= 0.8:
if budget_usage_percent >= 80:
# 중간 품질妥协
return "gemini-2.5-flash"
return "claude-sonnet-4"
# ===== 지연 시간 제약이 있는 경우 =====
if required_latency_ms:
if required_latency_ms <= 500:
candidates = [
m for m, cap in self.MODELS.items()
if cap.avg_latency_ms <= 500
]
if budget_usage_percent >= 80:
return min(candidates, key=lambda m: self.MODELS[m].cost_per_mtok)
return "gemini-2.5-flash-8b"
# ===== 태스크 유형별 최적 모델 선택 =====
task_model_map = {
"code_generation": {
"low": "deepseek-v3.2",
"medium": "gemini-2.5-flash",
"high": "claude-sonnet-4"
},
"code_review": {
"low": "gemini-2.5-flash",
"medium": "claude-haiku-4",
"high": "claude-sonnet-4"
},
"translation": {
"low": "deepseek-v3.2",
"medium": "gemini-2.5-flash",
"high": "gemini-2.5-flash"
},
"summarization": {
"low": "deepseek-v3.2",
"medium": "gemini-2.5-flash-8b",
"high": "claude-haiku-4"
},
"analysis": {
"low": "gemini-2.5-flash",
"medium": "claude-haiku-4",
"high": "claude-sonnet-4"
},
"simple_qa": {
"low": "deepseek-v3.2",
"medium": "gemini-2.5-flash-8b",
"high": "gemini-2.5-flash"
},
"creative": {
"low": "gemini-2.5-flash",
"medium": "gemini-2.5-flash",
"high": "claude-sonnet-4"
},
}
# 복잡도 레벨 결정
if complexity < 0.3:
level = "low"
elif complexity < 0.6:
level = "medium"
else:
level = "high"
# 예산 상태에 따른 추가 조정
if budget_usage_percent >= 80 and level != "high":
level = "low" # 예산 절약 모드
# 모델 선택
return task_model_map.get(task_type, task_model_map["general"]).get(level, "gemini-2.5-flash")
def get_route_explanation(
self,
prompt: str,
system_prompt: str = "",
budget_usage_percent: float = 0.0
) -> Dict:
"""라우팅 결정 이유 설명"""
task_type, complexity = self.classify_task(prompt, system_prompt)
selected_model = self.route(prompt, system_prompt, budget_usage_percent)
return {
"task_type": task_type,
"complexity_score": complexity,
"selected_model": selected_model,
"estimated_cost": self.MODELS[selected_model].cost_per_mtok,
"estimated_latency": self.MODELS[selected_model].avg_latency_ms,
"budget_status": "normal" if budget_usage_percent < 80 else "warning" if budget_usage_percent < 90 else "critical"
}
===== 테스트 =====
if __name__ == "__main__":
router = SmartRouter(budget_threshold_percent=80.0)
test_cases = [
{
"prompt": "Python에서 리스트를 정렬하는 방법을 알려주세요",
"expected": "simple task"
},
{
"prompt": "이 코드를 리뷰하고 보안 취약점이 있는지 분석해주세요. "
"또한 최적화 제안도 해주세요.",
"expected": "complex analysis"
},
{
"prompt": "``python\ndef hello():\n print('Hello')\n``\n"
"이 함수를 개선해주세요",
"expected": "code improvement"
}
]
print("=" * 70)
print("스마트 라우팅 테스트 결과")
print("=" * 70)
for i, case in enumerate(test_cases, 1):
result = router.get_route_explanation(
case["prompt"],
budget_usage_percent=30.0
)
print(f"\n[테스트 {i}] {case['expected']}")
print(f"분류: {result['task_type']}")
print(f"복잡도: {result['complexity_score']:.2f}")
print(f"선택 모델: {result['selected_model']}")
print(f"예상 비용: ${result['estimated_cost']:.2f}/MTok")
print(f"예상 지연: {result['estimated_latency']}ms")실전 모니터링 대시보드 구현
"""
AI Agent 비용 모니터링 대시보드
실시간 예산 추적 및 알림 시스템
"""
import time
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass, field
import json
from collections import defaultdict
@dataclass
class CostAlert:
level: str # "info", "warning", "critical"
message: str
timestamp: datetime = field(default_factory=datetime.now)
action_required: Optional[str] = None
class CostMonitor:
"""비용 모니터링 및 알림 시스템"""
def __init__(self, daily_limit_cents: float, monthly_limit_cents: float):
self.daily_limit = daily_limit_cents
self.monthly_limit = monthly_limit_cents
# 실시간 추적
self.daily_spent = 0.0
self.monthly_spent = 0.0
# 모델별 사용량
self.model_usage = defaultdict(lambda: {
"requests": 0,
"tokens": 0,
"cost_cents": 0.0
})
# 시간대별 사용량 (시간 단위)
self.hourly_usage = defaultdict(float)
# 알림 이력
self.alerts: List[CostAlert] = []
# 추적 시작 시간
self.start_time = datetime.now()
def record(
self,
model: str,
tokens: int,
cost_cents: float,
latency_ms: int
):
"""요청 기록"""
self.daily_spent += cost_cents
self.monthly_spent += cost_cents
# 모델별 집계
self.model_usage[model]["requests"] += 1
self.model_usage[model]["tokens"] += tokens
self.model_usage[model]["cost_cents"] += cost_cents
# 시간대별 집계
current_hour = datetime.now().strftime("%Y-%m-%d %H:00")
self.hourly_usage[current_hour] += cost_cents
# 알림 체크
self._check_thresholds()
def _check_thresholds(self):
"""임계값 체크 및 알림 생성"""
daily_percent = (self.daily_spent / self.daily_limit) * 100
monthly_percent = (self.monthly_spent / self.monthly_limit) * 100
# 일일 임계값 체크
if daily_percent >= 100:
alert = CostAlert(
level="critical",
message=f"일일 예산 초과! ${self.daily_spent/100:.2f} / ${self.daily_limit/100:.2f}",
action_required="서비스 일시 중지 또는 모델 전환 필요"
)
self.alerts.append(alert)
elif daily_percent >=