攀岩者

Day55——早课 AI应用部署与上线

作者：

climbing

在

系列教程说明

这是「AI零基础70天学习计划」系列教程的第55篇，共140篇（每天早课+晚课）。

上篇回顾

昨天我们学习了如何微调大模型，让AI更懂你的业务需求。今天我们将探讨如何把训练好的AI应用部署到线上，让真实用户能够使用。

什么是AI应用部署？

想象一下，你花了几周时间训练了一个AI模型，它在你的电脑上运行得很好。但是，如果你想让朋友也能用，让客户也能用，甚至让全世界的人都能用——你就需要把它部署到互联网上。

部署的核心目标：

  • 让AI服务24小时在线运行
  • 支持多人同时使用
  • 保证响应速度和稳定性
  • 控制成本，按需扩展

部署前的准备工作

确定你的应用类型

1. API服务型

  • 提供RESTful API接口
  • 其他应用通过HTTP调用你的AI
  • 适合：企业内部系统、开发者工具

2. Web应用型

  • 用户直接访问网页使用
  • 包含前端界面和后端服务
  • 适合：面向普通用户的产品

3. 嵌入式集成型

  • 集成到现有系统中
  • 作为功能模块存在
  • 适合：已有产品的AI增强

评估资源需求

计算资源：

  • CPU：轻量级推理任务
  • GPU：深度学习模型推理（推荐NVIDIA T4或A10）
  • 内存：模型加载和并发处理

存储需求：

  • 模型文件大小（几GB到几十GB）
  • 用户数据存储
  • 日志和监控数据

网络带宽：

  • 预估用户访问量
  • API请求频率
  • 数据传输量

主流部署方案对比

方案一：云服务器自建

适用场景：需要完全控制环境，有运维团队

步骤：

  1. 购买云服务器（阿里云、腾讯云、AWS等）
  2. 安装Python环境和依赖
  3. 配置Web框架（Flask/FastAPI）
  4. 设置反向代理（Nginx）
  5. 配置进程管理（Gunicorn/PM2）
  6. 申请域名和SSL证书

优点：

  • 完全掌控，自由度高
  • 可深度定制
  • 长期成本可控

缺点：

  • 需要运维能力
  • 扩展性需要自己实现
  • 故障恢复需要手动处理

方案二：容器化部署（Docker + Kubernetes）

适用场景：微服务架构，需要快速扩展

核心流程：

`dockerfile

# Dockerfile示例

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .

RUN pip install -r requirements.txt

COPY . .

EXPOSE 8000

CMD [“uvicorn”, “main:app”, “–host”, “0.0.0.0”, “–port”, “8000”]

`

优点：

  • 环境一致性好
  • 易于扩展和迁移
  • 生态丰富

缺点：

  • 学习曲线陡峭
  • 小项目可能过度工程化

方案三：Serverless无服务器

适用场景：请求量波动大，不想管理服务器

代表平台：

  • AWS Lambda
  • 阿里云函数计算
  • 腾讯云云函数

优点：

  • 按使用量付费
  • 自动扩缩容
  • 无需运维

缺点：

  • 冷启动延迟
  • 运行时间限制
  • 调试困难

方案四：AI专用平台

适用场景：快速上线AI应用，降低技术门槛

代表平台：

  • Hugging Face Spaces
  • Replicate
  • Modal
  • BentoCloud

优点：

  • 开箱即用
  • 内置GPU支持
  • 社区生态

缺点：

  • 定制性有限
  • 可能有平台锁定

实战：用FastAPI部署AI服务

让我们从最基础的API服务开始：

步骤1：创建API服务

`python

# main.py

from fastapi import FastAPI, HTTPException

from pydantic import BaseModel

import openai

app = FastAPI()

class ChatRequest(BaseModel):

message: str

user_id: str = “default”

class ChatResponse(BaseModel):

reply: str

tokens_used: int

@app.post(“/chat”, response_model=ChatResponse)

async def chat_endpoint(request: ChatRequest):

try:

# 调用OpenAI API

response = openai.ChatCompletion.create(

model=”gpt-3.5-turbo”,

messages=[

{“role”: “user”, “content”: request.message}

]

)

return ChatResponse(

reply=response.choices[0].message.content,

tokens_used=response.usage.total_tokens

)

except Exception as e:

raise HTTPException(status_code=500, detail=str(e))

@app.get(“/health”)

async def health_check():

return {“status”: “healthy”}

`

