Python连接ChatGPT接口实战：从认证到流式响应的完整解决方案

1次阅读

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

在对接 ChatGPT API 时，很多 Python 开发者会遇到以下典型问题：

认证失效 ：OAuth2.0 的 access token 有过期时间，需要处理 refresh token 的逻辑
流式响应解析困难 ：直接拼接 chunked encoding 数据可能导致消息不完整
速率限制 ：API 有严格的请求频率限制，突发流量容易被拒绝
错误重试复杂 ：网络波动或服务端错误时需要智能重试机制
性能瓶颈 ：同步请求在高并发场景下效率低下

import requests

response = requests.post(
    'https://api.openai.com/v1/chat/completions',
    headers={'Authorization': f'Bearer {token}'},
    json={'model': 'gpt-3.5-turbo', 'messages': [...]}
)

缺点：
– 阻塞式 I /O，无法并发处理多个请求
– 流式响应需要手动处理分块数据

import aiohttp

async with aiohttp.ClientSession() as session:
    async with session.post(
        'https://api.openai.com/v1/chat/completions',
        headers={'Authorization': f'Bearer {token}'},
        json={'model': 'gpt-3.5-turbo', 'messages': [...]}
    ) as response:
        data = await response.json()

优势：
– 非阻塞 I /O，适合高并发场景
– 内置支持流式响应处理
– 连接池复用降低延迟

class OpenAIAuth:
    def __init__(self, client_id, client_secret):
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = "https://api.openai.com/v1/auth/token"
        self._access_token = None
        self._refresh_token = None

    async def get_token(self):
        if self._access_token and not self._is_token_expired():
            return self._access_token

        async with aiohttp.ClientSession() as session:
            data = {
                'grant_type': 'client_credentials',
                'client_id': self.client_id,
                'client_secret': self.client_secret
            }
            async with session.post(self.token_url, data=data) as resp:
                result = await resp.json()
                self._access_token = result['access_token']
                self._refresh_token = result.get('refresh_token')
                return self._access_token

    def _is_token_expired(self):
        # 实际项目应解析 JWT 中的 exp 字段
        return False  # 简化示例

async def stream_response(session, prompt):
    url = "https://api.openai.com/v1/chat/completions"
    headers = {"Authorization": f"Bearer {token}",
        "Accept": "text/event-stream"
    }
    data = {
        "model": "gpt-3.5-turbo",
        "messages": [{"role": "user", "content": prompt}],
        "stream": True
    }

    full_response = ""
    async with session.post(url, headers=headers, json=data) as resp:
        async for line in resp.content:
            if line.startswith(b'data:'):
                chunk = line[6:].strip()
                if chunk == b'[DONE]':
                    break
                try:
                    json_chunk = json.loads(chunk)
                    delta = json_chunk['choices'][0]['delta'].get('content', '')
                    full_response += delta
                    print(delta, end='', flush=True)
                except json.JSONDecodeError:
                    continue
    return full_response

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
async def safe_api_call(session, prompt):
    try:
        return await stream_response(session, prompt)
    except aiohttp.ClientError as e:
        print(f"请求失败: {e}")
        raise

connector = aiohttp.TCPConnector(
    limit=20,  # 最大连接数
    limit_per_host=5,  # 单主机最大连接
    enable_cleanup_closed=True,  # 自动清理关闭的连接
    force_close=False  # 禁用强制关闭
)

async with aiohttp.ClientSession(connector=connector) as session:
    # 使用优化后的 session

方案	原始大小	压缩后大小	节省比例
JSON	1.2KB	1.1KB	8.3%
MsgPack	1.2KB	0.8KB	33.3%

在请求头中明确指定 API 版本：

headers = {
    "OpenAI-Version": "2023-05-15",
    "Authorization": f"Bearer {token}"
}

使用环境变量替代硬编码：

import os

client_id = os.getenv('OPENAI_CLIENT_ID')
client_secret = os.getenv('OPENAI_CLIENT_SECRET')

实现思路：

使用装饰器记录函数执行时间
将指标写入 Prometheus 或 StatsD
设置报警阈值（如 P99>500ms）

示例代码：

import time
from functools import wraps

def timed_request(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        start = time.monotonic()
        result = await func(*args, **kwargs)
        duration = time.monotonic() - start
        print(f"请求耗时: {duration:.2f}s")
        return result
    return wrapper

@timed_request
async def monitored_api_call(session, prompt):
    return await safe_api_call(session, prompt)

本文完整实现了 Python 对接 ChatGPT API 的关键技术点，包括：