Python requests调用ChatGPT API实战指南：从认证到异常处理

3次阅读

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

在直接使用 Python 的 requests 库调用 ChatGPT API 时，开发者常遇到三个典型痛点：动态 token 过期导致频繁重新认证、流式响应解析复杂容易出错，以及并发限制下的请求排队问题。本文将提供一套完整的解决方案，帮你绕过这些坑。

直接使用 requests 的优势在于轻量化和可控性，特别适合需要深度定制 HTTP 行为的场景：

无需安装额外依赖（官方 SDK 需要 openai>=0.27.0）
可精细控制连接池和超时设置
便于集成到现有请求中间件体系

而官方 SDK 更适合快速集成，它自动处理了：

token 自动刷新
流式响应封装
错误码标准化

下面是一个带自动重试和 token 刷新的 Client 实现（Python 3.8+）：

from typing import Optional, Generator
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

class ChatGPTClient:
    def __init__(self, api_key: str, organization: Optional[str] = None):
        self._api_key = api_key
        self._organization = organization
        self._session = self._create_session()

    def _create_session(self) -> requests.Session:
        session = requests.Session()
        # 配置指数退避重试策略
        retry = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[502, 503, 504]
        )
        session.mount('https://', HTTPAdapter(max_retries=retry))
        return session

    def _refresh_token(self) -> str:
        # 实际项目中应从安全存储获取新 token
        return self._api_key

    def chat_completion(
        self,
        messages: list[dict],
        model: str = "gpt-3.5-turbo",
        stream: bool = False
    ) -> Generator[dict, None, None] | dict:
        url = "https://api.openai.com/v1/chat/completions"
        headers = {"Authorization": f"Bearer {self._api_key}",
            "Content-Type": "application/json"
        }
        if self._organization:
            headers["OpenAI-Organization"] = self._organization

        data = {
            "model": model,
            "messages": messages,
            "stream": stream
        }

        try:
            response = self._session.post(
                url,
                headers=headers,
                json=data,
                timeout=30,
                stream=stream
            )
            response.raise_for_status()

            if stream:
                return self._handle_stream_response(response)
            return response.json()
        except requests.HTTPError as e:
            if e.response.status_code == 401:  # token 过期
                self._api_key = self._refresh_token()
                return self.chat_completion(messages, model, stream)
            raise

    def _handle_stream_response(self, response: requests.Response) -> Generator[dict, None, None]:
        for line in response.iter_lines():
            if line:
                decoded = line.decode('utf-8')
                if decoded.startswith('data:'):
                    chunk = decoded[6:]
                    if chunk != '[DONE]':
                        yield json.loads(chunk)

处理流式响应时，推荐使用生成器模式：

def print_stream_response(client: ChatGPTClient, prompt: str):
    messages = [{"role": "user", "content": prompt}]
    for chunk in client.chat_completion(messages, stream=True):
        content = chunk.get('choices', [{}])[0].get('delta', {}).get('content', '')
        if content:
            print(content, end='', flush=True)

实现令牌桶算法控制请求速率
监控 headers 中的 x-ratelimit-* 字段
重要业务逻辑添加队列缓冲

API 密钥使用 Vault 或 AWS Secrets Manager
临时 token 存放在内存而非环境变量
请求日志过滤敏感头字段

对于高并发场景，可替换为 aiohttp 实现：

import aiohttp

async def async_chat_completion(session: aiohttp.ClientSession, messages: list[dict]):
    async with session.post(
        'https://api.openai.com/v1/chat/completions',
        json={"model": "gpt-3.5-turbo", "messages": messages}
    ) as resp:
        return await resp.json()

使用 pytest-mock 进行测试：

import pytest

def test_token_refresh(mocker):
    client = ChatGPTClient("expired_key")
    mock_response = mocker.Mock()
    mock_response.status_code = 401

    # 第一次模拟 token 过期
    mocker.patch.object(client._session, 'post', 
        side_effect=[requests.HTTPError(response=mock_response),
            mocker.Mock(json=lambda: {'choices': [{'message': {'content': 'Hi'}}]})
        ]
    )

    # 应自动刷新 token 并重试
    result = client.chat_completion([{"role": "user", "content": "Hello"}])
    assert 'Hi' in result['choices'][0]['message']['content']

当需要持久化对话状态时，你会如何设计上下文管理系统？考虑以下维度：