步骤2：添加错误处理和日志

`python

import logging

from fastapi import Request

from fastapi.responses import JSONResponse

# 配置日志

logging.basicConfig(

level=logging.INFO,

format=’%(asctime)s – %(name)s – %(levelname)s – %(message)s’

)

logger = logging.getLogger(__name__)

# 全局异常处理

@app.exception_handler(Exception)

async def global_exception_handler(request: Request, exc: Exception):

logger.error(f”Unhandled exception: {exc}”)

return JSONResponse(

status_code=500,

content={“detail”: “Internal server error”}

)

# 请求日志中间件

@app.middleware(“http”)

async def log_requests(request: Request, call_next):

logger.info(f”Request: {request.method} {request.url}”)

response = await call_next(request)

logger.info(f”Response: {response.status_code}”)

return response

`

步骤3：配置环境变量和密钥

`python

import os

from dotenv import load_dotenv

load_dotenv()

# 从环境变量读取配置

OPENAI_API_KEY = os.getenv(“OPENAI_API_KEY”)

DATABASE_URL = os.getenv(“DATABASE_URL”)

DEBUG = os.getenv(“DEBUG”, “false”).lower() == “true”

if not OPENAI_API_KEY:

raise ValueError(“OPENAI_API_KEY environment variable is required”)

`

步骤4：部署到云服务器

`bash

# 1. 安装依赖

pip install fastapi uvicorn openai python-dotenv

# 2. 创建requirements.txt

pip freeze > requirements.txt

# 3. 启动服务

uvicorn main:app –host 0.0.0.0 –port 8000

# 4. 生产环境使用Gunicorn

gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000

`

性能优化策略

1. 模型优化

量化压缩：

  • 将FP32模型转为INT8或INT4
  • 减少内存占用，提升推理速度
  • 工具：ONNX Runtime、TensorRT

模型蒸馏：

  • 用大模型训练小模型
  • 保持性能的同时减小模型体积

2. 缓存策略

`python

from functools import lru_cache

import hashlib

# 简单的响应缓存

cache = {}

def get_cache_key(message: str) -> str:

return hashlib.md5(message.encode()).hexdigest()

async def chat_with_cache(message: str):

cache_key = get_cache_key(message)

if cache_key in cache:

return cache[cache_key]

# 调用AI接口

response = await call_ai_api(message)

cache[cache_key] = response

return response

`

3. 异步处理

`python

import asyncio

from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=4)

async def process_in_background(task):

loop = asyncio.get_event_loop()

result = await loop.run_in_executor(executor, heavy_computation, task)

return result

`

4. 负载均衡

Nginx配置示例：

`nginx

upstream ai_backend {

server 127.0.0.1:8001;

server 127.0.0.1:8002;

server 127.0.0.1:8003;

server 127.0.0.1:8004;

}

server {

listen 80;

server_name your-domain.com;

location / {

proxy_pass http://ai_backend;

proxy_set_header Host $host;

proxy_set_header X-Real-IP $remote_addr;

}

}

`

监控与运维

关键指标监控

1. 服务健康指标

  • 响应时间（P50、P95、P99）
  • 错误率（4xx、5xx）
  • 请求吞吐量（QPS）

2. 资源使用指标

  • CPU使用率
  • 内存占用
  • GPU利用率
  • 磁盘I/O

3. 业务指标

  • 用户活跃度
  • API调用次数
  • Token消耗量
  • 成本统计

监控工具推荐

开源方案：

  • Prometheus + Grafana：指标收集和可视化
  • ELK Stack：日志收集和分析
  • Sentry：错误追踪

云服务方案：

  • 阿里云ARMS
  • 腾讯云APM
  • AWS CloudWatch

告警设置

`yaml

# Prometheus告警规则示例

groups:

  • name: ai_service_alerts

rules:

  • alert: HighErrorRate

expr: rate(http_requests_total{status=~”5..”}[5m]) > 0.1

for: 2m

labels:

severity: critical

annotations:

summary: “High error rate detected”

  • alert: HighLatency

expr: histogram_quantile(0.95, rate(http_request_duration_seconds_bucket[5m])) > 2

for: 5m

labels:

severity: warning

annotations:

summary: “High latency detected”

`

