ChatGPT开放API接口实战：高并发场景下的稳定性优化方案

1次阅读

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

在实际开发中调用 ChatGPT API 时，高并发场景下常遇到三个核心问题：

HTTP 429 错误 ：API 的速率限制（默认每分钟 3,000-3,500 次请求）极易被突发流量触发
长尾延迟 ：P99 响应时间可能达到 5 -10 秒，导致下游服务雪崩
Token 消耗失控 ：max_tokens 参数设置不合理会导致响应截断或超额计费

通过实测发现（测试环境：4 核 8G 云服务器）：

同步调用（requests 库）单线程 QPS 约 15，10 并发时 P99 延迟达 4.2 秒
异步方案（aiohttp）单机 100 并发下 QPS 可达 1200，P99 延迟稳定在 1.3 秒内

import aiohttp
from typing import List, AsyncGenerator

async def batch_request(messages: List[str], 
    api_key: str,
    max_workers: int = 100
) -> AsyncGenerator[str, None]:
    """
    批量处理消息的异步生成器
    :param messages: 输入消息列表
    :param api_key: OpenAI API 密钥
    :param max_workers: 最大并发数
    """
    semaphore = asyncio.Semaphore(max_workers)

    async with aiohttp.ClientSession() as session:
        tasks = []
        for msg in messages:
            task = asyncio.create_task(_single_request(session, msg, api_key, semaphore)
            )
            tasks.append(task)

        for future in asyncio.as_completed(tasks):
            yield await future

async def _single_request(
    session: aiohttp.ClientSession,
    message: str,
    api_key: str,
    semaphore: asyncio.Semaphore
) -> str:
    """单个 API 请求封装"""
    async with semaphore:
        url = "https://api.openai.com/v1/chat/completions"
        headers = {"Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        payload = {
            "model": "gpt-3.5-turbo",
            "messages": [{"role": "user", "content": message}],
            "max_tokens": 1500  # 根据业务需求调整
        }

        try:
            async with session.post(url, json=payload, headers=headers) as resp:
                if resp.status == 429:
                    raise Exception("Rate limit exceeded")
                return await resp.json()
        except Exception as e:
            print(f"Request failed: {str(e)}")
            raise

import random
import time

async def request_with_retry(
    session: aiohttp.ClientSession,
    payload: dict,
    max_retries: int = 5,
    initial_delay: float = 0.1
) -> dict:
    """
    带指数退避的请求重试机制
    :param initial_delay: 初始延迟时间 (秒)
    :param max_retries: 最大重试次数
    """
    current_delay = initial_delay

    for attempt in range(max_retries):
        try:
            async with session.post(
                "https://api.openai.com/v1/chat/completions",
                json=payload
            ) as resp:
                if resp.status == 429:
                    jitter = random.uniform(0, current_delay/2)
                    await asyncio.sleep(current_delay + jitter)
                    current_delay *= 2  # 指数退避
                    continue
                return await resp.json()
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(current_delay)
            current_delay *= 2

    raise Exception("Max retries exceeded")

中文文本通常按 2.5 倍换算 token 数量（实测值）
推荐设置：max_tokens = 平均输入 token 数 + 期望输出 token 数 + 30% 缓冲

Prometheus 指标示例：

# metrics.yaml
rules:
- name: openai_cost
  rules:
  - record: openai:requests:total
    expr: sum(rate(openai_api_calls_total[5m])) by (endpoint)
  - record: openai:tokens:usage
    expr: sum(rate(openai_tokens_used[1h])) by (type)