ChatGPT部署实战：从零搭建到生产环境避坑指南

3次阅读

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

最近在帮团队部署 ChatGPT 服务时，发现新手常被这几个问题卡住：

冷启动延迟高：基础版 GPT- 3 模型加载需要 12GB+ 内存，首次响应经常超时
显存黑洞：对话长度超过 512token 时，显存占用会突然飙升
并发能力弱：单卡 GPU 在 16GB 显存下，只能维持 3 - 5 个并发会话
API 设计复杂：要同时处理流式响应、会话状态管理、超时重试等场景

我们测试发现，直接调用 OpenAI 官方 API 的成本是自建服务的 3 - 5 倍，这还不考虑数据隐私问题。

方案	优点	缺点
纯本地部署	数据可控，延迟稳定	硬件成本高，运维复杂
混合云	弹性伸缩灵活	网络延迟不可控
托管服务	开箱即用	黑箱操作，定制困难

FastAPI：异步支持好（ASGI），自带 OpenAPI 文档，我们的最终选择
Flask：生态成熟但同步架构，需要配合 Celery 实现异步
Django：太重，适合全栈项目但这里过度设计

FROM nvidia/cuda:11.8.0-base

# 最小化安装 Python 环境
RUN apt-get update && \
    apt-get install -y python3-pip && \
    rm -rf /var/lib/apt/lists/*

# 安装优化过的 transformers
RUN pip install torch --extra-index-url https://download.pytorch.org/whl/cu118
RUN pip install optimum[onnxruntime-gpu]

# 复制模型缓存（提前下载好）COPY ./models /app/models

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class ChatRequest(BaseModel):
    prompt: str
    max_tokens: int = 512

@app.post("/chat")
async def chat_completion(request: ChatRequest):
    try:
        # 实际调用示例
        response = generate_response(
            request.prompt, 
            max_length=request.max_tokens
        )
        return {"response": response}
    except RuntimeError as e:
        raise HTTPException(status_code=503, detail="GPU OOM")

在服务启动时预加载：

# 服务启动时执行
@app.on_event("startup")
async def load_model():
    global pipeline
    pipeline = pipeline(
        "text-generation", 
        model="/app/models/gpt-3",
        device=0
    )
    # 用空请求预热
    pipeline("", max_length=1)

使用 asyncio 实现请求队列：

from queue import Queue
from threading import Thread

request_queue = Queue()

def batch_worker():
    while True:
        batch = []
        # 攒够 10 个请求或等待 200ms
        while len(batch) < 10 and not request_queue.empty():
            batch.append(request_queue.get())
        if batch:
            process_batch(batch)

Thread(target=batch_worker, daemon=True).start()

JWT 鉴权：

from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

内容过滤：

banned_words = ["暴力", "色情"] # 实际应从数据库加载

def sanitize_input(text: str) -> bool:
    return any(word in text for word in banned_words)

速率限制：

from fastapi import Request
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.post("/chat")
@limiter.limit("5/minute")
async def chat_endpoint(request: Request):
    ...