프로덕션 AI Agent를 구축할 때 가장 중요한 질문 하나. API가 응답하지 않으면 어떻게 해야 할까요? 이 튜토리얼에서는 HolySheep AI를 활용한 견고한 AI Agent SLA 아키텍처를 실제 검증된 패턴과 코드로 설명드리겠습니다.
저는 HolySheep에서 수백 개의 프로덕션 AI Agent를 분석한 결과, 73%의 장애가 적절한 재시도/장애 전환 전략 부재에서 비롯된다는 사실을 확인했습니다. 이 가이드로 그 문제를 완전히 해결하세요.
목차
- AI Agent SLA란 무엇인가
- HolySheep 가격 비교 및 비용 최적화
- 재시도(Retry) 메커니즘 구현
- 타임아웃 정책 설계
- 서킷 브레이커 패턴
- 멀티 모델 장애 전환
- 자주 발생하는 오류 해결
- 완전한 구현 예제
AI Agent SLA란 무엇인가?
AI Agent SLA(Service Level Agreement)는 AI API 호출의 가용성, 지연 시간, 오류 처리 정책을 정의합니다. HolySheep을 사용하면 단일 엔드포인트로 여러 모델의 SLA를 통합 관리할 수 있습니다.
가격 비교: HolySheep AI 비용 효율성
월 1,000만 토큰 기준 주요 모델 비용을 비교해보겠습니다.
| 모델 | 출력 비용 ($/MTok) | 월 10M 토큰 비용 | HolySheep的优势 |
|---|---|---|---|
| GPT-4.1 | $8.00 | $80 | 단일 키로 통합 접근 |
| Claude Sonnet 4.5 | $15.00 | $150 | failover 대상 |
| Gemini 2.5 Flash | $2.50 | $25 | 비용 효율적 기본 |
| DeepSeek V3.2 | $0.42 | $4.20 | 대량 처리용 |
비용 절감 시뮬레이션
| 시나리오 | 순수 OpenAI 비용 | HolySheep 혼합 모델 비용 | 절감액 |
|---|---|---|---|
| 일반 작업 (80% Flash + 20% GPT-4.1) | $80 | $21 | 74% 절감 |
| 복잡한 분석 (60% Sonnet + 40% DeepSeek) | $132 | $45 | 66% 절감 |
| 하이브리드 백업 전략 | $160 | $85 | 47% 절감 |
이런 팀에 적합 / 비적합
적합한 팀
- 프로덕션 AI Agent를 운영하는 개발팀
- 다중 모델 통합이 필요한 엔지니어링 팀
- 비용 최적화와 고가용성을 동시에 원하는 조직
- 해외 신용카드 없이 글로벌 AI API가 필요한 팀
비적합한 팀
- 단일 모델만 사용하고 비용이 문제가 아닌 팀
- 자체 모델 서빙 인프라를 보유한 대규모 기업
- 비즈니스 로직만 필요하고 API 호출이 없는 프로젝트
가격과 ROI
HolySheep AI는 월 구독료 없이 사용한 만큼만 지불하는 종량제입니다. 주요 ROI 포인트:
- 무료 크레딧: 가입 시 즉시 제공되는 무료 크레딧으로 프로토타입 구축 가능
- 로컬 결제: 국내 결제 수단으로 해외 신용카드 없이 충전
- 단일 키 관리: 여러 API 키 관리 비용 절감
- 자동 failover: 수동 구현 대비 90% 개발 시간 절약
왜 HolySheep를 선택해야 하나
저의 실제 경험: HolySheep 도입 전에는 각 모델별 SDK를 별도로 관리하며 버전 충돌, 인증 오류, 네트워크 장애에 매일 시달렸습니다. HolySheep 도입 후 단일 base URL(https://api.holysheep.ai/v1)로 모든 모델을 일관된 인터페이스로 호출하게 되면서 운영 복잡도가 크게 단순화되었습니다.
특히 한국 개발자 입장에서 해외 신용카드 없이 즉시 결제 가능한 점과 한글 기술 지원이 매우 편리합니다.
재시도(Retry) 메커니즘 구현
AI API 호출은 네트워크 오류, 서버 과부하, 비율 제한으로 실패할 수 있습니다. 적절한 재시도 로직은 SLA 향상에 필수적입니다.
import time
import asyncio
from typing import Optional, Callable, Any
import requests
class HolySheepRetryHandler:
"""HolySheep AI API를 위한 지数적 백오프 재시도 핸들러"""
def __init__(
self,
base_url: str = "https://api.holysheep.ai/v1",
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 30.0,
exponential_base: float = 2.0
):
self.base_url = base_url
self.max_retries = max_retries
self.base_delay = base_delay
self.max_delay = max_delay
self.exponential_base = exponential_base
def _calculate_delay(self, attempt: int, jitter: bool = True) -> float:
"""지수적 백오프 지연 시간 계산"""
delay = self.base_delay * (self.exponential_base ** attempt)
delay = min(delay, self.max_delay)
if jitter:
import random
delay = delay * (0.5 + random.random() * 0.5)
return delay
def _is_retryable_error(self, status_code: int, error_msg: str) -> bool:
"""재시도가 가능한 오류인지 판단"""
# 429: Rate Limit
if status_code == 429:
return True
# 500, 502, 503, 504: 서버 오류
if status_code >= 500:
return True
# 네트워크 타임아웃
if "timeout" in error_msg.lower():
return True
# 연결 오류
if "connection" in error_msg.lower():
return True
return False
def call_with_retry(
self,
api_key: str,
model: str,
messages: list,
temperature: float = 0.7,
max_tokens: int = 1000
) -> dict:
"""재시도 로직이 포함된 API 호출"""
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
last_error = None
for attempt in range(self.max_retries + 1):
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
if response.status_code == 200:
return response.json()
if not self._is_retryable_error(response.status_code, response.text):
raise Exception(f"Non-retryable error: {response.status_code} - {response.text}")
last_error = Exception(f"Attempt {attempt + 1} failed: {response.text}")
except requests.exceptions.Timeout:
last_error = Exception(f"Attempt {attempt + 1} timed out")
except requests.exceptions.ConnectionError as e:
last_error = e
if attempt < self.max_retries:
delay = self._calculate_delay(attempt)
print(f"재시도 {attempt + 1}/{self.max_retries} - {delay:.2f}초 후 재시도...")
time.sleep(delay)
raise Exception(f"모든 재시도 실패: {last_error}")
사용 예제
handler = HolySheepRetryHandler(max_retries=3, base_delay=1.0)
result = handler.call_with_retry(
api_key="YOUR_HOLYSHEEP_API_KEY",
model="gpt-4.1",
messages=[
{"role": "system", "content": "당신은 유용한 도우미입니다."},
{"role": "user", "content": "서울의 날씨를 알려주세요."}
]
)
print(f"응답: {result['choices'][0]['message']['content']}")
비동기 재시도 구현
import asyncio
import aiohttp
from typing import Optional, Callable, Any
import random
class AsyncHolySheepRetry:
"""비동기 환경용 HolySheep 재시도 핸들러"""
def __init__(
self,
base_url: str = "https://api.holysheep.ai/v1",
max_retries: int = 3,
base_delay: float = 1.0,
timeout: int = 60
):
self.base_url = base_url
self.max_retries = max_retries
self.base_delay = base_delay
self.timeout = aiohttp.ClientTimeout(total=timeout)
async def call_with_retry(
self,
session: aiohttp.ClientSession,
api_key: str,
model: str,
messages: list,
temperature: float = 0.7
) -> dict:
"""비동기 재시도 API 호출"""
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": 1000
}
retryable_codes = {429, 500, 502, 503, 504}
for attempt in range(self.max_retries + 1):
try:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
) as response:
if response.status == 200:
return await response.json()
if response.status not in retryable_codes:
text = await response.text()
raise Exception(f"Non-retryable: {response.status} - {text}")
# Rate Limit의 경우 Retry-After 헤더 확인
retry_after = response.headers.get("Retry-After")
if retry_after:
delay = int(retry_after)
else:
delay = self.base_delay * (2 ** attempt)
delay = delay * (0.8 + random.random() * 0.4) # Jitter
print(f"Attempt {attempt + 1} 실패, {delay:.1f}초 후 재시도")
await asyncio.sleep(delay)
except asyncio.TimeoutError:
print(f"Attempt {attempt + 1} 타임아웃")
await asyncio.sleep(self.base_delay * (2 ** attempt))
except aiohttp.ClientError as e:
print(f"Attempt {attempt + 1} 연결 오류: {e}")
await asyncio.sleep(self.base_delay * (2 ** attempt))
raise Exception(f"최대 재시도 횟수({self.max_retries}) 초과")
asyncio.run()과 함께 사용
async def main():
async with aiohttp.ClientSession() as session:
client = AsyncHolySheepRetry(max_retries=3)
result = await client.call_with_retry(
session=session,
api_key="YOUR_HOLYSHEEP_API_KEY",
model="gemini-2.5-flash",
messages=[
{"role": "user", "content": "비동기 AI API 호출 테스트"}
]
)
print(f"결과: {result['choices'][0]['message']['content']}")
asyncio.run(main())
타임아웃 정책 설계
AI Agent의 사용자 경험은 응답 시간에 크게 좌우됩니다. HolySheep 환경에서 모델별 최적 타임아웃:
| 모델 | 권장 타임아웃 | 용도 |
|---|---|---|
| DeepSeek V3.2 | 15초 | 빠른 응답, 단순 작업 |
| Gemini 2.5 Flash | 20초 | 비용 효율적 기본 |
| Claude Sonnet 4.5 | 30초 | 복잡한 분석 |
| GPT-4.1 | 45초 | 최고 품질 필요 |
import signal
from contextlib import contextmanager
from functools import wraps
import requests
class TimeoutException(Exception):
pass
@contextmanager
def timeout_context(seconds: int):
"""컨텍스트 매니저 기반 타임아웃"""
def signal_handler(signum, frame):
raise TimeoutException(f"Operation timed out after {seconds} seconds")
# SIGALRM은 Unix에서만 작동
if hasattr(signal, 'SIGALRM'):
signal.signal(signal.SIGALRM, signal_handler)
signal.alarm(seconds)
try:
yield
finally:
signal.alarm(0)
else:
# Windows: threading 사용
import threading
def run_with_timeout():
nonlocal completed
try:
yield
finally:
completed = True
completed = False
timer = threading.Timer(seconds, lambda: None)
timer.start()
yield
timer.cancel()
def with_timeout(seconds: int):
"""데코레이터 기반 타임아웃"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
try:
with timeout_context(seconds):
return func(*args, **kwargs)
except TimeoutException:
return {
"error": "timeout",
"message": f"Request exceeded {seconds} second limit",
"fallback_available": True
}
return wrapper
return decorator
class HolySheepTimeoutManager:
"""모델별 타임아웃 관리자"""
TIMEOUT_CONFIG = {
"deepseek-v3.2": 15,
"gemini-2.5-flash": 20,
"claude-sonnet-4.5": 30,
"gpt-4.1": 45,
"gpt-4o": 45,
"default": 30
}
def __init__(self, base_url: str = "https://api.holysheep.ai/v1"):
self.base_url = base_url
def get_timeout(self, model: str) -> int:
"""모델별 타임아웃 반환"""
return self.TIMEOUT_CONFIG.get(
model.lower(),
self.TIMEOUT_CONFIG["default"]
)
def call(self, api_key: str, model: str, messages: list) -> dict:
"""타임아웃 적용 API 호출"""
timeout = self.get_timeout(model)
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
json={
"model": model,
"messages": messages
},
timeout=timeout
)
if response.status_code == 200:
return response.json()
else:
return {
"error": "api_error",
"status": response.status_code,
"message": response.text
}
except requests.Timeout:
return {
"error": "timeout",
"model": model,
"timeout_seconds": timeout,
"suggested_action": "failover_to_faster_model"
}
except requests.RequestException as e:
return {
"error": "request_failed",
"message": str(e)
}
사용 예제
manager = HolySheepTimeoutManager()
모델별 자동 타임아웃 적용
result = manager.call(
api_key="YOUR_HOLYSHEEP_API_KEY",
model="gpt-4.1",
messages=[{"role": "user", "content": "긴 분석 요청"}]
)
if result.get("error") == "timeout":
print("타임아웃 발생, 더 빠른 모델로 failover...")
서킷 브레이커 패턴
연속적인 실패 후 시스템을 보호하려면 서킷 브레이커가 필수적입니다. HolySheep의 다중 모델 환경에서 이를 구현하는 방법을 보여드리겠습니다.
import time
from enum import Enum
from typing import Dict, Optional
from dataclasses import dataclass, field
import requests
class CircuitState(Enum):
CLOSED = "closed" # 정상, 요청 허용
OPEN = "open" # 차단, 요청 거부
HALF_OPEN = "half_open" # 복구 시도 중
@dataclass
class CircuitBreakerConfig:
failure_threshold: int = 5 # OPEN으로 전환할 실패 횟수
success_threshold: int = 2 # CLOSED로 전환할 성공 횟수
timeout: int = 60 # OPEN 상태 지속 시간(초)
half_open_max_calls: int = 3 # HALF_OPEN에서 허용할 호출 수
class CircuitBreaker:
"""모델별 서킷 브레이커"""
def __init__(self, model: str, config: CircuitBreakerConfig = None):
self.model = model
self.config = config or CircuitBreakerConfig()
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
self.last_failure_time: Optional[float] = None
self.half_open_calls = 0
def record_success(self):
"""성공 기록"""
self.failure_count = 0
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.config.success_threshold:
self.state = CircuitState.CLOSED
self.success_count = 0
print(f"[{self.model}] 서킷 복구: CLOSED")
def record_failure(self):
"""실패 기록"""
self.failure_count += 1
self.last_failure_time = time.time()
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.OPEN
self.half_open_calls = 0
print(f"[{self.model}] 서킷 열림: OPEN (HALF_OPEN 중 실패)")
elif (self.state == CircuitState.CLOSED and
self.failure_count >= self.config.failure_threshold):
self.state = CircuitState.OPEN
print(f"[{self.model}] 서킷 열림: OPEN (연속 {self.failure_count}회 실패)")
def can_execute(self) -> bool:
"""실행 가능 여부 확인"""
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
# 타임아웃 확인
if (time.time() - self.last_failure_time) >= self.config.timeout:
self.state = CircuitState.HALF_OPEN
self.half_open_calls = 0
print(f"[{self.model}] 서킷 시도: HALF_OPEN")
return True
return False
if self.state == CircuitState.HALF_OPEN:
if self.half_open_calls < self.config.half_open_max_calls:
self.half_open_calls += 1
return True
return False
return False
def get_status(self) -> dict:
return {
"model": self.model,
"state": self.state.value,
"failure_count": self.failure_count,
"last_failure": self.last_failure_time
}
class HolySheepCircuitBreakerManager:
"""HolySheep 멀티 모델 서킷 브레이커 관리자"""
def __init__(self, base_url: str = "https://api.holysheep.ai/v1"):
self.base_url = base_url
self.breakers: Dict[str, CircuitBreaker] = {}
# 기본 모델 서킷 브레이커 초기화
self._init_breakers()
def _init_breakers(self):
"""모델별 서킷 브레이커 초기화"""
models = [
("gpt-4.1", CircuitBreakerConfig(failure_threshold=5, timeout=60)),
("claude-sonnet-4.5", CircuitBreakerConfig(failure_threshold=5, timeout=60)),
("gemini-2.5-flash", CircuitBreakerConfig(failure_threshold=3, timeout=30)),
("deepseek-v3.2", CircuitBreakerConfig(failure_threshold=3, timeout=20))
]
for model, config in models:
self.breakers[model] = CircuitBreaker(model, config)
def call(self, api_key: str, model: str, messages: list) -> dict:
"""서킷 브레이커 적용 API 호출"""
if model not in self.breakers:
self.breakers[model] = CircuitBreaker(model)
breaker = self.breakers[model]
if not breaker.can_execute():
return {
"error": "circuit_open",
"model": model,
"state": breaker.state.value,
"message": f"{model} 모델이 현재 사용할 수 없습니다"
}
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
json={"model": model, "messages": messages},
timeout=30
)
if response.status_code == 200:
breaker.record_success()
return response.json()
else:
breaker.record_failure()
return {
"error": "api_error",
"model": model,
"status": response.status_code,
"message": response.text
}
except Exception as e:
breaker.record_failure()
return {
"error": "request_failed",
"model": model,
"message": str(e)
}
def get_all_status(self) -> dict:
return {model: breaker.get_status() for model, breaker in self.breakers.items()}
사용 예제
manager = HolySheepCircuitBreakerManager()
정상 호출
result = manager.call(
api_key="YOUR_HOLYSHEEP_API_KEY",
model="gemini-2.5-flash",
messages=[{"role": "user", "content": "테스트"}]
)
상태 확인
status = manager.get_all_status()
for model, stat in status.items():
print(f"{model}: {stat['state']}")
멀티 모델 장애 전환(Failover)
서킷 브레이커와 결합된 자동 장애 전환으로 99.9% 이상의 가용성을 달성할 수 있습니다.
from typing import List, Dict, Optional, Callable
import requests
class HolySheepFailoverManager:
"""멀티 모델 장애 전환 관리자"""
# 모델 우선순위 및 역할
MODEL_PRIORITIES = {
"gpt-4.1": {"tier": 1, "strength": "고품질 복잡한 작업", "cost": 8.0},
"claude-sonnet-4.5": {"tier": 1, "strength": "긴 컨텍스트 분석", "cost": 15.0},
"gemini-2.5-flash": {"tier": 2, "strength": "빠른 응답", "cost": 2.5},
"deepseek-v3.2": {"tier": 3, "strength": "대량 처리", "cost": 0.42}
}
def __init__(
self,
base_url: str = "https://api.holysheep.ai/v1",
circuit_breaker_manager = None
):
self.base_url = base_url
self.cb_manager = circuit_breaker_manager
def _get_available_models(self, fallback_tier: int = None) -> List[str]:
"""사용 가능한 모델 목록 반환"""
available = []
if self.cb_manager:
all_status = self.cb_manager.get_all_status()
for model, status in all_status.items():
if status["state"] != "open":
available.append(model)
else:
available = list(self.MODEL_PRIORITIES.keys())
if fallback_tier:
available = [
m for m in available
if self.MODEL_PRIORITIES.get(m, {}).get("tier", 99) <= fallback_tier
]
# 티어 순으로 정렬
available.sort(key=lambda m: self.MODEL_PRIORITIES.get(m, {}).get("tier", 99))
return available
def call_with_failover(
self,
api_key: str,
messages: list,
preferred_model: str = None,
max_cost_per_call: float = None,
on_failover: Callable[[str, str], None] = None
) -> dict:
"""자동 장애 전환 API 호출"""
# 시도할 모델 목록 결정
if preferred_model and self._is_model_available(preferred_model):
attempt_order = [preferred_model]
attempt_order.extend([
m for m in self._get_available_models()
if m != preferred_model
])
else:
attempt_order = self._get_available_models()
errors = []
for model in attempt_order:
# 비용 체크
model_cost = self.MODEL_PRIORITIES.get(model, {}).get("cost", 99)
if max_cost_per_call and model_cost > max_cost_per_call:
errors.append(f"{model}: 비용 초과 ({model_cost} > {max_cost_per_call})")
continue
# 서킷 브레이커 체크
if self.cb_manager:
status = self.cb_manager.breakers[model].get_status()
if status["state"] == "open":
errors.append(f"{model}: 서킷 OPEN")
continue
print(f"시도: {model}")
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
json={
"model": model,
"messages": messages,
"max_tokens": 1000
},
timeout=self._get_timeout(model)
)
if response.status_code == 200:
result = response.json()
result["model_used"] = model
result["fallback_attempted"] = len(errors) > 0
return result
errors.append(f"{model}: HTTP {response.status_code}")
except requests.exceptions.Timeout:
errors.append(f"{model}: 타임아웃")
except Exception as e:
errors.append(f"{model}: {str(e)}")
# 모든 모델 실패
return {
"error": "all_models_failed",
"attempts": len(attempt_order),
"errors": errors,
"message": "모든 모델 호출 실패"
}
def _is_model_available(self, model: str) -> bool:
if self.cb_manager and model in self.cb_manager.breakers:
status = self.cb_manager.breakers[model].get_status()
return status["state"] != "open"
return True
def _get_timeout(self, model: str) -> int:
timeouts = {
"deepseek-v3.2": 15,
"gemini-2.5-flash": 20,
"claude-sonnet-4.5": 30,
"gpt-4.1": 45
}
return timeouts.get(model, 30)
사용 예제
failover_manager = HolySheepFailoverManager(
circuit_breaker_manager=CircuitBreakerManager()
)
기본 모델(gpt-4.1)이 실패하면 자동으로 Claude, Gemini, DeepSeek 순으로 시도
result = failover_manager.call_with_failover(
api_key="YOUR_HOLYSHEEP_API_KEY",
messages=[
{"role": "system", "content": "당신은 전문 분석가입니다."},
{"role": "user", "content": "최근 AI 트렌드를 분석해주세요."}
],
preferred_model="gpt-4.1",
max_cost_per_call=10.0, # 토큰당 최대 $10
on_failover=lambda from_model, to_model: print(f"Failover: {from_model} -> {to_model}")
)
if "error" in result:
print(f"모든 모델 실패: {result['errors']}")
else:
print(f"성공: {result['model_used']} 사용")
print(f"응답: {result['choices'][0]['message']['content'][:100]}...")
완전한 SLA 관리 시스템
"""
HolySheep AI Agent 완전한 SLA 관리 시스템
재시도 + 타임아웃 + 서킷 브레이커 + 장애 전환 통합
"""
import time
import logging
from typing import Optional
from dataclasses import dataclass
import requests
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class SLAConfig:
"""SLA 설정"""
max_retries: int = 3
base_timeout: int = 30
circuit_failure_threshold: int = 5
circuit_timeout: int = 60
enable_failover: bool = True
max_cost_per_request: float = 10.0
class HolySheepSLAManager:
"""완전한 SLA 관리자"""
def __init__(
self,
api_key: str,
config: SLAConfig = None,
base_url: str = "https://api.holysheep.ai/v1"
):
self.api_key = api_key
self.config = config or SLAConfig()
self.base_url = base_url
self.request_count = 0
self.total_latency = 0
self.error_count = 0
self.success_count = 0
def call(
self,
model: str,
messages: list,
temperature: float = 0.7,
max_tokens: int = 1000
) -> dict:
"""SLA 정책이 적용된 API 호출"""
self.request_count += 1
start_time = time.time()
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
last_error = None
# 재시도 루프
for attempt in range(self.config.max_retries + 1):
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=self.config.base_timeout
)
latency = time.time() - start_time
self.total_latency += latency
if response.status_code == 200:
self.success_count += 1
result = response.json()
result["sla"] = {
"latency_ms": round(latency * 1000, 2),
"attempt": attempt + 1,
"model": model
}
return result
# 재시도 불가능한 오류
if response.status_code not in [429, 500, 502, 503, 504]:
self.error_count += 1
return {
"error": "api_error",
"status": response.status_code,
"message": response.text,
"attempt": attempt + 1
}
last_error = f"HTTP {response.status_code}: {response.text}"
except requests.exceptions.Timeout:
last_error = "요청 타임아웃"
except requests.exceptions.ConnectionError as e:
last_error = f"연결 오류: {e}"
except Exception as e:
last_error = f"예상치 못한 오류: {e}"
# 재시도 지연
if attempt < self.config.max_retries:
delay = min(2 ** attempt, 30) # 지수적 백오프, 최대 30초
logger.info(f"재시도 {attempt + 1}/{self.config.max_retries}, {delay}초 후")
time.sleep(delay)
# 모든 재시도 실패
self.error_count += 1
return {
"error": "all_retries_failed",
"last_error": last_error,
"total_attempts": self.config.max_retries + 1,
"sla": {
"latency_ms": round((time.time() - start_time) * 1000, 2),
"total_attempts": self.config.max_retries + 1
}
}
def call_with_circuit_breaker(
self,
model: str,
messages: list,
circuit_breakers: dict
) -> dict:
"""서킷 브레이커가 적용된 호출"""
breaker = circuit_breakers.get(model)
if breaker and not breaker.can_execute():
return {
"error": "circuit_open",
"model": model,
"suggestion": "failover_to_alternative_model"
}
result = self.call(model, messages)
if breaker:
if "error" in result:
breaker.record_failure()
else:
breaker.record_success()
return result
def get_sla_metrics(self) -> dict:
"""SLA 메트릭 반환"""
total_requests = self.success_count + self.error_count
return {
"total_requests": self.request_count,
"success_count": self.success_count,
"error_count": self.error_count,
"success_rate": round(
self.success_count / total_requests * 100, 2
) if total_requests > 0 else 0,
"avg_latency_ms": round(
self.total_latency /