ChatGPT Python SDK 实战：如何解决 API 调用中的并发与超时问题

1次阅读

没有评论

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

在实际项目中，我们经常遇到需要批量处理大量请求的场景。例如，一个客服系统需要同时处理数百条用户咨询，每条咨询都需要调用 ChatGPT API 获取回复。这种情况下，最常见的故障模式包括：

429 Too Many Requests：短时间内发送过多请求触发了速率限制
连接中断：网络波动导致请求失败
超时未响应：服务器处理时间过长导致客户端等待超时

传统的同步请求方式使用 requests 库，代码简单但性能有限。我们做了一个简单的基准测试：

发送 100 个请求，同步方式耗时约 15 秒
同样的请求量，异步方式 (aiohttp) 仅需 2 秒

异步 IO 的优势在于可以同时处理多个请求，特别适合 API 调用这种 IO 密集型操作。

import aiohttp
from typing import AsyncIterator

class AsyncChatGPTClient:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.session = None

    async def __aenter__(self):
        connector = aiohttp.TCPConnector(limit_per_host=10)  # 控制每个主机的连接数
        self.session = aiohttp.ClientSession(connector=connector)
        return self

    async def __aexit__(self, exc_type, exc, tb):
        await self.session.close()

    async def chat_completion(self, messages: list) -> dict:
        headers = {"Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

        payload = {
            "model": "gpt-3.5-turbo",
            "messages": messages
        }

        async with self.session.post(
            "https://api.openai.com/v1/chat/completions",
            json=payload,
            headers=headers
        ) as response:
            return await response.json()

import time
import math

class RateLimiter:
    def __init__(self, max_rate: int, time_window: float = 60.0):
        self.max_rate = max_rate
        self.time_window = time_window
        self.tokens = max_rate
        self.last_refill = time.monotonic()

    async def wait_for_token(self):
        now = time.monotonic()
        elapsed = now - self.last_refill

        # 根据耗时补充 token
        if elapsed > 0:
            self.tokens = min(
                self.max_rate,
                self.tokens + (self.max_rate * elapsed / self.time_window)
            )
            self.last_refill = now

        if self.tokens >= 1:
            self.tokens -= 1
            return

        # 采用指数退避算法等待
        sleep_time = (self.time_window / self.max_rate) * 1.5
        await asyncio.sleep(sleep_time)
        await self.wait_for_token()

from functools import wraps
import asyncio

def retry(max_retries=3, base_delay=1.0):
    def decorator(f):
        @wraps(f)
        async def wrapper(*args, **kwargs):
            retries = 0
            while retries < max_retries:
                try:
                    return await f(*args, **kwargs)
                except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                    retries += 1
                    if retries >= max_retries:
                        raise

                    delay = base_delay * (2 ** retries)  # 指数退避
                    await asyncio.sleep(delay)
        return wrapper
    return decorator