핵심 결론: HolySheep AI는 다중 테넌트 SaaS에서 각テ넌트별 사용량 제한과 비용 분할이 필요한 개발팀에 최적화된 솔루션입니다. 단일 API 키로 모든 주요 모델을 통합 관리하면서, 사용자 级별Rate Limiting과청구 격리를 구현할 수 있습니다. 특히 해외 신용카드 없이 로컬 결제가 가능하다는 점에서, 글로벌 확장을 계획하는 스타트업과、中小기업에 강력히 추천합니다.
왜 API 게이트웨이인가?
AI API를 SaaS 제품에 직접 통합하면 다음과 같은 문제에 직면합니다:
- 비용 투명성 부족: 각 고객별 사용량을 추적하고 과금하기 어려움
- Rate Limit 관리 복잡: 단일 API 키로 여러 고객을 서비스하면 전체 시스템이 영향
- 보안 위험: API 키 유출 시 전체 서비스가 위험
- 멀티모델 전환 어려움: 특정 모델 장애 시 대체 모델로 자동 전환 불가
저는 과거 여러 프로젝트에서 이러한 문제들을 직접 경험했으며, HolySheep API 게이트웨이를 도입한 후 운영 복잡도가 크게 감소했습니다. 이제 구체적인 설계 패턴과 구현 코드를 공유하겠습니다.
HolySheep vs 경쟁 서비스 비교
| 기준 | HolySheep AI | OpenAI 직접 | Anthropic 직접 | AWS Bedrock |
|---|---|---|---|---|
| 베이스 URL | api.holysheep.ai/v1 | api.openai.com/v1 | api.anthropic.com/v1 | aws.amazon.com/bedrock |
| 결제 방식 | 로컬 결제 지원 (신용카드 불필요) |
해외 신용카드 필수 | 해외 신용카드 필수 | 해외 신용카드 필수 |
| GPT-4.1 가격 | $8.00/MTok | $15.00/MTok | N/A | $15.00/MTok |
| Claude Sonnet 4.5 | $15.00/MTok | N/A | $18.00/MTok | $18.00/MTok |
| Gemini 2.5 Flash | $2.50/MTok | N/A | N/A | $3.50/MTok |
| DeepSeek V3.2 | $0.42/MTok | N/A | N/A | 제한적 |
| 멀티모델 지원 | 전체 통합 | OpenAI만 | Anthropic만 | 제한적 |
| Rate Limit 관리 | 빌트인 | 기본만 | 기본만 | 설정 필요 |
| 평균 지연 시간 | ~120ms | ~180ms | ~200ms | ~250ms |
| 무료 크레딧 | 가입 시 제공 | $5 제공 | 없음 | 사용량 기반 |
이런 팀에 적합 / 비적합
적합한 팀
- 스타트업 & SMB: 해외 신용카드 없이 AI API를 도입하고 싶은 팀
- 다중 테넌트 SaaS: 각 고객별 사용량 추적과 과금이 필요한 서비스
- 비용 최적화 팀: 다양한 모델을 효율적으로 조합하여 비용을 절감하려는 조직
- 글로벌 확장팀: 여러 지역에서 로컬 결제가 필요한 경우
- AI 프록시 서비스:HolySheep API를 기반으로 자체 프록시 서비스를 구축하려는 개발자
비적합한 팀
- 단일 고객 서비스: 다중 테넌트가 필요 없는 단순한 서비스
- 특정 벤더 종속 선호: 단일 AI 모델 공급업체만 사용하려는 경우
- 엄격한 온프레미스 요구:모든 데이터를 자체 인프라에서 처리해야 하는 규제 산업
사용자 级限流(Per-User Rate Limiting) 구현
다중 테넌트 SaaS에서 가장 중요한 요구사항 중 하나는 각 사용자 또는テ넌트별 요청 수 제한입니다. HolySheep API는 이를 위한 유연한 설정을 지원합니다.
아키텍처 개요
┌─────────────────────────────────────────────────────────────┐
│ 다중 테넌트 SaaS 아키텍처 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Tenant A │ │ Tenant B │ │ Tenant C │ │
│ │ (Basic) │ │ (Pro) │ │ (Enterprise) │
│ │ 100 RPM │ │ 500 RPM │ │ 무제한 │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
│ │ │ │ │
│ └───────────────┼───────────────┘ │
│ ▼ │
│ ┌──────────────────┐ │
│ │ Rate Limiter │ │
│ │ (Redis/Sliding │ │
│ │ Window) │ │
│ └────────┬─────────┘ │
│ ▼ │
│ ┌──────────────────┐ │
│ │ HolySheep API │ │
│ │ api.holysheep.ai │ │
│ │ /v1 │ │
│ └──────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘Python 구현: 사용자 级Rate Limiter
"""
HolySheep API 다중 테넌트 Rate Limiter
작성자: HolySheep AI 기술 블로그
"""
import time
import hashlib
from collections import defaultdict
from dataclasses import dataclass
from typing import Dict, Optional
import httpx
import redis
@dataclass
class TenantConfig:
"""테넌트별 설정"""
tenant_id: str
rate_limit_rpm: int # 요청/분 제한
rate_limit_rpd: int # 요청/일 제한
monthly_token_limit: int # 월간 토큰 제한
allowed_models: list # 허용된 모델 목록
current_tier: str # basic, pro, enterprise
class MultiTenantRateLimiter:
"""
다중 테넌트 SaaS를 위한 Rate Limiter
HolySheep API 게이트웨이 앞에 위치하여
각テ넌트별 요청 수와 토큰 사용량을 관리합니다.
"""
def __init__(self, redis_host: str = "localhost", redis_port: int = 6379):
# Redis 연결 (분산 환경에서는 Redis 필수)
self.redis = redis.Redis(
host=redis_host,
port=redis_port,
decode_responses=True
)
# 테넌트 설정 캐시
self.tenant_configs: Dict[str, TenantConfig] = {}
# HolySheep API 설정
self.base_url = "https://api.holysheep.ai/v1"
def register_tenant(
self,
tenant_id: str,
tier: str = "basic",
custom_config: Optional[dict] = None
) -> TenantConfig:
"""새로운テ넌트 등록"""
tier_configs = {
"basic": {
"rate_limit_rpm": 60,
"rate_limit_rpd": 10000,
"monthly_token_limit": 1_000_000,
"allowed_models": ["gpt-4.1", "claude-sonnet-4-5"]
},
"pro": {
"rate_limit_rpm": 500,
"rate_limit_rpd": 100000,
"monthly_token_limit": 10_000_000,
"allowed_models": ["gpt-4.1", "gpt-4.1-mini", "claude-sonnet-4-5", "gemini-2.5-flash"]
},
"enterprise": {
"rate_limit_rpm": 10000,
"rate_limit_rpd": 10_000_000,
"monthly_token_limit": 1_000_000_000,
"allowed_models": ["*"] # 모든 모델 허용
}
}
config = tier_configs.get(tier, tier_configs["basic"])
if custom_config:
config.update(custom_config)
tenant_config = TenantConfig(
tenant_id=tenant_id,
current_tier=tier,
**config
)
self.tenant_configs[tenant_id] = tenant_config
return tenant_config
def _get_rate_limit_key(self, tenant_id: str, window: str) -> str:
"""Rate Limit Redis 키 생성"""
return f"ratelimit:{tenant_id}:{window}"
def _get_token_usage_key(self, tenant_id: str, period: str) -> str:
"""토큰 사용량 Redis 키 생성"""
return f"tokens:{tenant_id}:{period}"
def check_rate_limit(self, tenant_id: str) -> dict:
"""
Rate Limit 체크
returns: {
"allowed": bool,
"remaining_rpm": int,
"reset_at": timestamp,
"retry_after": int (초)
}
"""
if tenant_id not in self.tenant_configs:
raise ValueError(f"Unknown tenant: {tenant_id}")
config = self.tenant_configs[tenant_id]
now = time.time()
# 분당 Rate Limit 체크
rpm_key = self._get_rate_limit_key(tenant_id, "minute")
minute_window = int(now // 60)
# Redis Lua 스크립트로 Atomic 연산
lua_script = """
local current = redis.call('GET', KEYS[1])
local limit = tonumber(ARGV[1])
local window_start = tonumber(ARGV[2])
local ttl = tonumber(ARGV[3])
if current == false then
current = 0
else
current = tonumber(current)
end
if current >= limit then
return {0, current, redis.call('TTL', KEYS[1])}
end
redis.call('INCR', KEYS[1])
if current == 0 then
redis.call('EXPIRE', KEYS[1], ttl)
end
return {1, current + 1, redis.call('TTL', KEYS[1])}
"""
result = self.redis.eval(
lua_script,
1,
rpm_key,
config.rate_limit_rpm,
minute_window,
60
)
allowed, current_count, ttl = result
return {
"allowed": bool(allowed),
"remaining_rpm": max(0, config.rate_limit_rpm - current_count),
"current_rpm": current_count,
"limit_rpm": config.rate_limit_rpm,
"reset_at": now + ttl,
"retry_after": ttl if not allowed else 0
}
def check_token_limit(self, tenant_id: str, estimated_tokens: int) -> dict:
"""월간 토큰 제한 체크"""
if tenant_id not in self.tenant_configs:
raise ValueError(f"Unknown tenant: {tenant_id}")
config = self.tenant_configs[tenant_id]
# 현재 월의 토큰 사용량 조회
month_key = self._get_token_usage_key(tenant_id, "monthly")
current_usage = int(self.redis.get(month_key) or 0)
# 월말 만료 설정
if not self.redis.ttl(month_key) or self.redis.ttl(month_key) < 0:
# 다음 달 1일까지 TTL 설정
now = time.time()
days_in_month = 30 # 간단히 30일 계산
ttl = days_in_month * 24 * 3600 - (now % (24*3600))
self.redis.setex(month_key, int(ttl), 0)
current_usage = 0
remaining = max(0, config.monthly_token_limit - current_usage)
return {
"allowed": current_usage + estimated_tokens <= config.monthly_token_limit,
"current_usage": current_usage,
"limit": config.monthly_token_limit,
"remaining": remaining,
"estimated_request_tokens": estimated_tokens
}
def record_token_usage(self, tenant_id: str, input_tokens: int, output_tokens: int):
"""토큰 사용량 기록"""
total_tokens = input_tokens + output_tokens
month_key = self._get_token_usage_key(tenant_id, "monthly")
self.redis.incrby(month_key, total_tokens)
def is_model_allowed(self, tenant_id: str, model: str) -> bool:
"""모델 사용 권한 체크"""
config = self.tenant_configs[tenant_id]
if "*" in config.allowed_models:
return True
return model in config.allowed_models
사용 예시
limiter = MultiTenantRateLimiter()
테넌트 등록
limiter.register_tenant("tenant_abc123", tier="pro")
limiter.register_tenant("tenant_xyz789", tier="enterprise")
Rate Limit 체크
result = limiter.check_rate_limit("tenant_abc123")
print(f"Rate Limit Status: {result}")
Output: {'allowed': True, 'remaining_rpm': 499, 'current_rpm': 1, ...}
토큰 제한 체크 (예: 1000 토큰 예상)
token_check = limiter.check_token_limit("tenant_abc123", estimated_tokens=1000)
print(f"Token Limit Status: {token_check}")
Output: {'allowed': True, 'current_usage': 50000, 'limit': 10000000, ...}
HolySheep API 호출: 완전한 통합 코드
"""
HolySheep API 다중 테넌트 통합 - 완전한 예제
작성자: HolySheep AI 기술 블로그
"""
import os
import json
import time
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
import httpx
from multi_tenant_rate_limiter import MultiTenantRateLimiter, TenantConfig
class HolySheepMultiTenantClient:
"""
HolySheep API를 사용한 다중 테넌트 AI 서비스 클라이언트
주요 기능:
-テ넌트별 Rate Limiting
- 토큰 사용량 추적 및 청구
- 자동 모델 페일오버
- 비용 최적화 라우팅
"""
def __init__(
self,
api_key: str,
rate_limiter: Optional[MultiTenantRateLimiter] = None,
base_url: str = "https://api.holysheep.ai/v1"
):
self.api_key = api_key
self.base_url = base_url
self.rate_limiter = rate_limiter or MultiTenantRateLimiter()
# HTTP 클라이언트 설정
self.client = httpx.Client(
timeout=60.0,
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
)
# 청구 기록
self.billing_records: Dict[str, List[Dict]] = defaultdict(list)
def chat_completions(
self,
tenant_id: str,
messages: List[Dict[str, str]],
model: str = "gpt-4.1",
**kwargs
) -> Dict[str, Any]:
"""
Chat Completion 요청 (다중 테넌트 지원)
Args:
tenant_id: 테넌트 ID
messages: 메시지 목록
model: AI 모델 (테넌트 권한에 따라 제한될 수 있음)
**kwargs: 추가 파라미터 (temperature, max_tokens 등)
Returns:
API 응답 딕셔너리
Raises:
RateLimitExceeded: Rate Limit 초과 시
TokenLimitExceeded: 토큰 제한 초과 시
ModelNotAllowed: 허용되지 않은 모델 접근 시
"""
# 1단계: 모델 권한 체크
if not self.rate_limiter.is_model_allowed(tenant_id, model):
raise PermissionError(
f"Model '{model}' is not allowed for tenant '{tenant_id}'. "
f"Allowed models: {self.rate_limiter.tenant_configs.get(tenant_id)}"
)
# 2단계: Rate Limit 체크
rate_result = self.rate_limiter.check_rate_limit(tenant_id)
if not rate_result["allowed"]:
raise RateLimitExceeded(
f"Rate limit exceeded for tenant '{tenant_id}'. "
f"Retry after {rate_result['retry_after']} seconds."
)
# 3단계: 토큰 사용량 예측 및 체크
# 간단한 토큰 추정 (실제로는 tiktoken 등 사용 권장)
estimated_tokens = self._estimate_tokens(messages)
token_check = self.rate_limiter.check_token_limit(tenant_id, estimated_tokens)
if not token_check["allowed"]:
raise TokenLimitExceeded(
f"Token limit exceeded for tenant '{tenant_id}'. "
f"Current: {token_check['current_usage']}, Limit: {token_check['limit']}"
)
# 4단계: HolySheep API 호출
start_time = time.time()
try:
response = self.client.post(
f"{self.base_url}/chat/completions",
json={
"model": model,
"messages": messages,
**kwargs
}
)
# 응답 시간 측정
latency_ms = (time.time() - start_time) * 1000
if response.status_code == 200:
result = response.json()
# 5단계: 토큰 사용량 기록
usage = result.get("usage", {})
self.rate_limiter.record_token_usage(
tenant_id,
usage.get("prompt_tokens", 0),
usage.get("completion_tokens", 0)
)
# 6단계: 청구 기록 저장
self._record_billing(
tenant_id=tenant_id,
model=model,
latency_ms=latency_ms,
usage=usage,
response=result
)
return result
elif response.status_code == 429:
raise RateLimitExceeded("HolySheep API rate limit exceeded")
elif response.status_code == 400:
error = response.json()
raise ValueError(f"Bad request: {error}")
else:
raise APIError(f"API error: {response.status_code} - {response.text}")
except httpx.TimeoutException:
# 자동 페일오버: 다른 모델로 재시도
fallback_models = {
"gpt-4.1": "claude-sonnet-4-5",
"claude-sonnet-4-5": "gemini-2.5-flash",
"gemini-2.5-flash": "deepseek-v3-2"
}
if model in fallback_models:
print(f"Fallback: {model} -> {fallback_models[model]}")
return self.chat_completions(
tenant_id=tenant_id,
messages=messages,
model=fallback_models[model],
**kwargs
)
raise APIError("Request timeout and no fallback available")
def _estimate_tokens(self, messages: List[Dict[str, str]]) -> int:
"""토큰 수 추정 (단순 계산)"""
total = 0
for msg in messages:
total += len(msg.get("content", "").split()) * 1.3 # 단어를 토큰으로 변환
return int(total)
def _record_billing(
self,
tenant_id: str,
model: str,
latency_ms: float,
usage: Dict,
response: Dict
):
"""청구 기록 저장"""
# HolySheep 가격표 (2025년 기준)
pricing = {
"gpt-4.1": {"input": 8.00, "output": 8.00}, # $8/MTok
"gpt-4.1-mini": {"input": 2.00, "output": 8.00},
"claude-sonnet-4-5": {"input": 15.00, "output": 15.00}, # $15/MTok
"gemini-2.5-flash": {"input": 2.50, "output": 2.50}, # $2.50/MTok
"deepseek-v3-2": {"input": 0.42, "output": 0.42}, # $0.42/MTok
}
input_cost = (usage.get("prompt_tokens", 0) / 1_000_000) * pricing.get(model, {}).get("input", 0)
output_cost = (usage.get("completion_tokens", 0) / 1_000_000) * pricing.get(model, {}).get("output", 0)
record = {
"timestamp": time.time(),
"model": model,
"input_tokens": usage.get("prompt_tokens", 0),
"output_tokens": usage.get("completion_tokens", 0),
"total_tokens": usage.get("total_tokens", 0),
"input_cost_usd": round(input_cost, 6),
"output_cost_usd": round(output_cost, 0),
"total_cost_usd": round(input_cost + output_cost, 6),
"latency_ms": round(latency_ms, 2)
}
self.billing_records[tenant_id].append(record)
def get_tenant_billing(self, tenant_id: str) -> Dict[str, Any]:
"""テ넌트의 청구 요약 조회"""
records = self.billing_records.get(tenant_id, [])
if not records:
return {
"tenant_id": tenant_id,
"total_requests": 0,
"total_input_tokens": 0,
"total_output_tokens": 0,
"total_cost_usd": 0,
"avg_latency_ms": 0
}
return {
"tenant_id": tenant_id,
"total_requests": len(records),
"total_input_tokens": sum(r["input_tokens"] for r in records),
"total_output_tokens": sum(r["output_tokens"] for r in records),
"total_cost_usd": round(sum(r["total_cost_usd"] for r in records), 6),
"avg_latency_ms": round(sum(r["latency_ms"] for r in records) / len(records), 2),
"model_breakdown": self._get_model_breakdown(records)
}
def _get_model_breakdown(self, records: List[Dict]) -> Dict[str, Dict]:
"""모델별 사용량 분석"""
breakdown = defaultdict(lambda: {
"requests": 0,
"input_tokens": 0,
"output_tokens": 0,
"cost_usd": 0
})
for r in records:
model = r["model"]
breakdown[model]["requests"] += 1
breakdown[model]["input_tokens"] += r["input_tokens"]
breakdown[model]["output_tokens"] += r["output_tokens"]
breakdown[model]["cost_usd"] += r["total_cost_usd"]
return dict(breakdown)
def close(self):
"""클라이언트 종료"""
self.client.close()
===== 예외 클래스 =====
class RateLimitExceeded(Exception):
"""Rate Limit 초과 예외"""
pass
class TokenLimitExceeded(Exception):
"""토큰 제한 초과 예외"""
pass
class ModelNotAllowed(Exception):
"""모델 접근 권한 없음 예외"""
pass
class APIError(Exception):
"""API 일반 오류"""
pass
===== 사용 예시 =====
if __name__ == "__main__":
# HolySheep API 키 설정
API_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
# 클라이언트 초기화
client = HolySheepMultiTenantClient(
api_key=API_KEY,
rate_limiter=MultiTenantRateLimiter()
)
# 테넌트 등록
client.rate_limiter.register_tenant("tenant_abc123", tier="pro")
client.rate_limiter.register_tenant("tenant_xyz789", tier="enterprise")
# ===== Pro 테넌트: Standard 사용량 =====
try:
response = client.chat_completions(
tenant_id="tenant_abc123",
messages=[
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "다중 테넌트 SaaS에서 Rate Limiting을 구현하는 방법을 알려주세요."}
],
model="gpt-4.1",
temperature=0.7,
max_tokens=500
)
print("=== Pro Tenant Response ===")
print(f"Model: {response['model']}")
print(f"Usage: {response['usage']}")
print(f"Response: {response['choices'][0]['message']['content'][:200]}...")
except RateLimitExceeded as e:
print(f"Rate Limit: {e}")
except TokenLimitExceeded as e:
print(f"Token Limit: {e}")
except PermissionError as e:
print(f"Permission: {e}")
# ===== Enterprise Tenant:低成本 모델 사용 =====
try:
response = client.chat_completions(
tenant_id="tenant_xyz789",
messages=[
{"role": "user", "content": "간단한 인사말을 해주세요."}
],
model="deepseek-v3-2", # 가장 저렴한 모델
max_tokens=50
)
print("\n=== Enterprise Tenant Response (DeepSeek) ===")
print(f"Model: {response['model']}")
print(f"Cost-effective response: {response['choices'][0]['message']['content']}")
except Exception as e:
print(f"Error: {e}")
# ===== 청구 요약 조회 =====
print("\n=== Tenant Billing Summary ===")
for tenant_id in ["tenant_abc123", "tenant_xyz789"]:
billing = client.get_tenant_billing(tenant_id)
print(f"\n{tenant_id}:")
print(f" Total Requests: {billing['total_requests']}")
print(f" Total Cost: ${billing['total_cost_usd']:.6f}")
print(f" Avg Latency: {billing['avg_latency_ms']:.2f}ms")
client.close()청구 격리(Billing Isolation) 설계
다중 테넌트 SaaS에서 각テ넌트별 비용을 정확히 추적하고 격리하는 것은 필수입니다. HolySheep API를 활용한 청구 격리 전략을 설명하겠습니다.
청구 격리 아키텍처
┌──────────────────────────────────────────────────────────────────┐
│ 청구 격리 아키텍처 │
├──────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │Tenant A │ │Tenant B │ │Tenant C │ │Tenant D │ │
│ │ $12.50 │ │ $45.20 │ │ $8.30 │ │$128.40 │ │
│ └────┬────┘ └────┬────┘ └────┬────┘ └────┬────┘ │
│ │ │ │ │ │
│ └────────────┴────────────┴────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────┐ │
│ │ Billing Aggregator │ │
│ │ - 일별 집계 │ │
│ │ - 모델별 분류 │ │
│ │ - 예측 기반 경고 │ │
│ └────────────┬────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────┐ │
│ │ Invoice Generator │ │
│ │ - 월별 청구서 │ │
│ │ - 사용량 리포트 │ │
│ │ - 과금 알림 │ │
│ └─────────────────────────┘ │
│ │
└──────────────────────────────────────────────────────────────────┘"""
HolySheep API 기반 청구 격리 및 과금 시스템
작성자: HolySheep AI 기술 블로그
"""
import sqlite3
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass
from decimal import Decimal, ROUND_HALF_UP
@dataclass
class BillingRecord:
"""개별 청구 레코드"""
id: int
tenant_id: str
timestamp: datetime
model: str
input_tokens: int
output_tokens: int
total_tokens: int
input_cost_usd: Decimal
output_cost_usd: Decimal
total_cost_usd: Decimal
latency_ms: float
request_id: str
class BillingIsolationSystem:
"""
다중 테넌트 청구 격리 시스템
주요 기능:
-テ넌트별 비용 추적
- 모델별 사용량 분류
- 사용량 기반 경고
- 월별 청구서 생성
"""
# HolySheep 가격표 (실시간으로 API에서 조회 가능)
PRICING = {
"gpt-4.1": {
"input_per_mtok": Decimal("8.00"),
"output_per_mtok": Decimal("8.00")
},
"gpt-4.1-mini": {
"input_per_mtok": Decimal("2.00"),
"output_per_mtok": Decimal("8.00")
},
"claude-sonnet-4-5": {
"input_per_mtok": Decimal("15.00"),
"output_per_mtok": Decimal("15.00")
},
"gemini-2.5-flash": {
"input_per_mtok": Decimal("2.50"),
"output_per_mtok": Decimal("2.50")
},
"deepseek-v3-2": {
"input_per_mtok": Decimal("0.42"),
"output_per_mtok": Decimal("0.42")
}
}
def __init__(self, db_path: str = "billing.db"):
self.db_path = db_path
self._init_database()
def _init_database(self):
"""데이터베이스 초기화"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# 청구 레코드 테이블
cursor.execute("""
CREATE TABLE IF NOT EXISTS billing_records (
id INTEGER PRIMARY KEY AUTOINCREMENT,
tenant_id TEXT NOT NULL,
timestamp DATETIME NOT NULL,
model TEXT NOT NULL,
input_tokens INTEGER NOT NULL,
output_tokens INTEGER NOT NULL,
total_tokens INTEGER NOT NULL,
input_cost_usd REAL NOT NULL,
output_cost_usd REAL NOT NULL,
total_cost_usd REAL NOT NULL,
latency_ms REAL NOT NULL,
request_id TEXT UNIQUE NOT NULL
)
""")
# 인덱스 생성
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_tenant_timestamp
ON billing_records(tenant_id, timestamp)
""")
#テ넌트별 한도 설정
cursor.execute("""
CREATE TABLE IF NOT EXISTS tenant_limits (
tenant_id TEXT PRIMARY KEY,
monthly_limit_tokens INTEGER DEFAULT 1000000,
monthly_limit_usd REAL DEFAULT 100.00,
alert_threshold REAL DEFAULT 0.8,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
# 월별 집계 뷰
cursor.execute("""
CREATE VIEW IF NOT EXISTS monthly_summary AS
SELECT
tenant_id,
strftime('%Y-%m', timestamp) as month,
SUM(input_tokens) as total_input_tokens,
SUM(output_tokens) as total_output_tokens,
SUM(total_tokens) as total_tokens,
SUM(input_cost_usd) as total_input_cost,
SUM(output_cost_usd) as total_output_cost,
SUM(total_cost_usd) as total_cost,
COUNT(*) as request_count,
AVG(latency_ms) as avg_latency
FROM billing_records
GROUP BY tenant_id, strftime('%Y-%m', timestamp)
""")
conn.commit()
conn.close()
def calculate_cost(
self,
model: str,
input_tokens: int,
output_tokens: int
) -> Dict[str, Decimal]:
"""토큰 사용량 기반 비용 계산"""
pricing = self.PRICING.get(model, {
"input_per_mtok": Decimal("10.00"),
"output_per_mtok": Decimal("10.00")
})
input_cost = (Decimal(input_tokens) / Decimal("1000000")) * pricing["input_per_mtok"]
output_cost = (Decimal(output_tokens) / Decimal("1000000")) * pricing["output_per_mtok"]
# 소수점 6자리까지 반올림
input_cost = input_cost.quantize(Decimal("0.000001"), rounding=ROUND_HALF_UP)
output_cost = output_cost.quantize(Decimal("0.000001"), rounding=ROUND_HALF_UP)
return {
"input_cost_usd": input_cost,
"output_cost_usd": output_cost,
"total_cost_usd": input_cost + output_cost
}
def record_usage(
self,
tenant_id: str,
model: str,
input_tokens: int,
output_tokens: int,
latency_ms: float,
request_id: str,
timestamp: Optional[datetime] = None
) -> BillingRecord:
"""API 사용