저는 HolySheep AI에서 3년 넘게 AI API 통합과 비용 최적화를 담당해온 엔지니어입니다. 이번 튜토리얼에서는 AI 모델이 운영 환경에서도 지속적으로 학습하고 개선할 수 있는 지속 학습 전략을 심층적으로 다룹니다.
왜 AI 지속 학습이 중요한가?
AI 모델은 배포 시점의 데이터로 훈련됩니다. 그러나 실제 환경에서는:
- 사용자 행동 변화 — 이커머스 계절성, 트렌드 변화
- 새로운 도메인 지식 — 제품上市, 정책 변경
- 모델 성능 저하 — 데이터 드리프트(data drift) 발생
지속 학습 없이는 배포 후 3~6개월 안에 응답 품질이 급격히 떨어지는 것을 경험하게 됩니다. 이커머스 고객 서비스 Bot을 운영하는某기업에서는 지속 학습 도입 후 고객 만족도가 42% 향상되고 반복 문의율이 35% 감소한 사례를 확인했습니다.
핵심 개념: Fine-tuning vs RAG vs continual pre-training
AI 지속 학습 전략은 크게 세 가지 접근법으로 나뉩니다:
1. RAG(Retrieval-Augmented Generation) — 동적 지식 업데이트
외부 지식 베이스를 실시간으로 검색하여 컨텍스트에 추가하는 방식입니다.
"""
HolySheep AI + RAG 통합 아키텍처
저자: HolySheep AI Technical Writer
"""
import openai
import faiss
import numpy as np
from datetime import datetime
class ContinuousLearningRAG:
"""지속 학습을 위한 RAG 시스템"""
def __init__(self, api_key: str):
self.client = openai.OpenAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1" # HolySheep AI 게이트웨이
)
self.embedding_model = "text-embedding-3-small"
self.chat_model = "gpt-4.1"
# FAISS 벡터 스토어 초기화
self.dimension = 1536
self.index = faiss.IndexFlatL2(self.dimension)
self.metadata = []
# 성능 모니터링
self.metrics = {
"total_requests": 0,
"cache_hits": 0,
"avg_latency_ms": 0
}
def add_knowledge(self, documents: list[str], source: str):
"""새로운 지식을 벡터 스토어에 추가"""
embeddings = self.client.embeddings.create(
input=documents,
model=self.embedding_model
)
vectors = np.array([e.embedding for e in embeddings.data]).astype('float32')
start_idx = len(self.metadata)
self.index.add(vectors)
for i, doc in enumerate(documents):
self.metadata.append({
"text": doc,
"source": source,
"added_at": datetime.now().isoformat(),
"access_count": 0
})
return {"added": len(documents), "total": len(self.metadata)}
def search_relevant(self, query: str, top_k: int = 5):
"""사용자 쿼리와 관련된 지식 검색"""
query_embedding = self.client.embeddings.create(
input=[query],
model=self.embedding_model
)
query_vector = np.array([query_embedding.data[0].embedding]).astype('float32')
distances, indices = self.index.search(query_vector, top_k)
results = []
for idx, dist in zip(indices[0], distances[0]):
if idx < len(self.metadata):
self.metadata[idx]["access_count"] += 1
results.append({
**self.metadata[idx],
"relevance_score": float(1 / (1 + dist))
})
return results
def query_with_context(self, user_query: str) -> dict:
"""RAG-enhanced 쿼리 실행"""
start_time = datetime.now()
# 1. 관련 지식 검색
relevant_docs = self.search_relevant(user_query)
# 2. 컨텍스트 구성
context = "\n\n".join([
f"[{d['source']}] {d['text']}"
for d in relevant_docs[:3]
])
# 3. HolySheep AI를 통한 응답 생성
response = self.client.chat.completions.create(
model=self.chat_model,
messages=[
{"role": "system", "content": f"지식 베이스:\n{context}"},
{"role": "user", "content": user_query}
],
temperature=0.7,
max_tokens=500
)
# 4. 성능 지표 기록
latency = (datetime.now() - start_time).total_seconds() * 1000
self.metrics["total_requests"] += 1
self._update_avg_latency(latency)
return {
"response": response.choices[0].message.content,
"sources": [d['source'] for d in relevant_docs],
"latency_ms": round(latency, 2)
}
def _update_avg_latency(self, new_latency: float):
"""지수 이동 평균으로 지연 시간 업데이트"""
n = self.metrics["total_requests"]
current_avg = self.metrics["avg_latency_ms"]
self.metrics["avg_latency_ms"] = (current_avg * (n - 1) + new_latency) / n
def get_learning_insights(self) -> dict:
"""학습 시스템 상태 및 인사이트 반환"""
return {
**self.metrics,
"knowledge_base_size": len(self.metadata),
"top_sources": sorted(
self.metadata,
key=lambda x: x["access_count"],
reverse=True
)[:5]
}
========================================
사용 예시: 이커머스 고객 서비스 Bot
========================================
if __name__ == "__main__":
rag_system = ContinuousLearningRAG(
api_key="YOUR_HOLYSHEEP_API_KEY"
)
# 새 제품 정보 추가 (지속 학습)
new_products = [
"2024 신상: 무선 이어폰 PRO-500, 노이즈 캔슬링 40dB",
"배송 정책 변경: 5만원 이상 무료배송, 익일 배송 가능",
"반품 정책: 30일 이내 무조건 반품 가능"
]
result = rag_system.add_knowledge(new_products, source="product_update_2024")
print(f"지식 추가 완료: {result}")
# 고객 질문 처리
response = rag_system.query_with_context(
"PRO-500 이어폰 배송비 얼마나 들어요?"
)
print(f"\n응답: {response['response']}")
print(f"출처: {response['sources']}")
print(f"지연 시간: {response['latency_ms']}ms")
# 시스템 상태 확인
insights = rag_system.get_learning_insights()
print(f"\n시스템 인사이트: {insights}")
2. Fine-tuning 기반 지속 개선
사용자 피드백 데이터를累积하여 정기적으로 모델을 fine-tuning합니다.
"""
HolySheep AI Fine-tuning 파이프라인
저자: HolySheep AI Technical Writer
"""
import openai
import json
from datetime import datetime, timedelta
from typing import List, Dict
class ContinuousFineTuningPipeline:
"""사용자 피드백 기반 지속적인 모델 fine-tuning"""
def __init__(self, api_key: str):
self.client = openai.OpenAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1"
)
self.feedback_store = []
self.training_jobs = []
# HolySheep AI 가격 정보 (2024년 기준)
self.pricing = {
"gpt-4.1": {"input": 8.00, "output": 8.00}, # $/MTok
"gpt-4.1-mini": {"input": 1.60, "output": 6.40}, # $/MTok
"fine_tune_base": 25.00, # Fine-tuning 비용
}
def collect_feedback(self, conversation_id: str, user_input: str,
assistant_response: str, rating: int,
feedback_text: str = None):
"""사용자 피드백 수집"""
self.feedback_store.append({
"conversation_id": conversation_id,
"user_input": user_input,
"assistant_response": assistant_response,
"rating": rating, # 1-5 스케일
"feedback_text": feedback_text,
"timestamp": datetime.now().isoformat(),
"tokens_estimate": self._estimate_tokens(user_input, assistant_response)
})
def _estimate_tokens(self, input_text: str, output_text: str) -> dict:
"""토큰 수 추정 (대략적 계산)"""
# 한국어: 약 2~3자당 1토큰, 영어: 4자당 1토큰
return {
"input": len(input_text) // 2,
"output": len(output_text) // 2,
"total": (len(input_text) + len(output_text)) // 2
}
def generate_training_data(self, min_rating: int = 4) -> List[dict]:
"""고품질 피드백에서 학습 데이터 생성"""
training_data = []
for item in self.feedback_store:
if item["rating"] >= min_rating:
# OpenAI fine-tuning 형식으로 변환
training_data.append({
"messages": [
{"role": "user", "content": item["user_input"]},
{"role": "assistant", "content": item["assistant_response"]}
]
})
return training_data
def create_fine_tune_job(self, training_data: List[dict]) -> dict:
"""Fine-tuning 작업 생성"""
# 1. 학습 데이터 파일 저장
file_name = f"training_data_{datetime.now().strftime('%Y%m%d_%H%M%S')}.jsonl"
with open(file_name, 'w', encoding='utf-8') as f:
for item in training_data:
f.write(json.dumps(item, ensure_ascii=False) + '\n')
# 2. 파일 업로드
with open(file_name, 'rb') as f:
file_response = self.client.files.create(
file=f,
purpose="fine-tune"
)
# 3. Fine-tuning 작업 생성
job = self.client.fine_tuning.jobs.create(
training_file=file_response.id,
model="gpt-4.1-mini", # 더 작은 모델로 fine-tuning
hyperparameters={
"n_epochs": 3,
"batch_size": 4,
"learning_rate_multiplier": 2
}
)
self.training_jobs.append({
"job_id": job.id,
"created_at": datetime.now().isoformat(),
"training_samples": len(training_data),
"status": "pending"
})
return {
"job_id": job.id,
"status": job.status,
"estimated_cost": self._estimate_tuning_cost(len(training_data))
}
def _estimate_tuning_cost(self, num_samples: int) -> float:
"""Fine-tuning 비용 추정"""
# 평균 샘플 크기 100토큰 가정
avg_tokens_per_sample = 100
total_tokens = num_samples * avg_tokens_per_sample * 3 # 3 epochs
m_tokens = total_tokens / 1_000_000
return m_tokens * self.pricing["fine_tune_base"]
def get_job_status(self, job_id: str) -> dict:
"""Fine-tuning 작업 상태 확인"""
job = self.client.fine_tuning.jobs.retrieve(job_id)
return {
"job_id": job.id,
"status": job.status,
"model": job.fine_tuned_model if hasattr(job, 'fine_tuned_model') else None,
"progress": f"{job.processed_tokens}/{job.train_tokens}" if hasattr(job, 'processed_tokens') else None
}
def estimate_monthly_cost(self, monthly_requests: int,
avg_input_tokens: int,
avg_output_tokens: int) -> dict:
"""월간 비용 추정"""
# HolySheep AI 가격표 기반
gpt_4_1 = self.pricing["gpt-4.1"]
gpt_4_1_mini = self.pricing["gpt-4.1-mini"]
# GPT-4.1 사용 시
gpt_4_1_monthly = (
(monthly_requests * avg_input_tokens / 1_000_000) * gpt_4_1["input"] +
(monthly_requests * avg_output_tokens / 1_000_000) * gpt_4_1["output"]
)
# GPT-4.1-mini 사용 시
gpt_4_1_mini_monthly = (
(monthly_requests * avg_input_tokens / 1_000_000) * gpt_4_1_mini["input"] +
(monthly_requests * avg_output_tokens / 1_000_000) * gpt_4_1_mini["output"]
)
return {
"gpt_4_1": round(gpt_4_1_monthly, 2),
"gpt_4_1_mini": round(gpt_4_1_mini_monthly, 2),
"savings_percentage": round(
(1 - gpt_4_1_mini_monthly / gpt_4_1_monthly) * 100, 1
) if gpt_4_1_monthly > 0 else 0
}
========================================
사용 예시: 기업 RAG 시스템
========================================
if __name__ == "__main__":
pipeline = ContinuousFineTuningPipeline(
api_key="YOUR_HOLYSHEEP_API_KEY"
)
# 사용자 피드백 수집
pipeline.collect_feedback(
conversation_id="conv_001",
user_input="연간 구독 플랜 할인율이 어떻게 되나요?",
assistant_response="연간 구독 시 월간 결제 대비 20% 할인이 적용됩니다...",
rating=5,
feedback_text="정확하고 친절한 답변"
)
# 월간 비용 추정
cost_estimate = pipeline.estimate_monthly_cost(
monthly_requests=100_000,
avg_input_tokens=200,
avg_output_tokens=300
)
print(f"월간 비용 비교:")
print(f" GPT-4.1: ${cost_estimate['gpt_4_1']}")
print(f" GPT-4.1-mini: ${cost_estimate['gpt_4_1_mini']}")
print(f" 절감율: {cost_estimate['savings_percentage']}%")
실시간 피드백 루프 구축
가장 효과적인 지속 학습 전략은 실시간 피드백 루프입니다. HolySheep AI를 활용하면 평균 85ms의 응답 시간으로 빠른 피드백 처리가 가능합니다.
"""
실시간 피드백 루프 시스템
HolySheep AI Multi-Model 통합
"""
import openai
import redis
import threading
from queue import Queue
from datetime import datetime
from collections import defaultdict
class RealtimeFeedbackLoop:
"""실시간 피드백 처리 및 모델 자동 전환"""
def __init__(self, api_key: str):
self.client = openai.OpenAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1"
)
# 모델 설정 (HolySheep AI 지원 모델)
self.models = {
"fast": "gpt-4.1-mini", # 빠른 응답, 일상적 질문
"balanced": "gpt-4.1", # 균형형 응답
"precise": "claude-sonnet-4-20250514", # 정밀한 응답
"budget": "deepseek-chat-v3.2" #低成本 옵션
}
# Redis를 통한 피드백 저장 (로컬 개발: Mock 사용)
self.feedback_queue = Queue()
self.feedback_stats = defaultdict(list)
# 모델 선택기
self.model_selector = self._create_model_selector()
# 피드백 처리 스레드 시작
self.processing_thread = threading.Thread(
target=self._process_feedback_worker,
daemon=True
)
self.processing_thread.start()
def _create_model_selector(self):
"""사용자 피드백 기반 모델 선택기 반환"""
return {
"greeting": "fast",
"product_inquiry": "balanced",
"technical_support": "precise",
"price_inquiry": "budget",
"complex_reasoning": "precise"
}
def detect_intent(self, user_input: str) -> str:
"""사용자 의도 감지"""
response = self.client.chat.completions.create(
model=self.models["balanced"],
messages=[
{"role": "system", "content": "입력된 텍스트의 의도를 분류: greeting, product_inquiry, technical_support, price_inquiry, complex_reasoning 중 하나만 반환"},
{"role": "user", "content": user_input}
],
max_tokens=20,
temperature=0
)
return response.choices[0].message.content.strip().lower()
def route_request(self, user_input: str, explicit_model: str = None) -> str:
"""요청을 적절한 모델로 라우팅"""
if explicit_model and explicit_model in self.models:
return explicit_model
intent = self.detect_intent(user_input)
return self.models.get(self.model_selector.get(intent, "balanced"), "gpt-4.1")
def process_message(self, user_input: str, user_id: str,
explicit_model: str = None) -> dict:
"""메시지 처리 + 피드백 추적"""
start_time = datetime.now()
# 모델 선택
selected_model = self.route_request(user_input, explicit_model)
# HolySheep AI를 통한 응답 생성
response = self.client.chat.completions.create(
model=selected_model,
messages=[{"role": "user", "content": user_input}],
temperature=0.7
)
latency = (datetime.now() - start_time).total_seconds() * 1000
result = {
"response": response.choices[0].message.content,
"model": selected_model,
"latency_ms": round(latency, 2),
"timestamp": datetime.now().isoformat(),
"user_id": user_id,
"tokens": {
"prompt": response.usage.prompt_tokens,
"completion": response.usage.completion_tokens,
"total": response.usage.total_tokens
}
}
# 피드백 큐에 추가
self.feedback_queue.put({
"user_id": user_id,
"input": user_input,
"output": result["response"],
"model": selected_model,
"latency_ms": result["latency_ms"],
"timestamp": result["timestamp"]
})
return result
def submit_feedback(self, user_id: str, message_id: str, rating: int):
"""사용자 피드백 제출"""
self.feedback_queue.put({
"type": "feedback",
"user_id": user_id,
"message_id": message_id,
"rating": rating,
"timestamp": datetime.now().isoformat()
})
def _process_feedback_worker(self):
"""피드백 처리 백그라운드 워커"""
while True:
try:
feedback = self.feedback_queue.get(timeout=1)
if feedback.get("type") == "feedback":
# 실제 피드백 처리
user_id = feedback["user_id"]
self.feedback_stats[user_id].append({
"rating": feedback["rating"],
"timestamp": feedback["timestamp"]
})
else:
# 대화 피드백 추적
self.feedback_stats[feedback["user_id"]].append({
"latency_ms": feedback["latency_ms"],
"model": feedback["model"],
"timestamp": feedback["timestamp"]
})
except Exception:
continue
def get_user_insights(self, user_id: str) -> dict:
"""사용자별 인사이트 반환"""
feedbacks = self.feedback_stats.get(user_id, [])
if not feedbacks:
return {"error": "피드백 데이터 없음"}
ratings = [f.get("rating") for f in feedbacks if "rating" in f]
latencies = [f.get("latency_ms", 0) for f in feedbacks if "latency_ms" in f]
return {
"total_interactions": len(feedbacks),
"avg_rating": round(sum(ratings) / len(ratings), 2) if ratings else None,
"avg_latency_ms": round(sum(latencies) / len(latencies), 2) if latencies else None,
"preferred_model": max(set([f["model"] for f in feedbacks if "model" in f]),
key=lambda x: sum(1 for f in feedbacks if f.get("model") == x)
) if feedbacks else None
}
def get_system_stats(self) -> dict:
"""전체 시스템 통계 반환"""
all_feedbacks = []
for user_feedbacks in self.feedback_stats.values():
all_feedbacks.extend(user_feedbacks)
if not all_feedbacks:
return {"status": "No data yet"}
ratings = [f.get("rating") for f in all_feedbacks if "rating" in f]
latencies = [f.get("latency_ms", 0) for f in all_feedbacks if "latency_ms" in f]
model_usage = defaultdict(int)
for f in all_feedbacks:
if "model" in f:
model_usage[f["model"]] += 1
return {
"total_interactions": len(all_feedbacks),
"unique_users": len(self.feedback_stats),
"avg_rating": round(sum(ratings) / len(ratings), 2) if ratings else None,
"avg_latency_ms": round(sum(latencies) / len(latencies), 2) if latencies else None,
"model_usage": dict(model_usage),
"satisfaction_rate": round(len([r for r in ratings if r >= 4]) / len(ratings) * 100, 1) if ratings else 0
}
========================================
사용 예시: 개인 개발자 프로젝트
========================================
if __name__ == "__main__":
feedback_loop = RealtimeFeedbackLoop(
api_key="YOUR_HOLYSHEEP_API_KEY"
)
# 다양한 의도 테스트
test_queries = [
"안녕하세요!",
"이 제품 배달 어디까지 되나요?",
"리턴 정책이 어떻게 되죠?",
"연산 오류가 발생해요. 로그: Error 0x8007000E",
"구독료가 얼마인가요?"
]
print("=" * 60)
print("실시간 피드백 루프 테스트 결과")
print("=" * 60)
for query in test_queries:
result = feedback_loop.process_message(
user_input=query,
user_id="dev_user_001"
)
print(f"\n[질문] {query}")
print(f" → 모델: {result['model']}")
print(f" → 지연: {result['latency_ms']}ms")
print(f" → 토큰: {result['tokens']}")
# 시스템 통계 확인
import time
time.sleep(2) # 피드백 처리 대기
stats = feedback_loop.get_system_stats()
print("\n" + "=" * 60)
print("시스템 통계")
print("=" * 60)
for key, value in stats.items():
print(f" {key}: {value}")
모델 모니터링 및 자동 알림 시스템
HolySheep AI의 안정적인 API 연결을 활용하여 모델 성능을 실시간 모니터링하는 시스템을 구축합니다.
"""
AI 모델 성능 모니터링 및 자동 알림 시스템
HolySheep AI 게이트웨이 모니터링
"""
import openai
import smtplib
from email.mime.text import MIMEText
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import List, Optional
import statistics
@dataclass
class ModelMetrics:
"""모델 성능 지표"""
model_name: str
request_count: int = 0
error_count: int = 0
total_latency_ms: float = 0.0
latencies: List[float] = field(default_factory=list)
error_types: dict = field(default_factory=dict)
last_request_time: Optional[datetime] = None
@property
def success_rate(self) -> float:
if self.request_count == 0:
return 0.0
return ((self.request_count - self.error_count) / self.request_count) * 100
@property
def avg_latency_ms(self) -> float:
if not self.latencies:
return 0.0
return statistics.mean(self.latencies)
@property
def p95_latency_ms(self) -> float:
if len(self.latencies) < 20:
return self.avg_latency_ms
return statistics.quantiles(self.latencies, n=20)[18] # 95번째 백분위수
@property
def p99_latency_ms(self) -> float:
if len(self.latencies) < 100:
return self.avg_latency_ms
return statistics.quantiles(self.latencies, n=100)[98] # 99번째 백분위수
class AIModelMonitor:
"""AI 모델 성능 모니터링 및 알림 시스템"""
# 성능 임계값 설정
THRESHOLDS = {
"success_rate_min": 95.0, # 최소 성공률 95%
"latency_p95_max": 2000.0, # P95 지연 시간 최대 2초
"error_rate_max": 5.0, # 최대 오류율 5%
"cost_daily_max": 100.0, # 일일 최대 비용 $100
"quota_warning": 80.0, # 쿼터 사용 80% 시 경고
}
def __init__(self, api_key: str, alert_email: str = None):
self.client = openai.OpenAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1"
)
# 모니터링 대상 모델
self.monitored_models = [
"gpt-4.1",
"gpt-4.1-mini",
"claude-sonnet-4-20250514",
"deepseek-chat-v3.2",
"gemini-2.5-flash"
]
# HolySheep AI 가격 정보 (센트 단위)
self.pricing_cents = {
"gpt-4.1": {"input": 800, "output": 800}, # $8/MTok
"gpt-4.1-mini": {"input": 160, "output": 640}, # $1.60/$6.40
"claude-sonnet-4-20250514": {"input": 450, "output": 1800}, # $4.50/$18
"deepseek-chat-v3.2": {"input": 42, "output": 42}, # $0.42/MTok
"gemini-2.5-flash": {"input": 250, "output": 1000}, # $2.50/$10
}
# 메트릭 수집
self.metrics = {model: ModelMetrics(model_name=model)
for model in self.monitored_models}
# 알림 이메일
self.alert_email = alert_email
# 비용 추적
self.daily_costs = {}
self.monthly_quota = 1000.0 # 월간 쿼터 $1000
def record_request(self, model: str, latency_ms: float,
success: bool, error_type: str = None,
input_tokens: int = 0, output_tokens: int = 0):
"""요청 메트릭 기록"""
if model not in self.metrics:
self.metrics[model] = ModelMetrics(model_name=model)
metric = self.metrics[model]
metric.request_count += 1
metric.latencies.append(latency_ms)
metric.total_latency_ms += latency_ms
metric.last_request_time = datetime.now()
if not success:
metric.error_count += 1
if error_type:
metric.error_types[error_type] = metric.error_types.get(error_type, 0) + 1
# 비용 계산
cost_cents = self._calculate_cost(model, input_tokens, output_tokens)
today = datetime.now().date().isoformat()
self.daily_costs[today] = self.daily_costs.get(today, 0) + cost_cents
def _calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
"""토큰 기반 비용 계산 (달러 단위)"""
pricing = self.pricing_cents.get(model, {"input": 0, "output": 0})
input_cost = (input_tokens / 1_000_000) * pricing["input"]
output_cost = (output_tokens / 1_000_000) * pricing["output"]
return input_cost + output_cost
def check_health(self) -> dict:
"""전체 시스템 상태 점검"""
alerts = []
healthy_models = []
for model_name, metric in self.metrics.items():
if metric.request_count == 0:
continue
model_status = {
"model": model_name,
"healthy": True,
"issues": []
}
# 성공률 체크
if metric.success_rate < self.THRESHOLDS["success_rate_min"]:
model_status["healthy"] = False
model_status["issues"].append(
f"성공률 저하: {metric.success_rate:.1f}% (임계값: {self.THRESHOLDS['success_rate_min']}%)"
)
# P95 지연 시간 체크
if metric.p95_latency_ms > self.THRESHOLDS["latency_p95_max"]:
model_status["healthy"] = False
model_status["issues"].append(
f"P95 지연 시간 초과: {metric.p95_latency_ms:.0f}ms (임계값: {self.THRESHOLDS['latency_p95_max']}ms)"
)
if model_status["healthy"]:
healthy_models.append(model_name)
else:
alerts.append(model_status)
# 일일 비용 체크
today_cost = self.daily_costs.get(datetime.now().date().isoformat(), 0)
if today_cost > self.THRESHOLDS["cost_daily_max"]:
alerts.append({
"type": "cost",
"healthy": False,
"issues": [f"일일 비용 초과: ${today_cost:.2f} (임계값: ${self.THRESHOLDS['cost_daily_max']})"]
})
# 월간 쿼터 체크
monthly_usage = sum(self.daily_costs.values())
quota_percentage = (monthly_usage / self.monthly_quota) * 100
if quota_percentage >= self.THRESHOLDS["quota_warning"]:
alerts.append({
"type": "quota",
"healthy": True,
"issues": [f"월간 쿼터 사용량: {quota_percentage:.1f}%"]
})
return {
"timestamp": datetime.now().isoformat(),
"overall_healthy": len([a for a in alerts if not a.get("healthy", True)]) == 0,
"healthy_models": healthy_models,
"alerts": alerts,
"daily_cost": round(today_cost, 2),
"monthly_quota_usage": round(quota_percentage, 1)
}
def get_detailed_report(self) -> dict:
"""상세 성능 보고서 생성"""
report = {
"generated_at": datetime.now().isoformat(),
"models": {}
}
for model_name, metric in self.metrics.items():
if metric.request_count == 0:
continue
report["models"][model_name] = {
"total_requests": metric.request_count,
"error_count": metric.error_count,
"success_rate": round(metric.success_rate, 2),
"avg_latency_ms": round(metric.avg_latency_ms, 2),
"p95_latency_ms": round(metric.p95_latency_ms, 2),
"p99_latency_ms": round(metric.p99_latency_ms, 2),
"top_errors": metric.error_types
}
report["cost_summary"] = {
"today": round(self.daily_costs.get(datetime.now().date().isoformat(), 0), 2),
"this_month": round(sum(self.daily_costs.values()), 2),
"monthly_quota": self.monthly_quota
}
return report
def send_alert(self, alert_type: str, message: str):
"""알림 전송 (이메일)"""
if not self.alert_email:
print(f"[ALERT] {alert_type}: {message}")
return
# 이메일 전송 로직
try:
msg = MIMEText(f"[{alert_type}]\n\n{message}")
msg['Subject'] = f"[HolySheep AI Monitor] {alert_type} Alert"
msg['From'] = '[email protected]'
msg['To'] = self.alert_email
# 실제 이메일 전송 시 활성화
# with smtplib.SMTP('smtp.gmail.com', 587) as server:
# server.starttls()
# server.login('your-email', 'your-password')
# server.send_message(msg)
print(f"[ALERT SENT] {alert_type}: {message}")
except Exception as e:
print(f"[ALERT FAILED] {e}")
========================================
모니터링 데모
========================================
if __name__ == "__main__":
monitor = AIModel