ChatGPT模型官方接口网址接入实战：从认证到高并发调用的全流程解析

1次阅读

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

ChatGPT API 采用 API Key 进行身份验证，每个请求需要在 HTTP 头部添加 Authorization 字段：

Authorization: Bearer YOUR_API_KEY

API Key 可以在 OpenAI 账户设置中生成和管理
每个 Key 有独立的权限和限额
建议按环境（开发 / 生产）使用不同 Key

按实际使用的 token 数量计费
计费包含请求和响应中的所有 token
不同模型有不同的单价（如 gpt-3.5-turbo 比 gpt- 4 便宜）

开发者常见的挑战包括：

认证失败：API Key 错误或过期
响应延迟：复杂请求可能需要数秒响应
并发限制：免费账户每分钟 3 次请求
token 计算：长文本可能意外耗尽配额
错误处理：429、500 等错误需要妥善处理

import openai
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def chat_completion_with_retry(messages, model="gpt-3.5-turbo"):
    try:
        response = openai.ChatCompletion.create(
            model=model,
            messages=messages,
            temperature=0.7,
        )
        return response
    except openai.error.RateLimitError:
        print("Rate limit exceeded, retrying...")
        raise
    except openai.error.APIError as e:
        print(f"API error: {e}")
        raise

对于 Node.js 环境，可以使用 async 库控制并发：

const async = require('async');
const openai = require('openai-api');

const OPENAI = new openai(process.env.OPENAI_API_KEY);

async function batchProcessPrompts(prompts, concurrency = 3) {return new Promise((resolve, reject) => {
    async.mapLimit(
      prompts,
      concurrency,
      async (prompt) => {
        try {
          const response = await OPENAI.complete({
            engine: 'text-davinci-003',
            prompt: prompt,
            max_tokens: 100,
          });
          return response.data;
        } catch (error) {console.error(`Error processing prompt: ${prompt}`, error);
          return null;
        }
      },
      (err, results) => {if (err) return reject(err);
        resolve(results.filter(Boolean));
      }
    );
  });
}

Python 中处理流式响应：

response = openai.ChatCompletion.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "讲一个长篇故事"}],
    stream=True
)

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

方式	100 次请求耗时	资源占用
同步	~120s	CPU 15%
异步	~25s	CPU 45%

设置合理的 max_tokens 参数
使用 stop 序列提前终止
精简输入 prompt
对长文本先进行摘要

永远不要提交到版本控制
使用环境变量管理
设置 IP 白名单
定期轮换密钥

实现指数退避重试
监控 429 错误率
考虑分布式限流

def sanitize_input(text):
    sensitive_patterns = [r'\b\d{16}\b',  # 信用卡号
        r'\b\d{3}-\d{2}-\d{4}\b'  # SSN
    ]

    for pattern in sensitive_patterns:
        text = re.sub(pattern, '[REDACTED]', text)

    return text

可以考虑以下 fallback 机制：