# 模型序列化
import joblib
joblib.dump(model, ‘model.joblib’)

# 模型加载
model = joblib.load(‘model.joblib’)

# 使用ONNX格式
import onnx
import keras2onnx

# 转换模型为ONNX格式
onnx_model = keras2onnx.convert_keras(model, ‘mnist_model.onnx’)

# 保存ONNX模型
onnx.save_model(onnx_model, ‘mnist_model.onnx’)

# 使用TensorFlow SavedModel格式
model.save(‘saved_model’)

# 使用TensorFlow Lite
converter = tf.lite.TFLiteConverter.from_saved_model(‘saved_model’)
tflite_model = converter.convert()
with open(‘model.tflite’, ‘wb’) as f:
f.write(tflite_model)

# 使用FastAPI部署模型
from fastapi import FastAPI, UploadFile, File
import numpy as np
import tensorflow as tf
from PIL import Image

app = FastAPI()

# 加载模型
model = tf.keras.models.load_model(‘mnist_model.h5’)

@app.post(‘/predict’)
async def predict(file: UploadFile = File(…)):
# 读取图片
image = Image.open(file.file).convert(‘L’)
image = image.resize((28, 28))
image = np.array(image) / 255.0
image = image.reshape(1, 28, 28)

# 预测
prediction = model.predict(image)
class_id = np.argmax(prediction[0])

return {‘class_id’: int(class_id), ‘confidence’: float(np.max(prediction[0]))}

# 运行服务
if __name__ == ‘__main__’:
import uvicorn
uvicorn.run(app, host=’0.0.0.0′, port=8000)

# 使用TensorFlow Serving
$ docker pull tensorflow/serving
$ docker run -p 8501:8501 -v “$(pwd)/saved_model:/models/mnist” -e MODEL_NAME=mnist tensorflow/serving

# 测试API
$ curl -X POST http://localhost:8501/v1/models/mnist:predict -d ‘{“instances”: [[[[…]]]}’

# 使用Prometheus监控AI服务
from prometheus_client import Counter, Histogram, start_http_server

# 定义指标
REQUEST_COUNT = Counter(‘request_count’, ‘Number of requests received’)
REQUEST_LATENCY = Histogram(‘request_latency_seconds’, ‘Request latency in seconds’)
PREDICTION_COUNT = Counter(‘prediction_count’, ‘Number of predictions made’, [‘class’])

# 在API中使用指标
@app.post(‘/predict’)
async def predict(file: UploadFile = File(…)):
REQUEST_COUNT.inc()

with REQUEST_LATENCY.time():
# 处理图片和预测
image = Image.open(file.file).convert(‘L’)
image = image.resize((28, 28))
image = np.array(image) / 255.0
image = image.reshape(1, 28, 28)

prediction = model.predict(image)
class_id = np.argmax(prediction[0])

PREDICTION_COUNT.labels(class=str(class_id)).inc()

return {‘class_id’: int(class_id), ‘confidence’: float(np.max(prediction[0]))}

# 启动监控服务器
start_http_server(8001)

# 使用Grafana可视化监控数据
# 配置Prometheus数据源和仪表板

# 使用Kubernetes部署AI服务
$ nano deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: ai-service
spec:
replicas: 3
selector:
matchLabels:
app: ai-service
template:
metadata:
labels:
app: ai-service
spec:
containers:
– name: ai-service
image: fgedu/ai-service:v1
ports:
– containerPort: 8000
resources:
limits:
cpu: “1”
memory: “2Gi”
requests:
cpu: “500m”
memory: “1Gi”

# 应用部署
$ kubectl apply -f deployment.yaml

# 配置自动扩缩容
$ nano hpa.yaml
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: ai-service-hpa
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: ai-service
minReplicas: 3
maxReplicas: 10
metrics:
– type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70

# 应用HPA配置
$ kubectl apply -f hpa.yaml

# 查看HPA状态
$ kubectl get hpa

# 模型加密
from cryptography.fernet import Fernet

# 生成密钥
key = Fernet.generate_key()
cipher_suite = Fernet(key)

# 加载模型
model = tf.keras.models.load_model(‘mnist_model.h5’)

# 序列化模型
import pickle
model_bytes = pickle.dumps(model)

# 加密模型
encrypted_model = cipher_suite.encrypt(model_bytes)

# 保存加密模型
with open(‘encrypted_model.pkl’, ‘wb’) as f:
f.write(encrypted_model)

# 解密模型
with open(‘encrypted_model.pkl’, ‘rb’) as f:
encrypted_model = f.read()

model_bytes = cipher_suite.decrypt(encrypted_model)
model = pickle.loads(model_bytes)

# API安全
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBasic, HTTPBasicCredentials

security = HTTPBasic()

async def get_current_user(credentials: HTTPBasicCredentials = Depends(security)):
correct_username = “admin”
correct_password = “password”
if credentials.username != correct_username or credentials.password != correct_password:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=”Incorrect username or password”,
headers={“WWW-Authenticate”: “Basic”},
)
return credentials.username

@app.post(‘/predict’)
async def predict(file: UploadFile = File(…), username: str = Depends(get_current_user)):
# 处理预测
pass

# 模型版本管理
import mlflow

# 初始化MLflow
mlflow.set_tracking_uri(‘http://localhost:5000’)
mlflow.set_experiment(‘MNIST Model’)

# 记录模型训练
with mlflow.start_run():
# 训练模型
model.fit(x_train, y_train, epochs=10)

# 评估模型
loss, accuracy = model.evaluate(x_test, y_test)

# 记录参数和指标
mlflow.log_param(‘epochs’, 10)
mlflow.log_metric(‘loss’, loss)
mlflow.log_metric(‘accuracy’, accuracy)

# 保存模型
mlflow.keras.log_model(model, ‘model’)

# 模型注册
model_uri = “runs://model”
mlflow.register_model(model_uri, “MNIST-Model”)

# 模型监控
# 使用MLflow Model Registry和Prometheus监控模型性能