ChatGPT官网API集成实战：解决高并发场景下的稳定性与性能优化

1次阅读

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

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

请求限流 ：OpenAI 对 API 调用有严格的速率限制（如 RPM/TPM），超出限制会导致 429 错误
响应延迟 ：复杂查询或高峰期 API 响应时间波动大，直接影响用户体验
错误处理复杂 ：需要处理网络超时、内容过滤、额度耗尽等多种异常场景

使用 Celery+Redis 构建异步任务队列，避免阻塞主线程：

# tasks.py
from celery import Celery
import openai

app = Celery('chatgpt', broker='redis://localhost:6379/0')

@app.task(bind=True, max_retries=3)
def ask_chatgpt(self, 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 as exc:
        # 指数退避重试
        raise self.retry(exc=exc, countdown=2 ** self.request.retries)

# utils.py
import redis
import json
from datetime import timedelta

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

def cached_request(prompt, ttl=3600):
    cache_key = f"chatgpt:{hash(prompt)}"
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)

    response = ask_chatgpt.delay(prompt).get()
    r.setex(cache_key, timedelta(seconds=ttl), json.dumps(response))
    return response

# rate_limiter.py
import time

class SlidingWindow:
    def __init__(self, capacity, time_window):
        self.capacity = capacity
        self.time_window = time_window
        self.timestamps = []

    def allow_request(self):
        now = time.time()
        # 移除过期时间戳
        self.timestamps = [t for t in self.timestamps if t > now - self.time_window]

        if len(self.timestamps) < self.capacity:
            self.timestamps.append(now)
            return True
        return False

场景	QPS	错误率	平均响应时间
直接调用	23	18%	1200ms
优化后方案	156	0.3%	450ms

优化关键点：

批处理请求（合并相似问题）
预加载热点问题到缓存
动态调整并发度（根据错误率自动降级）

API 密钥管理 ：
使用环境变量或密钥管理系统（如 AWS Secrets Manager）
设置 IP 白名单
定期轮换密钥

敏感数据过滤 ：

def sanitize_input(text):
    # 移除 PII 信息
    patterns = [r'\d{3}-\d{2}-\d{4}', r'[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}']
    for pattern in patterns:
        text = re.sub(pattern, '[REDACTED]', text)
    return text