# 部署架构规划示例
架构组件：
– 模型服务：TensorFlow Serving / TorchServe / Triton Inference Server
– API网关：Kong / NGINX / Flask
– 负载均衡：HAProxy / 云负载均衡器
– 监控系统：Prometheus + Grafana
– 日志系统：ELK Stack
– 容器编排：Kubernetes

网络规划：
– 内部网络：10.0.0.0/16
– 模型服务网络：10.1.0.0/24
– API网关：10.2.0.0/24

# PyTorch模型量化示例
import torch
import torch.nn as nn

# 动态量化（最简单的量化方式）
model = load_model(‘fgedu_model.pth’)
quantized_model = torch.quantization.quantize_dynamic(
model,
{nn.Linear, nn.LSTM},
dtype=torch.qint8
)

# 静态量化（需要校准数据集）
model_static = load_model(‘fgedu_model.pth’)
model_static.eval()
model_static.qconfig = torch.quantization.get_default_qconfig(‘fbgemm’)
torch.quantization.prepare(model_static, inplace=True)
torch.quantization.convert(model_static, inplace=True)

# 保存量化后的模型
torch.save(quantized_model.state_dict(), ‘fgedu_model_quantized.pth’)

# TensorFlow模型剪枝示例
import tensorflow_model_optimization as tfmot

# 定义剪枝参数
prune_params = {
‘pruning_schedule’: tfmot.schedules.PolynomialDecay(
initial_sparsity=0.0,
final_sparsity=0.5,
begin_step=2000,
end_step=10000
)
}

# 应用剪枝
model = load_model(‘fgedu_model.h5′)
pruned_model = tfmot.sparsity.keras.prune_low_magnitude(model, **prune_params)

# 编译和训练
pruned_model.compile(optimizer=’adam’, loss=’categorical_crossentropy’)
pruned_model.fit(train_data, train_labels, epochs=10)

# 去除剪枝包装器
final_model = tfmot.sparsity.keras.strip_pruning(pruned_model)
final_model.save(‘fgedu_model_pruned.h5’)

# 知识蒸馏示例
import torch.nn.functional as F

class DistillationLoss(nn.Module):
def __init__(self, temperature=4.0, alpha=0.5):
super().__init__()
self.temperature = temperature
self.alpha = alpha

def forward(self, student_logits, teacher_logits, labels):
# 软目标损失
soft_teacher = F.softmax(teacher_logits / self.temperature, dim=-1)
soft_student = F.log_softmax(student_logits / self.temperature, dim=-1)
distillation_loss = F.kl_div(soft_student, soft_teacher, reduction=’batchmean’) * (self.temperature ** 2)

# 硬目标损失
hard_loss = F.cross_entropy(student_logits, labels)

# 组合损失
total_loss = self.alpha * distillation_loss + (1 – self.alpha) * hard_loss
return total_loss

# 蒸馏训练
teacher_model = load_teacher_model()
student_model = load_student_model()
distillation_criterion = DistillationLoss(temperature=4.0, alpha=0.7)

for epoch in range(num_epochs):
for batch in train_loader:
inputs, labels = batch

teacher_logits = teacher_model(inputs)
student_logits = student_model(inputs)

loss = distillation_criterion(student_logits, teacher_logits, labels)
optimizer.zero_grad()
loss.backward()
optimizer.step()

# Dockerfile示例 – TensorFlow Serving
FROM tensorflow/serving:latest-gpu

# 设置工作目录
WORKDIR /models

# 复制模型文件
COPY fgedu_classifier /models/fgedu_classifier

# 设置模型名称
ENV MODEL_NAME=fgedu_classifier

# 暴露端口
EXPOSE 8501 8500

# 启动服务
ENTRYPOINT [“tensorflow_model_server”, “–port=8500”, “–model_name=fgedu_classifier”, “–model_base_path=/models”]

# 构建镜像
$ docker build -t fgedu-ai-model:v1.0 .

# 运行容器
$ docker run -d -p 8501:8501 –name fgedu-model -v /models:/models fgedu-ai-model:v1.0

# 验证容器运行
$ docker ps
CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES
a1b2c3d4e5f6 fgedu-ai-model:v1.0 “tensorflow_model_se…” 5 seconds ago Up 4 seconds 8500-8501/tcp fgedu-model

# model-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: fgedu-ai-model-deployment
labels:
app: fgedu-ai-model
spec:
replicas: 3
selector:
matchLabels:
app: fgedu-ai-model
template:
metadata:
labels:
app: fgedu-ai-model
spec:
containers:
– name: model-server
image: fgedu-ai-model:v1.0
ports:
– containerPort: 8501
resources:
limits:
nvidia.com/gpu: 1
memory: “8Gi”
cpu: “4”
requests:
memory: “4Gi”
cpu: “2”
env:
– name: MODEL_NAME
value: “fgedu_classifier”
– name: TF Serving parameters
value: “–per_process_gpu_memory_fraction=0.5 –allow_growth=true”
livenessProbe:
httpGet:
path: /v1/models/fgedu_classifier
port: 8501
initialDelaySeconds: 60
periodSeconds: 10
readinessProbe:
httpGet:
path: /v1/models/fgedu_classifier
port: 8501
initialDelaySeconds: 30
periodSeconds: 5

