ChatGPT API获取实战：从认证到流式响应的完整解决方案

1次阅读

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

最近在项目中集成 ChatGPT API 时，遇到了几个让人头疼的问题。首先是 API 认证失败，尤其是在密钥轮换时容易出现 401 错误；其次是长文本响应经常超时，导致用户体验很差；最后是流式数据处理不够高效，内存占用过高。这些问题在生产环境中尤其突出，需要一套完整的解决方案。

在 Python 中，常用的 HTTP 客户端有 requests 和 aiohttp。requests 简单易用，同步阻塞的特性适合简单场景；aiohttp 则支持异步 IO，适合高并发和流式处理。考虑到 ChatGPT API 的流式响应特性，我们选择 aiohttp 作为 HTTP 客户端。

API 密钥是敏感信息，绝不能硬编码在代码中。最佳实践是使用环境变量存储密钥：

import os
from dotenv import load_dotenv

load_dotenv()  # 加载.env 文件
API_KEY = os.getenv('OPENAI_API_KEY')

网络请求难免会遇到临时故障，实现带指数退避的自动重试机制能显著提高可靠性：

import asyncio
import random

async def exponential_backoff_retry(func, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await func()
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            wait_time = min(2 ** attempt + random.uniform(0, 1), 10)
            await asyncio.sleep(wait_time)

下面是一个完整的异步流式处理示例：

import aiohttp
import json

async def stream_chat_completion(messages, model="gpt-3.5-turbo"):
    headers = {"Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }

    payload = {
        "model": model,
        "messages": messages,
        "stream": True
    }

    timeout = aiohttp.ClientTimeout(total=60)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.post(
            "https://api.openai.com/v1/chat/completions",
            headers=headers,
            json=payload
        ) as response:
            async for line in response.content:
                if line.startswith(b"data:"):
                    chunk = line[6:].strip()
                    if chunk == b"[DONE]":
                        break
                    try:
                        data = json.loads(chunk)
                        yield data
                    except json.JSONDecodeError:
                        continue

OpenAI 提供了使用量监控 API，建议定期调用并记录到监控系统中：

async def get_usage():
    headers = {"Authorization": f"Bearer {API_KEY}"}
    async with aiohttp.ClientSession() as session:
        async with session.get(
            "https://api.openai.com/v1/usage",
            headers=headers
        ) as response:
            return await response.json()

如果处理用户敏感数据，建议：
1. 实现数据匿名化
2. 遵守 GDPR 等隐私法规
3. 配置数据保留策略