저는 3개월 전 이커머스 스타트업에서 AI 고객 서비스 시스템을 구축했던 엔지니어입니다. 블랙프라이데이 시즌을 앞두고 급하게 AI 챗봇을 출시했는데, 예상치 못한 폭발적 트래픽과 함께 비용이 폭증하는 경험을 했습니다. 하루 만에 3일치 예산을 소진하면서 급하게 시스템을 정지시킨 경험이 있었죠. 이 글에서는 HolySheep AI를 활용하여 AI API 비용을 실시간으로 모니터링하고,预算阀值 초과 시 자동으로 경고하는 시스템을 구축하는 방법을 설명드리겠습니다.

왜 AI API 비용监控가 중요한가

AI API 사용 시 비용은 여러 요소에 의해 결정됩니다. 주요 요소들은 다음과 같습니다:

HolySheep AI의 현재 가격 정보를 확인해보면:

비용监控系统架构

제가 구축한 비용 모니터링 시스템은 크게 네 부분으로 구성됩니다:

핵심实现: Python成本监控クライアント

"""
HolySheep AI 비용 모니터링 및 알림 시스템
저자 경험: 이커머스 AI 고객 서비스 구축 시 실제 사용한 코드
"""

import requests
import time
import sqlite3
from datetime import datetime, timedelta
from collections import defaultdict
from dataclasses import dataclass
from typing import Optional
import json

@dataclass
class TokenUsage:
    """토큰 사용량 데이터 클래스"""
    model: str
    input_tokens: int
    output_tokens: int
    cost_usd: float
    latency_ms: int
    timestamp: datetime
    request_id: str

