2.17 部署上线

一句话总结：上线不是终点，而是起点——关注成本控制和可观测性。

📊 学习进度
状态：⬜ 未开始
上次实时更新：2026-07-03
预计时长：2-3 小时
已完成：0/3 个模块
在整体流程中的位置：AI 应用开发·第 5 阶段

📍 本章定位
服务方案：方案 1（重要 60%）/ 方案 2（重要 60%）
学习方式：🔥 推荐
在流程中的作用：把开发完成的应用部署到生产环境
核心知识点：推理服务、成本控制、架构设计
预计时长：2-3 小时
完成后能做什么：能将 AI 应用部署上线并控制成本

人机分工

环节	谁做	重要度	说明
架构设计	🧑 人	⭐⭐⭐⭐⭐	决定部署架构
成本策略	🧑 人	⭐⭐⭐⭐⭐	决定预算和模型选择
部署执行	🤖 AI	⭐⭐⭐	CI/CD 自动化
监控配置	🤖 AI	⭐⭐⭐	工具自动配置

两种部署模式

模式	适用场景	复杂度	成本结构	启动成本
调用 API	大多数场景、快速验证	低	按 Token 付费	¥0
自部署模型	数据隐私、高吞吐、定制化	高	GPU 固定成本	¥5,000+/月
混合方案	成本敏感、质量要求高	中	API + GPU	¥2,000+/月

部署平台对比（2025-2026）

平台	类型	特点	适用场景	月成本
Vercel	Serverless	自动扩缩、零运维	前端 + API	免费-¥200
Railway	PaaS	简单部署、按用量计费	后端服务	¥50-500
Fly.io	边缘计算	全球部署、低延迟	全球化应用	¥100-1000
AWS Lambda	Serverless	极致扩缩、按调用计费	事件驱动	按量计费
Modal	GPU Serverless	按秒计费 GPU	AI 推理服务	按量计费
Replicate	模型托管	一键部署开源模型	模型即服务	按量计费

数据来源：各平台官网定价，2025-2026。据 Vercel 2025 报告，AI 应用部署量同比增长 300%。

调用 API 型部署

架构：

用户 → CDN → 前端（Vercel） → API Gateway → 后端服务（Railway）
                                                  ↓
                                          AI API（Claude/OpenAI）
                                                  ↓
                                          缓存（Redis）/ 消息队列

关键配置

配置	说明	推荐方案	代码示例
重试	API 调用失败重试	指数退避 + 抖动	见下方代码
限流	控制调用频率	令牌桶算法	见下方代码
降级	API 不可用时的兜底	缓存/规则引擎	见下方代码
超时	单次调用超时	30-60 秒	见下方代码

重试与降级代码实现

python

"""
生产级 API 调用封装 - 包含重试、限流、降级
"""
import time
import asyncio
import anthropic
from functools import wraps
from typing import Optional, Callable
from collections import defaultdict
import hashlib
import json

