ChatGPT API接口调用实战：从认证到流式响应的完整指南

1次阅读

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

调用 ChatGPT API 时，开发者常遇到以下典型问题：

认证失败：API 密钥泄露或格式错误导致 401 错误
响应延迟：同步阻塞调用造成 UI 卡顿（平均响应时间 2 - 4 秒）
Token 限制：超过 max_tokens 参数限制触发 400 错误
速率限制：免费账户每分钟 3 次请求的限制（返回 429 状态码）
流式中断：网络波动导致 stream 响应中途断开

获取 API 密钥
登录 OpenAI 账户后从 API keys 页面生成
注意密钥格式以 sk- 开头

构造认证头

headers = {"Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}

请求参数配置

必填参数参考官方文档：

{
    "model": "gpt-3.5-turbo",
    "messages": [{"role": "user", "content": "你的问题"}],
    "temperature": 0.7
}

方式	适用场景	代码库推荐
同步调用	简单脚本 / 低并发场景	requests
异步调用	高并发 / 需要响应式 UI	httpx

同步示例：

import requests

response = requests.post(
    "https://api.openai.com/v1/chat/completions",
    headers=headers,
    json=payload
)

异步示例：

import httpx

async with httpx.AsyncClient() as client:
    response = await client.post(
        "https://api.openai.com/v1/chat/completions",
        headers=headers,
        json=payload
    )

启用 stream 参数后需要特殊处理响应块：

# 设置 stream=True
payload["stream"] = True

with requests.post(
    "https://api.openai.com/v1/chat/completions",
    headers=headers,
    json=payload,
    stream=True
) as response:
    for chunk in response.iter_lines():
        if chunk:
            print(chunk.decode("utf-8"))  # 实际处理需解析 JSON

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 重试策略配置
retry_strategy = Retry(
    total=3,
    backoff_factor=1,
    status_forcelist=[429, 500, 502, 503, 504]
)
adapter = HTTPAdapter(max_retries=retry_strategy)

# 带异常处理的调用
with requests.Session() as session:
    session.mount("https://", adapter)
    try:
        response = session.post(
            "https://api.openai.com/v1/chat/completions",
            headers=headers,
            json=payload,
            timeout=10
        )
        response.raise_for_status()
        print(response.json()["choices"][0]["message"]["content"])
    except requests.exceptions.RequestException as e:
        print(f"请求失败: {e}")

针对 429 状态码建议采用指数退避算法：

import time
import random

def exponential_backoff(retries):
    return random.uniform(0, (2 ** retries) - 1)

for i in range(3):
    try:
        response = requests.post(...)
        break
    except TooManyRequestsError:
        wait_time = exponential_backoff(i)
        time.sleep(wait_time)

通过 usage 字段计算消耗（1000 tokens≈$0.002）：

data = response.json()
input_tokens = data["usage"]["prompt_tokens"]
output_tokens = data["usage"]["completion_tokens"]
total_cost = (input_tokens + output_tokens) / 1000 * 0.002

编码问题
错误：响应内容包含非 UTF- 8 字符
解决：强制指定response.encoding = "utf-8"
代理配置
错误：企业网络拦截 API 请求

解决：

proxies = {
    "http": "http://proxy_ip:port",
    "https": "http://proxy_ip:port"
}

超时设置
错误：未设超时导致线程阻塞
解决：requests.post(..., timeout=(3.05, 27))
Stream 缓存
错误：流式响应未及时消费导致内存溢出
解决：使用 iter_content(chunk_size=1024) 逐块处理
Token 计算
错误：消息过长触发 400 错误

解决：使用 tiktoken 库预计算：

import tiktoken
encoder = tiktoken.encoding_for_model("gpt-3.5-turbo")
tokens = encoder.encode("你的文本")

挑战任务：
1. 实现一个自动切换同步 / 异步调用的智能封装类
2. 增加响应内容实时打字机效果（针对 stream 模式）
3. 设计基于 Redis 的请求限速中间件

对比工具：

# cURL 示例（同步）curl https://api.openai.com/v1/chat/completions \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"model":"gpt-3.5-turbo","messages": [{"role":"user","content":" 你好 "}]}'

通过以上方案，可实现日均百万级调用的稳定服务。建议结合业务场景选择合适的并发策略，并持续监控 API 使用情况。

正文完