共计 3255 个字符，预计需要花费 9 分钟才能阅读完成。
为什么需要 AI 网站
最近越来越多的场景需要 AI 能力接入：智能客服自动回复、图片内容识别、文档摘要生成 … 但把 AI 模型塞进网站会遇到几个头疼问题：
- 模型动不动好几 GB，普通服务器根本跑不动
- 用户等个结果要十几秒，体验极差
- 同时多人访问时 GPU 内存直接爆炸
下面我们就用最省资源的方案，一步步解决这些问题。
技术选型：框架怎么挑
选框架就像选工具，得看具体要干什么活：
-
Flask：轻量灵活，适合快速验证想法。但异步支持需要自己扩展
# 典型 Flask 路由示例 @app.route('/predict', methods=['POST']) def predict(): data = request.get_json() # 处理逻辑... -
Django：自带管理员界面和 ORM，适合需要复杂后台管理的项目。但略显笨重
-
FastAPI：当前最推荐！自动生成 API 文档，原生支持异步，性能堪比 Go
# FastAPI 异步端点 @app.post("/predict") async def predict(input: InputSchema): result = await async_model_run(input.text) return {"result": result}
模型部署实战
方案一：ONNX 运行时
把训练好的模型转成 ONNX 格式，体积能小一半：
-
安装依赖
pip install onnxruntime-gpu # 用 GPU 加速 -
加载模型
import onnxruntime as ort # 创建推理会话 sess = ort.InferenceSession("model.onnx", providers=['CUDAExecutionProvider']) -
运行预测
# 准备输入数据 inputs = {"input_name": np.array([your_data], dtype=np.float32)} # 执行推理 outputs = sess.run(None, inputs) # 返回列表形式的结果
方案二：TensorFlow Serving
适合已有 TensorFlow 模型的情况：
-
拉取 Docker 镜像
docker pull tensorflow/serving -
启动服务
docker run -p 8501:8501 \ --mount type=bind,source=/path/to/models,target=/models \ -e MODEL_NAME=your_model \ -t tensorflow/serving
异步任务处理
用户上传图片后立即返回『处理中』，后台用 Celery 慢慢跑：
-
安装消息队列
brew install redis # Mac sudo apt install redis-server # Ubuntu -
Celery 配置
# celery_app.py from celery import Celery app = Celery('tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/1') @app.task(bind=True) def predict_task(self, image_path): try: return model.predict(image_path) except Exception as exc: self.retry(exc=exc, countdown=60) # 失败后重试 -
接口调用
# 在 FastAPI 路由中 @app.post("/predict") async def predict(image: UploadFile): task = predict_task.delay(image.filename) return {"task_id": task.id}
性能优化技巧
冷启动预热
服务启动时先跑几个虚拟请求：
# 应用启动事件
@app.on_event("startup")
async def warmup():
fake_data = np.random.rand(1, 224, 224, 3)
sess.run(None, {"input": fake_data})请求批处理
攒够 10 个请求一起推理，吞吐量提升 5 倍：
from concurrent.futures import ThreadPoolExecutor
batch_size = 10
batch_buffer = []
async def handle_request(request):
batch_buffer.append(request)
if len(batch_buffer) >= batch_size:
with ThreadPoolExecutor() as pool:
results = await loop.run_in_executor(pool, process_batch, batch_buffer.copy())
batch_buffer.clear()
return results避坑指南
内存泄漏检测
用 memory_profiler 定期检查：
# 在可疑函数上加装饰器
@profile
def suspicious_function():
# 你的代码...运行时加参数：
python -m memory_profiler your_script.pyGPU 竞争处理
设置可见设备避免冲突：
import os
os.environ["CUDA_VISIBLE_DEVICES"] = "0" # 只使用第一块 GPU完整 API 示例
带错误处理的预测接口：
from fastapi import HTTPException
@app.post("/api/v1/predict")
async def api_predict(input: dict):
try:
# 参数校验
if not input.get("text"):
raise HTTPException(400, detail="Missing text input")
# 执行预测
result = await model_async_predict(input["text"])
return {"success": True, "data": result}
except ModelTimeoutError:
raise HTTPException(503, "Server busy, please retry later")
except Exception as e:
logger.error(f"Predict failed: {str(e)}")
raise HTTPException(500, "Internal server error")单元测试怎么写
用 pytest 测试 API 接口：
# test_api.py
def test_predict_endpoint():
# 模拟请求
response = client.post("/predict",
json={"text": "测试输入"})
assert response.status_code == 200
assert "result" in response.json()
# 测试错误案例
bad_response = client.post("/predict", json={})
assert bad_response.status_code == 400部署上线注意事项
-
用 gunicorn 多进程运行：
gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app -
Nginx 配置负载均衡：
upstream ai_server { server 127.0.0.1:8000; server 127.0.0.1:8001; } server { location / {proxy_pass http://ai_server;} } -
监控 GPU 使用情况：
watch -n 1 nvidia-smi
总结路线图
- 先用 FastAPI 搭出基础接口
- ONNX 优化模型体积
- Celery 处理长耗时任务
- 批处理和预热解决性能瓶颈
- 完善监控和错误处理
按照这个流程走下来，即使是小团队也能构建出支撑上千 QPS 的 AI 服务。最重要的是先跑通核心流程，再逐步优化各个模块。
正文完