成本控制

成本构成分析

1. 计算成本

  • 服务器/实例费用
  • GPU租赁费用
  • 按量计费 vs 包年包月

2. API调用成本

  • OpenAI/Anthropic等API费用
  • 按Token计费
  • 批量折扣

3. 存储成本

  • 模型文件存储
  • 用户数据存储
  • 日志存储

降本策略

1. 智能路由

  • 简单任务用小模型
  • 复杂任务用大模型
  • 动态选择最优模型

2. 缓存复用

  • 相似问题复用答案
  • 热点内容预生成
  • 本地缓存减少API调用

3. 弹性伸缩

  • 根据流量自动扩缩
  • 低峰期释放资源
  • 使用Spot实例降低成本

安全最佳实践

1. API安全

认证授权：

`python

from fastapi import Depends, HTTPException, status

from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):

token = credentials.credentials

if not validate_token(token):

raise HTTPException(

status_code=status.HTTP_401_UNAUTHORIZED,

detail=”Invalid authentication credentials”

)

return token

`

速率限制：

`python

from fastapi import Request

from datetime import datetime, timedelta

from collections import defaultdict

request_counts = defaultdict(list)

async def rate_limit(request: Request):

client_ip = request.client.host

now = datetime.now()

# 清理1分钟前的记录

request_counts[client_ip] = [

t for t in request_counts[client_ip]

if now – t < timedelta(minutes=1)

]

# 检查是否超过限制

if len(request_counts[client_ip]) >= 60:

raise HTTPException(status_code=429, detail=”Too many requests”)

request_counts[client_ip].append(now)

`

2. 数据安全

  • 传输加密（HTTPS）
  • 存储加密（AES-256）
  • 数据脱敏处理
  • 定期备份

3. 输入验证

`python

from pydantic import validator

class ChatRequest(BaseModel):

message: str

@validator(‘message’)

def validate_message(cls, v):

if len(v) > 10000:

raise ValueError(‘Message too long’)

if len(v.strip()) == 0:

raise ValueError(‘Message cannot be empty’)

# 过滤潜在的恶意输入

if ” in v.lower():

raise ValueError(‘Invalid characters in message’)

return v.strip()

`

今日行动项

动手实践：

  1. 用FastAPI创建一个简单的AI聊天API
  2. 添加健康检查和错误处理端点
  3. 使用Gunicorn部署到本地环境
  4. 测试并发请求下的表现

思考题：

  1. 你的AI应用预计有多少用户？需要什么级别的扩展能力？
  2. 你的预算范围是多少？哪种部署方案最经济？
  3. 你有哪些安全合规要求需要满足？

下篇预告

今晚5点：我们将进入第9周，探讨AI伦理与安全的第一个话题——AI偏见与公平性。了解AI系统可能存在的偏见来源，以及如何检测和缓解这些偏见。

地铁深读：Docker部署AI应用完整实战

为什么选择Docker？

传统部署最大的痛点是环境不一致——在你电脑上能跑的代码，到了服务器就报错。Docker通过容器化技术解决了这个问题：

  • 一次构建，到处运行
  • 环境隔离，互不干扰
  • 版本控制，快速回滚
  • 资源限制，安全可靠

完整的Docker化AI应用

项目结构：

`

ai-chat-app/

├── app/

│ ├── __init__.py

│ ├── main.py

│ ├── models/

│ ├── services/

│ └── utils/

├── tests/

├── Dockerfile

├── docker-compose.yml

├── requirements.txt

├── .env.example

└── README.md

`

Dockerfile优化版：

`dockerfile

# 多阶段构建，减小镜像体积

FROM python:3.11-slim as builder

WORKDIR /app

COPY requirements.txt .

RUN pip install –user –no-cache-dir -r requirements.txt

FROM python:3.11-slim

WORKDIR /app

COPY –from=builder /root/.local /root/.local

COPY . .

ENV PATH=/root/.local/bin:$PATH

EXPOSE 8000

# 健康检查

HEALTHCHECK –interval=30s –timeout=3s –start-period=5s –retries=3

CMD curl -f http://localhost:8000/health || exit 1

CMD [“gunicorn”, “app.main:app”, “-w”, “4”, “-k”, “uvicorn.workers.UvicornWorker”, “-b”, “0.0.0.0:8000”]

`

