ChatGPT API访问全指南：从认证到高并发优化的工程实践

1次阅读

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

在对接 ChatGPT API 时，开发者常遇到以下几个典型问题：

认证失败（401 错误）：API 密钥配置错误或过期
速率限制（429 错误）：超过每分钟 / 每天的请求配额
长文本处理困难：API 对输入长度有限制（如 4096 tokens）
高延迟：同步请求导致的性能瓶颈
流式响应处理复杂：大模型响应需要特殊的内存管理

登录 OpenAI 账户后，在 API keys 页面创建新密钥。建议为不同环境（开发 / 生产）创建独立密钥。

curl https://api.openai.com/v1/chat/completions \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"model":"gpt-3.5-turbo","messages": [{"role":"user","content":"Hello!"}]
  }'

import openai

# 推荐从环境变量读取 API 密钥
import os
openai.api_key = os.getenv("OPENAI_API_KEY")

response = openai.ChatCompletion.create(
  model="gpt-3.5-turbo",
  messages=[{"role": "user", "content": "Explain Python decorators"}]
)
print(response.choices[0].message.content)

当遇到 429 错误时，应采用指数退避算法自动重试。以下是 Python 实现：

import time
import random
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def call_chatgpt_api(prompt):
    try:
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}]
        )
        return response.choices[0].message.content
    except openai.error.RateLimitError:
        print("Rate limit reached, retrying...")
        raise
    except openai.error.APIError as e:
        print(f"API error: {e}")
        raise

Node.js 版本使用axios-retry:

const axios = require('axios');
const axiosRetry = require('axios-retry');

axiosRetry(axios, {
  retries: 3,
  retryDelay: (retryCount) => {return Math.pow(2, retryCount) * 1000 + Math.random() * 1000;}
});

import aiohttp
import asyncio

async def async_chat_completion(session, messages):
    headers = {"Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }

    async with session.post(
        "https://api.openai.com/v1/chat/completions",
        headers=headers,
        json={"model": "gpt-3.5-turbo", "messages": messages}
    ) as response:
        if response.status == 429:
            await asyncio.sleep(2 ** retry_count)
            return await async_chat_completion(session, messages)
        return await response.json()

async def batch_requests(prompts):
    connector = aiohttp.TCPConnector(limit=10)  # 控制并发连接数
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [async_chat_completion(session, [{"role":"user","content":p}]) 
                for p in prompts]
        return await asyncio.gather(*tasks, return_exceptions=True)

对于相似请求，可以使用 temperature=0 并合并为单个 API 调用：

# 将多个问题合并为一个对话
messages = [{"role": "user", "content": "Q1: What is Python?"},
    {"role": "user", "content": "Q2: Explain list comprehension"}
]

response = openai.ChatCompletion.create(
    model="gpt-3.5-turbo",
    messages=messages,
    n=2  # 生成 2 个独立回答
)

并发数	平均延迟(ms)	成功率
1	1200	100%
5	1500	100%
10	2300	98%
20	3500	95%

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

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

使用 AWS Secrets Manager 或 HashiCorp Vault 存储密钥
开发环境使用 .env 文件（加入.gitignore）

def sanitize_input(text):
    patterns = [r"\b\d{4}[-\.\s]?\d{4}[-\.\s]?\d{4}\b",  # 信用卡号
        r"\b\d{3}-\d{2}-\d{4}\b"  # SSN
    ]
    for pattern in patterns:
        text = re.sub(pattern, "[REDACTED]", text)
    return text