ChatGPT API调用实战：从基础接入到生产环境优化

1次阅读

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

在集成 ChatGPT API 时，开发者常遇到几个典型问题：

Token 管理复杂 ：API 按 token 数量计费，但准确计算消耗量困难
响应延迟波动 ：对话长度增加时，响应时间可能从几百毫秒骤增至秒级
速率限制陷阱 ：免费账号仅支持 3 RPM（每分钟请求数），付费账号也有分层限制
流式响应处理 ：直接输出 stream 数据可能导致信息不完整或格式混乱

import openai

response = openai.ChatCompletion.create(
    model="gpt-3.5-turbo",
    messages=[{"role": "user", "content": "解释量子计算"}]
)
print(response.choices[0].message.content)

适用场景 ：
– 简单问答场景
– 测试环境验证
– 不需要实时交互的批处理

import asyncio
import openai

async def async_query():
    resp = await openai.ChatCompletion.acreate(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": "写一首关于 AI 的诗"}]
    )
    return resp

result = asyncio.run(async_query())

优势：
– 适合高并发场景
– 避免阻塞主线程
– 天然适配流式响应

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_with_retry(prompt):
    try:
        return openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}],
            timeout=10  # 秒
        )
    except Exception as e:
        print(f"Attempt failed: {str(e)}")
        raise

response = openai.ChatCompletion.create(
    model="gpt-3.5-turbo",
    messages=[{"role": "user", "content": "用中文讲个笑话"}],
    stream=True
)

full_response = []
for chunk in response:
    content = chunk["choices"][0].get("delta", {}).get("content", "")
    if content:
        print(content, end="", flush=True)
        full_response.append(content)

final_output = "".join(full_response)

def safe_api_call(prompt):
    try:
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}],
            timeout=(3.05, 27)  # 连接超时 3 秒，读取超时 27 秒
        )
        return response.choices[0].message.content
    except openai.error.APIError as e:
        # 处理 API 级别错误
        return f"API Error: {str(e)}"
    except Exception as e:
        # 记录到监控系统
        sentry_sdk.capture_exception(e)
        return "Service unavailable"

使用 Semaphore 控制并发量：

import asyncio
from typing import List

semaphore = asyncio.Semaphore(5)  # 最大并发 5

async def limited_call(prompt: str) -> str:
    async with semaphore:
        return await chat_with_retry(prompt)

推荐方案：

使用环境变量存储 API Key
密钥轮换周期不超过 90 天
通过 Vault 或 KMS 管理密钥
禁用控制台日志输出

认证失败
检查 OPENAI_API_KEY 环境变量
验证密钥是否过期
确保没有多余空格或换行符
响应截断
设置 max_tokens 参数（默认 4096）
实现自动续接机制
速率限制
免费账号：3 RPM
付费账号：3500 RPM
建议实现请求队列
内容过滤
检查 moderation 端点
设置 temperature= 0 减少随机性
长文本处理
拆分超过 token 限制的文本
使用 gpt-3.5-turbo-16k 版本

import unittest
from unittest.mock import patch

class TestChatAPI(unittest.TestCase):
    @patch('openai.ChatCompletion.create')
    def test_normal_response(self, mock_create):
        mock_create.return_value = {"choices": [{"message": {"content": "测试响应"}}]
        }
        result = chat_with_retry("测试")
        self.assertIn("测试响应", result.choices[0].message.content)

在实际项目中，如何处理以下场景：