Appearance
章节4：生产级部署工程
本章导学
从开发到生产，AI服务的部署涉及推理加速、服务化、流式输出、负载均衡和成本优化等多个工程领域。本章将深入讲解vLLM推理加速框架的核心机制、FastAPI模型服务化的最佳实践、流式输出与SSE的实现、Docker/K8s部署方案，以及生产级的监控告警与成本优化策略。
学习目标
- 理解vLLM的PagedAttention和连续批处理原理
- 掌握FastAPI模型服务化的RESTful API设计
- 学会实现流式输出与SSE推送
- 了解Docker/K8s部署与弹性扩缩容方案
- 掌握生产环境监控告警与成本优化策略
4.1 vLLM推理加速框架
4.1.1 为什么需要推理加速
大模型推理的核心瓶颈在于显存效率和计算效率。
| 问题 | 说明 | vLLM的解决方案 |
|---|---|---|
| 显存碎片化 | KV Cache占用大量显存，碎片严重 | PagedAttention：按页管理显存 |
| 批处理效率低 | 动态批处理导致等待延迟 | 连续批处理：一个step一个batch |
| 长序列推理慢 | 序列越长，计算量越大 | 内存共享：多序列共享KV Cache |
4.1.2 PagedAttention原理
传统Transformer推理中，每个请求的KV Cache（Key-Value缓存）需要连续显存空间，导致：
- 显存碎片（类似操作系统的内存碎片）
- 无法高效共享（多个beam搜索或采样无法复用）
PagedAttention 借鉴操作系统分页管理思路：
传统方式（连续显存）：
┌──────────────────────────────┐
│ Request A: [K1|K2|K3|...|Kn] │ ← 要求连续块
└──────────────────────────────┘
PagedAttention（分页管理）：
┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐
│ Page 1 │ │ Page 2 │ │ Page 3 │ │ Page 4 │ ← 逻辑连续的物理页
│ K1 │V1 │ │ K2 │V2 │ │ K3 │V3 │ │ K4 │V4 │
└────────┘ └────────┘ └────────┘ └────────┘
│ │ │ │
└───────────┴───────────┴───────────┘
逻辑页表映射（非连续物理）核心优势：
- 零碎片：每次只分配必要大小的物理块
- 共享内存：多个采样结果共享相同的KV页（如beam search的公共前缀）
- 高效利用：显存利用率从~40%提升到~95%
4.1.3 连续批处理（Continuous Batching）
传统批处理需要等当前批次全部生成完成才能处理下一批：
传统批处理：
Batch 1: [Req1, Req2, Req3] 全部生成 -> Batch 2: [Req4, Req5]
▲ 空闲等待 ▲
vLLM连续批处理：
时间线: Req1 ██░░░██░░░██░░░██░░░██
Req2 ██████░░░██████░░░██
Req3 ██░░░░░░░██████░
Req4 ██░░░████
Req5 ██████
↑ 每个step动态调度，随时有新请求加入4.1.4 vLLM部署实战
bash
# 安装vLLM
pip install vllm
# 启动OpenAI兼容的API服务
python -m vllm.entrypoints.openai.api_server \
--model Qwen/Qwen2.5-7B-Instruct \
--tensor-parallel-size 1 \
--gpu-memory-utilization 0.9 \
--max-model-len 8192 \
--port 8000关键参数说明：
| 参数 | 说明 | 推荐值 |
|---|---|---|
--tensor-parallel-size | 张量并行度（GPU数） | 1-8 |
--pipeline-parallel-size | 流水线并行度 | 1 |
--gpu-memory-utilization | GPU显存利用率上限 | 0.85-0.95 |
--max-model-len | 最大序列长度 | 4096-32768 |
--max-num-seqs | 同时处理的最大请求数 | 256 |
--quantization | 量化方式（awq/gptq/fp8） | awq |
Python客户端调用：
python
from openai import OpenAI
# 连接到vLLM服务（兼容OpenAI API格式）
client = OpenAI(
base_url="http://localhost:8000/v1",
api_key="not-needed" # vLLM默认不校验API Key
)
# 流式调用
stream = client.chat.completions.create(
model="Qwen/Qwen2.5-7B-Instruct",
messages=[
{"role": "system", "content": "你是AI助手"},
{"role": "user", "content": "解释一下PagedAttention的原理"}
],
stream=True,
max_tokens=1024,
temperature=0.7
)
for chunk in stream:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="", flush=True)4.1.5 量化与加速对比
| 方案 | 显存占用 | 推理速度 | 质量损失 | 适用场景 |
|---|---|---|---|---|
| FP16 | 100% | 1x | 无 | 高精度需求 |
| INT8 | ~50% | ~1.5x | 极小 | 通用 |
| AWQ-4bit | ~30% | ~2x | 轻微 | 延迟敏感 |
| GPTQ-4bit | ~30% | ~1.8x | 轻微 | 离线批处理 |
| FP8 | ~50% | ~1.3x | 极小 | 新硬件支持 |
4.2 FastAPI模型服务化
4.2.1 架构设计
Client → Nginx/CDN → FastAPI Server → vLLM/LLM Service
│
├── Request Validator
├── Rate Limiter
├── Auth Middleware
├── Cache (Redis)
└── Token Tracker4.2.2 完整服务化实现
python
# pip install fastapi uvicorn pydantic redis
from fastapi import FastAPI, HTTPException, BackgroundTasks
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from typing import Optional, List
import time
import uuid
from contextlib import asynccontextmanager
# ========= 数据模型 =========
class ChatRequest(BaseModel):
"""聊天请求"""
model: str = "Qwen/Qwen2.5-7B-Instruct"
messages: List[dict]
temperature: float = Field(default=0.7, ge=0.0, le=2.0)
max_tokens: int = Field(default=1024, ge=1, le=32768)
stream: bool = False
user_id: Optional[str] = None
class Config:
json_schema_extra = {
"example": {
"model": "Qwen/Qwen2.5-7B-Instruct",
"messages": [{"role": "user", "content": "你好"}],
"temperature": 0.7,
"max_tokens": 1024,
"stream": False
}
}
class ChatResponse(BaseModel):
"""聊天响应"""
id: str
object: str = "chat.completion"
created: int
model: str
choices: list
usage: dict
class TokenUsage(BaseModel):
"""Token用量"""
prompt_tokens: int
completion_tokens: int
total_tokens: int
class ModelInfo(BaseModel):
"""模型信息"""
id: str
object: str = "model"
created: int
owned_by: str = "organization"
# ========= 服务实现 =========
app = FastAPI(
title="AI Model Serving API",
description="生产级大模型服务化接口",
version="1.0.0"
)
# CORS配置
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 模拟LLM推理引擎（实际接入vLLM）
class LLMEngine:
"""LLM推理引擎封装"""
def __init__(self):
self.model_name = "Qwen/Qwen2.5-7B-Instruct"
self.client = None # 实际接入: OpenAI(base_url="http://vllm:8000/v1")
async def generate(self, request: ChatRequest) -> dict:
"""非阻塞生成"""
# 模拟处理延迟
await asyncio.sleep(0.5)
prompt = request.messages[-1]["content"]
completion = f"这是对「{prompt[:20]}...」的模拟回复。"
prompt_tokens = len(prompt) * 2 # 粗略估算
completion_tokens = len(completion) * 2
return {
"id": f"chatcmpl-{uuid.uuid4().hex[:12]}",
"created": int(time.time()),
"model": request.model,
"choices": [{
"index": 0,
"message": {
"role": "assistant",
"content": completion
},
"finish_reason": "stop"
}],
"usage": {
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"total_tokens": prompt_tokens + completion_tokens
}
}
engine = LLMEngine()
# ========= API路由 =========
@app.get("/", tags=["健康检查"])
async def root():
return {"status": "ok", "service": "AI Model Serving", "version": "1.0.0"}
@app.get("/v1/models", tags=["模型管理"])
async def list_models():
"""列出可用模型"""
current_time = int(time.time())
return {
"object": "list",
"data": [
ModelInfo(
id=engine.model_name,
created=current_time
).model_dump()
]
}
@app.post("/v1/chat/completions",
response_model=ChatResponse,
tags=["推理服务"])
async def chat_completions(request: ChatRequest):
"""聊天完成接口（非流式）"""
try:
result = await engine.generate(request)
return result
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/v1/health", tags=["健康检查"])
async def health_check():
"""健康检查接口"""
return {
"status": "healthy",
"timestamp": int(time.time()),
"model": engine.model_name,
"uptime": "running"
}
# ========= 中间件：请求日志与限流 =========
@app.middleware("http")
async def log_requests(request, call_next):
"""记录所有请求"""
start = time.time()
response = await call_next(request)
duration = time.time() - start
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] "
f"{request.method} {request.url.path} "
f"- {response.status_code} ({duration:.3f}s)")
return response
# ========= 启动命令 =========
if __name__ == "__main__":
import uvicorn
uvicorn.run(
"main:app",
host="0.0.0.0",
port=8000,
workers=4, # 多Worker进程
loop="uvloop", # 使用uvloop加速
log_level="info",
reload=False # 生产环境关闭热重载
)4.2.3 异步处理与并发控制
python
import asyncio
from typing import AsyncGenerator
from contextlib import asynccontextmanager
class RequestQueue:
"""请求队列与并发控制"""
def __init__(self, max_concurrent: int = 10):
self.semaphore = asyncio.Semaphore(max_concurrent)
self.queue = asyncio.Queue()
self.active_requests = 0
async def process(self, request: ChatRequest) -> dict:
"""带并发控制的处理"""
async with self.semaphore:
self.active_requests += 1
try:
result = await engine.generate(request)
return result
finally:
self.active_requests -= 1
@property
def status(self) -> dict:
return {
"queue_size": self.queue.qsize(),
"active_requests": self.active_requests,
"max_concurrent": self.semaphore._value # 注意：内部属性
}
queue = RequestQueue(max_concurrent=10)
@app.post("/v1/chat/completions", tags=["推理服务"])
async def chat_completions_with_queue(request: ChatRequest):
"""带队列控制的聊天接口"""
return await queue.process(request)
@app.get("/v1/system/status", tags=["系统管理"])
async def system_status():
"""系统状态"""
return {
"queue": queue.status,
"memory": "256GB",
"gpu": "NVIDIA A100 80GB x4",
"model": engine.model_name
}4.3 流式输出与SSE
4.3.1 SSE（Server-Sent Events）原理
SSE是一种基于HTTP的单向服务器推送技术：
客户端：GET /v1/chat/completions (stream=true)
│
服务器： event: data
data: {"content": "我"}
│
event: data
data: {"content": "是"}
│
event: data
data: {"content": "AI"}
│
event: done
data: [DONE]相比WebSocket的优势：
| 特性 | SSE | WebSocket |
|---|---|---|
| 协议 | HTTP（简单） | WS（复杂） |
| 方向 | 服务器→客户端（单向） | 双向 |
| 自动重连 | 原生支持 | 需手动实现 |
| 兼容性 | 所有HTTP客户端 | 需要WebSocket支持 |
| 适用场景 | 流式输出、通知推送 | 实时双向通信 |
4.3.2 流式输出实现
python
from fastapi.responses import StreamingResponse
import asyncio
import json
class StreamingLLMEngine:
"""流式LLM引擎"""
async def generate_stream(self, request: ChatRequest) -> AsyncGenerator[str, None]:
"""流式生成，返回SSE格式的事件"""
prompt = request.messages[-1]["content"]
# 模拟流式输出（逐字返回）
response_text = f"这是对「{prompt[:20]}...」的流式回复。"
for char in response_text:
# 构造SSE数据格式
chunk = {
"id": f"chatcmpl-{uuid.uuid4().hex[:12]}",
"object": "chat.completion.chunk",
"created": int(time.time()),
"model": request.model,
"choices": [{
"index": 0,
"delta": {
"content": char
},
"finish_reason": None
}]
}
yield f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n"
await asyncio.sleep(0.05) # 模拟生成延迟
# 结束标志
yield "data: [DONE]\n\n"
stream_engine = StreamingLLMEngine()
@app.post("/v1/chat/completions", tags=["推理服务"])
async def chat_completions(request: ChatRequest):
"""支持流式和非流式"""
if request.stream:
# 流式返回
return StreamingResponse(
stream_engine.generate_stream(request),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no" # 禁用Nginx缓冲
}
)
else:
# 非流式
result = await engine.generate(request)
return result4.3.3 客户端流式接收
python
import requests
import json
def stream_chat(url: str, messages: list):
"""客户端流式接收"""
response = requests.post(
f"{url}/v1/chat/completions",
json={
"model": "Qwen/Qwen2.5-7B-Instruct",
"messages": messages,
"stream": True
},
stream=True
)
full_response = ""
for line in response.iter_lines():
if line:
line = line.decode("utf-8")
if line.startswith("data: "):
data = line[6:] # 去掉 "data: " 前缀
if data == "[DONE]":
break
chunk = json.loads(data)
if chunk["choices"][0]["delta"].get("content"):
content = chunk["choices"][0]["delta"]["content"]
print(content, end="", flush=True)
full_response += content
return full_response
# 使用
messages = [{"role": "user", "content": "写一首关于AI的诗"}]
response = stream_chat("http://localhost:8000", messages)4.4 负载均衡与弹性扩缩容
4.4.1 Docker容器化部署
dockerfile
# Dockerfile
FROM nvidia/cuda:12.1-runtime-ubuntu22.04
# 安装Python
RUN apt-get update && apt-get install -y python3.10 python3-pip && \
ln -s /usr/bin/python3 /usr/bin/python
# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 复制应用
COPY . /app
WORKDIR /app
# 暴露端口
EXPOSE 8000
# 启动命令
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]txt
# requirements.txt
fastapi==0.110.0
uvicorn[standard]==0.27.0
vllm==0.4.0
openai==1.12.0
redis==5.0.0
prometheus-client==0.19.0
pydantic==2.6.0yaml
# docker-compose.yml
version: '3.8'
services:
# 模型推理服务
llm-service:
image: llm-service:latest
build: .
deploy:
replicas: 3 # 3个副本
resources:
reservations:
devices:
- driver: nvidia
count: 1
capabilities: [gpu]
environment:
- MODEL_NAME=Qwen/Qwen2.5-7B-Instruct
- CUDA_VISIBLE_DEVICES=0
ports:
- "8000-8002:8000" # 每个副本映射不同端口
volumes:
- ./models:/models # 模型文件挂载
restart: unless-stopped
# Redis缓存
redis:
image: redis:7-alpine
ports:
- "6379:6379"
restart: unless-stopped
# Nginx负载均衡
nginx:
image: nginx:alpine
ports:
- "80:80"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf
depends_on:
- llm-service
restart: unless-stopped4.4.2 Nginx负载均衡配置
nginx
# nginx.conf
upstream llm_servers {
## 负载均衡策略
# least_conn; # 最少连接
# ip_hash; # IP哈希（保持会话）
random; # 随机
server llm-service:8000 max_fails=3 fail_timeout=30s;
server llm-service:8001 max_fails=3 fail_timeout=30s;
server llm-service:8002 max_fails=3 fail_timeout=30s;
}
server {
listen 80;
client_max_body_size 10m;
proxy_read_timeout 300s; # SSE需要长超时
proxy_send_timeout 300s;
location / {
proxy_pass http://llm_servers;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
# SSE支持
proxy_buffering off;
proxy_cache off;
}
# 健康检查端点
location /health {
proxy_pass http://llm_servers/v1/health;
access_log off;
}
}4.4.3 Kubernetes部署
yaml
# k8s-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: llm-service
namespace: ai-platform
spec:
replicas: 3
selector:
matchLabels:
app: llm-service
template:
metadata:
labels:
app: llm-service
spec:
containers:
- name: llm-service
image: registry.example.com/llm-service:latest
ports:
- containerPort: 8000
env:
- name: MODEL_NAME
value: "Qwen/Qwen2.5-7B-Instruct"
resources:
limits:
nvidia.com/gpu: 1
memory: "32Gi"
cpu: "8"
requests:
nvidia.com/gpu: 1
memory: "16Gi"
cpu: "4"
readinessProbe:
httpGet:
path: /v1/health
port: 8000
initialDelaySeconds: 30
periodSeconds: 10
livenessProbe:
httpGet:
path: /v1/health
port: 8000
initialDelaySeconds: 60
periodSeconds: 15
---
# 自动扩缩容
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: llm-service-hpa
namespace: ai-platform
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: llm-service
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
behavior:
scaleUp:
stabilizationWindowSeconds: 60
policies:
- type: Pods
value: 2
periodSeconds: 30
scaleDown:
stabilizationWindowSeconds: 180
policies:
- type: Pods
value: 1
periodSeconds: 60
---
apiVersion: v1
kind: Service
metadata:
name: llm-service
namespace: ai-platform
spec:
selector:
app: llm-service
ports:
- port: 80
targetPort: 8000
type: ClusterIP4.5 监控告警与成本优化
4.5.1 监控指标体系
| 类别 | 指标 | 说明 | 告警阈值 |
|---|---|---|---|
| 性能 | QPS | 每秒请求数 | < 预期值的50% |
| 性能 | P50/P99延迟 | 响应时间分位数 | P99 > 5s |
| 资源 | GPU利用率 | GPU计算/显存使用率 | > 90% |
| 资源 | Token吞吐量 | 每秒生成Token数 | < 预期值的50% |
| 成本 | Token消耗 | 每请求Token数 | 异常突增 |
| 错误 | 错误率 | 5xx/4xx比例 | > 1% |
| 业务 | 活跃用户 | DAU/MAU | 下降趋势 |
4.5.2 Prometheus + Grafana监控
python
# pip install prometheus-client
from prometheus_client import Counter, Histogram, Gauge, generate_latest
from fastapi.responses import Response
import time
# ========= 定义指标 =========
# 请求计数
REQUEST_COUNT = Counter(
'llm_requests_total',
'Total LLM requests',
['model', 'status'] # labels
)
# 延迟直方图
REQUEST_LATENCY = Histogram(
'llm_request_duration_seconds',
'LLM request latency in seconds',
['model'],
buckets=(0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0)
)
# Token消费
TOKEN_CONSUMPTION = Counter(
'llm_tokens_total',
'Total tokens consumed',
['model', 'type'] # type: prompt/completion
)
# GPU显存使用
GPU_MEMORY_USAGE = Gauge(
'llm_gpu_memory_usage_bytes',
'GPU memory usage in bytes',
['gpu_id']
)
# 活跃请求数
ACTIVE_REQUESTS = Gauge(
'llm_active_requests',
'Number of active requests'
)
# ========= Prometheus中间件 =========
@app.middleware("http")
async def metrics_middleware(request, call_next):
"""监控中间件"""
start = time.time()
ACTIVE_REQUESTS.inc()
try:
response = await call_next(request)
status = "success" if response.status_code < 400 else "error"
return response
except Exception as e:
status = "error"
raise
finally:
duration = time.time() - start
REQUEST_COUNT.labels(model="qwen2.5-7b", status=status).inc()
REQUEST_LATENCY.labels(model="qwen2.5-7b").observe(duration)
ACTIVE_REQUESTS.dec()
# Prometheus指标接口
@app.get("/metrics")
async def metrics():
"""Prometheus metrics endpoint"""
return Response(
content=generate_latest(),
media_type="text/plain"
)4.5.3 Token计费与成本优化
python
from datetime import datetime, timedelta
from typing import Dict
class TokenBillingSystem:
"""Token计费系统"""
# 定价模型（每1000 Token的价格）
PRICING = {
"gpt-4": {"prompt": 0.03, "completion": 0.06},
"gpt-4o": {"prompt": 0.01, "completion": 0.03},
"qwen2.5-7b": {"prompt": 0.002, "completion": 0.002},
"deepseek-v2": {"prompt": 0.001, "completion": 0.002}
}
def __init__(self):
self.usage_records = []
def calculate_cost(self, model: str, prompt_tokens: int, completion_tokens: int) -> float:
"""计算单次请求成本"""
pricing = self.PRICING.get(model, self.PRICING["qwen2.5-7b"])
prompt_cost = (prompt_tokens / 1000) * pricing["prompt"]
completion_cost = (completion_tokens / 1000) * pricing["completion"]
return round(prompt_cost + completion_cost, 6)
def record_usage(self, user_id: str, model: str, prompt_tokens: int, completion_tokens: int):
"""记录使用情况"""
cost = self.calculate_cost(model, prompt_tokens, completion_tokens)
record = {
"user_id": user_id,
"model": model,
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"total_tokens": prompt_tokens + completion_tokens,
"cost": cost,
"timestamp": datetime.now().isoformat()
}
self.usage_records.append(record)
return record
def get_daily_report(self, date: str = None) -> Dict:
"""获取日报"""
if not date:
date = datetime.now().strftime("%Y-%m-%d")
records = [r for r in self.usage_records
if r["timestamp"].startswith(date)]
total_cost = sum(r["cost"] for r in records)
total_tokens = sum(r["total_tokens"] for r in records)
return {
"date": date,
"total_requests": len(records),
"total_tokens": total_tokens,
"total_cost": round(total_cost, 4),
"avg_cost_per_request": round(total_cost / len(records), 6) if records else 0,
"by_model": self._group_by_model(records)
}
def _group_by_model(self, records: list) -> Dict:
"""按模型分组统计"""
groups = {}
for r in records:
model = r["model"]
if model not in groups:
groups[model] = {
"requests": 0,
"total_tokens": 0,
"cost": 0
}
groups[model]["requests"] += 1
groups[model]["total_tokens"] += r["total_tokens"]
groups[model]["cost"] += r["cost"]
return groups
# 使用
billing = TokenBillingSystem()
# 记录使用
billing.record_usage("user_001", "qwen2.5-7b", 500, 300)
billing.record_usage("user_001", "gpt-4", 1000, 500)
# 查看日报
report = billing.get_daily_report()
print(f"今日总花费：${report['total_cost']}")
print(f"总Token消耗：{report['total_tokens']}")4.5.4 成本优化策略
| 策略 | 说明 | 预期节省 |
|---|---|---|
| Prompt压缩 | 精简system prompt和对话历史 | 20-40% |
| 语义缓存 | 相似问题直接返回缓存结果 | 30-50% |
| 模型分级 | 简单问题用小模型，复杂问题用大模型 | 40-60% |
| 批处理 | 合并请求批量推理 | 2-3x吞吐量 |
| Token预算 | 限制每个用户的每日Token使用量 | 防止滥用 |
| 模型量化 | 使用量化后的模型 | 2-4x显存效率 |
小结与练习
知识小结
- vLLM推理加速：PagedAttention解决显存碎片，连续批处理提升吞吐量
- FastAPI服务化：RESTful API设计、异步处理、请求队列控制
- 流式输出SSE：逐Token返回，降低用户感知延迟
- 容器化部署：Docker + Nginx负载均衡 + K8s弹性扩缩容
- 监控与成本：Prometheus指标采集、Token计费、多级优化策略
练习
练习1：vLLM部署
部署一个本地vLLM服务（可选择Qwen2.5或Llama系列）：
1. 使用vLLM启动OpenAI兼容API
2. 编写客户端调用流式和非流式接口
3. 对比有无vLLM的性能差异练习2：FastAPI服务
基于FastAPI实现一个模型服务：
1. 实现聊天接口（支持流式和非流式）
2. 添加请求日志和限流中间件
3. 添加健康检查和监控指标接口
4. 使用Swagger UI测试所有接口练习3：Docker + Nginx部署
将练习2的服务容器化：
1. 编写Dockerfile
2. 配置docker-compose（3个副本+Redis+Nginx）
3. 配置Nginx负载均衡
4. 测试负载均衡功能练习4：监控与成本
实现一个完整的监控系统：
1. 集成Prometheus指标（QPS、延迟、Token数）
2. 实现Token计费系统
3. 添加成本优化策略（缓存+模型分级）
4. 模拟不同压力场景并分析指标下一章预告：章节5-结业项目实战 - 综合运用所学知识完成两个大型项目