从零搭建ChatGPT服务：技术选型与架构设计实战

3次阅读

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

当前开发者搭建私有 ChatGPT 服务时，主要面临三大挑战：第一是成本控制难题，商用 API 按 token 计费可能产生意外支出；第二是高并发场景下的响应延迟问题，尤其在处理长文本时更明显；第三是数据隐私顾虑，敏感信息通过第三方 API 传输存在泄露风险。这些痛点使得越来越多的团队开始考虑自建解决方案。

评估指标	OpenAI API	LLaMA2(7B/13B)
部署复杂度	无需部署	需 GPU 服务器
响应速度	200-500ms	1-5s(依赖硬件)
最大上下文长度	128K tokens	4K tokens
微调成本	按使用量计费	一次性训练投入
隐私保护	数据出域	完全可控

对于大多数中小团队，建议初期使用 OpenAI API 快速验证，业务稳定后再迁移到开源模型。

# app/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import jwt
from datetime import datetime, timedelta

app = FastAPI()
SECRET_KEY = "your_secure_key"

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

@app.post("/chat")
async def chat_endpoint(request: ChatRequest):
    try:
        # 实际处理逻辑替换为模型调用
        return {"response": "模拟响应"}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# app/auth.py
def create_access_token(data: dict):
    expire = datetime.utcnow() + timedelta(hours=1)
    return jwt.encode({"sub": data, "exp": expire}, SECRET_KEY)

async def get_current_user(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return payload["sub"]
    except:
        raise HTTPException(status_code=401, detail="Invalid token")

# /etc/nginx/sites-enabled/chatgpt
upstream chat_backend {
    server 127.0.0.1:8000;
    server 127.0.0.1:8001;
}

server {
    location / {
        proxy_pass http://chat_backend;
        proxy_set_header Host $host;
    }
}

启动命令：

uvicorn app.main:app --workers 4 --port 8000 &
uvicorn app.main:app --workers 4 --port 8001 &

# 使用 LRU 缓存最近对话
from functools import lru_cache

@lru_cache(maxsize=100)
def get_cached_response(user_id: str, prompt: str):
    return process_prompt(prompt)

# app/limiter.py
import redis
from fastapi import Request

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

def rate_limit(key: str, limit: int = 10):
    current = r.get(key)
    if current and int(current) > limit:
        raise HTTPException(429, "Too many requests")
    r.incr(key)
    r.expire(key, 60)

import re

blacklist = r"(暴力 | 色情 | 政治敏感词)"

def sanitize_input(text: str):
    if re.search(blacklist, text, re.I):
        raise ValueError("包含违规内容")
    return text