从零搭建ChatGPT应用：技术选型与架构设计实战

1次阅读

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

在搭建 ChatGPT 应用时，开发者常遇到以下几个核心挑战：

API 调用限制管理 ：OpenAI 的 API 有严格的速率限制（如每分钟请求数），需要设计合理的请求队列和重试机制。
对话上下文维护 ：多轮对话需要有效管理历史消息，但上下文长度受模型 token 限制（如 gpt-3.5-turbo 的 4096 tokens）。
流式响应处理 ：直接返回完整响应会导致用户等待时间过长，需要实现逐字输出优化体验。

以下是经过生产验证的 Python 调用示例，包含认证和基础请求处理：

import openai
from typing import List, Dict

class ChatGPTClient:
    def __init__(self, api_key: str):
        openai.api_key = api_key
        # 建议配置自定义域名代理（如需）# openai.api_base = "https://your-proxy.com/v1"

    def call_api(
        self,
        messages: List[Dict[str, str]],
        model: str = "gpt-3.5-turbo",
        temperature: float = 0.7,
        max_retries: int = 3
    ) -> str:
        """
        执行 API 调用并处理重试逻辑
        :param messages: 对话消息历史，格式参考 OpenAI 文档
        :param max_retries: 最大重试次数
        :return: 模型生成的回复内容
        """
        for attempt in range(max_retries):
            try:
                response = openai.ChatCompletion.create(
                    model=model,
                    messages=messages,
                    temperature=temperature
                )
                return response.choices[0].message.content
            except openai.error.RateLimitError:
                if attempt == max_retries - 1:
                    raise
                time.sleep(2 ** attempt)  # 指数退避
            except openai.error.APIError as e:
                print(f"API Error: {e}")
                raise

两种常见的上下文维护方案对比：

服务端有状态模式
优点：客户端无需处理复杂逻辑
缺点：服务端需要维护会话存储，扩展性差
客户端维护模式
优点：无状态服务更易扩展
缺点：需设计合理的 token 计数和截断策略

推荐采用混合方案：

def manage_context(history: List[Dict[str, str]],
    new_message: str,
    max_tokens: int = 3500
) -> List[Dict[str, str]]:
    """
    智能维护对话上下文
    :param history: 现有消息历史
    :param new_message: 用户新消息
    :param max_tokens: 允许的最大 token 数
    :return: 更新后的消息历史
    """
    # 添加新消息
    updated_history = history + [{"role": "user", "content": new_message}]

    # 简易 token 估算（实际应使用 tiktoken 库精确计算）while sum(len(m["content"]) for m in updated_history) > max_tokens:
        updated_history.pop(0)  # 移除最早的消息

    return updated_history

使用 Python 生成器实现逐字输出效果：

def stream_response(prompt: str) -> Generator[str, None, None]:
    """
    流式获取模型响应
    :yield: 按字返回的响应内容
    """
    try:
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}],
            stream=True
        )

        for chunk in response:
            content = chunk.choices[0].delta.get("content", "")
            if content:
                yield content

    except Exception as e:
        yield f"[ERROR] {str(e)}"

# 调用示例
for word in stream_response("你好"):
    print(word, end="", flush=True)

实施请求队列和优先级调度
监控每分钟请求量并动态调整
关键业务设置独立 API key 配额

from profanity_filter import ProfanityFilter

pf = ProfanityFilter()

def sanitize_input(text: str) -> str:
    """输入内容安全过滤"""
    # 基础敏感词过滤
    if pf.is_profane(text):
        raise ValueError("包含不当内容")

    # 移除 PII 信息（需自定义正则）return remove_pii(text)