ChatGPT API 实战指南：从接入到优化的全流程解析

1次阅读

没有评论

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

ChatGPT API 基于 OpenAI 的 GPT 模型提供对话服务，核心参数包括：

model：指定使用的模型版本（如 gpt-3.5-turbo）
messages：对话历史数组，每个对象包含 role(user/assistant) 和 content
temperature：控制回答随机性（0-2）

工作原理可简化为：客户端发送包含上下文的消息数组 → 服务端返回模型生成的回答。每次调用都是无状态的，上下文维护需开发者自行实现。

问题：API Key 直接硬编码在代码中会导致泄露风险。

解决方案 ：

使用环境变量存储密钥（推荐 python-dotenv 库）
密钥轮换机制（每月自动更新）
通过 AWS Secrets Manager 或 HashiCorp Vault 管理

# .env 文件示例
OPENAI_API_KEY=sk-your-key-here

# 安全加载示例
from dotenv import load_dotenv
import openai

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

方案对比 ：

方案	优点	缺点
本地存储	响应快，无网络开销	会话状态难跨设备同步
Redis 缓存	支持分布式，TTL 自动过期	需要额外基础设施

推荐实现 ：

# Redis 上下文存储示例
import redis

r = redis.Redis(host='localhost', port=6379, db=0)

def save_context(session_id, messages):
    r.setex(f"chat:{session_id}", 3600, json.dumps(messages))

def load_context(session_id):
    data = r.get(f"chat:{session_id}")
    return json.loads(data) if data else []

OpenAI 的速率限制通常为：

RPM（每分钟请求数）：3,000
TPM（每分钟 token 数）：250,000

令牌桶算法实现 ：

from ratelimit import limits, sleep_and_retry

# 限制为每分钟 60 次调用
@sleep_and_retry
@limits(calls=60, period=60)
def call_chatgpt_api(messages):
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=messages
    )
    return response

import openai
import os
from dotenv import load_dotenv
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type
)

# 初始化
load_dotenv()
openai.api_key = os.getenv('OPENAI_API_KEY')

# 带重试机制的 API 调用
@retry(stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10),
    retry=retry_if_exception_type(openai.error.APIError)
)
def get_chat_response(messages, max_tokens=500):
    try:
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=messages,
            max_tokens=max_tokens,
            temperature=0.7
        )
        return response.choices[0].message.content
    except openai.error.InvalidRequestError as e:
        print(f"无效请求: {e}")
    except openai.error.RateLimitError:
        print("触发速率限制")
        raise

# 使用示例
if __name__ == "__main__":
    chat_history = [{"role": "system", "content": "你是一个有帮助的助手"},
        {"role": "user", "content": "推荐三本 Python 入门书籍"}
    ]

    response = get_chat_response(chat_history)
    print(response)

模型	响应时间	适合场景
gpt-4	800-1200ms	高复杂度任务
gpt-3.5-turbo	300-500ms	常规对话
text-davinci-003	600-900ms	单轮生成

# 并行请求示例
import asyncio
import openai

async def async_chat_completion(messages):
    return await openai.ChatCompletion.acreate(
        model="gpt-3.5-turbo",
        messages=messages
    )

async def batch_process(queries):
    tasks = [async_chat_completion(q) for q in queries]
    return await asyncio.gather(*tasks)

错误码	原因	解决方案
429	速率限制	实现退避重试机制
400	无效请求	检查 messages 格式
503	服务不可用	等待后重试

def sanitize_input(text):
    sensitive_phrases = ["密码", "信用卡", "身份证"]
    for phrase in sensitive_phrases:
        if phrase in text:
            raise ValueError("包含敏感信息")
    return text