ChatGPT API 接入实战：从注册到生产环境部署的完整指南

2次阅读

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

企业接入 ChatGPT API 时通常会遇到以下几个典型问题：

认证复杂度高 ：需要管理 API Key 的轮换和安全存储
API 调用限制 ：免费账户有严格的速率限制（3- 5 次 / 分钟），付费账户也需要考虑配额
响应延迟不稳定 ：长文本生成时可能遇到超时问题
成本控制困难 ：Token 消耗难以精确预估，特别是处理大量请求时
数据合规风险 ：用户输入可能包含敏感信息需要过滤

访问 OpenAI 官网并创建账户
进入 API Keys 管理页面
点击 ”Create new secret key” 生成 API Key
安全存储 Key（推荐使用密钥管理服务）

REST API：
优势：语言无关，适合多语言环境
劣势：需要自行处理重试、错误处理等逻辑
SDK：
优势：官方维护，内置最佳实践
劣势：语言绑定，更新依赖较频繁

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(prompt):
    try:
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}],
            timeout=30  # 设置超时时间
        )
        return response.choices[0].message.content
    except Exception as e:
        print(f"API 调用失败: {str(e)}")
        raise

const {Configuration, OpenAIApi} = require('openai');
const rateLimit = require('express-rate-limit');

// 初始化 OpenAI 客户端
const configuration = new Configuration({apiKey: process.env.OPENAI_API_KEY});
const openai = new OpenAIApi(configuration);

// 限流中间件
const limiter = rateLimit({
  windowMs: 60 * 1000, // 1 分钟
  max: 60 // 每分钟最大请求数
});

app.post('/api/chat', limiter, async (req, res) => {
  try {
    const completion = await openai.createChatCompletion({
      model: "gpt-3.5-turbo",
      messages: req.body.messages
    });
    res.json(completion.data);
  } catch (error) {
    // 错误处理逻辑
    if (error.response?.status === 429) {res.status(429).json({error: "请求过于频繁"});
    } else {res.status(500).json({error: "服务内部错误"});
    }
  }
});

速率限制 ：
实现令牌桶算法控制请求速率
根据账户等级设置合理的 QPS
费用监控 ：
定期检查 API 使用情况
设置预算告警（可通过 OpenAI Dashboard 配置）

from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine

def sanitize_input(text):
    analyzer = AnalyzerEngine()
    analyzer_results = analyzer.analyze(text=text, language='en')

    anonymizer = AnonymizerEngine()
    sanitized_text = anonymizer.anonymize(
        text=text,
        analyzer_results=analyzer_results
    ).text

    return sanitized_text