ChatGPT官网API接入实战：从零开始的开发者避坑指南

1次阅读

没有评论

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

对于开发者来说，首次接入 ChatGPT 官网 API 时往往会遇到一些共性问题。这些问题如果没有提前了解，可能会在开发过程中耗费大量时间排查。

认证失败 ：API 密钥过期、无效或被撤销是常见问题。很多开发者会直接将密钥硬编码在代码中，这不仅不安全，还难以维护。
流式响应处理 ：ChatGPT 支持流式响应，但很多新手开发者不知道如何正确处理和拼接这些分块数据。
敏感信息过滤 ：在生产环境中，需要对用户输入和 AI 输出进行适当的过滤，以避免显示不当内容。

REST API：适合简单的请求 - 响应场景，实现简单，但无法实现实时交互。
WebSocket：适合需要持续对话的场景，可以实现低延迟的双向通信，但实现复杂度较高。

安全存储 API 密钥是接入 ChatGPT 的第一步。以下是推荐的做法：

使用环境变量存储密钥
对密钥进行加密
限制密钥的访问权限

import os
from cryptography.fernet import Fernet

# 生成加密密钥
key = Fernet.generate_key()
cipher_suite = Fernet(key)

# 加密 API 密钥
api_key = os.getenv('OPENAI_API_KEY')
encrypted_key = cipher_suite.encrypt(api_key.encode())

# 解密 API 密钥
decrypted_key = cipher_suite.decrypt(encrypted_key).decode()

const crypto = require('crypto');
const algorithm = 'aes-256-cbc';
const key = crypto.randomBytes(32);
const iv = crypto.randomBytes(16);

function encrypt(text) {let cipher = crypto.createCipheriv(algorithm, Buffer.from(key), iv);
    let encrypted = cipher.update(text);
    encrypted = Buffer.concat([encrypted, cipher.final()]);
    return {iv: iv.toString('hex'), encryptedData: encrypted.toString('hex') };
}

function decrypt(text) {let iv = Buffer.from(text.iv, 'hex');
    let encryptedText = Buffer.from(text.encryptedData, 'hex');
    let decipher = crypto.createDecipheriv(algorithm, Buffer.from(key), iv);
    let decrypted = decipher.update(encryptedText);
    decrypted = Buffer.concat([decrypted, decipher.final()]);
    return decrypted.toString();}

import openai
from tenacity import retry, stop_after_attempt, wait_exponential

openai.api_key = os.getenv('OPENAI_API_KEY')

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def chat_completion(messages):
    try:
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=messages,
            timeout=10  # 设置超时
        )
        return response
    except Exception as e:
        print(f"API 调用失败: {str(e)}")
        raise

const {Configuration, OpenAIApi} = require('openai');

const configuration = new Configuration({
  apiKey: process.env.OPENAI_API_KEY,
  timeout: 10000 // 10 秒超时
});

const openai = new OpenAIApi(configuration);

async function chatCompletion(messages) {
    try {
        const response = await openai.createChatCompletion({
            model: "gpt-3.5-turbo",
            messages: messages
        });
        return response.data;
    } catch (error) {console.error("API 调用失败:", error.response?.status, error.message);
        throw error;
    }
}

流式响应可以显著改善用户体验，特别是在生成长文本时。以下是处理流式响应的示例代码：

def handle_stream_response(stream):
    full_response = ""
    try:
        for chunk in stream:
            if chunk.choices and chunk.choices[0].delta and chunk.choices[0].delta.content:
                content = chunk.choices[0].delta.content
                full_response += content
                print(content, end="", flush=True)
    except Exception as e:
        print(f"\n 流式响应处理出错: {str(e)}")
    finally:
        return full_response

async function handleStreamResponse(stream) {
    let fullResponse = '';
    try {for await (const chunk of stream) {const content = chunk.choices[0]?.delta?.content || '';
            process.stdout.write(content);
            fullResponse += content;
        }
    } catch (error) {console.error('\n 流式响应处理出错:', error);
    } finally {return fullResponse;}
}

为了避免触发 ChatGPT 的速率限制（通常会导致 429 错误），建议实现一个请求限速器。

import time
from collections import deque

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

    def wait(self):
        now = time.time()
        while len(self.timestamps) >= self.max_calls:
            oldest = self.timestamps[0]
            if now - oldest < self.period:
                sleep_time = self.period - (now - oldest)
                time.sleep(sleep_time)
                now = time.time()
            else:
                self.timestamps.popleft()
        self.timestamps.append(now)

# 使用示例: 限制每分钟 60 次请求
limiter = RateLimiter(60, 60)

# 在每次 API 调用前
limiter.wait()
response = chat_completion(messages)

在构建对话系统时，上下文管理至关重要。以下是两种常见的实现方式：

内存存储 ：简单快速，适合小型应用或原型开发
数据库存储 ：持久化存储，适合生产环境

import redis
import json

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

def save_context(user_id, messages):
    r.set(f'chat:{user_id}', json.dumps(messages), ex=3600)  # 1 小时过期

def load_context(user_id):
    data = r.get(f'chat:{user_id}')
    return json.loads(data) if data else []

监控 API 调用的性能指标对于生产环境至关重要。以下是一些关键指标：

延迟（Latency）
P99 响应时间
错误码分布

可以使用 Prometheus 和 Grafana 来构建监控系统：

from prometheus_client import Summary, Counter

# 定义指标
API_LATENCY = Summary('openai_api_latency_seconds', 'API 调用延迟')
API_ERRORS = Counter('openai_api_errors', 'API 错误次数', ['status_code'])

@API_LATENCY.time()
def monitored_chat_completion(messages):
    try:
        response = chat_completion(messages)
        return response
    except Exception as e:
        status = getattr(e, 'status_code', 'unknown')
        API_ERRORS.labels(status_code=status).inc()
        raise

良好的代码规范可以提高代码的可维护性和可靠性。

from typing import List, Dict, Any, Optional

def chat_completion(messages: List[Dict[str, str]], 
                   model: str = "gpt-3.5-turbo",
                   temperature: float = 0.7) -> Optional[Dict[str, Any]]:
    """
    与 ChatGPT API 交互

    Args:
        messages: 对话消息列表
        model: 使用的模型
        temperature: 控制生成随机性的参数

    Returns:
        API 响应或 None（如果出错）Raises:
        ValueError: 如果输入参数无效
        Exception: 如果 API 调用失败
    """
    if not messages:
        raise ValueError("消息列表不能为空")

    try:
        response = openai.ChatCompletion.create(
            model=model,
            messages=messages,
            temperature=temperature
        )
        return response
    except Exception as e:
        print(f"API 调用失败: {str(e)}")
        raise

⚠️ 注意：以下操作会消耗 token 配额：