ChatGPT官网API集成实战：从认证到流式响应的全链路解析

1次阅读

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

集成 ChatGPT 官网 API 时，开发者常遇到三大痛点：认证流程复杂、流式响应解析困难、错误处理机制不完善。本文将深入解析这些难点，并提供可落地的解决方案。

ChatGPT 官网 API 采用 OAuth2.0 授权码模式进行认证。这是最安全的认证方式，但也带来了额外的复杂性。

sequenceDiagram
    participant Client
    participant AuthServer
    participant ResourceServer

    Client->>AuthServer: 1. 请求授权码
    AuthServer-->>Client: 2. 返回授权码
    Client->>AuthServer: 3. 用授权码换取 access_token
    AuthServer-->>Client: 4. 返回 access_token
    Client->>ResourceServer: 5. 携带 access_token 请求资源
    ResourceServer-->>Client: 6. 返回请求的资源

import requests
from authlib.integrations.requests_client import OAuth2Session

# 配置 OAuth2 客户端
client_id = 'your_client_id'
client_secret = 'your_client_secret'
redirect_uri = 'https://your-app.com/callback'
authorization_base_url = 'https://chatgpt.com/oauth/authorize'
token_url = 'https://chatgpt.com/oauth/token'

# 创建 OAuth2 会话
oauth = OAuth2Session(client_id, client_secret, redirect_uri=redirect_uri)

# 第一步：获取授权 URL
authorization_url, state = oauth.authorization_url(authorization_base_url)
print('请访问以下 URL 授权:', authorization_url)

# 第二步：用户授权后，获取授权码
redirect_response = input('请输入回调 URL:')

# 第三步：用授权码换取 access_token
token = oauth.fetch_token(token_url, authorization_response=redirect_response)
print('获取到的 token:', token)

必须确保 client_secret 的安全存储
建议实现 token 自动刷新机制
注意 token 有效期，设置合理的缓存策略

ChatGPT API 支持流式响应 (streaming response)，这对长文本生成特别有用。

特性	EventSource	WebSocket
协议	HTTP	独立协议
方向性	单向 (服务端 -> 客户端)	双向
重连机制	内置	需手动实现
数据格式	仅文本	任意二进制
浏览器支持	广泛	广泛

const {EventSource} = require('eventsource');

const es = new EventSource('https://api.chatgpt.com/v1/chat/completions', {
  headers: {'Authorization': `Bearer ${accessToken}`,
    'Content-Type': 'application/json',
    'Cache-Control': 'no-cache, must-revalidate'  // 重要：禁用缓存
  },
  body: JSON.stringify({
    model: 'gpt-4',
    messages: [{role: 'user', content: 'Hello!'}],
    stream: true
  })
});

es.on('message', (event) => {
  try {const data = JSON.parse(event.data);
    // 处理 UTF- 8 边界情况
    process.stdout.write(data.choices[0].delta.content || '');
  } catch (err) {console.error('解析错误:', err);
  }
});

es.on('error', (err) => {console.error('EventSource 错误:', err);
});

import requests

url = 'https://api.chatgpt.com/v1/chat/completions'
headers = {'Authorization': f'Bearer {access_token}',
    'Content-Type': 'application/json',
    'Cache-Control': 'no-cache, must-revalidate'
}

data = {
    'model': 'gpt-4',
    'messages': [{'role': 'user', 'content': 'Hello!'}],
    'stream': True
}

with requests.post(url, headers=headers, json=data, stream=True) as r:
    for chunk in r.iter_content(chunk_size=None):
        if chunk:
            # 处理分块数据
            decoded_chunk = chunk.decode('utf-8')
            for line in decoded_chunk.split('\n'):
                if line.startswith('data:'):
                    try:
                        json_data = json.loads(line[5:])
                        print(json_data['choices'][0]['delta'].get('content', ''), end='')
                    except json.JSONDecodeError:
                        pass

当 API 请求失败时，简单的立即重试可能加剧服务器负担。指数退避算法可以有效缓解这个问题。

import time
import random

def exponential_backoff_retry(func, max_retries=5, initial_delay=1):
    """指数退避重试装饰器"""
    def wrapper(*args, **kwargs):
        retries = 0
        while retries < max_retries:
            try:
                return func(*args, **kwargs)
            except Exception as e:
                if retries == max_retries - 1:
                    raise

                # 计算等待时间：基础延迟 * 2^ 重试次数 * (0.5~1.5 随机因子)
                delay = initial_delay * (2 ** retries) * (0.5 + random.random())
                time.sleep(delay)
                retries += 1
    return wrapper

# 使用示例
@exponential_backoff_retry
def call_chatgpt_api(prompt):
    # API 调用代码
    pass

服务端生成唯一 session token
客户端每次请求携带此 token
服务端维护 session 状态

from flask import Flask, request, jsonify
import uuid

app = Flask(__name__)
sessions = {}  # 实际生产环境应使用 Redis 等持久化存储

@app.route('/chat', methods=['POST'])
def chat():
    data = request.json
    session_token = data.get('session_token')

    if not session_token or session_token not in sessions:
        # 新会话
        session_token = str(uuid.uuid4())
        sessions[session_token] = {'messages': [data['message']],
            'created_at': time.time()}
    else:
        # 已有会话，追加消息
        sessions[session_token]['messages'].append(data['message'])

    # 调用 ChatGPT API...

    return jsonify({
        'response': api_response,
        'session_token': session_token
    })

使用 Redis 等内存数据库存储对话历史
设置合理的 TTL
适合高并发场景

import redis

r = redis.Redis(host='localhost', port=6379, db=0)

def get_chat_history(session_id):
    """从 Redis 获取对话历史"""
    history = r.get(f'chat:{session_id}')
    return json.loads(history) if history else []

def save_chat_history(session_id, messages):
    """保存对话历史到 Redis，设置 24 小时过期"""
    r.setex(f'chat:{session_id}', 24*3600, json.dumps(messages))

在高并发微服务架构中，如何设计 ChatGPT API 调用的熔断降级机制？