从零搭建ChatGPT应用：技术选型与生产环境避坑指南

3次阅读

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

在自建 ChatGPT 应用的过程中，开发者往往会遇到以下几个典型问题：

API 调用混乱：缺乏统一的封装管理，导致代码中散落着各种 API 调用逻辑
上下文丢失：对话过程中无法有效维护历史消息，影响用户体验
并发瓶颈：突发流量下 API 调用失败率飙升
安全风险：API 密钥硬编码、敏感信息泄露等安全隐患

优点：
简单直接，学习成本低
完全控制请求和响应流程
无需额外依赖
缺点：
需要自行处理上下文管理
缺乏高级抽象
并发控制需要额外开发

优点：
内置对话链管理
提供多种记忆组件
支持多种模型集成
缺点：
引入额外依赖
抽象层可能限制灵活性
学习曲线较陡

from typing import Optional
import openai
from pydantic import BaseModel

class ChatCompletionRequest(BaseModel):
    model: str = "gpt-3.5-turbo"
    messages: list[dict[str, str]]
    temperature: float = 0.7
    max_tokens: Optional[int] = None

class ChatGPTClient:
    def __init__(self, api_key: str):
        openai.api_key = api_key

    async def create_chat_completion(
        self, 
        request: ChatCompletionRequest,
        timeout: int = 30
    ) -> dict:
        try:
            response = await openai.ChatCompletion.acreate(
                model=request.model,
                messages=request.messages,
                temperature=request.temperature,
                max_tokens=request.max_tokens,
                request_timeout=timeout
            )
            return response
        except openai.error.APIError as e:
            # 处理 API 错误
            raise
        except Exception as e:
            # 处理其他异常
            raise

import json
from datetime import timedelta
import redis

class ConversationManager:
    def __init__(self, redis_client: redis.Redis, ttl: int = 3600):
        self.redis = redis_client
        self.ttl = ttl

    def _get_key(self, conversation_id: str) -> str:
        return f"conversation:{conversation_id}"

    def save_messages(self, conversation_id: str, messages: list[dict]) -> bool:
        key = self._get_key(conversation_id)
        serialized = json.dumps(messages)
        return self.redis.setex(key, timedelta(seconds=self.ttl), serialized)

    def load_messages(self, conversation_id: str) -> list[dict]:
        key = self._get_key(conversation_id)
        serialized = self.redis.get(key)
        if not serialized:
            return []
        return json.loads(serialized)

graph TD
    A[客户端请求] --> B[API 网关]
    B --> C[限流中间件]
    C --> D[对话上下文管理器]
    D --> E[ChatGPT 客户端]
    E --> F[响应处理器]
    F --> G[返回客户端]

使用密钥管理系统存储 API 密钥
定期自动轮换密钥（建议每月）
实现零停机切换机制

from fastapi import Request, Response
from fastapi.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware

class RateLimiterMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, limit: int = 60, window: int = 60):
        super().__init__(app)
        self.limit = limit
        self.window = window
        self.redis = redis.Redis()

    async def dispatch(self, request: Request, call_next):
        client_ip = request.client.host
        key = f"rate_limit:{client_ip}"

        current = self.redis.get(key)
        if current and int(current) >= self.limit:
            return Response("Too Many Requests", status_code=429)

        pipe = self.redis.pipeline()
        pipe.incr(key)
        pipe.expire(key, self.window)
        pipe.execute()

        return await call_next(request)

使用正则表达式匹配敏感信息
实现输入输出双向过滤
记录过滤日志用于审计

import re

class ContentFilter:
    def __init__(self):
        self.patterns = [r"\b(?:\d{4}[-\.\s]?){3}\d{4}\b",  # 信用卡号
            r"\b\d{3}[-\.\s]?\d{2}[-\.\s]?\d{4}\b"  # SSN
        ]

    def filter_text(self, text: str) -> str:
        for pattern in self.patterns:
            text = re.sub(pattern, "[REDACTED]", text)
        return text