ChatGPT Plus API 接入实战：从认证流程到最佳实践

1次阅读

没有评论

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

ChatGPT Plus 是 OpenAI 提供的付费订阅服务，相比免费版本，它在 API 调用方面提供了显著的优势。以下是主要差异点：

速率限制 ：Plus 用户享有更高的请求速率限制，具体数值取决于订阅等级。
模型权限 ：Plus 用户可以访问更强大的模型版本，例如 GPT-4，而免费用户通常只能使用 GPT-3.5。
优先级访问 ：在高负载时段，Plus 用户的请求会被优先处理。
配额管理 ：Plus 用户可以通过 API 更灵活地管理使用配额。

OAuth 2.0 是 ChatGPT Plus API 的认证标准，以下是完整的授权流程：

用户通过客户端向授权服务器请求授权码（Authorization Code）。
授权服务器返回授权码。
客户端使用授权码向令牌服务器请求访问令牌（Access Token）。
令牌服务器验证授权码并返回访问令牌和刷新令牌（Refresh Token）。

以下是使用 Python requests 库实现这一流程的示例代码：

import requests
from typing import Dict, Optional

def get_access_token(client_id: str, client_secret: str, auth_code: str) -> Optional[Dict]:
    """
    Exchange authorization code for access token.
    :param client_id: OAuth client ID
    :param client_secret: OAuth client secret
    :param auth_code: Authorization code from OAuth flow
    :return: JSON response containing access token or None if failed
    """token_url ="https://api.openai.com/v1/oauth/token"data = {"client_id": client_id,"client_secret": client_secret,"code": auth_code,"grant_type":"authorization_code"}

    try:
        response = requests.post(token_url, data=data, timeout=10)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"Token request failed: {e}")
        return None

403 Forbidden：通常表示无效的客户端凭据或授权码。
429 Too Many Requests：表示速率限制被触发，建议实现指数退避重试机制。

为避免频繁请求新令牌，可以将令牌存储在 Redis 中并设置自动刷新逻辑。

import redis
import time

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

def get_cached_token(client_id: str) -> Optional[str]:
    """
    Get cached access token from Redis.
    Automatically refresh if expired.
    """token_key = f"token:{client_id}"
    if r.exists(token_key):
        return r.get(token_key).decode('utf-8')
    # Refresh logic here

将多个请求合并为一个批次请求，可以显著减少 API 调用次数。

def batch_requests(messages: List[Dict]) -> List[Dict]:
    """Send multiple messages in a single API call."""
    batch_url = "https://api.openai.com/v1/chat/completions"
    headers = {"Authorization": f"Bearer {access_token}"}
    data = {"messages": messages, "model": "gpt-4"}
    response = requests.post(batch_url, json=data, headers=headers)
    return response.json()

实现滑动窗口算法来控制请求速率，避免触发 API 限制。

from collections import deque
import time

class RateLimiter:
    def __init__(self, max_calls: int, period: float):
        self.max_calls = max_calls
        self.period = period
        self.timestamps = deque()

    def __call__(self) -> bool:
        now = time.time()
        while self.timestamps and now - self.timestamps[0] > self.period:
            self.timestamps.popleft()
        if len(self.timestamps) < self.max_calls:
            self.timestamps.append(now)
            return True
        return False