LLM API 비용 모니터링

LLM 가격 구조

주요 모델 가격 (2024년 기준)

비용 계산기

def calculate_cost(
    model: str,
    input_tokens: int,
    output_tokens: int
) -> float:
    """토큰 수로 비용 계산"""

    # 가격표 (입력, 출력 per 1M tokens)
    prices = {
        "claude-": (15.00, 75.00),
        "claude-": (3.00, 15.00),
        "claude-": (0.80, 4.00),
        "gpt-4-turbo": (10.00, 30.00),
        "gpt-4o": (2.50, 10.00),
        "gpt-4o-mini": (0.15, 0.60),
        "gemini-1.5-pro": (3.50, 10.50),
        "gemini-1.5-flash": (0.075, 0.30),
    }

    if model not in prices:
        raise ValueError(f"Unknown model: {model}")

    input_price, output_price = prices[model]

    input_cost = (input_tokens / 1_000_000) * input_price
    output_cost = (output_tokens / 1_000_000) * output_price

    return input_cost + output_cost

# 사용 예시
cost = calculate_cost(
    model="claude-",
    input_tokens=5000,
    output_tokens=1000
)
print(f"비용: ${cost:.4f}")
# 출력: 비용: 변동

실제 사용 예시

사용량 추적

기본 로깅

Python 예제

import anthropic
import os
import json
import logging
from datetime import datetime

# 로깅 설정
logging.basicConfig(
    filename='llm_usage.log',
    level=logging.INFO,
    format='%(asctime)s - %(message)s'
)

class CostTracker:
    def __init__(self):
        self.client = anthropic.Anthropic(
            api_key=os.getenv("ANTHROPIC_API_KEY")
        )

    def log_usage(self, model, usage, cost):
        """API 사용량 로깅"""
        log_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "model": model,
            "input_tokens": usage.input_tokens,
            "output_tokens": usage.output_tokens,
            "cost_usd": cost,
        }
        logging.info(json.dumps(log_entry))

    def create_message(self, model, messages, **kwargs):
        """비용 추적과 함께 메시지 생성"""
        message = self.client.messages.create(
            model=model,
            messages=messages,
            **kwargs
        )

        # 비용 계산
        cost = calculate_cost(
            model=model,
            input_tokens=message.usage.input_tokens,
            output_tokens=message.usage.output_tokens
        )

        # 로깅
        self.log_usage(model, message.usage, cost)

        return message

# 사용
tracker = CostTracker()
message = tracker.create_message(
    model="claude-",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello!"}]
)

데이터베이스 저장

SQLite 예제

import sqlite3
from datetime import datetime

class UsageDatabase:
    def __init__(self, db_path="llm_usage.db"):
        self.conn = sqlite3.connect(db_path)
        self.create_table()

    def create_table(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS api_usage (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp TEXT NOT NULL,
                user_id TEXT,
                model TEXT NOT NULL,
                input_tokens INTEGER NOT NULL,
                output_tokens INTEGER NOT NULL,
                cost_usd REAL NOT NULL,
                request_id TEXT,
                metadata TEXT
            )
        """)
        self.conn.commit()

    def log_request(
        self,
        model: str,
        input_tokens: int,
        output_tokens: int,
        cost: float,
        user_id: str = None,
        request_id: str = None,
        metadata: dict = None
    ):
        self.conn.execute(
            """
            INSERT INTO api_usage
            (timestamp, user_id, model, input_tokens, output_tokens, cost_usd, request_id, metadata)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                datetime.utcnow().isoformat(),
                user_id,
                model,
                input_tokens,
                output_tokens,
                cost,
                request_id,
                json.dumps(metadata) if metadata else None
            )
        )
        self.conn.commit()

    def get_daily_cost(self, date: str = None) -> float:
        """특정 날짜의 총 비용"""
        if not date:
            date = datetime.utcnow().strftime("%Y-%m-%d")

        cursor = self.conn.execute(
            "SELECT SUM(cost_usd) FROM api_usage WHERE DATE(timestamp) = ?",
            (date,)
        )
        result = cursor.fetchone()[0]
        return result or 0.0

    def get_user_usage(self, user_id: str, days: int = 30):
        """사용자별 사용량 통계"""
        cursor = self.conn.execute(
            """
            SELECT
                model,
                COUNT(*) as requests,
                SUM(input_tokens) as total_input,
                SUM(output_tokens) as total_output,
                SUM(cost_usd) as total_cost
            FROM api_usage
            WHERE user_id = ?
              AND timestamp >= datetime('now', '-' || ? || ' days')
            GROUP BY model
            """,
            (user_id, days)
        )
        return cursor.fetchall()