class AIAPIClient:
    """生产级 AI API 客户端"""
    
    def __init__(self, api_key: str, max_retries: int = 3):
        self.client = anthropic.Anthropic(api_key=api_key)
        self.max_retries = max_retries
        self.cache = {}  # 简化版缓存，生产环境用 Redis
        self.rate_limiter = defaultdict(lambda: {"count": 0, "reset_time": 0})
    
    def _get_cache_key(self, messages: list, model: str) -> str:
        """生成缓存键"""
        content = json.dumps(messages, sort_keys=True) + model
        return hashlib.md5(content.encode()).hexdigest()
    
    def _check_cache(self, cache_key: str) -> Optional[str]:
        """检查缓存"""
        if cache_key in self.cache:
            entry = self.cache[cache_key]
            if time.time() < entry["expire_at"]:
                return entry["response"]
            del self.cache[cache_key]
        return None
    
    def _set_cache(self, cache_key: str, response: str, ttl: int = 3600):
        """设置缓存"""
        self.cache[cache_key] = {
            "response": response,
            "expire_at": time.time() + ttl,
        }
    
    def call_with_retry(
        self,
        messages: list,
        model: str = "claude-sonnet-4-20250514",
        max_tokens: int = 1024,
        use_cache: bool = True,
        fallback_response: str = "抱歉，服务暂时不可用，请稍后再试。",
    ) -> str:
        """
        带重试和降级的 API 调用
        
        重试策略：指数退避 + 抖动
        降级策略：返回缓存或默认响应
        """
        # 检查缓存
        if use_cache:
            cache_key = self._get_cache_key(messages, model)
            cached = self._check_cache(cache_key)
            if cached:
                return cached
        
        last_error = None
        for attempt in range(self.max_retries):
            try:
                response = self.client.messages.create(
                    model=model,
                    max_tokens=max_tokens,
                    messages=messages,
                    timeout=60,  # 60 秒超时
                )
                result = response.content[0].text
                
                # 缓存成功响应
                if use_cache:
                    self._set_cache(cache_key, result)
                
                return result
            
            except anthropic.RateLimitError as e:
                # 限流错误：等待后重试
                wait_time = (2 ** attempt) + (time.time() % 1)  # 指数退避 + 抖动
                print(f"限流，等待 {wait_time:.1f} 秒后重试...")
                time.sleep(wait_time)
                last_error = e
            
            except anthropic.APITimeoutError as e:
                # 超时错误：立即重试
                print(f"超时，第 {attempt + 1} 次重试...")
                last_error = e
            
            except anthropic.APIError as e:
                # 其他 API 错误：降级处理
                print(f"API 错误: {e}")
                return fallback_response
        
        # 所有重试都失败
        print(f"重试 {self.max_retries} 次后仍然失败: {last_error}")
        return fallback_response

# 使用示例
client = AIAPIClient(api_key="your-api-key")

# 正常调用
response = client.call_with_retry(
    messages=[{"role": "user", "content": "什么是 RAG？"}],
    model="claude-sonnet-4-20250514",
)

# 带降级的调用
response = client.call_with_retry(
    messages=[{"role": "user", "content": "紧急问题"}],
    fallback_response="客服忙，请拨打热线 400-xxx-xxxx",
)

Docker 容器化配置

dockerfile

# Dockerfile - FastAPI + Claude API 应用
FROM python:3.11-slim

WORKDIR /app

# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制代码
COPY . .

# 暴露端口
EXPOSE 8000

# 启动命令
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

yaml

# docker-compose.yml - 完整服务栈
version: '3.8'

services:
  # 后端 API 服务
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
      - REDIS_URL=redis://redis:6379
    depends_on:
      - redis
    restart: unless-stopped
    deploy:
      resources:
        limits:
          memory: 1G
  
  # Redis 缓存
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
    restart: unless-stopped
  
  # Nginx 反向代理
  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
      - ./ssl:/etc/nginx/ssl
    depends_on:
      - api
    restart: unless-stopped

volumes:
  redis_data:

自部署模型型部署

推理框架对比（2025-2026）

框架	特点	吞吐量	适用场景	学习曲线
vLLM	PagedAttention、连续批处理	最高	生产环境	中
Ollama	一键安装、简单易用	中等	本地开发、原型验证	低
TensorRT-LLM	NVIDIA 深度优化	极高	极致性能需求	高
llama.cpp	CPU 推理、GGUF 格式	低	无 GPU 环境	中
SGLang	结构化生成优化	高	结构化输出场景	中

数据来源：vLLM 官方基准测试，2025。据测试，vLLM 的吞吐量比 HuggingFace Transformers 高 10-20 倍。

vLLM 部署示例

bash

# 安装 vLLM
pip install vllm

