저는 3개월 전 이커머스 플랫폼의 AI 고객 서비스 시스템을 구축한 경험이 있습니다. 블랙프라이데이 특가 행사 중 AI 응답 지연이 5초를 초과하자 고객 이탈률이 급증하는 것을亲眼目击했습니다. 이 사건 이후 AI API 예외 자동 알림 시스템의 중요성을 절실히 깨달었고, 오늘 그 구축 방법을 상세히 공유하겠습니다.
왜 AI API 예외 알림이 필수인가?
AI API는与传统 REST API와 달리:
- 응답 시간 변동성: 모델 로딩, 토큰 생성량에 따라 200ms~30s까지 변동
- 비용 리스크: GPT-4.1은 $8/MTok, 잘못된 루프 시 수백 달러 손실 가능
- 토큰_limits 초과: 순간 트래픽 폭증 시 429 Too Many Requests 빈번
HolySheep AI는 단일 API 키로 모든 주요 모델을 통합管理하므로, 통합 알림 채널 하나로 모든 모델의 예외를 모니터링할 수 있습니다.
Python 기반 AI API 예외 자동 알림 시스템
실제 운영 환경에서 검증된 자동 알림 시스템 구축 방법을 설명드리겠습니다.
# requirements.txt
pip install requests psutil python-dotenv slack-sdk
import requests
import time
import logging
from datetime import datetime
from typing import Optional, Dict, Any
from slack_sdk import WebhookClient
from dataclasses import dataclass, field
import threading
import queue
HolySheep AI 설정
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
@dataclass
class AlertConfig:
"""알림 임계값 설정"""
max_response_time_ms: int = 5000 # 5초 초과 시 알림
max_retry_count: int = 3 # 3회 재시도 후 실패 시 알림
error_rate_threshold: float = 0.05 # 5% 이상 오류율 시 알림
cost_limit_usd: float = 100.0 # 시간당 $100 초과 시 경고
@dataclass
class APIMetrics:
"""API 메트릭 수집"""
total_requests: int = 0
failed_requests: int = 0
total_latency_ms: float = 0.0
total_cost_usd: float = 0.0
recent_errors: list = field(default_factory=list)
last_reset: datetime = field(default_factory=datetime.now)
class AIServiceMonitor:
"""AI API 서비스 모니터링 및 알림"""
def __init__(self, webhook_url: str, config: AlertConfig):
self.slack = WebhookClient(webhook_url)
self.config = config
self.metrics = APIMetrics()
self.alert_queue = queue.Queue()
self._start_alert_worker()
def _start_alert_worker(self):
"""비동기 알림 워커 스레드"""
def worker():
while True:
alert = self.alert_queue.get()
if alert:
self._send_slack_alert(alert)
time.sleep(1)
threading.Thread(target=worker, daemon=True).start()
def _calculate_cost(self, model: str, prompt_tokens: int,
completion_tokens: int) -> float:
"""토큰 기반 비용 계산"""
pricing = {
"gpt-4.1": {"prompt": 8.0, "completion": 8.0}, # $8/MTok
"claude-sonnet-4": {"prompt": 15.0, "completion": 15.0}, # $15/MTok
"gemini-2.5-flash": {"prompt": 2.5, "completion": 2.5}, # $2.50/MTok
"deepseek-v3": {"prompt": 0.42, "completion": 1.68}, # $0.42/$1.68 per MTok
}
model_key = model.lower().replace("-flash", "-flash")
rates = pricing.get(model_key, pricing["gpt-4.1"])
prompt_cost = (prompt_tokens / 1_000_000) * rates["prompt"]
completion_cost = (completion_tokens / 1_000_000) * rates["completion"]
return prompt_cost + completion_cost
def _send_slack_alert(self, alert: Dict[str, Any]):
"""Slack 알림 전송"""
severity_emoji = {
"critical": "🚨",
"warning": "⚠️",
"info": "ℹ️"
}
emoji = severity_emoji.get(alert.get("severity", "info"), "ℹ️")
message = f"""
{emoji} *AI API Alert - {alert['severity'].upper()}*
*시간:* {alert['timestamp']}
*모델:* {alert.get('model', 'N/A')}
*문제:* {alert['message']}
*현재 상태:*
• 총 요청: {self.metrics.total_requests}
• 실패율: {(self.metrics.failed_requests/max(self.metrics.total_requests,1))*100:.2f}%
• 누적 비용: ${self.metrics.total_cost_usd:.4f}
"""
try:
response = self.slack.send(text=message.strip())
logging.info(f"Alert sent: {alert['severity']}")
except Exception as e:
logging.error(f"Failed to send alert: {e}")
def call_ai_api(self, model: str, messages: list,
temperature: float = 0.7) -> Dict[str, Any]:
"""HolySheep AI API 호출 및 모니터링"""
start_time = time.time()
self.metrics.total_requests += 1
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature
}
for attempt in range(self.config.max_retry_count):
try:
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
latency_ms = (time.time() - start_time) * 1000
if response.status_code == 200:
data = response.json()
usage = data.get("usage", {})
# 비용 계산
cost = self._calculate_cost(
model,
usage.get("prompt_tokens", 0),
usage.get("completion_tokens", 0)
)
self.metrics.total_cost_usd += cost
self.metrics.total_latency_ms += latency_ms
# 지연 시간 알림 체크
if latency_ms > self.config.max_response_time_ms:
self.alert_queue.put({
"severity": "warning",
"timestamp": datetime.now().isoformat(),
"model": model,
"message": f"응답 지연 초과: {latency_ms:.0f}ms (임계값: {self.config.max_response_time_ms}ms)"
})
return {"success": True, "data": data, "latency_ms": latency_ms}
elif response.status_code == 429:
self.metrics.failed_requests += 1
self.metrics.recent_errors.append({
"time": datetime.now().isoformat(),
"code": 429,
"message": "Rate limit exceeded"
})
self.alert_queue.put({
"severity": "warning",
"timestamp": datetime.now().isoformat(),
"model": model,
"message": f"Rate Limit (429) - 재시도 {attempt + 1}/{self.config.max_retry_count}"
})
if attempt < self.config.max_retry_count - 1:
time.sleep(2 ** attempt) # 지수 백오프
continue
elif response.status_code == 500:
self.metrics.failed_requests += 1
self.metrics.recent_errors.append({
"time": datetime.now().isoformat(),
"code": 500,
"message": "Internal Server Error"
})
if attempt == self.config.max_retry_count - 1:
self.alert_queue.put({
"severity": "critical",
"timestamp": datetime.now().isoformat(),
"model": model,
"message": f"API Server Error (500) - {self.config.max_retry_count}회 재시도 실패"
})
else:
raise Exception(f"API Error: {response.status_code}")
except requests.exceptions.Timeout:
self.metrics.failed_requests += 1
self.metrics.recent_errors.append({
"time": datetime.now().isoformat(),
"type": "timeout",
"message": "Request timeout (>30s)"
})
self.alert_queue.put({
"severity": "critical",
"timestamp": datetime.now().isoformat(),
"model": model,
"message": "요청 타임아웃 (30초 초과)"
})
except requests.exceptions.ConnectionError as e:
self.metrics.failed_requests += 1
self.alert_queue.put({
"severity": "critical",
"timestamp": datetime.now().isoformat(),
"model": model,
"message": f"연결 오류: {str(e)[:100]}"
})
except Exception as e:
self.metrics.failed_requests += 1
self.alert_queue.put({
"severity": "warning",
"timestamp": datetime.now().isoformat(),
"model": model,
"message": f"예상치 못한 오류: {str(e)[:100]}"
})
return {"success": False, "error": "Max retries exceeded"}
def check_cost_alert(self):
"""비용 임계값 체크"""
if self.metrics.total_cost_usd > self.config.cost_limit_usd:
self.alert_queue.put({
"severity": "warning",
"timestamp": datetime.now().isoformat(),
"message": f"비용 임계값 초과: ${self.metrics.total_cost_usd:.2f} (제한: ${self.config.cost_limit_usd})"
})
self.reset_metrics()
def reset_metrics(self):
"""메트릭 초기화"""
self.metrics = APIMetrics()
logging.info("Metrics reset")
사용 예제
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
# HolySheep AI에 가입하여 webhook URL获取
monitor = AIServiceMonitor(
webhook_url="YOUR_SLACK_WEBHOOK_URL",
config=AlertConfig(
max_response_time_ms=5000,
max_retry_count=3,
error_rate_threshold=0.05,
cost_limit_usd=50.0
)
)
# 이커머스 고객 문의 응답
result = monitor.call_ai_api(
model="gpt-4.1",
messages=[
{"role": "system", "content": "당신은 이커머스 고객 서비스 챗봇입니다."},
{"role": "user", "content": "주문 배송 상태를 조회해주세요. 주문번호: ORD-12345"}
]
)
if result["success"]:
print(f"응답 시간: {result['latency_ms']:.0f}ms")
print(f"답변: {result['data']['choices'][0]['message']['content']}")
Node.js + Express 기반 실시간 알림 시스템
실시간 웹 애플리케이션 환경에서 WebSocket을 활용한 실시간 알림 대시보드 구축 방법입니다.
# package.json dependencies
npm install express ws axios dotenv prom-client @slack/webhook
const express = require('express');
const { WebSocketServer } = require('ws');
const axios = require('axios');
const { IncomingWebhook } = require('@slack/webhook');
const { Registry, Counter, Histogram, Gauge } = require('prom-client');
const app = express();
const PORT = 3000;
// HolySheep AI 설정
const BASE_URL = 'https://api.holysheep.ai/v1';
const API_KEY = process.env.HOLYSHEEP_API_KEY;
// Prometheus 메트릭 설정
const register = new Registry();
const apiRequestsTotal = new Counter({
name: 'ai_api_requests_total',
help: 'Total AI API requests',
labelNames: ['model', 'status'],
registers: [register]
});
const apiLatencyHistogram = new Histogram({
name: 'ai_api_latency_ms',
help: 'AI API latency in milliseconds',
buckets: [100, 500, 1000, 2000, 5000, 10000, 30000],
registers: [register]
});
const apiCostGauge = new Gauge({
name: 'ai_api_cost_usd',
help: 'Accumulated API cost in USD',
registers: [register]
});
// 알림 상태 관리
const alertState = {
recentErrors: [],
errorRates: [],
lastAlertTime: {},
alertingCooldown: 5 * 60 * 1000, // 5분 쿨다운
thresholds: {
maxLatency: 5000,
maxErrorRate: 0.05,
maxCostPerHour: 100,
maxConsecutiveErrors: 3
}
};
// Slack Webhook 초기화
const slackWebhook = new IncomingWebhook(process.env.SLACK_WEBHOOK_URL);
// WebSocket 클라이언트 관리
const wsClients = new Set();
function initWebSocket(server) {
const wss = new WebSocketServer({ server });
wss.on('connection', (ws) => {
console.log('Client connected to alert stream');
wsClients.add(ws);
ws.on('close', () => {
wsClients.delete(ws);
});
// 초기 상태 전송
ws.send(JSON.stringify({
type: 'initial_state',
data: getCurrentMetrics()
}));
});
return wss;
}
function broadcastAlert(alert) {
const message = JSON.stringify(alert);
wsClients.forEach(client => {
if (client.readyState === 1) { // WebSocket.OPEN
client.send(message);
}
});
}
async function sendSlackAlert(alert) {
try {
await slackWebhook.send({
text: ${alert.emoji} *AI API Alert - ${alert.severity.toUpperCase()}*,
blocks: [
{
type: 'section',
text: {
type: 'mrkdwn',
text: *${alert.message}*\n\n⏰ 시간: ${alert.timestamp}\n🤖 모델: ${alert.model || 'N/A'}
}
},
{
type: 'section',
fields: [
{ type: 'mrkdwn', text: *총 요청:* ${alert.metrics.totalRequests} },
{ type: 'mrkdwn', text: *실패율:* ${alert.metrics.errorRate}% },
{ type: 'mrkdwn', text: *평균 지연:* ${alert.metrics.avgLatency}ms },
{ type: 'mrkdwn', text: *누적 비용:* $${alert.metrics.totalCost} }
]
}
]
});
} catch (error) {
console.error('Slack webhook error:', error);
}
}
function shouldSendAlert(alertType) {
const now = Date.now();
const lastAlert = alertState.lastAlertTime[alertType] || 0;
return now - lastAlert > alertState.alertingCooldown;
}
function recordAlert(alertType) {
alertState.lastAlertTime[alertType] = Date.now();
}
function checkAndAlert(alertType, severity, message, model = null) {
if (!shouldSendAlert(alertType)) return;
recordAlert(alertType);
const emoji = severity === 'critical' ? '🚨' : severity === 'warning' ? '⚠️' : 'ℹ️';
const alert = {
type: 'alert',
severity,
message,
model,
timestamp: new Date().toISOString(),
emoji,
metrics: getCurrentMetrics()
};
broadcastAlert(alert);
sendSlackAlert(alert);
}
function getCurrentMetrics() {
return {
totalRequests: apiRequestsTotal.hashMap || 0,
errorRate: alertState.errorRates.length > 0
? (alertState.errorRates.reduce((a, b) => a + b, 0) / alertState.errorRates.length * 100).toFixed(2)
: '0.00',
avgLatency: 0,
totalCost: parseFloat(apiCostGauge.hashMap || 0).toFixed(4)
};
}
// HolySheep AI API 호출 미들웨어
async function aiProxy(req, res) {
const startTime = Date.now();
const { model, messages, temperature = 0.7, max_tokens = 1000 } = req.body;
try {
const response = await axios.post(
${BASE_URL}/chat/completions,
{
model,
messages,
temperature,
max_tokens
},
{
headers: {
'Authorization': Bearer ${API_KEY},
'Content-Type': 'application/json'
},
timeout: 30000
}
);
const latency = Date.now() - startTime;
// Prometheus 메트릭 업데이트
apiRequestsTotal.inc({ model, status: 'success' });
apiLatencyHistogram.observe({ model }, latency);
// 비용 계산 (간단한 추정)
const usage = response.data.usage || {};
const estimatedCost = calculateCost(model, usage);
apiCostGauge.add(estimatedCost);
// 지연 시간 알림 체크
if (latency > alertState.thresholds.maxLatency) {
checkAndAlert(
'high_latency',
'warning',
응답 지연 초과: ${latency}ms (임계값: ${alertState.thresholds.maxLatency}ms),
model
);
}
res.json({
success: true,
data: response.data,
latency_ms: latency,
estimated_cost_usd: estimatedCost
});
} catch (error) {
const latency = Date.now() - startTime;
apiRequestsTotal.inc({ model: req.body.model, status: 'error' });
alertState.recentErrors.push({ time: Date.now(), latency });
alertState.errorRates.push(1);
// 최근 100개 기준 오류율 계산
if (alertState.errorRates.length > 100) {
alertState.errorRates.shift();
}
// 연속 오류 체크
const recentErrors = alertState.recentErrors.filter(
e => Date.now() - e.time < 60000
);
if (recentErrors.length >= alertState.thresholds.maxConsecutiveErrors) {
checkAndAlert(
'consecutive_errors',
'critical',
연속 오류 발생: ${recentErrors.length}회 (최근 1분 내),
req.body.model
);
}
// 오류율 체크
const errorRate = alertState.errorRates.reduce((a, b) => a + b, 0) / alertState.errorRates.length;
if (errorRate > alertState.thresholds.maxErrorRate) {
checkAndAlert(
'high_error_rate',
'warning',
높은 오류율: ${(errorRate * 100).toFixed(1)}% (임계값: ${alertState.thresholds.maxErrorRate * 100}%),
req.body.model
);
}
const errorMessage = error.response?.data?.error?.message || error.message;
const statusCode = error.response?.status || 500;
res.status(statusCode).json({
success: false,
error: errorMessage,
latency_ms: latency,
retryable: [429, 500, 502, 503, 504].includes(statusCode)
});
}
}
function calculateCost(model, usage) {
const pricing = {
'gpt-4.1': { prompt: 8, completion: 8 },
'claude-sonnet-4': { prompt: 15, completion: 15 },
'gemini-2.5-flash': { prompt: 2.5, completion: 2.5 },
'deepseek-v3': { prompt: 0.42, completion: 1.68 }
};
const rates = pricing[model] || pricing['gpt-4.1'];
const promptCost = (usage.prompt_tokens || 0) / 1000000 * rates.prompt;
const completionCost = (usage.completion_tokens || 0) / 1000000 * rates.completion;
return promptCost + completionCost;
}
// API Routes
app.use(express.json());
app.post('/api/chat', aiProxy);
app.get('/api/metrics', async (req, res) => {
res.set('Content-Type', register.contentType);
res.send(await register.metrics());
});
app.get('/api/alerts/current', (req, res) => {
res.json({
recentErrors: alertState.recentErrors.slice(-10),
errorRate: alertState.errorRates.length > 0
? (alertState.errorRates.reduce((a, b) => a + b, 0) / alertState.errorRates.length * 100).toFixed(2)
: '0.00',
thresholds: alertState.thresholds
});
});
// 1시간마다 비용 리셋 (실제 운영에서는 DB 연동 권장)
setInterval(() => {
apiCostGauge.set(0);
alertState.errorRates = [];
console.log('Hourly metrics reset');
}, 60 * 60 * 1000);
const server = app.listen(PORT, () => {
console.log(AI API Proxy running on port ${PORT});
initWebSocket(server);
});
실전 모니터링 대시보드 구축
위 코드로 수집된 데이터를 시각화하는 간단한 대시보드 구성 방법입니다.
<!-- dashboard.html -->
<!DOCTYPE html>
<html lang="ko">
<head>
<meta charset="UTF-8">
<title>AI API Monitoring Dashboard</title>
<script src="https://cdn.tailwindcss.com"></script>
<script src="https://cdn.jsdelivr.net/npm/chart.js"></script>
<style>
.alert-critical { background: #fee2e2; border-left: 4px solid #ef4444; }
.alert-warning { background: #fef3c7; border-left: 4px solid #f59e0b; }
.alert-info { background: #dbeafe; border-left: 4px solid #3b82f6; }
</style>
</head>
<body class="bg-gray-100 p-6">
<div class="max-w-7xl mx-auto">
<h1 class="text-3xl font-bold mb-6">🤖 AI API Monitoring Dashboard</h1>
<!-- 실시간 메트릭 카드 -->
<div class="grid grid-cols-1 md:grid-cols-4 gap-4 mb-6">
<div class="bg-white rounded-lg shadow p-6">
<h3 class="text-gray-500 text-sm">총 요청 수</h3>
<p id="totalRequests" class="text-3xl font-bold">0</p>
</div>
<div class="bg-white rounded-lg shadow p-6">
<h3 class="text-gray-500 text-sm">평균 응답 시간</h3>
<p id="avgLatency" class="text-3xl font-bold">0ms</p>
</div>
<div class="bg-white rounded-lg shadow p-6">
<h3 class="text-gray-500 text-sm">오류율</h3>
<p id="errorRate" class="text-3xl font-bold text-red-500">0%</p>
</div>
<div class="bg-white rounded-lg shadow p-6">
<h3 class="text-gray-500 text-sm">누적 비용</h3>
<p id="totalCost" class="text-3xl font-bold text-green-500">$0.00</p>
</div>
</div>
<!-- 실시간 알림 로그 -->
<div class="grid grid-cols-1 lg:grid-cols-2 gap-6">
<div class="bg-white rounded-lg shadow p-6">
<h2 class="text-xl font-bold mb-4">🚨 실시간 알림</h2>
<div id="alerts" class="space-y-3 max-h-96 overflow-y-auto">
<p class="text-gray-500">아직 알림이 없습니다...</p>
</div>
</div>
<div class="bg-white rounded-lg shadow p-6">
<h2 class="text-xl font-bold mb-4">📊 응답 시간 분포</h2>
<canvas id="latencyChart"></canvas>
</div>
</div>
<!-- Prometheus 메트릭 링크 -->
<div class="mt-6 bg-white rounded-lg shadow p-6">
<a href="/api/metrics" class="text-blue-500 hover:underline">
📈 Prometheus 메트릭 확인 (/api/metrics)
</a>
</div>
</div>
<script>
const ws = new WebSocket('ws://' + location.host);
const alertsDiv = document.getElementById('alerts');
const latencyData = [];
ws.onmessage = (event) => {
const alert = JSON.parse(event.data);
if (alert.type === 'alert') {
addAlertToLog(alert);
updateMetrics(alert.metrics);
} else if (alert.type === 'initial_state') {
updateMetrics(alert.data);
}
};
function addAlertToLog(alert) {
const alertClass = alert-${alert.severity};
const emoji = alert.severity === 'critical' ? '🚨' :
alert.severity === 'warning' ? '⚠️' : 'ℹ️';
const alertHtml = `
<div class="p-4 rounded ${alertClass}">
<div class="flex items-center justify-between">
<span>${emoji} ${alert.message}</span>
<span class="text-sm text-gray-500">${new Date(alert.timestamp).toLocaleTimeString()}</span>
</div>
${alert.model ? <p class="text-sm mt-1">모델: ${alert.model}</p> : ''}
</div>
`;
alertsDiv.innerHTML = alertHtml + alertsDiv.innerHTML;
// 최대 20개 알림만 표시
if (alertsDiv.children.length > 20) {
alertsDiv.lastChild.remove();
}
}
function updateMetrics(data) {
document.getElementById('totalRequests').textContent = data.totalRequests || 0;
document.getElementById('avgLatency').textContent = (data.avgLatency || 0) + 'ms';
document.getElementById('errorRate').textContent = data.errorRate + '%';
document.getElementById('totalCost').textContent = '$' + (data.totalCost || 0);
}
// Chart.js 설정
const ctx = document.getElementById('latencyChart').getContext('2d');
new Chart(ctx, {
type: 'line',
data: {
labels: [],
datasets: [{
label: '응답 시간 (ms)',
data: [],
borderColor: '#3b82f6',
tension: 0.1
}]
},
options: {
responsive: true,
scales: {
y: { beginAtZero: true }
}
}
});
</script>
</body>
</html>
자주 발생하는 오류와 해결책
1. Rate Limit (429) 연속 발생
# 문제: HolySheep AI Rate Limit 초과로 모든 요청이 실패
해결: 지수 백오프 + 모델 폴백 전략
class ModelFallbackHandler:
def __init__(self):
self.models_priority = [
("gpt-4.1", 1.0), # 주력: 최고 품질
("claude-sonnet-4", 1.0), # 폴백 1
("gemini-2.5-flash", 0.5), # 폴백 2: 비용 절감
("deepseek-v3", 0.1) # 폴백 3: 초저비용
]
self.rate_limit_counts = {}
self.cooldown_seconds = 60
def get_available_model(self):
"""Rate limit 상태가 아닌 첫 번째 모델 반환"""
current_time = time.time()
for model, priority in self.models_priority:
last_failed = self.rate_limit_counts.get(model, 0)
# 쿨다운 기간 중이면 건너뜀
if current_time - last_failed < self.cooldown_seconds:
continue
return model
# 모든 모델이 쿨다운이면 cheapest 모델 반환
return self.models_priority[-1][0]
def mark_rate_limited(self, model):
"""Rate limit 발생 모델 기록"""
self.rate_limit_counts[model] = time.time()
print(f"[ALERT] Model {model} rate limited, switching fallback")
def call_with_fallback(self, messages, max_retries=5):
"""폴백策略으로 API 호출"""
for attempt in range(max_retries):
model = self.get_available_model()
try:
result = monitor.call_ai_api(model, messages)
if result.get("success"):
return result
error_code = result.get("error", {}).get("code")
if error_code == 429:
self.mark_rate_limited(model)
# 지수 백오프
wait_time = min(2 ** attempt, 30)
print(f"Waiting {wait_time}s before retry...")
time.sleep(wait_time)
except Exception as e:
if "rate limit" in str(e).lower():
self.mark_rate_limited(model)
return {"success": False, "error": "All models exhausted"}
2. 토큰 Limits 초과로 인한 트렁케이션
# 문제: 프롬프트가 너무 길어 응답이 잘림
해결: 토큰 카운팅 및 자동 프롬프트 압축
def count_tokens(text: str, model: str = "gpt-4") -> int:
"""대략적인 토큰 수 계산 (한글: 2자 ≈ 1토큰, 영어: 4자 ≈ 1토큰)"""
import re
# 한글字符 처리
korean_chars = len(re.findall(r'[가-힣]', text))
# 영어 및 기타 문자
other_chars = len(re.findall(r'[a-zA-Z0-9\s]', text))
# 추정 토큰: 한글 2자=1토큰, 영어 4자=1토큰
return int(korean_chars / 2 + other_chars / 4)
def truncate_conversation(messages: list, max_tokens: int = 100000,
model: str = "gpt-4.1") -> list:
"""대화 기록 자동 압축 (시스템 프롬프트 제외)"""
MAX_CONTEXT = {
"gpt-4.1": 128000,
"claude-sonnet-4": 200000,
"gemini-2.5-flash": 1000000,
"deepseek-v3": 64000
}
limit = MAX_CONTEXT.get(model, 128000) - max_tokens - 2000 # 안전 마진
total_tokens = 0
truncated_messages = []
# 시스템 메시지는 항상 유지
for msg in messages:
if msg["role"] == "system":
tokens = count_tokens(msg["content"])
total_tokens += tokens
truncated_messages.append(msg)
# 최근 메시지부터 추가 (FIFO)
user_assistant = [m for m in messages if m["role"] != "system"]
user_assistant.reverse()
for msg in user_assistant:
tokens = count_tokens(msg["content"])
if total_tokens + tokens <= limit:
truncated_messages.insert(0, msg)
total_tokens += tokens
else:
# 최대 길이로 자르기
remaining = limit - total_tokens
if remaining > 100: # 최소 100토큰
msg_copy = msg.copy()
# 대략적인 문자 수 계산
char_limit = int(remaining * 2.5) # 토큰→문자 변환
msg_copy["content"] = msg["content"][:char_limit] + "...[압축됨]"
truncated_messages.insert(0, msg_copy)
break
return truncated_messages
3. 연결 시간 초과 및 타임아웃 처리
# 문제: 네트워크 불안정으로 인한 연결 실패
해결: 재시도 로직 + 상태 검사
import socket
from functools import wraps
import httpx
def health_check_and_retry(func):
"""헬스 체크 + 자동 재시도 데코레이터"""
@wraps(func)
def wrapper(*args, **kwargs):
max_attempts = 3
base_timeout = 10
for attempt in range(max_attempts):
try:
# HolySheep AI 헬스 체크
response = httpx.get(
"https://api