# 사용
db = UsageDatabase()
db.log_request(
    model="claude-",
    input_tokens=5000,
    output_tokens=1000,
    cost=0.03,
    user_id="user123"
)

print(f"Today's cost: ${db.get_daily_cost():.2f}")

API 래퍼 클래스

class MonitoredAnthropicClient:
    """비용 추적 기능이 내장된 Anthropic 클라이언트"""

    def __init__(self, api_key: str, db: UsageDatabase):
        self.client = anthropic.Anthropic(api_key=api_key)
        self.db = db

    def messages_create(
        self,
        user_id: str = None,
        **kwargs
    ) -> anthropic.types.Message:
        """메시지 생성 + 자동 로깅"""

        # API 호출
        message = self.client.messages.create(**kwargs)

        # 비용 계산
        cost = calculate_cost(
            model=kwargs['model'],
            input_tokens=message.usage.input_tokens,
            output_tokens=message.usage.output_tokens
        )

        # DB 로깅
        self.db.log_request(
            model=kwargs['model'],
            input_tokens=message.usage.input_tokens,
            output_tokens=message.usage.output_tokens,
            cost=cost,
            user_id=user_id,
            request_id=message.id
        )

        return message

# 사용
db = UsageDatabase()
client = MonitoredAnthropicClient(
    api_key=os.getenv("ANTHROPIC_API_KEY"),
    db=db
)

message = client.messages_create(
    model="claude-",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello!"}],
    user_id="user123"
)

알림 설정

비용 임계값 알림

이메일 알림

import smtplib
from email.mime.text import MIMEText

class CostAlerter:
    def __init__(
        self,
        daily_limit: float = 100.0,
        monthly_limit: float = 1000.0,
        email_to: str = "admin@example.com"
    ):
        self.daily_limit = daily_limit
        self.monthly_limit = monthly_limit
        self.email_to = email_to
        self.db = UsageDatabase()

    def check_limits(self):
        """비용 제한 확인 및 알림"""
        daily_cost = self.db.get_daily_cost()
        monthly_cost = self.get_monthly_cost()

        # 일일 한도 초과
        if daily_cost > self.daily_limit:
            self.send_alert(
                subject="⚠️ Daily API cost limit exceeded",
                message=f"Daily cost: ${daily_cost:.2f} (limit: ${self.daily_limit})"
            )

        # 월간 한도 초과
        if monthly_cost > self.monthly_limit:
            self.send_alert(
                subject="🚨 Monthly API cost limit exceeded",
                message=f"Monthly cost: ${monthly_cost:.2f} (limit: ${self.monthly_limit})"
            )

        # 경고 (80% 도달)
        if daily_cost > self.daily_limit * 0.8:
            self.send_alert(
                subject="⚠️ Approaching daily limit",
                message=f"Daily cost: ${daily_cost:.2f} (80% of limit)"
            )

    def send_alert(self, subject: str, message: str):
        """이메일 알림 전송"""
        msg = MIMEText(message)
        msg['Subject'] = subject
        msg['From'] = 'alerts@example.com'
        msg['To'] = self.email_to

        with smtplib.SMTP('smtp.gmail.com', 587) as server:
            server.starttls()
            server.login(
                os.getenv('SMTP_USER'),
                os.getenv('SMTP_PASS')
            )
            server.send_message(msg)

    def get_monthly_cost(self) -> float:
        cursor = self.db.conn.execute(
            "SELECT SUM(cost_usd) FROM api_usage WHERE strftime('%Y-%m', timestamp) = strftime('%Y-%m', 'now')"
        )
        result = cursor.fetchone()[0]
        return result or 0.0

# 주기적으로 실행 (cron 또는 스케줄러)
alerter = CostAlerter(daily_limit=50.0, monthly_limit=500.0)
alerter.check_limits()

Slack 알림

import requests

def send_slack_alert(webhook_url: str, message: str):
    """Slack으로 알림 전송"""
    payload = {
        "text": message,
        "blocks": [
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": message
                }
            }
        ]
    }

    response = requests.post(webhook_url, json=payload)
    response.raise_for_status()

