从零搭建ChatGPT应用：架构设计与工程化实践指南

2次阅读

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

在自建 ChatGPT 类应用时，开发者通常会遇到以下三大挑战：

上下文丢失：大语言模型（LLM）通常有固定的上下文窗口（如 GPT-3.5 的 4096 tokens），当对话历史超过这个限制时，早期的对话内容会被截断，导致模型 ” 忘记 ” 之前的交流。
响应延迟：特别是在直接调用 OpenAI API 的情况下，网络延迟和模型推理时间会导致用户感知的响应时间变长，影响对话流畅度。
token 成本控制：API 调用按 token 计费，长对话和复杂查询会快速消耗预算，需要有效的 token 管理和优化策略。

搭建 ChatGPT 类应用时，首先面临的技术决策是：使用云 API 还是自托管 LLM。以下是关键维度对比：

维度	OpenAPI 直接调用	自托管 LLM
延迟	较高（网络 + 服务端处理）	较低（本地 / 内网调用）
成本	按 token 计费，长期成本高	前期硬件投入大，边际成本低
可定制性	有限（只能调整有限参数）	完全可控（可微调模型权重）
维护复杂度	低（无需管理基础设施）	高（需维护 GPU 集群等）

对于大多数中小团队，建议从 OpenAPI 开始，待业务稳定后再考虑自托管方案。

import openai
from functools import lru_cache
import logging

logging.basicConfig(level=logging.INFO)

@lru_cache(maxsize=1024)
def get_chat_completion(messages, model="gpt-3.5-turbo"):
    try:
        response = openai.ChatCompletion.create(
            model=model,
            messages=messages,
            temperature=0.7,
            top_p=0.9
        )
        return response.choices[0].message.content
    except Exception as e:
        logging.error(f"API 调用失败: {str(e)}")
        raise

这个实现使用了 LRU 缓存来避免对相同输入重复调用 API，同时加入了基本的错误处理和日志记录。

dialog_history = []

def add_to_history(role, content):
    dialog_history.append({"role": role, "content": content})
    # 简单的 token 计数和截断
    if len(dialog_history) > 10:  # 示例阈值
        dialog_history.pop(0)

def summarize_history(history):
    # 调用 API 生成摘要的简化示例
    prompt = f"请用 100 字以内总结以下对话:\n{str(history)}"
    summary = get_chat_completion([{"role": "user", "content": prompt}])
    return [{"role": "system", "content": "对话摘要:" + summary}]

from sentence_transformers import SentenceTransformer
import numpy as np

encoder = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')

def retrieve_relevant_history(current_query, full_history, top_k=3):
    query_embedding = encoder.encode(current_query)
    similarities = []
    for msg in full_history:
        msg_embedding = encoder.encode(msg["content"])
        sim = np.dot(query_embedding, msg_embedding)
        similarities.append(sim)
    indices = np.argsort(similarities)[-top_k:]
    return [full_history[i] for i in indices]

const express = require('express');
const OpenAI = require('openai');
const app = express();

app.get('/chat-stream', async (req, res) => {res.setHeader('Content-Type', 'text/event-stream');
    res.setHeader('Cache-Control', 'no-cache');
    res.setHeader('Connection', 'keep-alive');

    const stream = await openai.chat.completions.create({
        model: "gpt-3.5-turbo",
        messages: [{role: "user", content: req.query.q}],
        stream: true,
    });

    for await (const chunk of stream) {const content = chunk.choices[0]?.delta?.content || '';
        res.write(`data: ${JSON.stringify({content})}\n\n`);
    }

    res.write('data: [DONE]\n\n');
    res.end();});

令牌桶算法实现限流：
每个 API key 限制为 60 RPM（每分钟请求数）
突发流量允许短时超限，但不超过 120 RPM
指数退避重试：

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 safe_api_call(prompt):
    return get_chat_completion(prompt)

推荐使用两级过滤：

前置过滤：基于关键词和正则的简单规则
后置过滤：调用内容审核 API（如 OpenAI 的 moderation 端点）

必备的四类指标：

延迟指标：P50/P95/P99 响应时间
流量指标：并发会话数、RPS
质量指标：用户满意度评分（如有）、API 错误率
成本指标：日均 token 消耗、每会话平均 token

现象：用户长时间对话后，机器人开始给出无关回复。

解决方案：
1. 实现 token 计数器，实时计算对话历史 token 数
2. 当接近限制时（如 3800/4096），触发摘要压缩
3. 保留最后 3 - 5 轮原始对话确保连贯性

现象：促销活动期间大量用户涌入，API 返回 429 错误。

解决方案：
1. 实现多 API key 轮询
2. 在前端加入排队机制和优雅降级
3. 使用 CDN 缓存常见问答

现象：用户通过特殊编码绕过过滤规则。

解决方案：
1. 统一 unicode 标准化处理
2. 增加基于语义的 LLM 二次审核
3. 建立人工审核兜底流程

from locust import HttpUser, task, between

class ChatUser(HttpUser):
    wait_time = between(1, 3)

    @task
    def chat_request(self):
        self.client.post("/chat", 
            json={"messages": [{"role": "user", "content": "你好"}]},
            headers={"Authorization": "Bearer API_KEY"}
        )