# 启动 API 服务（兼容 OpenAI 格式）
python -m vllm.entrypoints.openai.api_server \
    --model Qwen/Qwen2.5-7B-Instruct \
    --host 0.0.0.0 \
    --port 8000 \
    --gpu-memory-utilization 0.9 \
    --max-model-len 8192 \
    --dtype auto \
    --quantization awq  # 使用 AWQ 量化

python

# 调用 vLLM API（兼容 OpenAI 格式）
from openai import OpenAI

client = OpenAI(
    base_url="http://localhost:8000/v1",
    api_key="not-needed",  # vLLM 不需要 API Key
)

response = client.chat.completions.create(
    model="Qwen/Qwen2.5-7B-Instruct",
    messages=[{"role": "user", "content": "你好！"}],
    max_tokens=512,
    temperature=0.7,
)
print(response.choices[0].message.content)

量化方法对比

方法	显存节省	质量损失	推理速度	适用场景
GPTQ	4x	小（1-2%）	快	GPU 推理
AWQ	4x	小（1-2%）	最快	GPU 推理（推荐）
GGUF	2-4x	中（2-5%）	中	CPU 推理
bitsandbytes	2-4x	小	中	训练+推理
FP8	2x	极小	最快	H100/H200 GPU

数据来源：HuggingFace 量化基准测试，2025。AWQ 在保持 98% 精度的同时，推理速度比 GPTQ 快 10-20%。

Ollama 本地部署示例

bash

# 安装 Ollama（macOS/Linux）
curl -fsSL https://ollama.ai/install.sh | sh

# 拉取模型
ollama pull qwen2.5:7b

# 运行模型
ollama run qwen2.5:7b

# 启动 API 服务
ollama serve
# API 默认在 http://localhost:11434

python

# 调用 Ollama API
import requests

response = requests.post(
    "http://localhost:11434/api/generate",
    json={
        "model": "qwen2.5:7b",
        "prompt": "什么是 RAG？",
        "stream": False,
    },
)
print(response.json()["response"])

成本控制策略

人机分工：🧑 人制定策略 / 🤖 AI 执行优化

成本优化四层架构

语义缓存实现

python

"""
语义缓存 - 相似问题不重复调用 API
节省比例：30-50%（据 LangChain 2025 测试）
"""
import hashlib
import numpy as np
from typing import Optional, List, Dict
from sentence_transformers import SentenceTransformer
import chromadb

class SemanticCache:
    """语义缓存：相似问题命中缓存"""
    
    def __init__(self, similarity_threshold: float = 0.92):
        self.model = SentenceTransformer("all-MiniLM-L6-v2")
        self.client = chromadb.Client()
        self.collection = self.client.create_collection("cache")
        self.threshold = similarity_threshold
        self.cache_data = {}
    
    def get(self, question: str) -> Optional[str]:
        """查询缓存"""
        embedding = self.model.encode(question).tolist()
        
        results = self.collection.query(
            query_embeddings=[embedding],
            n_results=1,
        )
        
        if results["distances"][0] and results["distances"][0][0] < (1 - self.threshold):
            cache_key = results["ids"][0][0]
            return self.cache_data.get(cache_key)
        
        return None
    
    def set(self, question: str, answer: str, ttl: int = 3600):
        """设置缓存"""
        cache_key = hashlib.md5(question.encode()).hexdigest()
        embedding = self.model.encode(question).tolist()
        
        self.collection.add(
            ids=[cache_key],
            embeddings=[embedding],
            metadatas=[{"question": question, "ttl": ttl}],
        )
        self.cache_data[cache_key] = answer

# 使用示例
cache = SemanticCache(similarity_threshold=0.92)

# 查询缓存
cached = cache.get("什么是 RAG？")
if cached:
    print(f"缓存命中: {cached}")
else:
    # 调用 API
    answer = call_api("什么是 RAG？")
    cache.set("什么是 RAG？", answer)
    print(f"API 调用: {answer}")

成本控制策略详解

