안녕하세요, 저는 HolySheep AI의 시니어 엔지니어입니다. 이번 튜토리얼에서는 AI API 사용량을 추적하고 팀별, 프로젝트별, 기능별로 비용을 정확히 배분하는 시스템을 구축하는 방법을 다루겠습니다. HolySheep AI를 사용하면 단일 API 키로 여러 모델을 통합 관리할 수 있어, 비용 추적과 할당 작업이 한결 수월해집니다.
왜 AI API 사용량 추적과 비용 배분이 중요한가
AI API 비용은 예측하기 어렵습니다. 토큰 소비량은 입력/출력 길이, 모델 선택, 프롬프트 최적화 여부에 따라 수 배까지 달라질 수 있습니다. HolySheep AI의 대시보드에서는 실시간 사용량을 확인 가능하지만, 복잡한 마이크로서비스 환경에서는 팀 단위, 기능 단위의 세밀한 비용 배분이 필요합니다.
본 가이드에서 구축할 시스템은:
- API 호출별 토큰 소비량 자동 기록
- 태그 기반 비용 배분
- 실시간 비용 대시보드
- 예산 초과 알림
- 비용 최적화 제안
아키텍처 설계
비용 추적 시스템의 핵심은 미들웨어 패턴입니다. 모든 API 호출이 프록시를 통해 라우팅되며, 호출 정보가 비동기적으로 로그 시스템에 기록됩니다.
시스템 구성 요소
┌─────────────────────────────────────────────────────────────────────┐
│ HolySheep AI Gateway │
│ https://api.holysheep.ai/v1 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Team A │ │ Team B │ │ Team C │ │ Internal │ │
│ │ (Marketing)│ │ (R&D) │ │ (Support)│ │ Services │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
│ │ │ │ │ │
│ └───────────────┴───────┬───────┴───────────────┘ │
│ │ │
│ ┌───────────▼───────────┐ │
│ │ Usage Tracking │ │
│ │ Middleware Layer │ │
│ └───────────┬───────────┘ │
│ │ │
│ ┌──────────────────────┼──────────────────────┐ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │
│ │PostgreSQL│ │ Redis │ │Analytics DB │ │
│ │(Storage) │ │ (Cache) │ │ (TimescaleDB)│ │
│ └──────────┘ └──────────┘ └──────────────┘ │
│ │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ Cost Allocation Dashboard │ │
│ └──────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────┘핵심 구현: HolySheep AI 비용 추적 미들웨어
먼저 HolySheep AI API 호출을 래핑하는 비용 추적 클래스를 구현하겠습니다. HolySheep AI의 단일 엔드포인트로 여러 모델을 호출하므로, 모델별 가격 정책이 자동으로 적용됩니다.
import httpx
import asyncio
import time
import json
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Optional, Dict, List, Any
from contextlib import asynccontextmanager
import hashlib
HolySheep AI 모델별 가격 (USD per 1M tokens)
HOLYSHEEP_MODEL_PRICES = {
"gpt-4.1": {"input": 8.00, "output": 8.00},
"gpt-4.1-mini": {"input": 0.40, "output": 1.60},
"claude-sonnet-4-5": {"input": 15.00, "output": 15.00},
"claude-sonnet-4": {"input": 5.00, "output": 15.00},
"gemini-2.5-flash": {"input": 2.50, "output": 10.00},
"gemini-2.0-flash": {"input": 0.10, "output": 0.40},
"deepseek-v3.2": {"input": 0.42, "output": 0.42},
"deepseek-chat": {"input": 0.27, "output": 1.10},
}
@dataclass
class UsageRecord:
"""단일 API 호출의 사용량 기록"""
timestamp: datetime
model: str
team_id: str
project_id: str
feature: str
input_tokens: int
output_tokens: int
request_id: str
latency_ms: float
cost_usd: float = 0.0
metadata: Dict[str, Any] = field(default_factory=dict)
def calculate_cost(self) -> float:
"""HolySheep AI 가격 정책 기반 비용 계산"""
prices = HOLYSHEEP_MODEL_PRICES.get(self.model, {"input": 0, "output": 0})
input_cost = (self.input_tokens / 1_000_000) * prices["input"]
output_cost = (self.output_tokens / 1_000_000) * prices["output"]
self.cost_usd = round(input_cost + output_cost, 6)
return self.cost_usd
class HolySheepCostTracker:
"""
HolySheep AI API용 비용 추적 시스템
특징:
- 태그 기반 비용 배분 (team_id, project_id, feature)
- 실시간 비용 계산
- 비동기 배치 저장로직
- 예산 초과 자동 알림
"""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
storage_batch_size: int = 100,
flush_interval_seconds: int = 30
):
self.api_key = api_key
self.base_url = base_url
self.batch_size = storage_batch_size
self.flush_interval = flush_interval_seconds
self._usage_buffer: List[UsageRecord] = []
self._buffer_lock = asyncio.Lock()
self._client = httpx.AsyncClient(
base_url=base_url,
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
timeout=httpx.Timeout(60.0, connect=10.0)
)
# Budget configuration
self._budgets: Dict[str, Dict] = {}
self._budget_alerts: Dict[str, Dict] = {}
# Start background flush task
self._flush_task: Optional[asyncio.Task] = None
self._running = False
async def start(self):
"""배경 플러시 태스크 시작"""
self._running = True
self._flush_task = asyncio.create_task(self._periodic_flush())
async def stop(self):
""" Graceful shutdown """
self._running = False
if self._flush_task:
self._flush_task.cancel()
try:
await self._flush_task
except asyncio.CancelledError:
pass
await self._flush_buffer() # 최종 플러시
await self._client.aclose()
@asynccontextmanager
async def track(
self,
team_id: str,
project_id: str,
feature: str,
model: str = "deepseek-v3.2"
):
"""
컨텍스트 매니저로 API 호출 추적
Usage:
async with tracker.track("team_alpha", "chatbot", "user_response"):
response = await self._client.post("/chat/completions", json=payload)
"""
start_time = time.perf_counter()
request_id = hashlib.sha256(
f"{time.time()}{team_id}{feature}".encode()
).hexdigest()[:16]
record = UsageRecord(
timestamp=datetime.utcnow(),
model=model,
team_id=team_id,
project_id=project_id,
feature=feature,
input_tokens=0,
output_tokens=0,
request_id=request_id,
latency_ms=0
)
try:
yield record
finally:
record.latency_ms = (time.perf_counter() - start_time) * 1000
async def record_usage(
self,
team_id: str,
project_id: str,
feature: str,
model: str,
input_tokens: int,
output_tokens: int,
latency_ms: float,
metadata: Optional[Dict] = None
) -> UsageRecord:
"""사용량 기록 - HolySheep AI 응답 파싱 후 호출"""
record = UsageRecord(
timestamp=datetime.utcnow(),
model=model,
team_id=team_id,
project_id=project_id,
feature=feature,
input_tokens=input_tokens,
output_tokens=output_tokens,
request_id=hashlib.sha256(
f"{time.time()}{team_id}{feature}".encode()
).hexdigest()[:16],
latency_ms=latency_ms,
metadata=metadata or {}
)
record.calculate_cost()
# 버퍼에 추가
async with self._buffer_lock:
self._usage_buffer.append(record)
# 버퍼 크기 도달 시 플러시
if len(self._usage_buffer) >= self.batch_size:
await self._flush_buffer()
# 예산 체크
await self._check_budget(record)
return record
async def chat_completion(
self,
messages: List[Dict],
team_id: str,
project_id: str,
feature: str,
model: str = "deepseek-v3.2",
**kwargs
) -> Dict:
"""
HolySheep AI 채팅 완성 API 호출 및 사용량 추적
모델 선택 가이드:
- 비용 최적화: deepseek-v3.2 ($0.42/MTok)
- 균형 잡힌 성능: gemini-2.5-flash ($2.50/MTok)
- 최고 품질: claude-sonnet-4.5 ($15/MTok)
"""
start_time = time.perf_counter()
payload = {
"model": model,
"messages": messages,
**kwargs
}
# 토큰 수 추정
input_tokens = self._estimate_tokens(messages)
try:
response = await self._client.post("/chat/completions", json=payload)
response.raise_for_status()
data = response.json()
# 실제 토큰 사용량 추출 (HolySheep AI 응답)
usage = data.get("usage", {})
actual_input_tokens = usage.get("prompt_tokens", input_tokens)
actual_output_tokens = usage.get("completion_tokens", 0)
latency_ms = (time.perf_counter() - start_time) * 1000
# 사용량 기록
await self.record_usage(
team_id=team_id,
project_id=project_id,
feature=feature,
model=model,
input_tokens=actual_input_tokens,
output_tokens=actual_output_tokens,
latency_ms=latency_ms,
metadata={
"model_alias": data.get("model"),
"finish_reason": usage.get("finish_reason")
}
)
return data
except httpx.HTTPStatusError as e:
# 재시도 로직
if e.response.status_code == 429:
await asyncio.sleep(2 ** kwargs.get("retries", 0))
return await self.chat_completion(
messages, team_id, project_id, feature, model,
retries=kwargs.get("retries", 0) + 1
)
raise
def _estimate_tokens(self, messages: List[Dict]) -> int:
"""토큰 수 추정 (gpt-3.5-turbo 기준 근사치)"""
total = 0
for msg in messages:
total += len(msg.get("content", "").split()) * 1.3
return int(total)
async def _flush_buffer(self):
"""버퍼 내용을 저장소로 플러시"""
async with self._buffer_lock:
if not self._usage_buffer:
return
records_to_store = self._usage_buffer.copy()
self._usage_buffer.clear()
# 실제 환경에서는 PostgreSQL/TimescaleDB에 저장
# 예: await self._db.insert_usage_records(records_to_store)
total_cost = sum(r.cost_usd for r in records_to_store)
print(f"[CostTracker] Flushed {len(records_to_store)} records, "
f"total cost: ${total_cost:.4f}")
async def _periodic_flush(self):
"""정기적 플러시 태스크"""
while self._running:
await asyncio.sleep(self.flush_interval)
await self._flush_buffer()
# ===== 예산 관리 =====
def set_budget(
self,
identifier: str,
monthly_limit_usd: float,
alert_threshold: float = 0.8,
webhook_url: Optional[str] = None
):
"""
예산 설정
identifier 형식: "team:{team_id}" 또는 "project:{project_id}"
"""
self._budgets[identifier] = {
"monthly_limit": monthly_limit_usd,
"alert_threshold": alert_threshold,
"webhook_url": webhook_url,
"period_start": datetime.utcnow().replace(day=1, hour=0, minute=0, second=0)
}
async def _check_budget(self, record: UsageRecord):
"""예산 초과 체크"""
identifiers = [
f"team:{record.team_id}",
f"project:{record.project_id}",
f"feature:{record.feature}"
]
for identifier in identifiers:
if identifier not in self._budgets:
continue
budget = self._budgets[identifier]
current_spend = await self._get_period_spend(identifier)
if current_spend >= budget["monthly_limit"]:
await self._trigger_budget_alert(identifier, current_spend, budget)
async def _get_period_spend(self, identifier: str) -> float:
"""기간 내 총 지출 조회"""
# 실제 환경에서는 DB 쿼리
# 임시 구현: 버퍼 내 레코드 합산
total = 0.0
async with self._buffer_lock:
for record in self._usage_buffer:
if identifier in [
f"team:{record.team_id}",
f"project:{record.project_id}",
f"feature:{record.feature}"
]:
total += record.cost_usd
return total
async def _trigger_budget_alert(
self, identifier: str, current: float, budget: Dict
):
"""예산 초과 알림 발송"""
if budget["webhook_url"]:
async with httpx.AsyncClient() as client:
await client.post(budget["webhook_url"], json={
"alert": "Budget Exceeded",
"identifier": identifier,
"current_spend": current,
"limit": budget["monthly_limit"]
})
print(f"[ALERT] Budget exceeded for {identifier}: ${current:.2f}")
# ===== 비용 분석 =====
async def get_cost_breakdown(
self,
team_id: Optional[str] = None,
project_id: Optional[str] = None,
start_date: Optional[datetime] = None,
end_date: Optional[datetime] = None
) -> Dict:
"""비용 분해 조회"""
# 실제 환경에서는 DB 쿼리
breakdown = {
"by_model": {},
"by_feature": {},
"total_cost": 0.0,
"total_tokens": {"input": 0, "output": 0},
"request_count": 0,
"avg_latency_ms": 0.0
}
# ... DB 쿼리 로직
return breakdown
def suggest_model_optimization(self, current_spend: float) -> List[Dict]:
"""비용 최적화 제안"""
suggestions = []
if current_spend > 1000: # 월 $1000 이상 사용 시
suggestions.append({
"priority": "high",
"action": "Consider Gemini 2.5 Flash",
"savings_potential": "60-80%",
"reason": "Gemini 2.5 Flash는 DeepSeek V3.2 대비 6배 저렴"
})
suggestions.append({
"priority": "medium",
"action": "Implement caching layer",
"savings_potential": "20-40%",
"reason": "반복 질문에 대한 캐싱으로 중복 API 호출 제거"
})
return suggestions
===== 사용 예시 =====
async def main():
# HolySheep AI 초기화
tracker = HolySheepCostTracker(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
await tracker.start()
# 예산 설정
tracker.set_budget(
identifier="team:marketing",
monthly_limit_usd=500.0,
alert_threshold=0.8,
webhook_url="https://your-app.com/webhooks/budget-alert"
)
try:
# 마케팅팀 - 고객 응답 생성
response = await tracker.chat_completion(
messages=[
{"role": "system", "content": "당신은 친절한 고객 서비스 담당자입니다."},
{"role": "user", "content": "배송 조회를 하고 싶습니다."}
],
team_id="marketing",
project_id="customer-service",
feature="user_response",
model="deepseek-v3.2",
temperature=0.7,
max_tokens=500
)
print(f"Response: {response['choices'][0]['message']['content']}")
print(f"Usage: {response['usage']}")
# R&D팀 - 코드 리뷰
response = await tracker.chat_completion(
messages=[
{"role": "user", "content": "다음 코드를 리뷰해주세요: " + "x = 1" * 1000}
],
team_id="rd",
project_id="code-review",
feature="auto_review",
model="claude-sonnet-4",
temperature=0.1
)
finally:
await tracker.stop()
if __name__ == "__main__":
asyncio.run(main())비용 배분 대시보드 구축
수집된 사용량 데이터를 시각화하는 대시보드를 구축하겠습니다. 이 대시보드는 팀별, 프로젝트별, 모델별 비용을 실시간으로 추적합니다.
import asyncio
from dataclasses import dataclass
from typing import List, Dict, Optional
from datetime import datetime, timedelta
import asyncpg
from fastapi import FastAPI, HTTPException, Depends
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from collections import defaultdict
app = FastAPI(title="HolySheep AI Cost Allocation Dashboard")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
===== 데이터베이스 스키마 =====
CREATE_TABLES_SQL = """
-- 사용량 로그 테이블
CREATE TABLE IF NOT EXISTS usage_logs (
id BIGSERIAL PRIMARY KEY,
timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW(),
team_id VARCHAR(50) NOT NULL,
project_id VARCHAR(50) NOT NULL,
feature VARCHAR(100) NOT NULL,
model VARCHAR(50) NOT NULL,
input_tokens INTEGER NOT NULL,
output_tokens INTEGER NOT NULL,
total_tokens INTEGER GENERATED ALWAYS AS (input_tokens + output_tokens) STORED,
cost_usd DECIMAL(12, 6) NOT NULL,
latency_ms DECIMAL(10, 2) NOT NULL,
request_id VARCHAR(32) UNIQUE NOT NULL,
metadata JSONB
);
-- 시계열 최적화를 위한 인덱스
CREATE INDEX IF NOT EXISTS idx_usage_logs_timestamp ON usage_logs (timestamp DESC);
CREATE INDEX IF NOT EXISTS idx_usage_logs_team ON usage_logs (team_id, timestamp DESC);
CREATE INDEX IF NOT EXISTS idx_usage_logs_project ON usage_logs (project_id, timestamp DESC);
-- Hypertable 생성 (TimescaleDB 사용 시)
-- SELECT create_hypertable('usage_logs', 'timestamp', if_not_exists => TRUE);
-- 예산 테이블
CREATE TABLE IF NOT EXISTS budgets (
id SERIAL PRIMARY KEY,
identifier VARCHAR(100) UNIQUE NOT NULL, -- team:{id}, project:{id}, feature:{id}
monthly_limit_usd DECIMAL(12, 2) NOT NULL,
alert_threshold DECIMAL(3, 2) DEFAULT 0.8,
webhook_url VARCHAR(500),
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW()
);
-- 예산 알림 로그
CREATE TABLE IF NOT EXISTS budget_alerts (
id BIGSERIAL PRIMARY KEY,
budget_id INTEGER REFERENCES budgets(id),
triggered_at TIMESTAMPTZ DEFAULT NOW(),
current_spend_usd DECIMAL(12, 2),
percentage DECIMAL(5, 2),
notification_sent BOOLEAN DEFAULT FALSE
);
"""
===== API 응답 모델 =====
@dataclass
class CostBreakdown:
"""비용 분해 데이터"""
team_id: str
project_id: str
total_cost: float
total_tokens: int
request_count: int
avg_latency_ms: float
by_model: Dict[str, Dict]
class DashboardSummary(BaseModel):
period: str
start_date: datetime
end_date: datetime
total_cost_usd: float
total_requests: int
total_input_tokens: int
total_output_tokens: int
avg_cost_per_request: float
teams: List[Dict]
top_models: List[Dict]
cost_trend: List[Dict]
class TeamCostDetail(BaseModel):
team_id: str
projects: List[Dict]
total_cost: float
budget_status: Optional[Dict]
===== 데이터베이스 연결 =====
class DatabaseManager:
def __init__(self, dsn: str):
self.dsn = dsn
self.pool: Optional[asyncpg.Pool] = None
async def connect(self):
self.pool = await asyncpg.create_pool(self.dsn, min_size=5, max_size=20)
async def disconnect(self):
if self.pool:
await self.pool.close()
async def init_schema(self):
"""테이블 초기화"""
async with self.pool.acquire() as conn:
await conn.execute(CREATE_TABLES_SQL)
async def get_dashboard_summary(
self,
start_date: datetime,
end_date: datetime
) -> DashboardSummary:
"""대시보드 요약 조회"""
async with self.pool.acquire() as conn:
# 전체 요약
summary = await conn.fetchrow("""
SELECT
COUNT(*) as total_requests,
COALESCE(SUM(input_tokens), 0) as total_input,
COALESCE(SUM(output_tokens), 0) as total_output,
COALESCE(SUM(cost_usd), 0) as total_cost,
COALESCE(AVG(latency_ms), 0) as avg_latency
FROM usage_logs
WHERE timestamp BETWEEN $1 AND $2
""", start_date, end_date)
# 팀별 분해
team_breakdown = await conn.fetch("""
SELECT
team_id,
COUNT(*) as requests,
COALESCE(SUM(cost_usd), 0) as cost,
COALESCE(SUM(total_tokens), 0) as tokens
FROM usage_logs
WHERE timestamp BETWEEN $1 AND $2
GROUP BY team_id
ORDER BY cost DESC
""", start_date, end_date)
# 모델별 분해
model_breakdown = await conn.fetch("""
SELECT
model,
COUNT(*) as requests,
COALESCE(SUM(cost_usd), 0) as cost,
COALESCE(SUM(total_tokens), 0) as tokens,
COALESCE(AVG(latency_ms), 0) as avg_latency
FROM usage_logs
WHERE timestamp BETWEEN $1 AND $2
GROUP BY model
ORDER BY cost DESC
LIMIT 10
""", start_date, end_date)
# 일별 비용 추이
daily_trend = await conn.fetch("""
SELECT
DATE(timestamp) as date,
COALESCE(SUM(cost_usd), 0) as cost,
COUNT(*) as requests
FROM usage_logs
WHERE timestamp BETWEEN $1 AND $2
GROUP BY DATE(timestamp)
ORDER BY date
""", start_date, end_date)
return DashboardSummary(
period=f"{start_date.date()} ~ {end_date.date()}",
start_date=start_date,
end_date=end_date,
total_cost_usd=float(summary['total_cost']),
total_requests=summary['total_requests'],
total_input_tokens=summary['total_input'],
total_output_tokens=summary['total_output'],
avg_cost_per_request=(
float(summary['total_cost']) / summary['total_requests']
if summary['total_requests'] > 0 else 0
),
teams=[
{
"team_id": r['team_id'],
"requests": r['requests'],
"cost_usd": float(r['cost']),
"tokens": r['tokens']
}
for r in team_breakdown
],
top_models=[
{
"model": r['model'],
"requests": r['requests'],
"cost_usd": float(r['cost']),
"tokens": r['tokens'],
"avg_latency_ms": float(r['avg_latency'])
}
for r in model_breakdown
],
cost_trend=[
{
"date": r['date'].isoformat(),
"cost_usd": float(r['cost']),
"requests": r['requests']
}
for r in daily_trend
]
)
async def get_team_detail(
self,
team_id: str,
start_date: datetime,
end_date: datetime
) -> TeamCostDetail:
"""팀별 상세 비용"""
async with self.pool.acquire() as conn:
# 프로젝트별 분해
project_breakdown = await conn.fetch("""
SELECT
project_id,
COUNT(*) as requests,
COALESCE(SUM(cost_usd), 0) as cost,
COALESCE(SUM(total_tokens), 0) as tokens,
jsonb_object_agg(
feature,
jsonb_build_object(
'cost', COALESCE(SUM(cost_usd) FILTER (WHERE feature = f.feature), 0),
'requests', COUNT(*) FILTER (WHERE feature = f.feature)
)
) as features
FROM usage_logs
CROSS JOIN LATERAL unnest(ARRAY[feature]) AS f(feature)
WHERE team_id = $1 AND timestamp BETWEEN $2 AND $3
GROUP BY project_id
ORDER BY cost DESC
""", team_id, start_date, end_date)
# 예산 상태
budget = await conn.fetchrow("""
SELECT * FROM budgets WHERE identifier = $1
""", f"team:{team_id}")
current_spend = await conn.fetchval("""
SELECT COALESCE(SUM(cost_usd), 0)
FROM usage_logs
WHERE team_id = $1
AND timestamp >= DATE_TRUNC('month', NOW())
""", team_id)
budget_status = None
if budget:
percentage = (float(current_spend) / float(budget['monthly_limit_usd'])) * 100
budget_status = {
"limit": float(budget['monthly_limit_usd']),
"current": float(current_spend),
"remaining": float(budget['monthly_limit_usd']) - float(current_spend),
"percentage": round(percentage, 2),
"status": "exceeded" if percentage >= 100 else
"warning" if percentage >= 80 else "normal"
}
return TeamCostDetail(
team_id=team_id,
projects=[
{
"project_id": r['project_id'],
"requests": r['requests'],
"cost_usd": float(r['cost']),
"tokens": r['tokens']
}
for r in project_breakdown
],
total_cost=sum(float(r['cost']) for r in project_breakdown),
budget_status=budget_status
)
async def get_cost_optimization_report(
self,
start_date: datetime,
end_date: datetime
) -> Dict:
"""비용 최적화 보고서"""
async with self.pool.acquire() as conn:
# 모델별 사용량 및 비용
model_usage = await conn.fetch("""
SELECT
model,
COUNT(*) as requests,
COALESCE(SUM(input_tokens), 0) as input_tokens,
COALESCE(SUM(output_tokens), 0) as output_tokens,
COALESCE(SUM(cost_usd), 0) as total_cost
FROM usage_logs
WHERE timestamp BETWEEN $1 AND $2
GROUP BY model
""", start_date, end_date)
recommendations = []
total_cost = sum(float(r['total_cost']) for r in model_usage)
for record in model_usage:
model = record['model']
cost = float(record['total_cost'])
cost_percentage = (cost / total_cost * 100) if total_cost > 0 else 0
# Claude Sonnet 사용 시 Gemini Flash 권장
if "claude-sonnet" in model and cost_percentage > 30:
recommendations.append({
"current_model": model,
"recommended_model": "gemini-2.5-flash",
"reason": "Gemini 2.5 Flash는 동일한 품질 대비 6배 저렴",
"estimated_savings": round(cost * 0.7, 2),
"savings_percentage": "70%",
"priority": "high"
})
# GPT-4.1 사용 시 DeepSeek 권장
if "gpt-4.1" in model and cost_percentage > 20:
recommendations.append({
"current_model": model,
"recommended_model": "deepseek-v3.2",
"reason": "DeepSeek V3.2는 GPT-4.1 대비 19배 저렴",
"estimated_savings": round(cost * 0.95, 2),
"savings_percentage": "95%",
"priority": "high"
})
# 반복 호출 감지
repeat_requests = await conn.fetchval("""
SELECT COUNT(*) / COUNT(DISTINCT request_id)
FROM usage_logs
WHERE model = $1 AND timestamp BETWEEN $2 AND $3
""", model, start_date, end_date)
if repeat_requests > 1.5:
recommendations.append({
"issue": "High repeat request rate",
"suggestion": "Implement semantic caching layer",
"potential_savings": round(cost * 0.25, 2),
"priority": "medium"
})
return {
"total_current_cost": round(total_cost, 2),
"potential_savings": sum(
r.get('estimated_savings', 0) for r in recommendations
),
"recommendations": recommendations
}
===== FastAPI 엔드포인트 =====
db: Optional[DatabaseManager] = None
@app.on_event("startup")
async def startup():
global db
db = DatabaseManager(
dsn="postgresql://user:pass@localhost:5432/holysheep_costs"
)
await db.connect()
await db.init_schema()
@app.on_event("shutdown")
async def shutdown():
if db:
await db.disconnect()
@app.get("/api/dashboard/summary")
async def get_summary(
days: int = 30,
start_date: Optional[datetime] = None,
end_date: Optional[datetime] = None
):
"""대시보드 요약"""
if not start_date:
end_date = datetime.utcnow()
start_date = end_date - timedelta(days=days)
return await db.get_dashboard_summary(start_date, end_date)
@app.get("/api/teams/{team_id}/costs")
async def get_team_costs(
team_id: str,
days: int = 30,
start_date: Optional[datetime] = None,
end_date: Optional[datetime] = None
):
"""팀별 비용 상세"""
if not start_date:
end_date = datetime.utcnow()
start_date = end_date - timedelta(days=days)
return await db.get_team_detail(team_id, start_date, end_date)
@app.get("/api/optimization/report")
async def get_optimization_report(days: int = 30):
"""비용 최적화 보고서"""
end_date = datetime.utcnow()
start_date = end_date - timedelta(days=days)
return await db.get_cost_optimization_report(start_date, end_date)
@app.post("/api/budgets")
async def create_budget(
identifier: str,
monthly_limit_usd: float,
alert_threshold: float = 0.8,
webhook_url: Optional[str] = None
):
"""예산 생성"""
async with db.pool.acquire() as conn:
await conn.execute("""
INSERT INTO budgets (identifier, monthly_limit_usd, alert_threshold, webhook_url)
VALUES ($1, $2, $3, $4)
ON CONFLICT (identifier) DO UPDATE SET
monthly_limit_usd = EXCLUDED.monthly_limit_usd,
alert_threshold = EXCLUDED.alert_threshold,
webhook_url = EXCLUDED.webhook_url,
updated_at = NOW()
""", identifier, monthly_limit_usd, alert_threshold, webhook_url)
return {"status": "success", "identifier": identifier}
===== 실행 =====
uvicorn main:app --host 0.0.0.0 --port 8000
print("""
================================================================================
HolySheep AI 비용 추적 대시보드 API 시작
================================================================================
프론트엔드 연동 예시:
fetch('/api/dashboard/summary?days=30')
.then(r => r.json())
.then(data => {
// Chart.js로 시각화
new Chart(ctx, {
type: 'line',
data: {
labels: data.cost_trend.map(d => d.date),
datasets: [{
label: '일별 비용