# 사용
if daily_cost > 100:
    send_slack_alert(
        webhook_url=os.getenv("SLACK_WEBHOOK_URL"),
        message=f"🚨 *API Cost Alert*\n\nDaily cost: ${daily_cost:.2f}\nLimit: 변동"
    )

실시간 알림

class RealtimeMonitor:
    def __init__(self, threshold_per_request: float = 1.0):
        self.threshold = threshold_per_request

    def monitor_request(self, cost: float, model: str, user_id: str):
        """단일 요청이 임계값 초과 시 즉시 알림"""
        if cost > self.threshold:
            self.alert_expensive_request(cost, model, user_id)

    def alert_expensive_request(self, cost: float, model: str, user_id: str):
        message = f"""
⚠️ Expensive API Request Detected

User: {user_id}
Model: {model}
Cost: ${cost:.2f}
Threshold: ${self.threshold:.2f}

This single request exceeded the per-request cost threshold.
Please review the usage pattern.
        """

        send_slack_alert(
            webhook_url=os.getenv("SLACK_WEBHOOK_URL"),
            message=message
        )

비용 최적화 전략

프롬프트 캐싱

# 캐싱 없이 (매번 전체 비용 지불)
message = client.messages.create(
    model="claude-",
    max_tokens=1024,
    system="You are a code reviewer. [긴 가이드라인...]",  # 10,000 토큰
    messages=[{"role": "user", "content": "Review this code..."}]
)
# 비용: 변동 (10,000 입력 토큰)

# 캐싱 사용 (첫 요청 후 90% 절감)
message = client.messages.create(
    model="claude-",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "You are a code reviewer. [긴 가이드라인...]",
            "cache_control": {"type": "ephemeral"}  # 캐싱 활성화
        }
    ],
    messages=[{"role": "user", "content": "Review this code..."}]
)
# 첫 요청: 변동 (캐시 쓰기 비용 25% 추가)
# 이후 요청: 변동 (캐시 읽기 비용 90% 절감)

모델 선택 전략

계층적 모델 사용

def smart_generate(prompt: str, complexity: str = "auto"):
    """복잡도에 따라 적절한 모델 선택"""

    if complexity == "auto":
        # 간단한 휴리스틱으로 복잡도 판단
        if len(prompt) < 500:
            complexity = "simple"
        elif "analyze" in prompt.lower() or "complex" in prompt.lower():
            complexity = "complex"
        else:
            complexity = "medium"

    # 모델 선택
    model_map = {
        "simple": "claude-",      # 변동/변동 per 1M
        "medium": "claude-",     # 변동/변동 per 1M
        "complex": "claude-",      # 변동/변동 per 1M
    }

    model = model_map[complexity]

    message = client.messages.create(
        model=model,
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    )

    return message

# 사용
# 간단한 작업 → Haiku (저렴)
response = smart_generate("What is 2+2?")

# 복잡한 작업 → Opus (비싸지만 정확)
response = smart_generate("Analyze this complex algorithm...", complexity="complex")

컨텍스트 최적화

def summarize_context(long_text: str, max_tokens: int = 2000) -> str:
    """긴 컨텍스트를 요약하여 토큰 절약"""

    # 이미 충분히 짧으면 그대로 반환
    estimated_tokens = len(long_text) / 4  # 대략적인 추정
    if estimated_tokens < max_tokens:
        return long_text

    # Haiku로 요약 (저렴)
    message = client.messages.create(
        model="claude-",
        max_tokens=max_tokens,
        messages=[{
            "role": "user",
            "content": f"Summarize the following in {max_tokens} tokens or less:\n\n{long_text}"
        }]
    )

    return message.content[0].text

# 사용
large_codebase = read_all_files()  # 100,000 토큰
summary = summarize_context(large_codebase, max_tokens=5000)  # 5,000 토큰으로 축소

# 요약된 컨텍스트로 메인 작업 수행
result = client.messages.create(
    model="claude-",
    max_tokens=2048,
    messages=[{
        "role": "user",
        "content": f"Based on this codebase summary:\n{summary}\n\nGenerate a README."
    }]
)
# 총 비용: 요약(변동) + 메인(변동) = 변동
# vs 전체 전달: 변동 (3배 절감!)

배치 처리

