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

2次阅读

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

初次接触 ChatGPT API 的开发者常会遇到以下几个典型问题：

API 版本混淆：OpenAI 的 API 版本迭代较快，不同版本的参数和功能略有差异，容易导致请求失败或返回意外结果。
token 计算误差：未正确计算 token 数量可能导致请求被截断或超额收费。
流式响应截断：在流式传输场景下，不恰当的处理方式可能导致响应不完整或连接中断。

安装必要的库

pip install openai tiktoken tenacity

配置 API 密钥

建议将 API 密钥存储在环境变量中，而非直接硬编码在代码里：

import openai
import os

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

from tenacity import retry, stop_after_attempt, wait_exponential
import openai

@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,
            temperature=0.7,
        )
        return response.choices[0].message.content
    except Exception as e:
        print(f"请求失败: {str(e)}")
        raise

import asyncio

async def async_chat_completion(messages):
    try:
        response = await openai.ChatCompletion.acreate(
            model="gpt-3.5-turbo",
            messages=messages,
            temperature=0.7,
        )
        return response.choices[0].message.content
    except Exception as e:
        print(f"异步请求失败: {str(e)}")
        raise

messages 参数是一个字典列表，每个字典包含以下字段：

role: 可以是 ”system”、”user” 或 ”assistant”
content: 消息内容

示例：

messages = [{"role": "system", "content": "你是一个有帮助的助手。"},
    {"role": "user", "content": "今天天气怎么样？"}
]

使用 tiktoken 库可以准确计算 token 数量：

import tiktoken

def num_tokens_from_messages(messages, model="gpt-3.5-turbo"):
    encoding = tiktoken.encoding_for_model(model)
    num_tokens = 0
    for message in messages:
        num_tokens += 4  # 每个消息有 4 个 token 的开销
        for key, value in message.items():
            num_tokens += len(encoding.encode(value))
            if key == "name":
                num_tokens += -1  # 如果有 name 字段，调整
    num_tokens += 2  # 回复的开销
    return num_tokens

对于长时间运行的对话，使用流式响应可以显著改善用户体验：

response = openai.ChatCompletion.create(
    model="gpt-3.5-turbo",
    messages=messages,
    stream=True,
)

for chunk in response:
    print(chunk.choices[0].delta.get("content", ""), end="")

在发送请求前，建议对用户输入进行敏感数据过滤：

def filter_sensitive_data(text):
    sensitive_keywords = [...]  # 定义敏感词列表
    for keyword in sensitive_keywords:
        text = text.replace(keyword, "[FILTERED]")
    return text

OpenAI API 有调用频率限制，建议实现限速策略：

import time

class RateLimiter:
    def __init__(self, calls_per_minute):
        self.calls_per_minute = calls_per_minute
        self.last_call_time = 0
        self.call_count = 0

    def wait_if_needed(self):
        now = time.time()
        if now - self.last_call_time >= 60:
            self.call_count = 0
            self.last_call_time = now
        self.call_count += 1
        if self.call_count > self.calls_per_minute:
            time.sleep(60 - (now - self.last_call_time))
            self.call_count = 0
            self.last_call_time = time.time()

为降低 token 消耗，可以设计上下文缓存机制，只保留最近几轮对话：

def trim_context(messages, max_tokens=1000):
    while num_tokens_from_messages(messages) > max_tokens and len(messages) > 1:
        messages.pop(1)  # 保留系统消息，删除最早的对话
    return messages