ChatGPT API接入实战：从认证到流式响应的完整指南

1次阅读

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

在实际开发中，直接调用 ChatGPT API 可能会遇到一些常见问题：

token 计算误差 ：OpenAI 的 token 计算方式与常见的分词器不同，容易导致请求超出限制
流式响应中断 ：网络波动或超时可能导致流式响应中途断开
上下文长度限制 ：长对话场景下需要合理管理上下文，避免超出模型限制

REST API：
优点：实现简单，适合低频请求
缺点：每次请求都需要建立新连接，延迟较高
WebSocket：
优点：保持长连接，适合高频交互场景
缺点：实现复杂度高，服务器资源消耗大

获取 API Key：登录 OpenAI 平台创建 API Key
安全存储：推荐使用环境变量或密钥管理服务
请求认证：在请求头中添加 Authorization: Bearer <your_api_key>

import openai
import asyncio

async def chat_completion(messages):
    try:
        response = await openai.ChatCompletion.acreate(
            model="gpt-3.5-turbo",
            messages=messages,
            stream=True
        )
        async for chunk in response:
            yield chunk['choices'][0]['delta'].get('content', '')
    except Exception as e:
        # 错误重试逻辑
        print(f"Error: {e}")
        await asyncio.sleep(1)
        await chat_completion(messages)

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

const config = new Configuration({apiKey: process.env.OPENAI_API_KEY});

const openai = new OpenAIApi(config);

async function streamChat(messages) {
  const response = await openai.createChatCompletion({
    model: "gpt-3.5-turbo",
    messages,
    stream: true
  }, {responseType: 'stream'});

  let buffer = '';
  response.data.on('data', (chunk) => {const payloads = chunk.toString().split('\n\n');
    for (const payload of payloads) {if (payload.includes('[DONE]')) return;
      if (!payload.startsWith('data:')) continue;

      try {const data = JSON.parse(payload.replace('data:', ''));
        const content = data.choices[0]?.delta?.content || '';
        buffer += content;
      } catch (e) {console.error('Error parsing chunk:', e);
      }
    }
  });

  return buffer;
}

from collections import deque
import time

class RateLimiter:
    def __init__(self, max_requests, time_window):
        self.max_requests = max_requests
        self.time_window = time_window
        self.request_times = deque()

    async def wait(self):
        now = time.time()
        # 移除超过时间窗口的请求
        while self.request_times and now - self.request_times[0] > self.time_window:
            self.request_times.popleft()

        if len(self.request_times) >= self.max_requests:
            sleep_time = self.time_window - (now - self.request_times[0])
            await asyncio.sleep(sleep_time)
            return await self.wait()

        self.request_times.append(now)

全量携带 ：每次请求都携带完整对话历史
优点：上下文完整
缺点：token 消耗大
摘要压缩 ：将历史对话压缩为摘要
优点：节省 token
缺点：可能丢失细节

对用户输入进行过滤和转义
使用系统消息明确角色设定
考虑在 API 网关层添加防护

import openai
from datetime import datetime

def track_cost(response):
    usage = response.get('usage', {})
    timestamp = datetime.now().isoformat()
    cost = calculate_cost(usage)

    log_entry = {
        'timestamp': timestamp,
        'prompt_tokens': usage.get('prompt_tokens', 0),
        'completion_tokens': usage.get('completion_tokens', 0),
        'total_tokens': usage.get('total_tokens', 0),
        'estimated_cost': cost
    }

    # 写入日志系统或数据库
    save_to_logging_system(log_entry)

在实际项目中接入 ChatGPT API 时，除了关注基础功能实现外，还需要特别注意生产环境中的稳定性、性能和成本控制。建议从小规模测试开始，逐步增加负载，同时建立完善的监控系统，确保服务的可靠性。

对于需要长期维护的项目，可以考虑封装统一的 SDK，将认证、限流、错误处理等通用逻辑集中管理，这样可以大大提高开发效率和系统稳定性。

正文完