def batch_reviews(files: list[str]) -> list[str]:
    """여러 파일을 한 번에 리뷰하여 오버헤드 감소"""

    # 개별 요청 (비효율)
    # for file in files:
    #     review(file)  # 각각 시스템 프롬프트 전송

    # 배치 요청 (효율적)
    combined_content = "\n\n---\n\n".join([
        f"## File: {file}\n```\n{read_file(file)}\n```"
        for file in files
    ])

    message = client.messages.create(
        model="claude-",
        max_tokens=8192,
        system="You are a code reviewer...",  # 한 번만 전송
        messages=[{
            "role": "user",
            "content": f"Review these {len(files)} files:\n\n{combined_content}"
        }]
    )

    # 파일별 리뷰 파싱
    reviews = message.content[0].text.split("## File:")
    return reviews[1:]  # 첫 번째는 빈 문자열

로컬 모델 활용

import ollama

def hybrid_generation(prompt: str, require_accuracy: bool = False):
    """간단한 작업은 Ollama, 복잡한 작업은 Claude"""

    if not require_accuracy:
        # Ollama 사용 (무료)
        response = ollama.generate(
            model='llama3.2:3b',
            prompt=prompt
        )
        return response['response']
    else:
        # Claude 사용 (유료, 더 정확)
        message = client.messages.create(
            model="claude-",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}]
        )
        return message.content[0].text

# 사용
# 간단한 질문 → Ollama (무료)
answer = hybrid_generation("Explain what a for loop is")

# 중요한 작업 → Claude (유료, 정확)
review = hybrid_generation(
    "Review this security-critical code...",
    require_accuracy=True
)

대시보드 구축

Streamlit 대시보드

dashboard.py

import streamlit as st
import pandas as pd
import plotly.express as px
from datetime import datetime, timedelta

st.set_page_config(page_title="LLM Cost Dashboard", layout="wide")

# 데이터 로드
db = UsageDatabase()

# 제목
st.title("🤖 LLM API Cost Dashboard")

# 날짜 범위 선택
col1, col2 = st.columns(2)
with col1:
    start_date = st.date_input("Start Date", datetime.now() - timedelta(days=30))
with col2:
    end_date = st.date_input("End Date", datetime.now())

# 전체 통계
st.header("📊 Overall Statistics")

cursor = db.conn.execute(
    """
    SELECT
        COUNT(*) as total_requests,
        SUM(input_tokens) as total_input,
        SUM(output_tokens) as total_output,
        SUM(cost_usd) as total_cost
    FROM api_usage
    WHERE DATE(timestamp) BETWEEN ? AND ?
    """,
    (start_date.isoformat(), end_date.isoformat())
)
stats = cursor.fetchone()

col1, col2, col3, col4 = st.columns(4)
col1.metric("Total Requests", f"{stats[0]:,}")
col2.metric("Total Tokens", f"{(stats[1] + stats[2]):,}")
col3.metric("Total Cost", f"${stats[3]:.2f}")
col4.metric("Avg Cost/Request", f"${stats[3]/max(stats[0],1):.4f}")

# 일별 비용 그래프
st.header("📈 Daily Cost Trend")

df = pd.read_sql_query(
    """
    SELECT
        DATE(timestamp) as date,
        model,
        SUM(cost_usd) as cost
    FROM api_usage
    WHERE DATE(timestamp) BETWEEN ? AND ?
    GROUP BY DATE(timestamp), model
    ORDER BY date
    """,
    db.conn,
    params=(start_date.isoformat(), end_date.isoformat())
)

if not df.empty:
    fig = px.line(
        df,
        x="date",
        y="cost",
        color="model",
        title="Daily Cost by Model"
    )
    st.plotly_chart(fig, use_container_width=True)

# 모델별 통계
st.header("🤖 Cost by Model")

model_stats = pd.read_sql_query(
    """
    SELECT
        model,
        COUNT(*) as requests,
        SUM(cost_usd) as total_cost,
        AVG(cost_usd) as avg_cost
    FROM api_usage
    WHERE DATE(timestamp) BETWEEN ? AND ?
    GROUP BY model
    ORDER BY total_cost DESC
    """,
    db.conn,
    params=(start_date.isoformat(), end_date.isoformat())
)

if not model_stats.empty:
    fig = px.pie(
        model_stats,
        values="total_cost",
        names="model",
        title="Cost Distribution by Model"
    )
    st.plotly_chart(fig, use_container_width=True)

    st.dataframe(model_stats, use_container_width=True)

