ChatGPT App下载与集成指南：从零搭建智能对话应用

1次阅读

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

直接调用 ChatGPT API 时，开发者常遇到三个典型问题：

上下文丢失 ：原始 API 默认无状态，连续对话需手动维护历史消息
响应延迟 ：跨地区请求可能产生 200ms 以上的网络延迟
成本失控 ：突发流量可能导致超额 Token 消耗

以电商客服场景为例，用户连续询问 ” 这件衣服有货吗？” 和 ” 什么时候能送到？” 时，若未正确处理对话上下文，AI 将无法理解 ” 送到 ” 指代的是前文提到的衣服。

方案类型	QPS(峰值)	平均延迟	成本 ($/ 百万 Token)
官方 SDK 直连	300	350ms	2.00
自建代理集群	1500	180ms	1.20
本地缓存 +SDK	5000	50ms	0.80

注：测试环境为 AWS us-east- 1 区域，使用 gpt-3.5-turbo 模型

# requirements.txt
# openai>=1.0.0
# redis>=4.5.0

from typing import List, Dict
import openai
from redis import Redis

class ChatSession:
    """维护对话上下文的会话管理器"""
    def __init__(self, redis_client: Redis, system_prompt: str = "你是一个智能助手"):
        self.history: List[Dict] = [{"role": "system", "content": system_prompt}
        ]
        self.redis = redis_client

    async def respond(self, user_input: str, session_id: str) -> str:
        """处理用户输入并返回 AI 响应"""
        # 从 Redis 读取历史记录（如有）cached = self.redis.get(f"chat:{session_id}")
        if cached:
            self.history = json.loads(cached)

        # 添加用户消息
        self.history.append({"role": "user", "content": user_input})

        try:
            # TODO: 添加重试逻辑
            response = await openai.ChatCompletion.create(
                model="gpt-3.5-turbo",
                messages=self.history,
                temperature=0.7
            )
            ai_message = response.choices[0].message.content

            # 更新历史记录
            self.history.append({"role": "assistant", "content": ai_message})
            self.redis.setex(f"chat:{session_id}", 
                3600,  # 1 小时过期
                json.dumps(self.history)
            )
            return ai_message
        except Exception as e:
            # TODO: 实现降级策略
            raise RuntimeError(f"API 调用失败: {str(e)}")

def cache_frequent_qa(question: str, answer: str):
    """缓存常见问答对"""
    question_hash = hashlib.md5(question.encode()).hexdigest()
    redis_client.setex(f"qa:{question_hash}", 
        86400,  # 24 小时
        answer
    )

def get_cached_answer(question: str) -> Optional[str]:
    """检查问题是否已有缓存答案"""
    question_hash = hashlib.md5(question.encode()).hexdigest()
    return redis_client.get(f"qa:{question_hash}")

动态计算消耗 ：

def calculate_token_usage(messages: List[Dict]) -> int:
    """估算消息列表的 Token 数"""
    # 简化计算：1 个汉字≈1.33 Token
    return sum(len(msg["content"])*1.33 for msg in messages)

令牌桶算法实现 ：

from ratelimit import limits, sleep_and_retry

# 每分钟最多 10000 Token
@sleep_and_retry
@limits(calls=10000, period=60)
def api_call_with_limit(messages):
    return openai.ChatCompletion.create(messages=messages)

import re

def sanitize_log(text: str) -> str:
    """移除敏感信息"""
    # 隐藏电话号码
    text = re.sub(r'\d{3}-\d{4}-\d{4}', '[PHONE]', text)
    # 隐藏邮箱
    text = re.sub(r'[\w\.-]+@[\w\.-]+\.\w+', '[EMAIL]', text)
    return text

混合架构方案 ：
– 高频简单问题：本地轻量模型（如 GGML 格式的 Alpaca）
– 复杂逻辑问题：转发至 ChatGPT API
– 知识库查询：内置向量数据库（FAISS）

用 FastAPI 实现带速率限制的代理端点：

# main.py
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from redis import Redis

app = FastAPI()
redis = Redis(host='localhost', port=6379)

# 允许跨域
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.post("/chat")
async def chat_endpoint(request: Request):
    """带速率限制的聊天代理"""
    ip = request.client.host
    # 检查 IP 请求频率
    if redis.get(f"rate_limit:{ip}"):
        return {"error": "请求过于频繁"}

    # 设置 10 秒冷却期
    redis.setex(f"rate_limit:{ip}", 10, "1")

    data = await request.json()
    # 实际处理逻辑...
    return {"response": "模拟响应"}

启动服务：