构建ChatGPT应用的最佳实践：从架构设计到生产环境部署

2次阅读

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

在开发生产级 ChatGPT 应用时，开发者常常面临 API 速率限制、长文本处理、对话状态维护等挑战。本文将分享一套完整的解决方案，从架构设计到代码实现，再到性能优化和安全考量，帮助开发者构建稳定、高效的 ChatGPT 应用。

API 速率限制：OpenAI 对 API 调用有严格的速率限制，超出限制会导致请求失败。
长文本处理：GPT 模型对输入长度有限制，处理长文本时需要分段或压缩。
对话状态维护：在多轮对话中，如何有效管理对话历史以避免重复或丢失上下文。
成本控制：API 调用按 token 计费，如何优化提示词和结果缓存以降低成本。
高并发场景：如何在高并发下保证响应速度和稳定性。

直接调用 API：
优点：无需维护模型，节省硬件资源。
缺点：受速率限制，延迟较高，成本随调用量增加。
自托管模型：
优点：完全控制模型，无速率限制，延迟低。
缺点：需要大量硬件资源，维护成本高。

[客户端] -> [负载均衡] -> [API 网关] -> [消息队列] -> [Worker] -> [OpenAI API]
                                  |
                                  v
                           [缓存层 (Redis)]

负载均衡：分散请求压力，避免单点故障。
API 网关：统一处理认证、限流和日志。
消息队列：异步处理高并发请求，平滑流量峰值。
Worker：执行实际 API 调用，处理重试逻辑。
缓存层：存储常用结果，减少 API 调用。

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

说明：使用tenacity 库实现指数退避重试，最多重试 3 次，等待时间随失败次数增加。

def compress_history(history):
    # 只保留最近 3 轮对话和首轮提示
    if len(history) > 6:  # 每轮对话包含 user 和 assistant 两条消息
        compressed = history[:2] + history[-4:]  # 保留首轮和最近两轮
        return compressed
    return history

说明：通过截断早期对话历史，减少 token 消耗，同时保留关键上下文。

from flask import Flask, Response
import json

app = Flask(__name__)

@app.route('/stream')
def stream():
    def generate():
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": "Hello!"}],
            stream=True
        )
        for chunk in response:
            yield f"data: {json.dumps(chunk)}\n\n"
    return Response(generate(), mimetype="text/event-stream")

说明：使用 Server-Sent Events (SSE) 实现流式响应，提升用户体验。

将常用提示词模板化，避免重复编写。例如：

templates = {"summary": "请用不超过 50 字总结以下文本: {text}",
    "translate": "将以下文本翻译成英文: {text}"
}

使用 Redis 缓存常见问题的回答，减少 API 调用。

import redis

r = redis.Redis()

def get_cached_response(prompt):
    cached = r.get(prompt)
    if cached:
        return cached.decode('utf-8')
    response = call_chatgpt(prompt)
    r.setex(prompt, 3600, response)  # 缓存 1 小时
    return response

将多个独立请求合并为一个批量请求，减少 API 调用次数。

import asyncio

async def batch_requests(prompts):
    tasks = [asyncio.create_task(call_chatgpt(prompt)) for prompt in prompts]
    return await asyncio.gather(*tasks)