策略	说明	节省比例	实现复杂度
语义缓存	相似问题不重复调用	30-50%	中
模型路由	简单问题用小模型	40-60%	低
Prompt 精简	减少不必要的 Token	10-20%	低
批处理	非实时请求批量处理	20-40%	中
Token 预算	限制输入长度	15-25%	低
异步调用	并发处理提升吞吐	间接节省	中

成本对比（月调用 100 万次）：

策略组合	月成本	节省幅度
无优化	¥30,000	基准
仅缓存	¥18,000	40%
缓存 + 路由	¥10,000	67%
缓存 + 路由 + Prompt 精简	¥7,000	77%

CI/CD 自动化部署

GitHub Actions 部署流程

yaml

# .github/workflows/deploy.yml - CI/CD 配置
name: Deploy AI App

on:
  push:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - run: pip install -r requirements.txt
      - run: pytest tests/ -v
  
  deploy:
    needs: test
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      # 构建 Docker 镜像
      - name: Build Docker image
        run: docker build -t myapp:${{ github.sha }} .
      
      # 推送到 Docker Hub
      - name: Push to Docker Hub
        run: |
          docker tag myapp:${{ github.sha }} myuser/myapp:latest
          docker push myuser/myapp:latest
      
      # 部署到 Railway
      - name: Deploy to Railway
        uses: bervProject/railway-deploy@main
        with:
          railway_token: ${{ secrets.RAILWAY_TOKEN }}
          service: myapp
      
      # 健康检查
      - name: Health Check
        run: |
          for i in $(seq 1 10); do
            if curl -s https://myapp.railway.app/health | grep -q "ok"; then
              echo "部署成功！"
              exit 0
            fi
            sleep 10
          done
          echo "健康检查失败，触发回滚"
          exit 1

健康检查与优雅关闭

生产环境必须实现健康检查和优雅关闭，防止请求丢失和服务中断。

python

"""
生产级 FastAPI 应用 - 健康检查 + 优雅关闭 + 中间件
"""
from fastapi import FastAPI, Request
from contextlib import asynccontextmanager
import signal
import asyncio
import time

# 全局状态：标记服务是否正在关闭
is_shutting_down = False

