ChatGPT API 购买与集成实战：从选型到生产环境避坑指南

1次阅读

没有评论

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

在直接使用 ChatGPT 网页版时，开发者常遇到功能限制、无法自动化集成等问题。而通过 API 调用则能实现更灵活的集成和定制化需求。但自行搭建代理服务存在以下风险：

违反服务条款可能导致账户封禁
难以保证服务的稳定性和响应速度
缺乏官方支持，问题排查困难

价格透明，按实际使用量计费
功能全面，支持最新模型版本
QPS 限制相对严格，需要合理规划调用频率

集成到现有云服务中，便于统一管理
可能提供更高的 QPS 限制
价格可能略高于官方 API，且功能更新可能有延迟

建议根据项目需求和现有技术栈进行选择。若已在 AWS/Azure 上运行其他服务，托管版本可能更合适。

登录 OpenAI 官网，进入 API 页面
创建新的 API Key，妥善保存
在 Billing 页面设置付费计划和预算限制

import aiohttp
import asyncio
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

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

    try:
        async with aiohttp.ClientSession() as session:
            async with session.post(
                'https://api.openai.com/v1/chat/completions',
                headers=headers,
                json=data,
                timeout=30
            ) as response:
                if response.status == 200:
                    return await response.json()
                else:
                    logger.error(f'API call failed with status {response.status}')
                    return None
    except Exception as e:
        logger.error(f'Error calling API: {str(e)}')
        return None

const axios = require('axios');
const {Readable} = require('stream');

async function streamChatResponse(prompt, apiKey) {
    const response = await axios({
        method: 'post',
        url: 'https://api.openai.com/v1/chat/completions',
        headers: {'Authorization': `Bearer ${apiKey}`,
            'Content-Type': 'application/json'
        },
        data: {
            model: 'gpt-3.5-turbo',
            messages: [{role: 'user', content: prompt}],
            stream: true
        },
        responseType: 'stream'
    });

    return new Readable({read() {},
        destroy(err, callback) {response.data.destroy();
            callback(err);
        }
    }).pipe(response.data);
}

对于 429 状态码（Too Many Requests），建议采用指数退避策略：

首次重试延迟 1 秒
每次失败后延迟时间翻倍
最大重试次数不超过 5 次

import redis
import time

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

def check_rate_limit(api_key, limit=60, window=60):
    current_time = int(time.time())
    key = f'rate_limit:{api_key}'

    pipe = r.pipeline()
    pipe.zadd(key, {current_time: current_time})
    pipe.zremrangebyscore(key, 0, current_time - window)
    pipe.zcard(key)
    _, _, count = pipe.execute()

    return count <= limit

建议监控以下指标：

API 调用次数
平均响应时间
错误率
花费预估

实现令牌桶算法控制请求速率：

import time

class TokenBucket:
    def __init__(self, capacity, fill_rate):
        self.capacity = float(capacity)
        self.tokens = float(capacity)
        self.fill_rate = float(fill_rate)
        self.last_time = time.time()

    def consume(self, tokens=1):
        now = time.time()
        delta = now - self.last_time
        self.last_time = now

        self.tokens = min(self.capacity, self.tokens + delta * self.fill_rate)

        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        return False

在请求 API 前，建议：