—
# model-service.yaml
apiVersion: v1
kind: Service
metadata:
name: fgedu-ai-model-service
spec:
selector:
app: fgedu-ai-model
ports:
– protocol: TCP
port: 80
targetPort: 8501
type: LoadBalancer

# 部署到Kubernetes
$ kubectl apply -f model-deployment.yaml
deployment.apps/fgedu-ai-model-deployment created

$ kubectl apply -f model-service.yaml
service/fgedu-ai-model-service created

# 检查部署状态
$ kubectl get pods -l app=fgedu-ai-model
NAME READY STATUS RESTARTS AGE
fgedu-ai-model-deployment-7b8f9c-d4e5f 1/1 Running 0 2m
fgedu-ai-model-deployment-7b8f9c-g6h7i 1/1 Running 0 2m
fgedu-ai-model-deployment-7b8f9c-j8k9l 1/1 Running 0 2m

# 模型版本管理示例
models/
├── fgedu_classifier/
│ ├── 1/
│ │ └── saved_model.pb
│ ├── 2/
│ │ └── saved_model.pb
│ └── 3/
│ └── saved_model.pb

# TensorFlow Serving多版本配置
model_config_list {
config {
name: ‘fgedu_classifier’
base_path: ‘/models/fgedu_classifier’
model_platform: ‘tensorflow’
model_version_policy {
specific {
versions: 1
versions: 2
versions: 3
}
}
}
}

# 滚动更新模型版本
$ kubectl set image deployment/fgedu-ai-model-deployment model-server=fgedu-ai-model:v2.0

# 回滚到上一个版本
$ kubectl rollout undo deployment/fgedu-ai-model-deployment

# 查看版本历史
$ kubectl rollout history deployment/fgedu-ai-model-deployment

# Prometheus监控指标配置
– name: requests_total
help: Total number of inference requests
type: counter
metric_labels:
– model_name
– model_version
– status

– name: inference_duration_seconds
help: Inference duration in seconds
type: histogram
buckets: [0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0]
metric_labels:
– model_name

– name: predictions_total
help: Total number of predictions by class
type: counter
metric_labels:
– model_name
– class

– name: model_load_status
help: Model loading status (1=success, 0=failure)
type: gauge
metric_labels:
– model_name
– model_version

# Grafana仪表板关键指标
1. QPS (每秒请求数) – 评估服务负载
2. 平均推理延迟 – 评估响应时间
3. P99延迟 – 评估极端情况性能
4. GPU利用率 – 评估硬件使用效率
5. 错误率 – 评估服务稳定性
6. 预测分布 – 评估模型输出分布

# 查看监控数据
$ curl http://model-server:8501/v1/models/fgedu_classifier
{
“model_version_status”: {
“1”: {
“state”: “AVAILABLE”,
“status”: “OK”
}
}
}

# 结构化日志配置
import json
import logging
from datetime import datetime

class AILogger:
def __init__(self, model_name):
self.model_name = model_name
self.logger = logging.getLogger(model_name)

def log_inference(self, request_id, input_data, output_data, duration, status):
log_entry = {
“timestamp”: datetime.utcnow().isoformat(),
“request_id”: request_id,
“model_name”: self.model_name,
“input_shape”: str(input_data.shape) if hasattr(input_data, ‘shape’) else str(len(input_data)),
“output_shape”: str(output_data.shape) if hasattr(output_data, ‘shape’) else str(len(output_data)),
“duration_ms”: round(duration * 1000, 2),
“status”: status
}
self.logger.info(json.dumps(log_entry))

def log_error(self, request_id, error_message, stack_trace):
log_entry = {
“timestamp”: datetime.utcnow().isoformat(),
“request_id”: request_id,
“model_name”: self.model_name,
“error”: error_message,
“stack_trace”: stack_trace
}
self.logger.error(json.dumps(log_entry))

# 使用日志记录器
logger = AILogger(‘fgedu_classifier’)
logger.log_inference(‘req-001’, input_array, predictions, 0.025, ‘success’)

# 日志输出示例
$ tail -f /var/log/ai-models/fgedu_classifier.log
{“timestamp”: “2026-04-03T10:30:25.123Z”, “request_id”: “req-001”, “model_name”: “fgedu_classifier”, “input_shape”: “[1, 224, 224, 3]”, “output_shape”: “[1, 1000]”, “duration_ms”: 25.34, “status”: “success”}
{“timestamp”: “2026-04-03T10:30:26.456Z”, “request_id”: “req-002”, “model_name”: “fgedu_classifier”, “input_shape”: “[1, 224, 224, 3]”, “output_shape”: “[1, 1000]”, “duration_ms”: 24.89, “status”: “success”}