@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用生命周期管理"""
    # 启动时：初始化资源
    print("服务启动中...")
    yield
    # 关闭时：清理资源
    global is_shutting_down
    is_shutting_down = True
    print("服务正在优雅关闭，等待当前请求完成...")
    await asyncio.sleep(5)  # 等待当前请求完成
    print("服务已关闭")

app = FastAPI(lifespan=lifespan)

@app.middleware("http")
async def check_shutdown(request: Request, call_next):
    """关闭期间拒绝新请求"""
    if is_shutting_down and request.url.path != "/health":
        return {"status": "error", "message": "服务正在关闭"}
    return await call_next(request)

@app.get("/health")
async def health_check():
    """健康检查端点"""
    return {
        "status": "healthy" if not is_shutting_down else "shutting_down",
        "timestamp": time.time(),
        "version": "1.0.0",
    }

@app.get("/ready")
async def readiness_check():
    """就绪检查：验证依赖服务是否可用"""
    checks = {
        "api": True,      # AI API 是否可用
        "cache": True,     # Redis 是否连接
        "database": True,  # 数据库是否连接
    }
    all_healthy = all(checks.values())
    return {"status": "ready" if all_healthy else "not_ready", "checks": checks}

健康检查配置（Docker）：

yaml

# docker-compose.yml 中添加健康检查
services:
  api:
    build: .
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s      # 每 30 秒检查一次
      timeout: 10s        # 超时 10 秒
      retries: 3          # 连续 3 次失败标记为不健康
      start_period: 40s   # 启动后 40 秒开始检查

部署检查清单

阶段	检查项	状态
部署前	单元测试通过	⬜
部署前	集成测试通过	⬜
部署前	环境变量配置正确	⬜
部署中	Docker 镜像构建成功	⬜
部署中	服务启动正常	⬜
部署后	健康检查通过	⬜
部署后	API 响应正常	⬜
部署后	监控告警就绪	⬜

常见问题

问题	原因	解决方案
响应慢	模型太大/网络延迟	流式响应+CDN+就近部署
成本高	Token 消耗过多	缓存+模型路由+Prompt 精简
服务不稳定	API 限流	重试+降级+多 Provider 兜底
部署失败	依赖冲突	Docker 容器化+固定版本
冷启动慢	Serverless 初始化	预热机制+保持实例
内存溢出	并发过高	限流+队列+资源监控
SSL 证书过期	未配置自动续期	使用 Let's Encrypt + certbot 自动续期
日志丢失	未配置日志收集	使用 ELK 或 Loki 收集日志
环境变量泄露	代码中硬编码	使用 .env 文件 + 密钥管理服务

实操案例：AI 写作助手部署

场景

一个 AI 写作助手产品，日活 1000+ 用户，日均 API 调用 5 万次。

部署架构

部署成本对比

组件	方案	月成本
前端	Vercel Pro	¥150
后端	Railway	¥200
数据库	Supabase	¥150
AI API	Claude（带路由）	¥3,000
缓存	Redis Cloud	¥100
总计		¥3,600

前后对比

指标	之前（手动部署）	之后（自动化）	改善
部署时间	30 分钟	3 分钟	90%↓
部署频率	每周 1 次	每天 3-5 次	15-25x
回滚时间	15 分钟	30 秒	97%↓
故障率	5%	0.5%	90%↓

下一步

完成部署后，进入阶段 6：监控与迭代

参考与延伸

[1] vLLM. "Documentation"（2025）— 高性能推理框架，PagedAttention 技术

[2] Ollama（2025）— 本地模型运行工具，一键部署

[3] Docker. "Best Practices for Python"（2025）— Python 容器化最佳实践

[4] GitHub Actions. "Documentation"（2025）— CI/CD 自动化部署

[5] Railway. "Documentation"（2025）— PaaS 部署平台

[6] Vercel. "AI Applications"（2025）— AI 应用部署平台

[7] HuggingFace. "Quantization Guide"（2025）— 模型量化方法对比

[8] SGLang. "Documentation"（2025）— 结构化生成优化框架，支持 JSON 约束输出

[9] Modal. "GPU Cloud"（2025）— GPU Serverless 平台，按秒计费，适合 AI 推理

2.17 部署上线 ​

人机分工 ​

两种部署模式 ​

部署平台对比（2025-2026） ​

调用 API 型部署 ​

关键配置 ​

重试与降级代码实现 ​

Docker 容器化配置 ​

自部署模型型部署 ​

推理框架对比（2025-2026） ​

vLLM 部署示例 ​

量化方法对比 ​

Ollama 本地部署示例 ​

成本控制策略 ​

成本优化四层架构 ​

语义缓存实现 ​

成本控制策略详解 ​

CI/CD 自动化部署 ​

GitHub Actions 部署流程 ​

健康检查与优雅关闭 ​

部署检查清单 ​

常见问题 ​

实操案例：AI 写作助手部署 ​

场景 ​

部署架构 ​

部署成本对比 ​

前后对比 ​

下一步 ​

参考与延伸 ​

2.17 部署上线

人机分工

两种部署模式

部署平台对比（2025-2026）

调用 API 型部署

关键配置

重试与降级代码实现

Docker 容器化配置

自部署模型型部署

推理框架对比（2025-2026）

vLLM 部署示例

量化方法对比

Ollama 本地部署示例

成本控制策略

成本优化四层架构

语义缓存实现

成本控制策略详解

CI/CD 自动化部署

GitHub Actions 部署流程

健康检查与优雅关闭

部署检查清单

常见问题

实操案例：AI 写作助手部署

场景

部署架构

部署成本对比

前后对比

下一步

参考与延伸