class HolySheepCostMonitor:
    """HolySheep AI 비용 모니터링 클라이언트"""
    
    # HolySheep AI 모델별 가격표 (2024년 기준)
    PRICING = {
        "gpt-4.1": {"input": 2.00, "output": 8.00},  # $/MTok
        "gpt-4.1-mini": {"input": 0.50, "output": 2.00},
        "claude-sonnet-4": {"input": 3.00, "output": 15.00},
        "gemini-2.5-flash": {"input": 0.35, "output": 2.50},
        "deepseek-v3.2": {"input": 0.14, "output": 0.42},
    }
    
    def __init__(self, api_key: str, db_path: str = "cost_monitor.db"):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.db_path = db_path
        self._init_database()
        self.daily_budget = 100.00  # 기본 일일 예산 100달러
        self.monthly_budget = 2000.00  # 기본 월간 예산 2000달러
        self.alert_callbacks = []
        
    def _init_database(self):
        """SQLite 데이터베이스 초기화"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS api_calls (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                request_id TEXT UNIQUE,
                model TEXT,
                input_tokens INTEGER,
                output_tokens INTEGER,
                cost_usd REAL,
                latency_ms INTEGER,
                timestamp TEXT,
                status TEXT
            )
        """)
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS daily_summary (
                date TEXT PRIMARY KEY,
                total_cost REAL,
                total_calls INTEGER,
                avg_latency_ms REAL
            )
        """)
        conn.commit()
        conn.close()
        
    def _calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
        """토큰 수 기반으로 비용 계산"""
        pricing = self.PRICING.get(model, {"input": 1.0, "output": 1.0})
        input_cost = (input_tokens / 1_000_000) * pricing["input"]
        output_cost = (output_tokens / 1_000_000) * pricing["output"]
        return round(input_cost + output_cost, 6)
    
    def _save_usage(self, usage: TokenUsage):
        """토큰 사용량 데이터베이스 저장"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("""
            INSERT OR REPLACE INTO api_calls 
            (request_id, model, input_tokens, output_tokens, cost_usd, latency_ms, timestamp, status)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        """, (
            usage.request_id,
            usage.model,
            usage.input_tokens,
            usage.output_tokens,
            usage.cost_usd,
            usage.latency_ms,
            usage.timestamp.isoformat(),
            "success"
        ))
        conn.commit()
        conn.close()
    
    def call_model(self, model: str, messages: list, **kwargs) -> dict:
        """HolySheep AI API 호출 및 비용 추적"""
        start_time = time.time()
        request_id = f"req_{int(start_time * 1000)}_{hash(str(messages)) % 1000000}"
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "messages": messages,
            **kwargs
        }
        
        try:
            response = requests.post(
                f"{self.base_url}/chat/completions",
                headers=headers,
                json=payload,
                timeout=30
            )
            response.raise_for_status()
            
            end_time = time.time()
            latency_ms = int((end_time - start_time) * 1000)
            
            result = response.json()
            
            # 토큰 사용량 추출 (HolySheep AI 응답 형식)
            usage = result.get("usage", {})
            input_tokens = usage.get("prompt_tokens", 0)
            output_tokens = usage.get("completion_tokens", 0)
            cost = self._calculate_cost(model, input_tokens, output_tokens)
            
            usage_record = TokenUsage(
                model=model,
                input_tokens=input_tokens,
                output_tokens=output_tokens,
                cost_usd=cost,
                latency_ms=latency_ms,
                timestamp=datetime.now(),
                request_id=request_id
            )
            
            self._save_usage(usage_record)
            self._check_budget_alerts(cost)
            
            return {
                "success": True,
                "data": result,
                "usage": {
                    "input_tokens": input_tokens,
                    "output_tokens": output_tokens,
                    "cost_usd": cost,
                    "latency_ms": latency_ms
                }
            }
            
        except requests.exceptions.RequestException as e:
            return {
                "success": False,
                "error": str(e),
                "request_id": request_id
            }
    
    def _check_budget_alerts(self, current_cost: float):
        """예산 초과 여부 확인 및 알림"""
        today = datetime.now().date()
        total_today = self.get_daily_cost(today)
        
        # 일일 예산 초과 체크 (80%, 100% 임계값)
        daily_percentage = (total_today / self.daily_budget) * 100
        
        if daily_percentage >= 100:
            self._trigger_alert("CRITICAL", f"일일 예산 초과! {total_today:.2f}달러 사용")
        elif daily_percentage >= 80:
            self._trigger_alert("WARNING", f"일일 예산 80% 초과! {total_today:.2f}달러 사용")
    
    def _trigger_alert(self, level: str, message: str):
        """알림 발송"""
        alert = {
            "level": level,
            "message": message,
            "timestamp": datetime.now().isoformat()
        }
        for callback in self.alert_callbacks:
            callback(alert)
        print(f"[{level}] {message}")
    
    def register_alert_callback(self, callback):
        """알림 콜백 등록"""
        self.alert_callbacks.append(callback)
    
    def get_daily_cost(self, date: datetime.date) -> float:
        """특정 날짜의 총 비용 조회"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("""
            SELECT COALESCE(SUM(cost_usd), 0) 
            FROM api_calls 
            WHERE DATE(timestamp) = ?
        """, (date.isoformat(),))
        result = cursor.fetchone()[0]
        conn.close()
        return result
    
    def get_cost_breakdown(self, days: int = 7) -> dict:
        """일별 비용 상세 분석"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute("""
            SELECT 
                DATE(timestamp) as date,
                model,
                COUNT(*) as calls,
                SUM(input_tokens) as input_total,
                SUM(output_tokens) as output_total,
                SUM(cost_usd) as cost
            FROM api_calls
            WHERE timestamp >= ?
            GROUP BY DATE(timestamp), model
            ORDER BY date DESC
        """, ((datetime.now() - timedelta(days=days)).isoformat(),))
        
        rows = cursor.fetchall()
        conn.close()
        
        breakdown = defaultdict(lambda: {"calls": 0, "cost": 0})
        for row in rows:
            date, model, calls, input_tok, output_tok, cost = row
            breakdown[model]["calls"] += calls
            breakdown[model]["cost"] += cost
            
        return dict(breakdown)
    
    def get_cost_forecast(self) -> dict:
        """월간 비용 예측"""
        today = datetime.now()
        days_in_month = 30  # 대략적
        days_passed = today.day
        
        daily_avg = self.get_daily_cost(today.date()) if days_passed > 0 else 0
        
        return {
            "today_cost": daily_avg,
            "projected_monthly": daily_avg * days_in_month,
            "days_passed": days_passed,
            "remaining_budget": self.monthly_budget - (daily_avg * days_passed)
        }



사용 예시

if __name__ == "__main__":
    monitor = HolySheepCostMonitor(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        db_path="production_costs.db"
    )
    
    # 슬랙 알림 콜백 예시
    def slack_notification(alert):
        # 실제 구현 시 슬랙 웹훅 URL로 요청 전송
        print(f"슬랙 알림: [{alert['level']}] {alert['message']}")
    
    monitor.register_alert_callback(slack_notification)
    
    # AI API 호출 예시
    response = monitor.call_model(
        model="deepseek-v3.2",  # 비용 효율적인 모델 선택
        messages=[
            {"role": "system", "content": "당신은 친절한 고객 서비스 담당자입니다."},
            {"role": "user", "content": "주문 취소 방법은 무엇인가요?"}
        ],
        temperature=0.7,
        max_tokens=500
    )
    
    if response["success"]:
        print(f"토큰 사용량: 입력 {response['usage']['input_tokens']}, 출력 {response['usage']['output_tokens']}")
        print(f"비용: ${response['usage']['cost_usd']:.4f}")
        print(f"응답 지연시간: {response['usage']['latency_ms']}ms")

실시간 대시보드: Streamlit可视化

"""
AI API 비용 모니터링 대시보드
Streamlit 기반 실시간 비용 추적 및 알림 대시보드
"""

import streamlit as st
import sqlite3
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go
from datetime import datetime, timedelta
from holySheep_monitor import HolySheepCostMonitor

st.set_page_config(page_title="AI API 비용 모니터링", page_icon="💰", layout="wide")

def load_data(db_path: str, days: int = 7) -> pd.DataFrame:
    """데이터베이스에서 데이터 로드"""
    conn = sqlite3.connect(db_path)
    query = f"""
        SELECT 
            timestamp,
            model,
            input_tokens,
            output_tokens,
            cost_usd,
            latency_ms
        FROM api_calls
        WHERE timestamp >= datetime('now', '-{days} days')
        ORDER BY timestamp DESC
    """
    df = pd.read_sql_query(query, conn, parse_dates=["timestamp"])
    conn.close()
    return df

def main():
    st.title("🩺 HolySheep AI API 비용 모니터링 대시보드")
    st.markdown("실시간 API 사용량 추적 및 비용 최적화를 위한 모니터링 시스템")
    
    # 사이드바 설정
    st.sidebar.header("설정")
    db_path = st.sidebar.text_input("데이터베이스 경로", "production_costs.db")
    days_filter = st.sidebar.slider("조회 기간 (일)", 1, 30, 7)
    api_key = st.sidebar.text_input("HolySheep API Key", type="password")
    
    if api_key and len(api_key) > 10:
        monitor = HolySheepCostMonitor(api_key=api_key, db_path=db_path)
        
        # 실시간 비용 현황
        col1, col2, col3, col4 = st.columns(4)
        
        today = datetime.now().date()
        df = load_data(db_path, days_filter)
        
        if not df.empty:
            today_df = df[df["timestamp"].dt.date == today]
            
            with col1:
                today_cost = today_df["cost_usd"].sum()
                st.metric("오늘의 비용", f"${today_cost:.2f}", 
                         delta=f"${monitor.daily_budget - today_cost:.2f} 잔여")
            
            with col2:
                week_cost = df["cost_usd"].sum()
                st.metric("주간 누적 비용", f"${week_cost:.2f}")
            
            with col3:
                total_calls = len(df)
                st.metric("총 API 호출", f"{total_calls:,}회")
            
            with col4:
                avg_latency = df["latency_ms"].mean()
                st.metric("평균 응답시간", f"{avg_latency:.0f}ms")
            
            # 비용 예측
            st.subheader("📊 월간 비용 예측")
            forecast = monitor.get_cost_forecast()
            
            pred_col1, pred_col2 = st.columns(2)
            with pred_col1:
                st.info(f"**예상 월간 비용**: ${forecast['projected_monthly']:.2f}")
            with pred_col2:
                if forecast['projected_monthly'] > monitor.monthly_budget:
                    st.warning(f"⚠️ 월간 예산 초과 예상! ($ {forecast['remaining_budget']:.2f} 남음)")
                else:
                    st.success(f"✅ 월간 예산 범위 내 ($ {forecast['remaining_budget']:.2f} 여유)")
            
            # 모델별 비용 분포
            st.subheader("🏷️ 모델별 비용 분석")
            model_costs = df.groupby("model").agg({
                "cost_usd": "sum",
                "latency_ms": "mean",
                "timestamp": "count"
            }).rename(columns={"timestamp": "calls"}).round(4)
            
            col_a, col_b = st.columns(2)
            
            with col_a:
                fig_pie = px.pie(
                    model_costs.reset_index(), 
                    values="cost_usd", 
                    names="model",
                    title="모델별 비용 비중"
                )
                st.plotly_chart(fig_pie, use_container_width=True)
            
            with col_b:
                fig_bar = px.bar(
                    model_costs.reset_index(),
                    x="model",
                    y="cost_usd",
                    color="model",
                    title="모델별 총 비용"
                )
                st.plotly_chart(fig_bar, use_container_width=True)
            
            # 일별 추이 그래프
            st.subheader("📈 일별 비용 추이")
            daily_df = df.set_index("timestamp").resample("D")["cost_usd"].sum().reset_index()
            daily_df.columns = ["날짜", "비용"]
            
            fig_line = px.line(
                daily_df,
                x="날짜",
                y="비용",
                markers=True,
                title="일별 비용 추이"
            )
            fig_line.add_hline(
                y=monitor.daily_budget, 
                line_dash="dash", 
                annotation_text="일일 예산 한도",
                line_color="red"
            )
            st.plotly_chart(fig_line, use_container_width=True)
            
            # 토큰 사용량 테이블
            st.subheader("📋 최근 API 호출 내역")
            st.dataframe(
                df[["timestamp", "model", "input_tokens", "output_tokens", "cost_usd", "latency_ms"]]
                .head(20)
                .style.format({
                    "cost_usd": "${:.6f}",
                    "latency_ms": "{}ms"
                })
            )
            
            # 비용 최적화 추천
            st.subheader("💡 비용 최적화 추천")
            
            expensive_models = model_costs[model_costs["cost_usd"] > 10].index.tolist()
            if expensive_models:
                st.warning(f"**고비용 모델 사용 경고**: {', '.join(expensive_models)}")
                st.markdown("""
                **추천**: 
                - 간단한 질의에는 deepseek-v3.2 ($0.42/MTok 출력) 사용 고려
                - 대량 처리에는 gemini-2.5-flash ($2.50/MTok 출력) 검토
                - gpt-4.1 ($8/MTok 출력)은 복잡한 작업만 제한적으로 사용
                """)
            
            # 예산 초과 시 자동 알림 설정
            st.subheader("🔔 알림 설정")
            new_daily_budget = st.number_input("일일 예산 ($)", value=100.0, step=10.0)
            new_monthly_budget = st.number_input("월간 예산 ($)", value=2000.0, step=100.0)
            
            if st.button("예산 설정 업데이트"):
                monitor.daily_budget = new_daily_budget
                monitor.monthly_budget = new_monthly_budget
                st.success("예산 설정이 업데이트되었습니다.")
                
        else:
            st.info("아직 API 호출 데이터가 없습니다. API를 호출하면 데이터가 표시됩니다.")
    
    else:
        st.info("사이드바에서 HolySheep AI API Key를 입력해주세요.")
        st.markdown("""
        ### HolySheep AI 시작하기
        
        아직 HolySheep AI 계정이 없으신가요?
        
        1. [HolySheep AI 가입](https://www.holysheep.ai/register) - 무료 크레딧 제공
        2. API Key 발급
        3. 위 입력창에 API Key 입력
        
        **주요 모델 비용 비교**:
        - DeepSeek V3.2: $0.42/MTok (출력) - **최고 가성비**
        - Gemini 2.5 Flash: $2.50/MTok (출력)
        - Claude Sonnet 4: $15/MTok (출력)
        - GPT-4.1: $8/MTok (출력)
        """)

if __name__ == "__main__":
    main()

告警通知系统: 슬랙 연동

"""
AI API 비용 알림 시스템
슬랙, 이메일, 웹훅을 통한 실시간 알림
"""

import requests
import json
from enum import Enum
from typing import List, Callable
from dataclasses import dataclass
from datetime import datetime

class AlertLevel(Enum):
    """알림 수준枚举"""
    INFO = "info"
    WARNING = "warning"
    CRITICAL = "critical"
    
@dataclass
class Alert:
    """알림 데이터 클래스"""
    level: AlertLevel
    title: str
    message: str
    current_cost: float
    budget_limit: float
    percentage: float
    timestamp: datetime
    metadata: dict = None

class AlertManager:
    """알림 관리자"""
    
    def __init__(self):
        self.handlers: List[Callable] = []
        self.alert_history: List[Alert] = []
        self.cooldown_seconds = 300  # 5분간 중복 알림 방지
        
    def add_handler(self, handler: Callable):
        """알림 핸들러 추가"""
        self.handlers.append(handler)
        
    def send_alert(self, alert: Alert):
        """알림 발송"""
        # 쿨다운 체크
        recent_alerts = [
            a for a in self.alert_history 
            if a.level == alert.level 
            and (datetime.now() - a.timestamp).seconds < self.cooldown_seconds
        ]
        
        if recent_alerts:
            return  # 쿨다운 중
        
        self.alert_history.append(alert)
        
        for handler in self.handlers:
            try:
                handler(alert)
            except Exception as e:
                print(f"알림 핸들러 오류: {e}")

class SlackAlertHandler:
    """슬랙 웹훅 알림 핸들러"""
    
    def __init__(self, webhook_url: str, channel: str = "#ai-alerts"):
        self.webhook_url = webhook_url
        self.channel = channel
        
    def __call__(self, alert: Alert):
        """슬랙 메시지 발송"""
        
        # 알림 수준별 이모지 및 색상
        level_config = {
            AlertLevel.INFO: {"emoji": "ℹ️", "color": "#36a64f"},
            AlertLevel.WARNING: {"emoji": "⚠️", "color": "#ff9800"},
            AlertLevel.CRITICAL: {"emoji": "🚨", "color": "#f44336"}
        }
        
        config = level_config[alert.level]
        
        payload = {
            "channel": self.channel,
            "username": "AI Cost Monitor",
            "icon_emoji": ":money_with_wings:",
            "attachments": [{
                "color": config["color"],
                "blocks": [
                    {
                        "type": "header",
                        "text": {
                            "type": "plain_text",
                            "text": f"{config['emoji']} {alert.title}",
                            "emoji": True
                        }
                    },
                    {
                        "type": "section",
                        "fields": [
                            {
                                "type": "mrkdwn",
                                "text": f"*알림 수준:*\n{alert.level.value.upper()}"
                            },
                            {
                                "type": "mrkdwn",
                                "text": f"*현재 비용:*\n${alert.current_cost:.2f}"
                            },
                            {
                                "type": "mrkdwn",
                                "text": f"*예산 한도:*\n${alert.budget_limit:.2f}"
                            },
                            {
                                "type": "mrkdwn",
                                "text": f"*사용률:*\n{alert.percentage:.1f}%"
                            }
                        ]
                    },
                    {
                        "type": "section",
                        "text": {
                            "type": "mrkdwn",
                            "text": f"*상세 내용:*\n{alert.message}"
                        }
                    },
                    {
                        "type": "context",
                        "elements": [
                            {
                                "type": "mrkdwn",
                                "text": f"발송 시간: {alert.timestamp.strftime('%Y-%m-%d %H:%M:%S')} UTC"
                            }
                        ]
                    }
                ]
            }]
        }
        
        response = requests.post(self.webhook_url, json=payload, timeout=10)
        response.raise_for_status()

class EmailAlertHandler:
    """이메일 알림 핸들러"""
    
    def __init__(self, smtp_server: str, smtp_port: int, 
                 username: str, password: str, 
                 from_addr: str, to_addrs: List[str]):
        self.smtp_server = smtp_server
        self.smtp_port = smtp_port
        self.username = username
        self.password = password
        self.from_addr = from_addr
        self.to_addrs = to_addrs
        
    def __call__(self, alert: Alert):
        """이메일 발송"""
        import smtplib
        from email.mime.text import MIMEText
        from email.mime.multipart import MIMEMultipart
        
        msg = MIMEMultipart("alternative")
        msg["Subject"] = f"[{alert.level.value.upper()}] AI API 비용 알림: {alert.title}"
        msg["From"] = self.from_addr
        msg["To"] = ", ".join(self.to_addrs)
        
        html_content = f"""
        
        
            
{alert.title}

            

                

                    항목
                    값
                

                

                    알림 수준
                    {alert.level.value.upper()}
                

                

                    현재 비용
                    ${alert.current_cost:.2f}
                

                

                    예산 한도
                    ${alert.budget_limit:.2f}
                

                

                    사용률
                    {alert.percentage:.1f}%
                

            

            
상세 내용:

            
{alert.message}

            

            

                이 메일은 AI API 비용 모니터링 시스템에 의해 자동 발송되었습니다.

                발송 시간: {alert.timestamp.strftime('%Y-%m-%d %H:%M:%S')} UTC
            

        
        
        """
        
        msg.attach(MIMEText(html_content, "html"))
        
        with smtplib.SMTP(self.smtp_server, self.smtp_port) as server:
            server.starttls()
            server.login(self.username, self.password)
            server.send_message(msg)

class WebhookAlertHandler:
    """일반 웹훅 알림 핸들러"""
    
    def __init__(self, url: str, headers: dict = None):
        self.url = url
        self.headers = headers or {"Content-Type": "application/json"}
        
    def __call__(self, alert: Alert):
        """웹훅 요청 발송"""
        payload = {
            "event": "ai_cost_alert",
            "level": alert.level.value,
            "title": alert.title,
            "message": alert.message,
            "data": {
                "current_cost": alert.current_cost,
                "budget_limit": alert.budget_limit,
                "percentage": alert.percentage,
                "timestamp": alert.timestamp.isoformat()
            },
            "metadata": alert.metadata or {}
        }
        
        response = requests.post(
            self.url, 
            headers=self.headers,
            data=json.dumps(payload),
            timeout=10
        )
        response.raise_for_status()



사용 예시

if __name__ == "__main__":
    # 알림 관리자 초기화
    alert_manager = AlertManager()
    
    # 슬랙 핸들러 추가
    slack_handler = SlackAlertHandler(
        webhook_url="YOUR_SLACK_WEBHOOK_URL",
        channel="#ai-api-alerts"
    )
    alert_manager.add_handler(slack_handler)
    
    # 이메일 핸들러 추가 (선택적)
    # email_handler = EmailAlertHandler(
    #     smtp_server="smtp.gmail.com",
    #     smtp_port=587,
    #     username="[email protected]",
    #     password="your-app-password",
    #     from_addr="[email protected]",
    #     to_addrs=["[email protected]", "[email protected]"]
    # )
    # alert_manager.add_handler(email_handler)
    
    # 테스트 알림 발송
    test_alert = Alert(
        level=AlertLevel.WARNING,
        title="일일 예산 80% 초과",
        message="AI API 사용량이 일일 예산의 80%를 초과했습니다. 사용량을 확인해주세요.",
        current_cost=80.50,
        budget_limit=100.00,
        percentage=80.5,
        timestamp=datetime.now(),
        metadata={"user_id": "admin", "region": "ap-northeast-1"}
    )
    
    alert_manager.send_alert(test_alert)
    print("테스트 알림 발송 완료")

비용 최적화实战策略

저의 경험상 AI API 비용을 60% 이상 절감할 수 있었던 핵심 전략들을 공유드립니다:

1. 모델 선택 최적화

작업의 특성에 맞는 모델을 선택하는 것이 가장 효과적입니다:

"""
지능형 모델 선택기
작업 복잡도에 따라 최적의 모델 자동 선택
"""

class SmartModelSelector:
    """작업 복잡도에 따른 모델 자동 선택"""
    
    COMPLEXITY_THRESHOLDS = {
        "simple": {
            "max_tokens": 200,
            "requires_reasoning": False,
            "preferred_model": "deepseek-v3.2",
            "fallback_model": "gemini-2.5-flash"
        },
        "medium": {
            "max_tokens": 1000,
            "requires_reasoning": True,
            "preferred_model": "gemini-2.5-flash",
            "fallback_model": "claude-sonnet-4"
        },
        "complex": {
            "max_tokens": 4000,
            "requires_reasoning": True,
            "preferred_model": "claude-sonnet-4",
            "fallback_model": "gpt-4.1"
        }
    }
    
    def __init__(self, monitor: HolySheepCostMonitor):
        self.monitor = monitor
        
    def select_model(self, task_type: str, estimated_tokens: int = None) -> str:
        """작업 유형에 따른 최적 모델 선택"""
        
        if estimated_tokens and estimated_tokens > 3000:
            return self.COMPLEXITY_THRESHOLDS["complex"]["preferred_model"]
        
        complexity_map = {
            "qa": "simple",
            "summary": "simple",
            "classification": "simple",
            "translation": "simple",
            "chat": "medium",
            "writing": "medium",
            "analysis": "medium",
            "code_generation": "complex",
            "reasoning": "complex",
            "creative": "complex"
        }
        
        complexity = complexity_map.get(task_type, "medium")
        config = self.COMPLEXITY_THRESHOLDS[complexity]
        
        # 현재 예산 상태에 따라 모델 재선택
        today_cost = self.monitor.get_daily_cost(datetime.now().date())
        budget_usage = (today_cost / self.monitor.daily_budget) * 100
        
        if budget_usage > 90:
            # 예산 초과 위험 시 가장 저렴한 모델 강제 사용
            return "deepseek-v3.2"
        elif budget_usage > 70:
            # 예산 사용率高 시 중급 모델만 사용
            return config.get("fallback_model", "gemini-2.5-flash")
        
        return config["preferred_model"]
    
    def estimate_cost(self, model: str, prompt_tokens: int, completion_tokens: int) -> float:
        """비용 예측"""
        return self.monitor._calculate_cost(model, prompt_tokens, completion_tokens)



사용 예시

if __name__ == "__main__":
    monitor = HolySheepCostMonitor(api_key="YOUR_HOLYSHEEP_API_KEY")
    selector = SmartModelSelector(monitor)
    
    # 작업별 최적 모델 선택
    print(f"간단한 QA: {selector.select_model('qa')}")
    print(f"코드 생성: {selector.select_model('code_generation')}")
    print(f"긴 컨텍스트 요약: {selector.select_model('summary', estimated_tokens=5000)}")

2. 토큰使用량 최적화

제가 효과적으로 사용한 토큰 절약 기법들입니다: