Python调用ChatGPT代理服务的完整指南：从API封装到生产环境部署

2次阅读

共计 2292 个字符，预计需要花费 6 分钟才能阅读完成。

直接调用 OpenAI 官方 API 时，开发者常遇到三类典型问题：

网络稳定性：跨国 API 调用受网络波动影响大，偶发连接超时需要手动重试
速率限制：免费账号每分钟仅 3 次请求，付费账号也有 TPM（每分钟 tokens）限制
错误处理：API 返回 429/503 等状态码时缺乏标准化处理流程

根据实际测试数据（相同并发量下）：

客户端	同步 / 异步	长连接支持	性能（QPS）
requests	同步	❌	120
aiohttp	异步	✅	850
httpx	双模	✅	780

推荐选择：生产环境优先使用 aiohttp，需要同步 / 异步切换时选择 httpx

from functools import wraps
import random
import time

def retry_with_backoff(
    max_retries: int = 3,
    initial_delay: float = 1.0,
    max_delay: float = 10.0,
    jitter: bool = True
):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            delay = initial_delay
            for attempt in range(max_retries):
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise

                    # 计算下一次延迟时间
                    sleep_time = min(delay * 2 ** attempt, max_delay)
                    if jitter:
                        sleep_time *= random.uniform(0.5, 1.5)

                    await asyncio.sleep(sleep_time)
        return wrapper
    return decorator

import aiohttp
from typing import Optional, Dict, Any

class ChatGPTProxy:
    def __init__(self, api_keys: list[str]):
        self.api_keys = api_keys
        self.current_key_idx = 0
        self.session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30)
        )

    @retry_with_backoff()
    async def send_request(
        self, 
        prompt: str,
        model: str = "gpt-3.5-turbo"
    ) -> Optional[Dict[str, Any]]:
        headers = {"Authorization": f"Bearer {self._get_current_key()}",
            "Content-Type": "application/json"
        }

        payload = {
            "model": model,
            "messages": [{"role": "user", "content": prompt}]
        }

        try:
            async with self.session.post(
                "https://api.openai.com/v1/chat/completions",
                headers=headers,
                json=payload
            ) as resp:
                if resp.status == 429:
                    self._rotate_key()
                return await resp.json()
        except aiohttp.ClientError as e:
            log_error(f"Request failed: {str(e)}")
            raise

    def _get_current_key(self) -> str:
        return self.api_keys[self.current_key_idx]

    def _rotate_key(self) -> None:
        self.current_key_idx = (self.current_key_idx + 1) % len(self.api_keys)

建议使用 circuitbreaker 包：

from circuitbreaker import circuit

@circuit(
    failure_threshold=5,
    recovery_timeout=60,
    expected_exception=aiohttp.ClientError
)
async def safe_request():
    # 业务代码