Python统一捕获与格式化AI API的各类异常的方法

try:
    response = client.chat.completions.create(...)
except Exception as e:
    print("出错了:", e)

from dataclasses import dataclass
from typing import Optional
from openai import (
    APIConnectionError,
    APITimeoutError,
    RateLimitError,
    AuthenticationError,
    PermissionDeniedError,
    NotFoundError,
    BadRequestError,
    InternalServerError,
    APIError,
)

@dataclass
class APIResult:
    success: bool
    data: Optional[str] = None
    error_code: Optional[str] = None
    error_message: Optional[str] = None
    retryable: bool = False

def handle_ai_exception(e: Exception) -> APIResult:
    """
    统一将 OpenAI SDK 的各类异常转换为结构化的 APIResult
    """
    if isinstance(e, APITimeoutError):
        return APIResult(
            success=False, 
            error_code="TIMEOUT", 
            error_message="AI 接口响应超时，请稍后重试", 
            retryable=True
        )
    
    elif isinstance(e, APIConnectionError):
        return APIResult(
            success=False, 
            error_code="CONNECTION_ERROR", 
            error_message="无法连接到 AI 服务端，请检查网络或代理设置", 
            retryable=True
        )
        
    elif isinstance(e, RateLimitError):
        return APIResult(
            success=False, 
            error_code="RATE_LIMIT", 
            error_message="请求过于频繁或账户额度受限，触发限流", 
            retryable=True
        )
        
    elif isinstance(e, AuthenticationError):
        return APIResult(
            success=False, 
            error_code="UNAUTHORIZED", 
            error_message="API Key 鉴权失败，请检查密钥是否正确", 
            retryable=False
        )
        
    elif isinstance(e, (PermissionDeniedError, NotFoundError)):
        return APIResult(
            success=False, 
            error_code="INVALID_REQUEST", 
            error_message=f"请求的模型不存在或无权访问: {e.message if hasattr(e, 'message') else str(e)}", 
            retryable=False
        )
        
    elif isinstance(e, BadRequestError):
        return APIResult(
            success=False, 
            error_code="BAD_REQUEST", 
            error_message=f"请求参数不合法: {e.message if hasattr(e, 'message') else str(e)}", 
            retryable=False
        )
        
    elif isinstance(e, InternalServerError):
        return APIResult(
            success=False, 
            error_code="UPSTREAM_ERROR", 
            error_message="AI 服务商内部错误，请稍后重试", 
            retryable=True
        )
        
    elif isinstance(e, APIError):
        # 兜底其他标准 API 错误
        return APIResult(
            success=False, 
            error_code="API_ERROR", 
            error_message=f"AI 接口返回错误 (Status {e.status_code}): {e.message}", 
            retryable=bool(e.status_code and e.status_code >= 500)
        )
        
    else:
        # 非 AI 客户端引发的未知异常（如代码 Bug、内存溢出等）
        return APIResult(
            success=False, 
            error_code="INTERNAL_UNKNOWN", 
            error_message=f"系统未知异常: {str(e)}", 
            retryable=False
        )

from openai import OpenAI

client = OpenAI(
    api_key="your-api-key",
    base_url="https://your-api-domain.com/v1"
)

def safe_ask_llm(prompt: str) -> APIResult:
    try:
        response = client.chat.completions.create(
            model="your-model-name",
            messages=[{"role": "user", "content": prompt}],
            timeout=15.0
        )
        content = response.choices[0].message.content
        return APIResult(success=True, data=content)
        
    except Exception as e:
        # 统一交由异常处理器分类
        result = handle_ai_exception(e)
        
        # 可以在这里记录结构化日志
        print(f"[日志记录] 错误码: {result.error_code}, 是否可重试: {result.retryable}, 详情: {result.error_message}")
        
        return result

# 运行测试
if __name__ == "__main__":
    res = safe_ask_llm("你好")
    if res.success:
        print("回答内容:", res.data)
    else:
        print(f"调用失败 [{res.error_code}]: {res.error_message}")
        if res.retryable:
            print("提示：该错误属于临时故障，可以触发自动重试。")
        else:
            print("提示：该错误属于致命配置问题，请检查代码或密钥。")

def should_retry_based_on_result(result: APIResult) -> bool:
    return result.retryable