ChatGPT在线问答系统架构设计与性能优化实战

1次阅读

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

构建一个稳定的 ChatGPT 在线问答系统，开发者通常会面临几个核心问题：

响应延迟 ：LLM 模型推理耗时随输入长度增长而增加，用户感知延迟明显
高并发崩溃 ：突发流量导致 GPU 内存溢出，服务不可用
上下文管理 ：多轮对话的上下文拼接消耗大量计算资源
成本控制 ：GPU 实例费用高昂，需提高资源利用率

传统同步架构（左）与优化后的异步架构（右）对比：

@startuml
left to right direction

rectangle "同步架构" {
  component "Web 服务器" as sync_web
  component "Worker 进程" as sync_worker
  sync_web -> sync_worker : 阻塞调用
}

rectangle "异步架构" {
  component "ASGI 服务" as async_web
  component "任务队列" as async_queue
  component "模型实例" as async_model
  async_web -> async_queue : 非阻塞投递
  async_queue -> async_model : 批量处理
}
@enduml

关键差异点：

同步架构每个请求独占线程 / 进程资源
异步架构通过消息队列实现请求批处理
模型实例保持常驻内存，避免重复加载

针对大模型内存占用问题，采用分层加载方案：

基础层常驻内存 ：Embedding 矩阵等基础参数
动态加载层 ：根据请求类型按需加载特定领域参数
CPU 卸载 ：将部分层暂时卸载到主机内存

from fastapi import FastAPI
import uvicorn
from typing import List

app = FastAPI()

@app.post("/chat")
async def handle_chat(prompts: List[str]):
    """
    批量处理聊天请求
    :param prompts: 最多支持 8 条并发请求
    :return: 按输入顺序返回响应
    """
    batch_results = await model.batch_predict(prompts)
    return {"results": batch_results}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

import torch
from transformers import AutoModelForCausalLM

class BatchProcessor:
    def __init__(self):
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        self.model = AutoModelForCausalLM.from_pretrained("gpt-3.5")

    async def batch_predict(self, texts: List[str]):
        inputs = self.tokenizer(texts, return_tensors="pt", padding=True, truncation=True).to(self.device)
        with torch.no_grad():
            outputs = self.model.generate(**inputs, max_new_tokens=256)
        return [self.tokenizer.decode(out, skip_special_tokens=True) for out in outputs]

import redis
from functools import wraps

r = redis.Redis(host='localhost', port=6379)

def cache_response(ttl=300):
    """
    缓存高频问答对
    :param ttl: 缓存存活时间 (秒)
    """
    def decorator(func):
        @wraps(func)
        async def wrapper(prompt: str):
            cache_key = f"chat:{hash(prompt)}"
            cached = r.get(cache_key)
            if cached:
                return cached.decode()

            result = await func(prompt)
            r.setex(cache_key, ttl, result)
            return result
        return wrapper
    return decorator