免费ChatGPT网页服务的架构解析与自建指南

2次阅读

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

当前自建 AI 对话服务面临三个核心挑战：商业 API 调用成本居高不下，尤其在流量激增时费用不可控；传统同步响应模式导致用户等待时间过长，体验下降；单一 GPU 实例的并发处理能力有限，难以应对突发流量。

框架	QPS（Llama2-7B）	显存占用（GB）	支持量化	流式响应
HuggingFace TGI	35	14.2	✅	✅
vLLM	52	10.8	✅	✅
原生 Transformers	18	16.4	❌	❌

测试环境：NVIDIA A10G 显卡，输入长度 256 tokens。实际部署推荐 vLLM 框架，其基于 PagedAttention 技术显著提升吞吐量。

from fastapi import FastAPI, Request
from sse_starlette.sse import EventSourceResponse

app = FastAPI()

async def generate_stream(prompt: str):
    for i in range(10):
        yield {'data': f'Chunk {i} of response'}
        await asyncio.sleep(0.1)

@app.get('/stream')
async def stream_response(request: Request, token: str = Depends(validate_jwt)):
    try:
        return EventSourceResponse(generate_stream(request.query_params.get('q')),
            headers={'X-Accel-Buffering': 'no'}
        )
    except Exception as e:
        logging.error(f'Stream error: {str(e)}')
        raise HTTPException(500)

关键点说明：

使用 Server-Sent Events(SSE) 协议实现文本流式传输
JWT 验证通过 Depends 注入实现
必须关闭 Nginx 等代理的缓冲功能

import redis
from datetime import timedelta

r = redis.Redis(host='redis', decode_responses=True)

def save_session(session_id: str, messages: list[dict]):
    try:
        r.setex(f'session:{session_id}',
            timedelta(hours=2),
            json.dumps(messages)
        )
    except redis.RedisError as e:
        logging.warning(f'Session save failed: {e}')

使用 bitsandbytes 进行 8 -bit 量化：

from transformers import AutoModelForCausalLM

model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-2-7b-chat-hf",
    load_in_8bit=True,
    device_map='auto'
)

实测显存占用从 16GB 降至 9GB，推理速度损失约 15%。4-bit 量化可进一步降至 6GB，但需要配套使用 GPTQ 算法。

对比三种并发模式（测试 100 并发请求）：

同步模式：平均响应时间 2.3s，错误率 12%
多线程模式：平均响应时间 1.7s，错误率 8%
Asyncio 协程：平均响应时间 0.9s，错误率 0%

推荐使用 uvicorn 运行 FastAPI，配合 –workers 参数充分利用多核 CPU。

import re

blacklist = re.compile(r'( 恶意关键词 1 | 敏感词 2)', flags=re.IGNORECASE)

def sanitize_input(text: str) -> str:
    if blacklist.search(text):
        raise ContentFilterError()
    return text[:2000]  # 限制输入长度

基于令牌桶算法实现：

from fastapi import HTTPException

class RateLimiter:
    def __init__(self, rate: int, capacity: int):
        self.tokens = capacity
        self.last_update = time.time()

    async def check(self, key: str) -> bool:
        now = time.time()
        elapsed = now - self.last_update
        self.tokens = min(
            self.capacity,
            self.tokens + elapsed * self.rate
        )
        self.last_update = now

        if self.tokens < 1:
            raise HTTPException(429)
        self.tokens -= 1
        return True