Комплексная обработка ошибок API ChatGPT: стратегии интеграции

{
"error": {
"message": "You exceeded your current quota, please check your plan and billing details.",
"type": "insufficient_quota",
"param": null,
"code": "quota_exceeded"
}
}

import time
import random
import openai
from tenacity import (
retry,
stop_after_attempt,
wait_exponential_jitter,
retry_if_exception_type
)

@retry(
wait=wait_exponential_jitter(multiplier=1, max=60, jitter=random.random),
stop=stop_after_attempt(5),
retry=retry_if_exception_type((openai.error.APIError, openai.error.RateLimitError))
)
def generate_text_with_retry(prompt, model="gpt-3.5-turbo"):
try:
response = openai.ChatCompletion.create(
model=model,
messages=[{"role": "user", "content": prompt}],
timeout=10 # Таймаут в секундах
)
return response.choices[0].message.content
except openai.error.Timeout:
print("Request timed out. Retrying...")
raise
except openai.error.APIError as e:
print(f"API error: {e}")
raise
except openai.error.RateLimitError:
print("Rate limit exceeded. Applying backoff...")
raise
except Exception as e:
print(f"Unexpected error: {e}")
# Не повторяем запрос приunexpected ошибках
return None

import asyncio
import aiohttp
import backoff

@backoff.on_exception(
backoff.expo,
(aiohttp.ClientError, asyncio.TimeoutError),
max_tries=5,
max_time=30
)
async def async_openai_request(prompt):
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {API_KEY}"
}
payload = {
"model": "gpt-3.5-turbo",
"messages": [{"role": "user", "content": prompt}]
}

async with aiohttp.ClientSession() as session:
async with session.post(
"https://api.openai.com/v1/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=60)
) as response:
if response.status == 200:
result = await response.json()
return result["choices"][0]["message"]["content"]
else:
error_detail = await response.text()
print(f"Error {response.status}: {error_detail}")
response.raise_for_status()

import sentry_sdk
import openai

# Инициализация Sentry
sentry_sdk.init(
dsn="ваш_dsn_ключ",
traces_sample_rate=1.0,
environment="production"
)

def call_chatgpt_api(prompt, model="gpt-3.5-turbo"):
# Добавляем контекст к потенциальным ошибкам
with sentry_sdk.configure_scope() as scope:
scope.set_tag("api_type", "chatgpt")
scope.set_tag("model", model)
scope.set_context("api_params", {
"prompt": prompt[:100] + "..." if len(prompt) > 100 else prompt,
"model": model,
"max_tokens": 1000
})

try:
response = openai.ChatCompletion.create(
model=model,
messages=[{"role": "user", "content": prompt}],
max_tokens=1000
)
return response
except openai.error.APIError as e:
# Фиксируем API ошибки с детализацией
sentry_sdk.set_context("api_error", {
"status_code": getattr(e, "http_status", None),
"error_code": getattr(e, "code", None),
"error_type": type(e).__name__
})
sentry_sdk.capture_exception(e)
raise
except openai.error.RateLimitError as e:
# Особая обработка для ошибок с кодом 429 и 403
sentry_sdk.set_tag("error_category", "rate_limit")
sentry_sdk.capture_exception(e)
# Можно добавить метрику для отслеживания частоты rate limit ошибок
raise
except Exception as e:
sentry_sdk.capture_exception(e)
raise

import logging
import json
from datetime import datetime
import sentry_sdk
from sentry_sdk.integrations.logging import LoggingIntegration

# Настройка логирования
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("chatgpt_api")

# Интеграция логирования с Sentry
sentry_logging = LoggingIntegration(
level=logging.INFO, # Уровень для захвата через логгер
event_level=logging.ERROR # Уровень для отправки событий в Sentry
)

sentry_sdk.init(
dsn="ваш_dsn_ключ",
integrations=[sentry_logging],
traces_sample_rate=1.0
)

def log_api_error(error, request_params, response=None):
"""Расширенное логирование ошибок API с интеграцией Sentry"""

# Извлечение деталей ошибки
error_type = type(error).__name__
status_code = getattr(error, "http_status", None)
error_code = getattr(error, "code", None)
error_message = str(error)

# Создание структурированного лога
error_log = {
"timestamp": datetime.utcnow().isoformat(),
"error_type": error_type,
"status_code": status_code,
"error_code": error_code,
"message": error_message,
"request_params": {k: v for k, v in request_params.items() if k != "api_key"},
"response": response
}

# Добавление идентификатора запроса, если доступен
request_id = getattr(error, "request_id", None)
if request_id:
error_log["request_id"] = request_id

# Логирование с разным уровнем в зависимости от типа ошибки
if status_code in (401, 403):
logger.error(f"Authentication error: {json.dumps(error_log)}")
elif status_code == 429:
logger.warning(f"Rate limit exceeded: {json.dumps(error_log)}")
elif status_code and 500 <= status_code < 600:
logger.error(f"Server error: {json.dumps(error_log)}")
else:
logger.info(f"API error: {json.dumps(error_log)}")

# Отправка в Sentry для ошибок высокого приоритета
if status_code in (401, 403, 429) or (status_code and 500 <= status_code < 600):
with sentry_sdk.configure_scope() as scope:
scope.set_tag("error_type", error_type)
scope.set_tag("status_code", status_code)
scope.set_context("api_error", error_log)
sentry_sdk.capture_exception(error)

