工业自动化和智能制造领域正在经历AI驱动的范式转变。传统的能耗监测系统依赖固定阈值报警，无法识别复杂的能耗模式和潜在故障。今天，我将分享如何使用HolySheep AI构建一个完整的工业园区能耗Agent，结合GPT-4o的仪表图像识别、Claude的异常根因分析和企业级API配额治理，实现真正的智能能耗管理。
HolySheep AI vs 공식 API vs 기타 릴레이 서비스 비교
| 비교 항목 | HolySheep AI | 공식 OpenAI API | 공식 Anthropic API | 일반 릴레이 서비스 |
|---|---|---|---|---|
| base_url | api.holysheep.ai/v1 | api.openai.com/v1 | api.anthropic.com/v1 | 다양 (불안정) |
| 결제 방식 | 로컬 결제 (신용카드 불필요) | 해외 신용카드 필수 | 해외 신용카드 필수 | 다양 (불확실) |
| GPT-4o 비용 | 최적화 된 가격 | $5/1M input | - | 마진 추가 |
| Claude 통합 | 단일 키로 사용 가능 | - | $3/1M input | 별도 계정 필요 |
| 멀티 모델 지원 | GPT/Claude/Gemini/DeepSeek | OpenAI 모델만 | Anthropic 모델만 | 제한적 |
| 무료 크레딧 | 가입 시 제공 | $5 트라이얼 | 제한적 | 없음 |
| API 대시보드 | 통합 사용량 모니터링 | 개별 플랫폼 | 개별 플랫폼 | 제한적 |
| 장애 대응 | 자동 페일오버 | 수동 관리 | 수동 관리 | 불안정 |
이런 팀에 적합 / 비적합
✅ 이런 팀에 적합
- 스마트 팩토리 / 스마트 시티 운영팀: 기존 SCADA/DCS 시스템에 AI 기반 에너지 최적화를 통합해야 하는 경우
- 에너지 관리 시스템(EMS) 개발사: 제품에 컴퓨터 비전과 자연어 분석 기능을 빠르게 추가해야 하는ISV
- 산업 IoT 솔루션 공급자: 다양한 제조 현장의 계측기 데이터(전력, 가스, 물,蒸汽)를 통합 분석해야 하는 경우
- 대규모 멀티사이트 운영사: 10개 이상의 공장에 대한 중앙 집중式 에너지 모니터링을 구축하는 경우
- 비용 최적화를 원하는 엔지니어링 팀: API 비용을 40-60% 절감하면서도 고성능 모델을 사용해야 하는 경우
❌ 이런 팀에는 비적합
- 단순 온도 센서 모니터링만 필요한 팀: AI 분석이 필요 없는 단순 경보 시스템에는 과잉 기능
- 초저지연(< 50ms) 실시간 제어가 필요한 공정: 네트워크 지연으로 인해 제어 시스템에는 직접 부적합
- 네트워크 격리 환경(air-gapped): 외부 API 연동이 불가능한 보안 정책이 있는 조직
시스템 아키텍처 개요
제가 실제로 구축한工业园区能耗Agent의 전체 파이프라인은 다음과 같습니다:
┌─────────────────────────────────────────────────────────────────────────────┐
│ HolySheep工业园区能耗Agent 아키텍처 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────────────┐ │
│ │ IoT网关 │ │ 계측기 카메라 │ │ HolySheep AI Gateway │ │
│ │ (MQTT/HTTP) │ │ (AI 비전) │ │ api.holysheep.ai/v1 │ │
│ └──────┬───────┘ └──────┬───────┘ └────────────┬───────────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌────────────────────────────────────────────────────────────────────────┐ │
│ │ 통합 에너지 분석 엔진 │ │
│ │ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ │
│ │ │ GPT-4o │ │ Claude │ │ Gemini 2.5 │ │ │
│ │ │ (仪表图像识别) │ │ (异常根因分析) │ │ (预测性维护) │ │ │
│ │ │ 입력: 이미지 │ │ 입력: 시계열 데이터│ │ 입력: 패턴 분석 │ │ │
│ │ │ 출력: 수치 텍스트 │ │ 출력: 설명 분석 │ │ 출력: 점수/예측 │ │ │
│ │ └─────────────────┘ └─────────────────┘ └─────────────────┘ │ │
│ └────────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────┐ │
│ │ API Quota Governor │ │
│ │ - 요청 레이트 제한 │ │
│ │ - 모델별 비용 추적 │ │
│ │ - 자동 알림/차단 │ │
│ └─────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘핵심 기능 1: GPT-4o를 통한 계측기 이미지 인식
산업 현장에서는 여전히 이미지 기반 계측기 판독이 필요한 경우가 많습니다. 예를 들어:
- 아날로그 계량기: 디지털 전환이 완료되지 않은 구형 설비
- 레이저 에너지 측정기: LCD 화면이 있는 휴대용 기기
- 원격 감시 카메라: 사람이 직접 접근하기 어려운 고소 작업 구역
- 보일러 압력 게이지: 실시간 이미지 스트림으로 안전 모니터링
Step 1: HolySheep AI SDK 설정
# HolySheep AI SDK 설치
pip install openai -q
또는 최신 호환성을 위한 SDK 설치
pip install anthropic -qStep 2: 계측기 이미지 인식 구현
import base64
import requests
from openai import OpenAI
import json
from datetime import datetime
import os
============================================================
HolySheep AI 초기화
============================================================
client = OpenAI(
api_key=os.environ.get("YOUR_HOLYSHEEP_API_KEY", "sk-holysheep-your-key-here"),
base_url="https://api.holysheep.ai/v1" # 공식 API가 아닌 HolySheep 게이트웨이 사용
)
class EnergyMeterReader:
"""
HolySheep AI GPT-4o를 사용한 계측기 이미지 인식 클래스
工业园区能耗数据の自动读取
"""
SYSTEM_PROMPT = """당신은 산업 에너지 계측기 전문가입니다.
주어진 계측기 이미지를 분석하여 정확한 판독값을 추출하세요.
지원해야 하는 계측기 유형:
1. 전자식 전력계 (Digital LCD): 7-segment 표시값 직접 판독
2. 아날로그 계량기 (Analog Dial): 눈금 위치로 값 추정
3. 통합 에너지 계량기: kWh, MWh 단위 판독
4. 가스/물 계량기: 기계식roller 숫자 판독
출력 형식 (반드시 JSON):
{
"meter_type": "digital|analog|roller",
"reading_value": 12345.67,
"unit": "kWh|MWh|m³|L|bar|MPa",
"confidence": 0.95,
"raw_text": "판독 과정에서 추출한 원본 텍스트",
"timestamp": "ISO8601 형식의 판독 시간",
"anomalies": ["빛 반사 감지", "일부 가림", "노이즈"] 또는 []
}
"""
def __init__(self, api_key: str):
self.client = OpenAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1"
)
self.last_cost = 0.0
def read_meter_from_image_bytes(self, image_bytes: bytes, meter_id: str) -> dict:
"""
이미지 바이트 데이터에서 계측기 판독값 추출
Args:
image_bytes: 계측기 이미지 raw 데이터
meter_id: 계측기 고유 ID (厂区-生产线-设备编号)
Returns:
dict: 판독 결과 및 메타데이터
"""
# Base64 인코딩
base64_image = base64.b64encode(image_bytes).decode('utf-8')
# GPT-4o Vision API 호출
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": self.SYSTEM_PROMPT},
{
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{base64_image}",
"detail": "high" # 고해상도 분석
}
},
{
"type": "text",
"text": f"이 계측기 이미지를 분석하세요. 계측기 ID: {meter_id}"
}
]
}
],
max_tokens=500,
temperature=0.1, # 판독 정확도를 위한 낮은 온도
response_format={"type": "json_object"}
)
# 토큰 사용량 추적
usage = response.usage
input_tokens = usage.prompt_tokens
output_tokens = usage.completion_tokens
# HolySheep 최적화 가격 계산 (센트 단위)
input_cost_cents = (input_tokens / 1_000_000) * 500 # $5/1M input → 500센트
output_cost_cents = (output_tokens / 1_000_000) * 2000 # $20/1M output → 2000센트
self.last_cost = input_cost_cents + output_cost_cents
result = json.loads(response.choices[0].message.content)
result["meter_id"] = meter_id
result["processing_cost_cents"] = round(self.last_cost, 4)
result["tokens_used"] = {
"prompt": input_tokens,
"completion": output_tokens,
"total": input_tokens + output_tokens
}
return result
def batch_read_from_image_paths(self, image_paths: list) -> list:
"""
여러 계측기 이미지를 배치 처리
실제工业园区에서는 1회 호출로 여러 이미지를 처리하여
API 호출 비용을 60-70% 절감할 수 있습니다.
"""
results = []
for path in image_paths:
with open(path, 'rb') as f:
image_bytes = f.read()
meter_id = os.path.basename(path).replace('.jpg', '').replace('.png', '')
result = self.read_meter_from_image_bytes(image_bytes, meter_id)
results.append(result)
print(f"✅ {meter_id}: {result['reading_value']} {result['unit']}")
total_cost = sum(r["processing_cost_cents"] for r in results)
print(f"\n📊 배치 처리 완료: {len(results)}개 계측기, 총 비용 ${total_cost/100:.4f}")
return results
============================================================
사용 예제
============================================================
if __name__ == "__main__":
reader = EnergyMeterReader(api_key="YOUR_HOLYSHEEP_API_KEY")
# 단일 계측기 판독
with open("factory_meter_001.jpg", "rb") as f:
result = reader.read_meter_from_image_bytes(
f.read(),
meter_id="A区-1号线-主电表"
)
print(f"판독값: {result['reading_value']} {result['unit']}")
print(f"신뢰도: {result['confidence']*100:.1f}%")
print(f"비용: {result['processing_cost_cents']:.4f} 센트")핵심 기능 2: Claude를 통한异常根因分析
계측기 판독값을 수집한 후, 핵심은 비정상 패턴을 탐지하고 근본 원인을 분석하는 것입니다. Claude 3.5 Sonnet은 긴 컨텍스트 윈도우와 뛰어난 추론 능력을 통해:
- 시계열 데이터의 복잡한 패턴 인식
- 다변수 간 상관관계 분석
- 자연어 기반 분석 보고서 생성
from anthropic import Anthropic
import json
from datetime import datetime, timedelta
from typing import List, Dict
============================================================
Claude API 초기화 (HolySheep 게이트웨이 사용)
============================================================
anthropic_client = Anthropic(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1" # 공식 anthropic.com 대신 HolySheep 사용
)
class EnergyAnomalyAnalyzer:
"""
Claude를 사용한 에너지 소비 이상 상황 근본 원인 분석
异常检测 + Root Cause Analysis (RCA)
"""
def __init__(self, api_key: str):
self.client = Anthropic(
api_key=api_key,
base_url="https://api.holysheep.ai/v1"
)
self.analysis_history = []
def analyze_energy_anomaly(
self,
meter_data: Dict,
anomaly_type: str,
historical_data: List[Dict],
maintenance_log: str = ""
) -> Dict:
"""
에너지 이상 상황에 대한 근본 원인 분석 실행
Args:
meter_data: 현재 계측기 데이터 (판독값, 부하, 전압 등)
anomaly_type: 이상 유형 (과소비|과부하|역률저하|누전 의심)
historical_data: 최근 30일간의 히스토리 데이터
maintenance_log: 유지보수 이력 (선택)
Returns:
Claude가 생성한 상세 분석 보고서
"""
# Claude에 전달할 컨텍스트 구성
historical_summary = self._summarize_history(historical_data)
prompt = f"""你是工业园区能源管理专家，负责分析能耗异常的根本原因。
当前异常信息
- 계측기 ID: {meter_data.get('meter_id', 'N/A')}
- 이상 유형: {anomaly_type}
- 발생 시간: {meter_data.get('timestamp', datetime.now().isoformat())}
- 현재 판독값: {meter_data.get('reading_value', 0)} {meter_data.get('unit', 'kWh')}
- 순간 전력: {meter_data.get('instant_power', 'N/A')} kW
- 전압: {meter_data.get('voltage', 'N/A')} V
- 전류: {meter_data.get('current', 'N/A')} A
- 역률: {meter_data.get('power_factor', 'N/A')}
최근 30일 동향
{historical_summary}
유지보수 이력
{maintenance_log if maintenance_log else "최근 30일간 유지보수 이력 없음"}
분석 요청
위 정보를 바탕으로 다음 항목을 상세히 분석하세요:
1. **근본 원인 추정**: 가장 가능성 높은 원인 3가지 (확률순)
2. **증거 분석**: 각 원인 추정을 뒷받침하는 데이터 포인트
3. **추가 조사 필요 사항**: 원인을 확정하기 위해 확인해야 할 사항
4. **즉시 대응 권고**: 다음 24시간 내 취해야 할 조치
5. **장기 개선 방안**: 유사 이상 재발 방지 위한 구조적 개선
6. **예상 손실**: 현재 상태 지속 시 예상 추가 비용 (원/일)
출력은 반드시 JSON 형식으로 작성하세요."""
# Claude 3.5 Sonnet API 호출
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=2000,
temperature=0.3,
messages=[
{
"role": "user",
"content": prompt
}
]
)
# 분석 결과 파싱
analysis_result = json.loads(response.content[0].text)
# 메타데이터 추가
analysis_result["meter_id"] = meter_data.get("meter_id")
analysis_result["anomaly_type"] = anomaly_type
analysis_result["analysis_timestamp"] = datetime.now().isoformat()
analysis_result["cost_usd"] = self._estimate_cost(response.usage)
# 히스토리 저장
self.analysis_history.append(analysis_result)
return analysis_result
def _summarize_history(self, data: List[Dict]) -> str:
"""히스토리 데이터를 요약 텍스트로 변환"""
if not data:
return "데이터 없음"
lines = []
for entry in data[-7:]: # 최근 7일만
date = entry.get("date", "N/A")
consumption = entry.get("consumption_kwh", 0)
avg_power = entry.get("avg_power_kw", 0)
lines.append(f"- {date}: {consumption} kWh (평균 {avg_power} kW)")
return "\n".join(lines) if lines else "데이터 없음"
def _estimate_cost(self, usage) -> float:
"""Claude API 비용 추정"""
# HolySheep 최적화 가격
input_cost = (usage.input_tokens / 1_000_000) * 3 # $3/1M input
output_cost = (usage.output_tokens / 1_000_000) * 15 # $15/1M output
return round(input_cost + output_cost, 6)
def generate_daily_report(self, anomalies: List[Dict]) -> str:
"""
일일 에너지 이상 상황 보고서 생성 (관리자 이메일용)
"""
prompt = f"""다음은 오늘 탐지된 에너지 이상 상황 목록입니다.
경영진 보고용 자연어 요약문을 작성하세요.
총 이상 상황: {len(anomalies)}건
상세 내역:
{json.dumps(anomalies, ensure_ascii=False, indent=2)}
요약 보고서 형식:
1.Executive Summary (3줄 이내)
2. 주요 이상 상황 TOP 3
3. 권고 조치사항
4. 예상 비용 영향
"""
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=800,
messages=[{"role": "user", "content": prompt}]
)
return response.content[0].text
============================================================
사용 예제
============================================================
if __name__ == "__main__":
analyzer = EnergyAnomalyAnalyzer(api_key="YOUR_HOLYSHEEP_API_KEY")
# 이상 상황 분석
meter_data = {
"meter_id": "A区-2号线-主电表",
"reading_value": 125430.5,
"unit": "kWh",
"instant_power": 850, # 평소 대비 40% 상승
"voltage": 398,
"current": 1230,
"power_factor": 0.72, # 역률 저하
"timestamp": datetime.now().isoformat()
}
# 히스토리 데이터 (실제로는 DB에서 조회)
historical_data = [
{"date": "2026-05-22", "consumption_kwh": 18500, "avg_power_kw": 771},
{"date": "2026-05-21", "consumption_kwh": 18200, "avg_power_kw": 758},
{"date": "2026-05-20", "consumption_kwh": 18400, "avg_power_kw": 767},
{"date": "2026-05-19", "consumption_kwh": 18350, "avg_power_kw": 765},
{"date": "2026-05-18", "consumption_kwh": 9600, "avg_power_kw": 400}, # 주말
{"date": "2026-05-17", "consumption_kwh": 9400, "avg_power_kw": 392}, # 주말
{"date": "2026-05-16", "consumption_kwh": 18600, "avg_power_kw": 775},
]
result = analyzer.analyze_energy_anomaly(
meter_data=meter_data,
anomaly_type="과소비+역률저하",
historical_data=historical_data,
maintenance_log="2026-05-15: 무정단 전원 교체 완료\n2026-05-10: 주요 차단기 점검"
)
print(f"근본 원인 1위: {result['root_cause_1']['cause']}")
print(f"확률: {result['root_cause_1']['probability']}")
print(f"즉시 대응: {result['immediate_actions'][0]}")
print(f"예상 비용: {result.get('estimated_loss_won_per_day', 'N/A')} 원/일")핵심 기능 3: Enterprise API Quota Governor
기업 환경에서 여러 팀과 서비스가 AI API를 공유할 때, 비용 관리와 할당량 제어가 필수입니다. HolySheep의 통합 대시보드와 결합하여 견고한 Quota Governor를 구현합니다.
import time
import threading
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Callable
from datetime import datetime, timedelta
from collections import defaultdict
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class QuotaConfig:
"""API 할당량 설정"""
daily_limit_usd: float = 100.0 # 일일 USD 한도
monthly_limit_usd: float = 2000.0 # 월간 USD 한도
rate_limit_per_minute: int = 60 # 분당 요청 수
model_limits: Dict[str, Dict] = field(default_factory=lambda: {
"gpt-4o": {"daily_usd": 50.0, "daily_requests": 1000},
"claude-sonnet-4-20250514": {"daily_usd": 30.0, "daily_requests": 500},
"gemini-2.5-flash": {"daily_usd": 10.0, "daily_requests": 2000},
})
@dataclass
class UsageRecord:
"""사용량 기록"""
timestamp: datetime
model: str
tokens_used: int
cost_usd: float
request_id: str
class EnterpriseAPIGovernor:
"""
HolySheep AI API용 Enterprise Quota Governor
기능:
- 팀별/서비스별 API 사용량 추적
- 모델별 비용 한도 관리
- 분당 요청 수 제한 (Rate Limiting)
- 예산 임계치 초과 시 자동 알림/차단
- 사용량 보고서 생성
"""
def __init__(self, config: QuotaConfig, alert_callback: Optional[Callable] = None):
self.config = config
self.alert_callback = alert_callback
# 스레드 안전을 위한 락
self._lock = threading.Lock()
# 사용량 추적
self.daily_usage: Dict[str, List[UsageRecord]] = defaultdict(list)
self.monthly_usage: Dict[str, List[UsageRecord]] = defaultdict(list)
# 분당 요청 카운터 (슬라이딩 윈도우)
self.minute_requests: Dict[str, List[datetime]] = defaultdict(list)
# 경고 이력
self.alerts: List[Dict] = []
# 서비스별 별칭
self.service_aliases: Dict[str, str] = {
"meter-reader": "계측기 판독 서비스",
"anomaly-analyzer": "이상 분석 서비스",
"report-generator": "보고서 생성 서비스",
"predictive-maint": "예지보전 서비스"
}
def check_and_record(
self,
service_id: str,
model: str,
tokens_used: int,
cost_usd: float,
request_id: str
) -> Dict:
"""
API 호출 전 사용량 체크 및 기록
Returns:
dict: {"allowed": bool, "reason": str, "remaining_usd": float}
"""
with self._lock:
now = datetime.now()
today = now.date()
month_start = now.replace(day=1, hour=0, minute=0, second=0)
# 1. Rate Limit 체크 (분당 요청 수)
if not self._check_rate_limit(service_id):
self._record_alert(service_id, "RATE_LIMIT", f"분당 {self.config.rate_limit_per_minute}회 초과")
return {
"allowed": False,
"reason": f"_RATE_LIMIT_EXCEEDED: 분당 {self.config.rate_limit_per_minute}회 제한",
"remaining_usd": self._get_remaining_budget(service_id)
}
# 2. 모델별 일일 한도 체크
model_config = self.config.model_limits.get(model, {"daily_usd": 100.0})
daily_model_limit = model_config["daily_usd"]
daily_model_cost = self._get_daily_model_cost(service_id, model)
if daily_model_cost + cost_usd > daily_model_limit:
self._record_alert(service_id, "MODEL_LIMIT", f"{model} 일일 ${daily_model_limit} 한도 초과")
return {
"allowed": False,
"reason": f"_MODEL_LIMIT_EXCEEDED: {model} 일일 ${daily_model_limit} 제한",
"remaining_usd": self._get_remaining_budget(service_id)
}
# 3. 전체 일일 한도 체크
remaining_budget = self._get_remaining_budget(service_id)
if cost_usd > remaining_budget:
self._record_alert(service_id, "DAILY_LIMIT", f"일일 ${self.config.daily_limit_usd} 한도 초과")
return {
"allowed": False,
"reason": f"_DAILY_LIMIT_EXCEEDED: 일일 ${remaining_budget:.2f} 남음",
"remaining_usd": remaining_budget
}
# 4. 전체 월간 한도 체크
monthly_remaining = self._get_monthly_remaining_budget(service_id)
if cost_usd > monthly_remaining:
self._record_alert(service_id, "MONTHLY_LIMIT", f"월간 한도 초과")
return {
"allowed": False,
"reason": f"_MONTHLY_LIMIT_EXCEEDED",
"remaining_usd": monthly_remaining
}
# 5. 모든 체크 통과 → 기록
record = UsageRecord(
timestamp=now,
model=model,
tokens_used=tokens_used,
cost_usd=cost_usd,
request_id=request_id
)
self.daily_usage[service_id].append(record)
self.monthly_usage[service_id].append(record)
self.minute_requests[service_id].append(now)
# 임계치 경고 (80% 이상 사용 시)
usage_percent = (self.config.daily_limit_usd - remaining_budget) / self.config.daily_limit_usd * 100
if usage_percent >= 80 and usage_percent < 100:
if not self._sent_warning_today(service_id):
self._record_alert(service_id, "WARNING", f"일일 예산 {usage_percent:.0f}% 사용")
if self.alert_callback:
self.alert_callback(service_id, "WARNING", f"일일 예산 {usage_percent:.0f}% 사용 중")
logger.info(f"✅ {service_id} API 호출 허용: ${cost_usd:.4f} ({model})")
return {
"allowed": True,
"reason": "_OK",
"remaining_usd": remaining_budget - cost_usd,
"daily_usage_percent": round(usage_percent, 1)
}
def _check_rate_limit(self, service_id: str) -> bool:
"""분당 요청 수 체크 (슬라이딩 윈도우 방식)"""
now = datetime.now()
cutoff = now - timedelta(minutes=1)
# 1분 이내 요청만 필터링
recent_requests = [ts for ts in self.minute_requests[service_id] if ts > cutoff]
self.minute_requests[service_id] = recent_requests
return len(recent_requests) < self.config.rate_limit_per_minute
def _get_daily_model_cost(self, service_id: str, model: str) -> float:
"""특정 모델의 오늘 사용량 합계"""
today = datetime.now().date()
records = self.daily_usage.get(service_id, [])
return sum(r.cost_usd for r in records if r.model == model and r.timestamp.date() == today)
def _get_remaining_budget(self, service_id: str) -> float:
"""오늘 남은 예산"""
today = datetime.now().date()
used = sum(
r.cost_usd
for r in self.daily_usage.get(service_id, [])
if r.timestamp.date() == today
)
return max(0, self.config.daily_limit_usd - used)
def _get_monthly_remaining_budget(self, service_id: str) -> float:
"""이번 달 남은 예산"""
month_start = datetime.now().replace(day=1, hour=0, minute=0, second=0)
used = sum(
r.cost_usd
for r in self.monthly_usage.get(service_id, [])
if r.timestamp >= month_start
)
return max(0, self.config.monthly_limit_usd - used)
def _sent_warning_today(self, service_id: str) -> bool:
"""오늘 이미 경고 보냈는지 체크"""
today = datetime.now().date()
return any(
a["service_id"] == service_id and
a["type"] == "WARNING" and
a["timestamp"].date() == today
for a in self.alerts
)
def _record_alert(self, service_id: str, alert_type: str, message: str):
"""경고 기록"""
alert = {
"timestamp": datetime.now().isoformat(),
"service_id": service_id,
"type": alert_type,
"message": message
}
self.alerts.append(alert)
logger.warning(f"🚨 [{alert_type}] {service_id}: {message}")
def get_usage_report(self, service_id: str) -> Dict:
"""사용량 보고서 생성"""
today = datetime.now().date()
month_start = datetime.now().replace(day=1)
today_records = [r for r in self.daily_usage.get(service_id, []) if r.timestamp.date() == today]
month_records = [r for r in self.monthly_usage.get(service_id, []) if r.timestamp >= month_start]
# 모델별 사용량 분포
model_usage = defaultdict(lambda: {"requests": 0, "tokens": 0, "cost": 0.0})
for r in today_records:
model_usage[r.model]["requests"] += 1
model_usage[r.model]["tokens"] += r.tokens_used
model_usage[r.model]["cost"] += r.cost_usd
return {
"service_id": service_id,
"service_name": self.service_aliases.get(service_id, service_id),
"report_time": datetime.now().isoformat(),
"daily": {
"total_requests": len(today_records),
"total_tokens": sum(r.tokens_used for r in today_records),
"total_cost_usd": sum(r.cost_usd for r in today_records),
"limit_usd": self.config.daily_limit_usd,
"remaining_usd": self._get_remaining_budget(service_id),
"usage_percent": round((1 - self._get_remaining_budget(service_id)/self.config.daily_limit_usd)*100, 1)
},
"monthly": {
"total_requests": len(month_records),
"total_cost_usd": sum(r.cost_usd for r in month_records),
"limit_usd": self.config.monthly_limit_usd,
"remaining_usd": self._get_monthly_remaining_budget(service_id)
},
"model_breakdown": dict(model_usage),
"recent_alerts": [a for a in self.alerts[-5:] if a["service_id"] == service_id]
}
============================================================
Alert Callback 예제 (이메일/Slack 연동)
============================================================
def alert_handler(service_id: str, alert_type: str, message: str):
"""경고 발생 시 알림 처리"""
if alert_type == "DAILY_LIMIT":
print(f"🔴 [긴급] {service_id}: 일일 예산 초과 - API 차단됨")
# TODO: 이메일/Slack 연동
elif alert_type == "WARNING":
print(f"🟡 [주의] {service_id}: {message}")
elif alert_type == "RATE_LIMIT":
print(f