Building resilient AI-powered applications requires more than just sending requests and parsing responses. After three years of integrating large language models into production systems at scale, I've learned that exception handling separates hobby projects from enterprise-grade deployments. This guide walks through battle-tested patterns for handling API errors, timeouts, rate limits, and transient failures when calling AI endpoints.
Why Exception Handling Matters for AI APIs
AI API calls differ from typical REST endpoints in several critical ways:
- Latency variability: Generation time scales with output length, making timeout prediction difficult
- Rate limiting complexity: Tokens-per-minute limits interact with concurrent requests
- Cost volatility: A single malformed request can waste dollars in token consumption
- Provider inconsistency: Error formats and retry semantics vary across API versions
HolySheep AI addresses several of these concerns with sub-50ms cold-start latency and transparent transparent pricing starting at ¥1=$1 equivalent—85% cheaper than the ¥7.3 benchmark. They support WeChat and Alipay for Chinese market convenience, and new accounts receive free credits to validate these production patterns.
Architecture: The Retry Circuit Breaker Pattern
For production systems, I recommend implementing a three-layer exception strategy:
- Immediate retry with exponential backoff: For transient network errors and 5xx responses
- Circuit breaker: Prevents cascade failures when a provider is degraded
- Fallback handling: Graceful degradation to cached responses or alternative models
Implementing the HolyShehe SDK Client
The following implementation uses the official OpenAI-compatible client with HolySheep AI's endpoint at https://api.holysheep.ai/v1:
import os
import time
import logging
from functools import wraps
from typing import Optional, TypeVar, Callable, Any
from dataclasses import dataclass
from enum import Enum
import openai
from openai import APIError, RateLimitError, Timeout, APIConnectionError
Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
T = TypeVar('T')
class CircuitState(Enum):
CLOSED = "closed" # Normal operation
OPEN = "open" # Failing, reject requests
HALF_OPEN = "half_open" # Testing recovery
@dataclass
class RetryConfig:
max_retries: int = 3
base_delay: float = 1.0
max_delay: float = 30.0
exponential_base: float = 2.0
jitter: bool = True
class CircuitBreaker:
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: float = 60.0,
expected_exception: type = Exception
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.expected_exception = expected_exception
self.failure_count = 0
self.last_failure_time: Optional[float] = None
self.state = CircuitState.CLOSED
def call(self, func: Callable[..., T], *args, **kwargs) -> T:
if self.state == CircuitState.OPEN:
if time.time() - self.last_failure_time >= self.recovery_timeout:
self.state = CircuitState.HALF_OPEN
logger.info("Circuit breaker entering HALF_OPEN state")
else:
raise CircuitBreakerOpenError("Circuit breaker is OPEN")
try:
result = func(*args, **kwargs)
self._on_success()
return result
except self.expected_exception as e:
self._on_failure()
raise
def _on_success(self):
self.failure_count = 0
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.CLOSED
logger.info("Circuit breaker restored to CLOSED state")
def _on_failure(self):
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
logger.warning(f"Circuit breaker opened after {self.failure_count} failures")
class CircuitBreakerOpenError(Exception):
pass
class HolySheepAIClient:
def __init__(
self,
api_key: Optional[str] = None,
base_url: str = "https://api.holysheep.ai/v1",
timeout: float = 120.0,
max_retries: int = 3
):
self.client = openai.OpenAI(
api_key=api_key or os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"),
base_url=base_url,
timeout=timeout,
max_retries=max_retries
)
self.circuit_breaker = CircuitBreaker(
failure_threshold=5,
recovery_timeout=60.0,
expected_exception=(APIError, RateLimitError, APIConnectionError)
)
def chat_completion_with_retry(
self,
model: str = "deepseek-v3.2",
messages: list[dict],
temperature: float = 0.7,
max_tokens: Optional[int] = None,
retry_config: Optional[RetryConfig] = None
) -> dict:
if retry_config is None:
retry_config = RetryConfig()
last_exception = None
for attempt in range(retry_config.max_retries + 1):
try:
response = self.circuit_breaker.call(
self._make_completion_request,
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens
)
return response
except (RateLimitError, APIConnectionError, Timeout) as e:
last_exception = e
if attempt < retry_config.max_retries:
delay = self._calculate_delay(retry_config, attempt)
logger.warning(
f"Attempt {attempt + 1} failed: {type(e).__name__}. "
f"Retrying in {delay:.2f}s"
)
time.sleep(delay)
else:
logger.error(f"All {retry_config.max_retries + 1} attempts exhausted")
except CircuitBreakerOpenError as e:
logger.error(f"Circuit breaker rejected request: {e}")
raise
except APIError as e:
# Non-retryable errors (400 Bad Request, 401 Unauthorized, etc.)
if e.status_code in {400, 401, 403, 404, 422}:
logger.error(f"Non-retryable API error: {e.status_code} - {e.message}")
raise
last_exception = e
if attempt < retry_config.max_retries:
delay = self._calculate_delay(retry_config, attempt)
logger.warning(f"Attempt {attempt + 1} failed with {e.status_code}. Retrying...")
time.sleep(delay)
raise AIAPIException(
f"Failed after {retry_config.max_retries + 1} attempts",
last_exception
)
def _make_completion_request(self, **kwargs) -> dict:
return self.client.chat.completions.create(**kwargs).model_dump()
def _calculate_delay(self, config: RetryConfig, attempt: int) -> float:
delay = min(
config.base_delay * (config.exponential_base ** attempt),
config.max_delay
)
if config.jitter:
import random
delay = delay * (0.5 + random.random())
return delay
class AIAPIException(Exception):
def __init__(self, message: str, original_exception: Optional[Exception] = None):
super().__init__(message