Python调用ChatGPT API的实战指南：从认证到流式响应处理

1次阅读

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

ChatGPT API 采用标准的 RESTful 设计，遵循 HTTP 协议规范。认证机制使用 Bearer Token 方式，即在请求头中添加Authorization: Bearer {your_api_key}。这种设计具有以下特点：

无状态性：每个请求独立处理，服务端不保存客户端状态
资源导向：通过不同的 URL 路径访问不同功能（如 /v1/chat/completions）
JSON 交互：请求和响应体均采用 JSON 格式

实际开发中常遇到以下问题：

token 限制处理：
对话越长 token 消耗越多，可能触发 4096 或 8192 的硬限制
需要实时计算 token 用量（可用 tiktoken 库）
流式响应解析：
启用 stream=True 时响应变为多个 chunk
需要特殊处理增量数据拼接和中间状态维护
速率限制规避：
免费账号每分钟 3 次请求限制
异步调用时需控制并发量避免 429 错误

import os
import requests
from requests.adapters import HTTPAdapter

class ChatGPTClient:
    def __init__(self):
        self.api_key = os.getenv('OPENAI_API_KEY')
        self.session = requests.Session()
        # 配置重试策略
        adapter = HTTPAdapter(max_retries=3)
        self.session.mount('https://', adapter)

    def send_request(self, prompt):
        headers = {'Authorization': f'Bearer {self.api_key}',
            'Content-Type': 'application/json'
        }
        payload = {
            'model': 'gpt-3.5-turbo',
            'messages': [{'role': 'user', 'content': prompt}],
            'temperature': 0.7
        }

        try:
            with self.session.post(
                'https://api.openai.com/v1/chat/completions',
                headers=headers,
                json=payload,
                timeout=10
            ) as response:
                response.raise_for_status()
                return response.json()['choices'][0]['message']['content']
        except requests.exceptions.RequestException as e:
            print(f'Request failed: {e}')
            return None

import aiohttp
import asyncio

class AsyncChatGPTClient:
    def __init__(self):
        self.api_key = os.getenv('OPENAI_API_KEY')

    async def send_async_request(self, prompt):
        headers = {'Authorization': f'Bearer {self.api_key}',
            'Content-Type': 'application/json'
        }
        payload = {
            'model': 'gpt-3.5-turbo',
            'messages': [{'role': 'user', 'content': prompt}]
        }

        async with aiohttp.ClientSession() as session:
            try:
                async with session.post(
                    'https://api.openai.com/v1/chat/completions',
                    headers=headers,
                    json=payload,
                    timeout=aiohttp.ClientTimeout(total=15)
                ) as response:
                    if response.status == 200:
                        data = await response.json()
                        return data['choices'][0]['message']['content']
                    else:
                        error = await response.text()
                        raise Exception(f'API error: {error}')
            except Exception as e:
                print(f'Async request failed: {e}')
                return None

指标	requests 同步	aiohttp 异步
100 次请求耗时	12.7s	3.2s
内存占用	较低	较高
CPU 利用率	单核	多核

连接池配置：

conn = aiohttp.TCPConnector(
    limit=30,  # 最大连接数
    limit_per_host=10,  # 单主机连接数
    enable_cleanup_closed=True  # 自动清理关闭连接
)

幂等性设计：
为每条消息生成唯一 ID
服务端可通过 message_id 去重
响应缓存：
对高频问题结果缓存
使用 LRU 策略控制缓存大小

密钥管理：

# 错误做法
API_KEY = 'sk-xxx...'

# 正确做法
import os
from dotenv import load_dotenv
load_dotenv()
API_KEY = os.getenv('OPENAI_API_KEY')

异步改造：

# 同步阻塞写法（不推荐）result = client.send_request(prompt)

# 异步改进（推荐）async def process_prompts(prompts):
    tasks = [client.send_async_request(p) for p in prompts]
    return await asyncio.gather(*tasks, return_exceptions=True)