ChatGPT API实战指南：从接入到生产环境优化的完整解决方案

1次阅读

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

在开发基于 ChatGPT API 的应用时，开发者常常会遇到以下典型问题：

认证失败 ：API 密钥过期或无效导致请求被拒绝
上下文丢失 ：多轮对话中无法有效维护历史消息上下文
限流问题 ：突发流量导致 API 调用被限流
响应延迟 ：高并发场景下 API 响应时间不稳定
错误处理不足 ：网络波动或服务端错误导致请求失败

根据应用场景选择适当的调用方式：

同步调用 ：适用于简单请求 - 响应模式，实现简单但吞吐量有限
异步调用 ：适合高并发场景，可通过事件循环或协程提高吞吐量

import time
import jwt
from datetime import datetime, timedelta

class AuthManager:
    def __init__(self, api_key):
        self.api_key = api_key
        self.token = None
        self.expires_at = None

    def get_token(self):
        if self.token and datetime.now() < self.expires_at:
            return self.token

        # 生成新 token
        payload = {
            'iss': 'your_service',
            'exp': datetime.now() + timedelta(minutes=30),
            'iat': datetime.now()}
        self.token = jwt.encode(payload, self.api_key, algorithm='HS256')
        self.expires_at = datetime.now() + timedelta(minutes=25)  # 提前 5 分钟刷新
        return self.token

const jwt = require('jsonwebtoken');

class AuthManager {constructor(apiKey) {
    this.apiKey = apiKey;
    this.token = null;
    this.expiresAt = null;
  }

  getToken() {if (this.token && new Date() < this.expiresAt) {return this.token;}

    const payload = {
      iss: 'your_service',
      exp: Math.floor(Date.now() / 1000) + 1800, // 30 分钟后过期
      iat: Math.floor(Date.now() / 1000)
    };

    this.token = jwt.sign(payload, this.apiKey);
    this.expiresAt = new Date(Date.now() + 25 * 60 * 1000); // 提前 5 分钟刷新
    return this.token;
  }
}

设计架构：

每个对话分配唯一 session_id
将对话历史存储在 Redis 有序集合中
使用 TTL 自动清理过期对话

import redis
import json

class DialogueManager:
    def __init__(self):
        self.redis = redis.Redis(host='localhost', port=6379, db=0)

    def add_message(self, session_id, role, content):
        message = {'role': role, 'content': content}
        self.redis.zadd(f'dialogue:{session_id}', {json.dumps(message): time.time()})
        self.redis.expire(f'dialogue:{session_id}', 3600)  # 1 小时过期

    def get_history(self, session_id, max_messages=10):
        messages = self.redis.zrevrange(f'dialogue:{session_id}', 0, max_messages-1)
        return [json.loads(msg) for msg in messages]

import requests
import time
import logging
from requests.exceptions import RequestException

class ChatGPTAPI:
    def __init__(self, api_key, base_url='https://api.openai.com/v1'):
        self.auth = AuthManager(api_key)
        self.base_url = base_url
        self.session = requests.Session()

    def call_with_retry(self, endpoint, payload, max_retries=3):
        url = f"{self.base_url}/{endpoint}"
        headers = {'Authorization': f"Bearer {self.auth.get_token()}",
            'Content-Type': 'application/json'
        }

        for attempt in range(max_retries):
            try:
                response = self.session.post(
                    url,
                    headers=headers,
                    json=payload,
                    timeout=30
                )

                if response.status_code == 429:  # Rate limited
                    retry_after = int(response.headers.get('Retry-After', 5))
                    time.sleep(retry_after + (2 ** attempt))  # Exponential backoff
                    continue

                response.raise_for_status()
                return response.json()

            except RequestException as e:
                logging.error(f"Attempt {attempt + 1} failed: {str(e)}")
                if attempt == max_retries - 1:
                    raise
                time.sleep(2 ** attempt)  # Exponential backoff

from collections import deque
import time

class RateLimiter:
    def __init__(self, max_calls, period):
        self.max_calls = max_calls
        self.period = period
        self.calls = deque()

    def check_limit(self):
        now = time.time()

        # 移除过期的调用记录
        while self.calls and self.calls[0] <= now - self.period:
            self.calls.popleft()

        if len(self.calls) >= self.max_calls:
            return False

        self.calls.append(now)
        return True

import re

def sanitize_input(text):
    # 移除信用卡号
    text = re.sub(r'\b(?:\d[ -]*?){13,16}\b', '[REDACTED]', text)

    # 移除邮箱
    text = re.sub(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', '[REDACTED]', text)

    return text

测试环境：
– 4 核 CPU/8GB 内存云服务器
– Python 3.8 + requests 库
– 测试 QPS: 50