# Kubernetes HPA配置
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: fgedu-ai-model-hpa
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: fgedu-ai-model-deployment
minReplicas: 2
maxReplicas: 10
metrics:
– type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70
– type: Pods
pods:
metric:
name: requests_per_second
target:
type: AverageValue
averageValue: “100”
behavior:
scaleDown:
stabilizationWindowSeconds: 300
policies:
– type: Percent
value: 10
periodSeconds: 60
scaleUp:
stabilizationWindowSeconds: 60
policies:
– type: Percent
value: 100
periodSeconds: 15
– type: Pods
value: 4
periodSeconds: 15
selectPolicy: Max

# 应用HPA配置
$ kubectl apply -f model-hpa.yaml
horizontalpodautoscaler.autoscaling/fgedu-ai-model-hpa created

# 查看HPA状态
$ kubectl get hpa fgedu-ai-model-hpa
NAME REFERENCE TARGETS MINPODS MAXPODS REPLICAS AGE
fgedu-ai-model-hpa Deployment/fgedu-ai-model-deployment 65%/70% 2 10 3 5m

# 输入验证和过滤
import numpy as np

def validate_input(data, expected_shape, value_range=(0, 1)):
# 检查数据类型
if not isinstance(data, np.ndarray):
raise ValueError(“Input must be numpy array”)

# 检查形状
if data.shape != expected_shape:
raise ValueError(f”Expected shape {expected_shape}, got {data.shape}”)

# 检查数值范围
if np.any(data < value_range[0]) or np.any(data > value_range[1]):
raise ValueError(f”Input values must be in range {value_range}”)

# 检查异常值
if np.any(np.isnan(data)) or np.any(np.isinf(data)):
raise ValueError(“Input contains NaN or Inf values”)

return True

# API认证和授权
from functools import wraps
import jwt

def verify_token(token):
try:
payload = jwt.decode(token, ‘fgedu-secret-key’, algorithms=[‘HS256’])
return payload
except jwt.ExpiredSignatureError:
raise ValueError(“Token expired”)
except jwt.InvalidTokenError:
raise ValueError(“Invalid token”)

# 请求限流
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

limiter = Limiter(key_func=get_remote_address, default_limits=[“200 per day”, “50 per hour”])

@app.route(‘/predict’, methods=[‘POST’])
@limiter.limit(“10 per minute”)
def predict():
# 验证令牌
auth_header = request.headers.get(‘Authorization’)
if not auth_header:
return jsonify({‘error’: ‘No token provided’}), 401

token = auth_header.split(‘ ‘)[1]
verify_token(token)

# 处理请求
data = request.json
return jsonify({‘prediction’: model.predict(data)})

# 数据脱敏处理
import hashlib
import re

def anonymize_data(data):
“””对敏感数据进行脱敏处理”””
anonymized = data.copy()

# 脱敏邮箱
if ’email’ in anonymized:
email = anonymized[’email’]
username, domain = email.split(‘@’)
anonymized[’email’] = username[:2] + ‘***@’ + domain

# 脱敏手机号
if ‘phone’ in anonymized:
phone = anonymized[‘phone’]
anonymized[‘phone’] = phone[:3] + ‘****’ + phone[-4:]

# 哈希处理身份证号
if ‘id_card’ in anonymized:
anonymized[‘id_card’] = hashlib.sha256(
anonymized[‘id_card’].encode()
).hexdigest()[:16]

return anonymized

# 差分隐私配置
import diffprivlib.models as dp

# 使用差分隐私的逻辑回归
model = dp.LogisticRegression(epsilon=1.0, data_norm=1.0)
model.fit(train_data, train_labels)

# 预测时也加入噪声
def predict_with_privacy(model, data, epsilon=1.0):
prediction = model.predict(data)
noise = np.random.laplace(0, 1/epsilon, prediction.shape)
return prediction + noise

# 审计日志配置
import boto3
from datetime import datetime

class ModelAuditor:
def __init__(self, bucket_name):
self.s3_client = boto3.client(‘s3’)
self.bucket_name = bucket_name

def log_request(self, model_name, user_id, input_hash, output_hash, timestamp):
audit_entry = {
‘model_name’: model_name,
‘user_id’: user_id,
‘input_hash’: input_hash,
‘output_hash’: output_hash,
‘timestamp’: timestamp,
‘action’: ‘inference’
}

# 保存到S3
key = f”audit/{model_name}/{timestamp}.json”
self.s3_client.put_object(
Bucket=self.bucket_name,
Key=key,
Body=json.dumps(audit_entry)
)

def query_audit(self, model_name, user_id, start_time, end_time):
# 查询特定用户或时间段的审计记录
pass

# 使用审计系统
auditor = ModelAuditor(‘fgedu-ai-audit-logs’)

@app.route(‘/predict’, methods=[‘POST’])
def predict():
input_data = request.json
input_hash = hashlib.sha256(str(input_data).encode()).hexdigest()

# 执行推理
output = model.predict(input_data)
output_hash = hashlib.sha256(str(output).encode()).hexdigest()

# 记录审计日志
auditor.log_request(
model_name=’fgedu_classifier’,
user_id=get_current_user_id(),
input_hash=input_hash,
output_hash=output_hash,
timestamp=datetime.utcnow().isoformat()
)

return jsonify({‘result’: output.tolist()})