ChatGPT API 接入实战：从鉴权到高并发优化的全链路解决方案

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

背景痛点分析

在接入 ChatGPT API 时，开发者常遇到以下几个典型问题：

1. 鉴权流程复杂 ：OpenAI 的 API Key 需要妥善管理，且每个请求都需要携带正确的 Authorization 头。
2. 网络延迟高 ：由于服务器通常位于海外，网络延迟可能严重影响响应速度。
3. API 限流严格 ：免费 tier 和付费 tier 都有严格的速率限制，超出限制会导致 429 错误。
4. 错误处理复杂 ：API 可能返回多种错误（如 503 服务不可用），需要合理的重试机制。

技术方案

1. 使用 aiohttp 实现异步请求

异步请求可以显著提高吞吐量，尤其是在高并发场景下。aiohttp 是一个优秀的异步 HTTP 客户端库，适合与 ChatGPT API 交互。

import aiohttp

async def fetch_chat_response(session, prompt):
    url = "https://api.openai.com/v1/chat/completions"
    headers = {"Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }
    data = {
        "model": "gpt-3.5-turbo",
        "messages": [{"role": "user", "content": prompt}]
    }
    async with session.post(url, headers=headers, json=data) as response:
        return await response.json()

2. 基于 Tenacity 的指数退避重试机制

Tenacity 是一个强大的重试库，可以轻松实现指数退避策略，避免因短暂错误导致请求失败。

from tenacity import retry, stop_after_attempt, wait_exponential
import aiohttp

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
async def fetch_with_retry(session, prompt):
    try:
        return await fetch_chat_response(session, prompt)
    except aiohttp.ClientError as e:
        print(f"Request failed: {e}")
        raise

3. 连接池配置优化

通过调整 TCP Keep-Alive 和连接池大小，可以进一步提高性能。

conn = aiohttp.TCPConnector(
    keepalive_timeout=30,
    limit=100  # 最大连接数
)

async with aiohttp.ClientSession(connector=conn) as session:
    response = await fetch_with_retry(session, "Hello, ChatGPT!")

代码示例

完整的异步上下文管理器实现

import aiohttp
from contextlib import asynccontextmanager

@asynccontextmanager
async def get_chat_session():
    conn = aiohttp.TCPConnector(limit=100)
    async with aiohttp.ClientSession(connector=conn) as session:
        try:
            yield session
        finally:
            await session.close()

流式响应处理

对于长文本生成，流式响应可以显著改善用户体验。

async def stream_response(session, prompt):
    url = "https://api.openai.com/v1/chat/completions"
    data = {
        "model": "gpt-3.5-turbo",
        "messages": [{"role": "user", "content": prompt}],
        "stream": True
    }
    async with session.post(url, headers=headers, json=data) as response:
        async for chunk in response.content:
            print(chunk.decode(), end="", flush=True)

错误分类处理

async def handle_errors(response):
    if response.status == 429:
        print("Rate limit exceeded, please wait...")
    elif response.status == 503:
        print("Service unavailable, retrying...")
    else:
        print(f"Unexpected error: {response.status}")

性能考量

同步 vs 异步 QPS 对比

• 同步请求 ：约 10-20 QPS（受限于网络延迟）。
• 异步请求 ：可达 100+ QPS（取决于连接池配置和服务器性能）。

Token 消耗监控

def calculate_token_cost(response):
    usage = response.get("usage", {})
    prompt_tokens = usage.get("prompt_tokens", 0)
    completion_tokens = usage.get("completion_tokens", 0)
    total_cost = (prompt_tokens * 0.002 + completion_tokens * 0.002) / 1000
    print(f"Total cost: ${total_cost:.4f}")

避坑指南

敏感信息加密存储

使用环境变量或加密工具存储 API Key，避免硬编码。

import os
from cryptography.fernet import Fernet

key = Fernet.generate_key()
cipher_suite = Fernet(key)
encrypted_key = cipher_suite.encrypt(b"your_api_key_here")

# 存储到环境变量
os.environ["OPENAI_API_KEY"] = encrypted_key.decode()

对话上下文管理

保持上下文完整，避免重复发送历史消息。

messages = [{"role": "system", "content": "You are a helpful assistant."},
    {"role": "user", "content": "Who won the world series in 2020?"}
]

# 后续请求只需追加新消息
messages.append({"role": "user", "content": "Where was it played?"})

防止 Prompt 注入

对用户输入进行过滤，避免恶意指令。

def sanitize_input(prompt):
    forbidden_phrases = ["ignore previous", "system", "role"]
    for phrase in forbidden_phrases:
        if phrase in prompt.lower():
            raise ValueError("Invalid input detected")
    return prompt

总结

通过异步请求、合理的重试机制和连接池优化，可以显著提升 ChatGPT API 的调用效率和稳定性。同时，注意错误处理、成本监控和安全性，确保生产环境中的可靠运行。