中科院ChatGPT网页版入口实现方案：从零搭建高可用对话系统

2次阅读

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

在自建对话系统的过程中，开发者通常会面临几个关键问题：

响应延迟 ：大语言模型推理需要消耗大量计算资源，导致用户等待时间过长
上下文管理 ：多轮对话需要有效维护对话历史，传统数据库方案性能不足
并发瓶颈 ：单节点服务难以应对突发流量，需要合理的负载均衡策略
部署复杂度 ：从开发环境到生产环境的迁移往往涉及大量配置调整

中科院开源方案相比主流项目如 FastChat 具有以下特点：

特性	中科院方案	FastChat
模型微调	中文领域专用优化	通用英文模型
API 兼容性	OpenAI 风格接口	自有协议
部署复杂度	提供完整 Docker 支持	依赖 Kubernetes
中文处理	专有分词器	通用 Tokenizer
硬件要求	支持 CPU/GPU 混合部署	仅 GPU 支持

from flask import Flask, request, jsonify
from flask_jwt_extended import JWTManager, create_access_token

app = Flask(__name__)
app.config['JWT_SECRET_KEY'] = 'your-secret-key'
jwt = JWTManager(app)

@app.route('/api/login', methods=['POST'])
def login():
    username = request.json.get('username')
    password = request.json.get('password')
    # 实际项目应使用数据库验证
    if username != 'admin' or password != 'password':
        return jsonify({'error': 'Invalid credentials'}), 401

    access_token = create_access_token(identity=username)
    return jsonify({'access_token': access_token})

@app.route('/api/chat', methods=['POST'])
def chat():
    @after_this_request
    def add_header(response):
        response.headers['Content-Type'] = 'text/event-stream'
        response.headers['Cache-Control'] = 'no-cache'
        response.headers['Connection'] = 'keep-alive'
        return response

    def generate():
        prompt = request.json['prompt']
        for chunk in model.stream_generate(prompt):
            yield f'data: {json.dumps({"text": chunk})}\n\n'

    return Response(generate(), mimetype='text/event-stream')

import redis
from datetime import timedelta

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

def save_context(user_id: str, messages: list) -> bool:
    try:
        r.setex(f'ctx_{user_id}', timedelta(hours=1), json.dumps(messages))
        return True
    except redis.RedisError as e:
        app.logger.error(f'Redis error: {str(e)}')
        return False

version: '3.8'
services:
  web:
    build: .
    ports:
      - "5000:5000"
    environment:
      - REDIS_HOST=redis
    depends_on:
      - redis

  redis:
    image: redis:alpine
    volumes:
      - redis_data:/data
    ports:
      - "6379:6379"

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
    depends_on:
      - web

volumes:
  redis_data:

测试环境：AWS c5.2xlarge (8 vCPU, 16GB 内存)

并发数	平均响应时间 (ms)	错误率	吞吐量 (req/s)
50	320	0%	156
100	480	0%	208
200	920	2%	217
500	超时	85%	35

中文乱码解决 ：

@app.after_request
def set_charset(response):
    response.headers['Content-Type'] = 'application/json; charset=utf-8'
    return response

GPU 显存不足 ：

from transformers import AutoModelForCausalLM, BitsAndBytesConfig

bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_use_double_quant=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype=torch.bfloat16
)
model = AutoModelForCausalLM.from_pretrained("model_path", quantization_config=bnb_config)

日志脱敏 ：

import re

def sanitize_log(text: str) -> str:
    # 移除邮箱
    text = re.sub(r'[\w\.-]+@[\w\.-]+', '[EMAIL]', text)
    # 移除手机号
    text = re.sub(r'1[3-9]\d{9}', '[PHONE]', text)
    return text