# 사용자별 통계 (있는 경우)
st.header("👤 Top Users by Cost")

user_stats = pd.read_sql_query(
    """
    SELECT
        user_id,
        COUNT(*) as requests,
        SUM(cost_usd) as total_cost
    FROM api_usage
    WHERE user_id IS NOT NULL
      AND DATE(timestamp) BETWEEN ? AND ?
    GROUP BY user_id
    ORDER BY total_cost DESC
    LIMIT 10
    """,
    db.conn,
    params=(start_date.isoformat(), end_date.isoformat())
)

if not user_stats.empty:
    st.dataframe(user_stats, use_container_width=True)
else:
    st.info("No user data available")

# 최근 요청
st.header("🕐 Recent Requests")

recent = pd.read_sql_query(
    """
    SELECT
        timestamp,
        user_id,
        model,
        input_tokens,
        output_tokens,
        cost_usd
    FROM api_usage
    ORDER BY timestamp DESC
    LIMIT 50
    """,
    db.conn
)

st.dataframe(recent, use_container_width=True)

실행

# 의존성 설치
pip install streamlit pandas plotly

# 대시보드 실행
streamlit run dashboard.py

# 브라우저에서 http://localhost:8501 열기

Grafana + Prometheus

prometheus.yml

global:
  scrape_interval: 15s

scrape_configs:
  - job_name: 'llm-api'
    static_configs:
      - targets: ['localhost:8000']

FastAPI 메트릭 엔드포인트

from fastapi import FastAPI
from prometheus_client import Counter, Histogram, generate_latest

app = FastAPI()

# 메트릭 정의
api_requests = Counter(
    'llm_api_requests_total',
    'Total API requests',
    ['model', 'status']
)

api_cost = Counter(
    'llm_api_cost_usd_total',
    'Total cost in USD',
    ['model']
)

api_tokens = Counter(
    'llm_api_tokens_total',
    'Total tokens used',
    ['model', 'type']  # type: input or output
)

api_latency = Histogram(
    'llm_api_latency_seconds',
    'API request latency',
    ['model']
)

@app.get("/metrics")
def metrics():
    return generate_latest()

@app.post("/chat")
async def chat(request: ChatRequest):
    with api_latency.labels(model=request.model).time():
        try:
            message = client.messages.create(
                model=request.model,
                max_tokens=1024,
                messages=request.messages
            )

            # 메트릭 업데이트
            api_requests.labels(model=request.model, status='success').inc()

            cost = calculate_cost(
                model=request.model,
                input_tokens=message.usage.input_tokens,
                output_tokens=message.usage.output_tokens
            )
            api_cost.labels(model=request.model).inc(cost)

            api_tokens.labels(model=request.model, type='input').inc(message.usage.input_tokens)
            api_tokens.labels(model=request.model, type='output').inc(message.usage.output_tokens)

            return {"response": message.content[0].text}

        except Exception as e:
            api_requests.labels(model=request.model, status='error').inc()
            raise

종합 베스트 프랙티스

모니터링

• 실시간 추적: 모든 API 호출 로깅
• 일일 리포트: 매일 비용 요약 이메일
• 대시보드: 시각화된 통계 확인
• 알림 설정: 임계값 초과 시 즉시 알림

최적화

• 캐싱: 반복 컨텍스트는 Prompt Caching 사용
• 모델 선택: 작업에 맞는 최소 모델 사용
• 컨텍스트 축소: 불필요한 정보 제거
• 배치 처리: 여러 작업을 한 번에 처리
• 로컬 모델: 간단한 작업은 Ollama 활용

예산 관리

• 일일/월간 한도: 명확한 예산 설정
• 사용자별 할당: 팀원별 예산 분배
• 자동 차단: 한도 초과 시 API 호출 중단
• 정기 리뷰: 주간/월간 비용 분석

핵심 정리

• LLM API 비용 모니터링의 핵심 개념과 흐름을 정리합니다.
• LLM 가격 구조를 단계별로 이해합니다.
• 실전 적용 시 기준과 주의점을 확인합니다.

실무 팁

• 입력/출력 예시를 고정해 재현성을 확보하세요.
• LLM API 비용 모니터링 범위를 작게 잡고 단계적으로 확장하세요.
• LLM 가격 구조 조건을 문서화해 대응 시간을 줄이세요.