ChatGPT API调用实战：从基础接入到高并发优化

1次阅读

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

在当今 AI 技术快速发展的背景下，ChatGPT API 已成为开发者构建智能应用的重要工具。然而，在实际使用过程中，我们常常会遇到各种挑战。本文将分享我在实际项目中总结的 ChatGPT API 调用经验，从基础接入到高并发优化，希望能帮助开发者更高效地使用这一强大工具。

在开始技术实现之前，我们先来看看 ChatGPT API 调用的三大核心痛点：

认证密钥的轮换管理 ：API 密钥需要定期更换以保证安全，但手动管理既繁琐又容易出错。
速率限制处理 ：当请求超过限制时会返回 429 错误，需要优雅地处理这种限制。
长文本对话的上下文维护 ：保持对话连贯性需要有效管理上下文，这对内存和性能都是挑战。

在 Python 生态中，我们有多个 HTTP 客户端可以选择：

requests：同步请求，简单易用但性能有限
aiohttp：异步请求，适合高并发场景
httpx：同时支持同步和异步，API 设计友好

对于 ChatGPT API 调用，我推荐使用 httpx，因为它兼具灵活性和性能。

以下是带 JWT 自动刷新的认证装饰器示例：

from functools import wraps
from datetime import datetime, timedelta
import jwt

def auth_required(f):
    @wraps(f)
    async def wrapper(*args, **kwargs):
        # 检查 token 是否即将过期
        if datetime.now() > kwargs.get('token_expiry', datetime.min):
            kwargs['api_key'] = refresh_token()
            kwargs['token_expiry'] = datetime.now() + timedelta(hours=1)
        return await f(*args, **kwargs)
    return wrapper

使用 tenacity 库可以轻松实现带指数退避的重试机制：

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
async def call_chatgpt(prompt: str) -> str:
    # API 调用实现
    pass

对于批量消息处理，我们可以采用以下策略：

import asyncio
from typing import List

async def batch_process(messages: List[str], batch_size: int = 5):
    semaphore = asyncio.Semaphore(batch_size)

    async def process_one(message: str):
        async with semaphore:
            return await call_chatgpt(message)

    return await asyncio.gather(*[process_one(m) for m in messages])

我们在生产环境中进行了性能对比测试：