from pybreaker import CircuitBreaker, CircuitBreakerError
import openai
import time
import logging

logger = logging.getLogger(__name__)

# Настройка Circuit Breaker
chatgpt_breaker = CircuitBreaker(
fail_max=5, # Количество ошибок до размыкания цепи
reset_timeout=60, # Время до попытки восстановления в секундах
exclude=[openai.error.InvalidRequestError] # Исключения, не вызывающие размыкание
)

class ApiKeyManager:
def __init__(self, primary_key, backup_keys=None):
self.primary_key = primary_key
self.backup_keys = backup_keys or []
self.current_key_index = 0
self.keys = [primary_key] + self.backup_keys

def get_current_key(self):
return self.keys[self.current_key_index]

def rotate_key(self):
self.current_key_index = (self.current_key_index + 1) % len(self.keys)
logger.info(f"Rotating to API key {self.current_key_index + 1}/{len(self.keys)}")
return self.get_current_key()

# Инициализация менеджера ключей
key_manager = ApiKeyManager(
primary_key="your-primary-api-key",
backup_keys=["backup-key-1", "backup-key-2"]
)

@chatgpt_breaker
def generate_text_with_circuit_breaker(prompt, model="gpt-3.5-turbo"):
try:
# Установка текущего API-ключа
openai.api_key = key_manager.get_current_key()

response = openai.ChatCompletion.create(
model=model,
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
except openai.error.RateLimitError:
# При превышении лимита пытаемся сменить ключ
logger.warning("Rate limit exceeded, attempting to rotate API key")
openai.api_key = key_manager.rotate_key()

# Если это был последний ключ, используем фолбэк
if key_manager.current_key_index == 0:
logger.error("All API keys have reached their rate limits")
return fallback_response(prompt)

# Повторная попытка с новым ключом
time.sleep(1) # Небольшая пауза перед повторным запросом
return generate_text_with_circuit_breaker(prompt, model)
except openai.error.APIError as e:
# Серверная ошибка OpenAI, может потребоваться повтор
logger.error(f"API error: {str(e)}")
raise
except CircuitBreakerError:
# Circuit breaker разомкнут, используем фолбэк
logger.warning("Circuit breaker open, using fallback")
return fallback_response(prompt)
except Exception as e:
logger.exception(f"Unexpected error: {str(e)}")
raise

def fallback_response(prompt):
"""Фолбэк-ответ при недоступности API"""
# Здесь может быть локальная модель, предопределенные ответы и т.д.
return "I apologize, but I'm currently experiencing connectivity issues. Please try again later."

import hashlib
import json
from functools import lru_cache
import redis
import pickle
from datetime import timedelta

# Настройка Redis для второго уровня кэша
redis_client = redis.Redis(host='localhost', port=6379, db=0)

def generate_cache_key(model, messages, temperature=None):
"""Генерация уникального ключа кэша на основе параметров запроса"""
cache_data = {
'model': model,
'messages': messages,
'temperature': temperature
}
serialized = json.dumps(cache_data, sort_keys=True)
return hashlib.md5(serialized.encode()).hexdigest()

# Первый уровень кэша (в памяти)
@lru_cache(maxsize=1000)
def get_from_memory_cache(cache_key):
"""Получение результата из кэша в памяти"""
return None # LRU-кэш обрабатывается декоратором

# Второй уровень кэша (Redis)
def get_from_redis_cache(cache_key):
"""Получение результата из Redis"""
cached_value = redis_client.get(cache_key)
if cached_value:
return pickle.loads(cached_value)
return None

def set_in_redis_cache(cache_key, value, ttl_hours=24):
"""Сохранение результата в Redis"""
redis_client.setex(
cache_key,
timedelta(hours=ttl_hours),
pickle.dumps(value)
)

# Комплексная функция доступа к кэшу
def get_cached_response(model, messages, temperature=None):
cache_key = generate_cache_key(model, messages, temperature)

# Проверяем первый уровень кэша
memory_result = get_from_memory_cache(cache_key)
if memory_result:
return memory_result, 'memory_cache'

# Проверяем второй уровень кэша
redis_result = get_from_redis_cache(cache_key)
if redis_result:
# Обновляем кэш в памяти
_ = get_from_memory_cache.cache_clear() # Инвалидация
_ = get_from_memory_cache(cache_key)
return redis_result, 'redis_cache'

return None, None

def cache_response(model, messages, response, temperature=None):
"""Сохранение ответа API во всех уровнях кэша"""
cache_key = generate_cache_key(model, messages, temperature)

# Сохраняем в памяти
_ = get_from_memory_cache.cache_clear()
_ = get_from_memory_cache(cache_key)

# Сохраняем в Redis
set_in_redis_cache(cache_key, response)

def generate_text_with_caching(prompt, model="gpt-3.5-turbo", temperature=0.7):
messages = [{"role": "user", "content": prompt}]

# Пытаемся получить из кэша
cached_response, cache_source = get_cached_response(model, messages, temperature)
if cached_response:
print(f"Cache hit from {cache_source}")
return cached_response

# Если нет в кэше, делаем запрос к API
try:
response = openai.ChatCompletion.create(
model=model,
messages=messages,
temperature=temperature
)
result = response.choices[0].message.content

# Кэшируем результат
cache_response(model, messages, result, temperature)
return result
except Exception as e:
# Обработка ошибок
print(f"Error: {str(e)}")
raise