ChatGPT接口调用实战：高并发场景下的性能优化与错误处理

1次阅读

没有评论

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

在实际开发中，调用 ChatGPT 接口时经常会遇到几个典型问题：

429 限流错误：当请求频率超过 API 限制时，服务端会返回 429 状态码，导致请求失败
长文本响应超时：处理大段文本时，生成响应时间可能超过客户端或服务端的超时设置
token 计算偏差：开发者预估的 token 数量与 API 实际消耗不一致，导致请求被拒绝或响应被截断

这些问题在高并发场景下会被放大，严重影响服务可用性。

在吞吐量要求高的场景下，异步调用模式明显优于同步调用：

同步调用会阻塞线程，导致资源利用率低下
异步调用可以复用连接，显著提高并发处理能力
异步模式更易于实现请求批处理和并发控制

以下是使用 aiohttp 管理连接池的示例代码（含 JWT 鉴权）：

import aiohttp
from jwt import encode

async def create_session(api_key):
    """
    创建带 JWT 认证的 aiohttp 会话
    :param api_key: OpenAI API 密钥
    :return: aiohttp.ClientSession
    """
    # 生成 JWT token
    token = encode({"api_key": api_key}, "secret", algorithm="HS256")

    # 配置连接池
    connector = aiohttp.TCPConnector(
        limit=100,  # 最大连接数
        limit_per_host=50,  # 单主机最大连接
        enable_cleanup_closed=True
    )

    # 创建会话
    headers = {"Authorization": f"Bearer {token}"}
    return aiohttp.ClientSession(
        connector=connector,
        headers=headers,
        timeout=aiohttp.ClientTimeout(total=30)
    )

使用 tenacity 库实现带随机抖动的重试策略：

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential_jitter,
    retry_if_exception_type
)
import aiohttp

@retry(stop=stop_after_attempt(5),
    wait=wait_exponential_jitter(initial=1, max=60),
    retry=retry_if_exception_type(aiohttp.ClientError)
)
async def call_chatgpt(session, payload):
    """带重试机制的 API 调用"""
    async with session.post(
        "https://api.openai.com/v1/chat/completions",
        json=payload
    ) as response:
        if response.status == 429:
            retry_after = int(response.headers.get("Retry-After", 1))
            await asyncio.sleep(retry_after)
            raise aiohttp.ClientError("Rate limited")
        return await response.json()

将多个独立请求合并为一个批量请求：

async def batch_request(session, messages_list):
    """处理批量消息请求"""
    payload = {
        "messages": messages_list,
        "model": "gpt-3.5-turbo",
        "max_tokens": 1000
    }
    return await call_chatgpt(session, payload)

使用 tiktoken 库准确计算 token 数量：

import tiktoken

def count_tokens(text, model="gpt-3.5-turbo"):
    """精确计算文本的 token 数量"""
    encoding = tiktoken.encoding_for_model(model)
    return len(encoding.encode(text))