docker-compose.yml：

`yaml

version: ‘3.8’

services:

ai-app:

build: .

ports:

  • “8000:8000”

environment:

  • OPENAI_API_KEY=${OPENAI_API_KEY}
  • DATABASE_URL=${DATABASE_URL}

volumes:

  • ./models:/app/models

restart: unless-stopped

redis:

image: redis:7-alpine

ports:

  • “6379:6379”

nginx:

image: nginx:alpine

ports:

  • “80:80”
  • “443:443”

volumes:

  • ./nginx.conf:/etc/nginx/nginx.conf
  • ./ssl:/etc/nginx/ssl

depends_on:

  • ai-app

`

部署到Kubernetes

Deployment配置：

`yaml

apiVersion: apps/v1

kind: Deployment

metadata:

name: ai-chat-app

spec:

replicas: 3

selector:

matchLabels:

app: ai-chat

template:

metadata:

labels:

app: ai-chat

spec:

containers:

  • name: ai-app

image: your-registry/ai-chat-app:latest

ports:

  • containerPort: 8000

resources:

requests:

memory: “512Mi”

cpu: “250m”

limits:

memory: “1Gi”

cpu: “500m”

env:

  • name: OPENAI_API_KEY

valueFrom:

secretKeyRef:

name: ai-secrets

key: openai-key

livenessProbe:

httpGet:

path: /health

port: 8000

initialDelaySeconds: 30

periodSeconds: 10

readinessProbe:

httpGet:

path: /health

port: 8000

initialDelaySeconds: 5

periodSeconds: 5

`

Service配置：

`yaml

apiVersion: v1

kind: Service

metadata:

name: ai-chat-service

spec:

selector:

app: ai-chat

ports:

  • protocol: TCP

port: 80

targetPort: 8000

type: LoadBalancer

`

自动扩缩容

`yaml

apiVersion: autoscaling/v2

kind: HorizontalPodAutoscaler

metadata:

name: ai-chat-hpa

spec:

scaleTargetRef:

apiVersion: apps/v1

kind: Deployment

name: ai-chat-app

minReplicas: 2

maxReplicas: 10

metrics:

  • type: Resource

resource:

name: cpu

target:

type: Utilization

averageUtilization: 70

  • type: Resource

resource:

name: memory

target:

type: Utilization

averageUtilization: 80

`

CI/CD自动化

GitHub Actions示例：

`yaml

name: Deploy AI App

on:

push:

branches: [main]

jobs:

deploy:

runs-on: ubuntu-latest

steps:

  • uses: actions/checkout@v3
  • name: Login to Docker Hub

uses: docker/login-action@v2

with:

username: ${{ secrets.DOCKER_HUB_USERNAME }}

password: ${{ secrets.DOCKER_HUB_TOKEN }}

  • name: Build and push

uses: docker/build-push-action@v4

with:

push: true

tags: your-registry/ai-chat-app:${{ github.sha }}

  • name: Deploy to K8s

uses: azure/k8s-deploy@v4

with:

manifests: |

k8s/deployment.yaml

k8s/service.yaml

images: |

your-registry/ai-chat-app:${{ github.sha }}

`

实战建议

1. 从小开始

  • 先部署单机版验证功能
  • 逐步添加监控、日志、告警
  • 最后考虑高可用和扩展

2. 文档先行

  • 记录部署流程
  • 编写故障排查手册
  • 维护配置变更日志

3. 安全第一

  • 定期更新依赖
  • 扫描镜像漏洞
  • 实施最小权限原则

4. 成本意识

  • 监控资源使用率
  • 清理未使用的资源
  • 选择合适的实例类型

今日收获：

  • 理解了AI应用部署的核心概念和挑战
  • 掌握了主流部署方案的优缺点对比
  • 学会了用FastAPI构建生产级AI服务
  • 了解了性能优化、监控和安全的最佳实践

明天早上8点：Day56——早课 AI产品设计思维

评论

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注

更多文章