企业级AI中台架构完全教程
概述
随着人工智能技术在企业中的广泛应用，越来越多的组织面临着AI项目碎片化、重复建设、难以规模化落地的困境。AI中台作为企业AI能力的集中化平台，旨在解决这些问题，实现AI资源的统一管理、能力的复用共享和业务的快速赋能。
本教程将全面深入地讲解企业级AI中台的架构设计、核心组件、技术实现和最佳实践，帮助你从零开始构建一套完整的AI中台体系。
什么是AI中台
AI中台是介于AI基础设施和业务应用之间的中间层平台，它将AI能力进行抽象、封装和标准化，提供统一的接口和服务，使得业务团队能够快速、便捷地使用AI能力，而无需关注底层的技术细节。
AI中台的核心价值包括：
- 能力复用：将通用AI能力沉淀为可复用的组件，避免重复建设
- 效率提升：提供标准化的开发、训练、部署流程，加速AI项目交付
- 资源优化：统一管理和调度AI资源，提高资源利用率
- 质量保障：建立统一的质量标准和监控体系，确保AI服务的可靠性
- 成本控制：通过资源共享和自动化运维，降低AI项目的总体成本
AI中台与传统IT中台的区别
| 维度 | 传统IT中台 | AI中台 |
|---|---|---|
| 核心资产 | 业务服务、API | 模型、特征、数据 |
| 资源类型 | 计算、存储 | GPU、TPU、大内存 |
| 迭代周期 | 周/月 | 天/周 |
| 质量指标 | 可用性、延迟 | 准确率、漂移检测 |
| 运维重点 | 服务监控 | 模型监控、数据监控 |
第一章：AI中台整体架构设计
1.1 四层架构模型
企业级AI中台通常采用四层架构设计：
┌─────────────────────────────────────────────────────┐
│ 应用层 (Application Layer) │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ 智能客服 │ │ 推荐系统 │ │ 风控系统 │ │ 智能营销 │ │
│ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │
├─────────────────────────────────────────────────────┤
│ 服务层 (Service Layer) │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │API网关 │ │模型服务 │ │特征服务 │ │调度服务 │ │
│ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │
├─────────────────────────────────────────────────────┤
│ 模型层 (Model Layer) │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │模型训练 │ │模型管理 │ │实验管理 │ │模型评估 │ │
│ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │
├─────────────────────────────────────────────────────┤
│ 数据层 (Data Layer) │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │数据存储 │ │特征平台 │ │数据管道 │ │数据治理 │ │
│ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │
└─────────────────────────────────────────────────────┘
1.2 数据层设计
数据层是AI中台的基础，负责数据的采集、存储、处理和管理。
from abc import ABC, abstractmethod
from typing import Dict, List, Any, Optional
from datetime import datetime
from enum import Enum
import json
import hashlib
class DataSourceType(Enum):
DATABASE = "database"
FILE_SYSTEM = "file_system"
STREAMING = "streaming"
API = "api"
OBJECT_STORAGE = "object_storage"
class DataFormat(Enum):
PARQUET = "parquet"
CSV = "csv"
JSON = "json"
AVRO = "avro"
DELTA = "delta"
class DataSource:
"""数据源定义"""
def __init__(self, source_id: str, name: str, source_type: DataSourceType,
connection_config: Dict, description: str = ""):
self.source_id = source_id
self.name = name
self.source_type = source_type
self.connection_config = connection_config
self.description = description
self.created_at = datetime.now().isoformat()
self.metadata = {}
def to_dict(self):
return {
"source_id": self.source_id,
"name": self.name,
"source_type": self.source_type.value,
"connection_config": self.connection_config,
"description": self.description,
"created_at": self.created_at,
"metadata": self.metadata
}
class DataCatalog:
"""数据目录"""
def __init__(self):
self.sources: Dict[str, DataSource] = {}
self.tables: Dict[str, Dict] = {}
self.lineage: Dict[str, List[str]] = {}
def register_source(self, source: DataSource):
"""注册数据源"""
self.sources[source.source_id] = source
def register_table(self, table_id: str, source_id: str, schema: Dict,
description: str = "", tags: List[str] = None):
"""注册数据表"""
self.tables[table_id] = {
"table_id": table_id,
"source_id": source_id,
"schema": schema,
"description": description,
"tags": tags or [],
"created_at": datetime.now().isoformat(),
"row_count": 0,
"size_bytes": 0
}
def update_table_stats(self, table_id: str, row_count: int, size_bytes: int):
"""更新表统计信息"""
if table_id in self.tables:
self.tables[table_id]["row_count"] = row_count
self.tables[table_id]["size_bytes"] = size_bytes
self.tables[table_id]["updated_at"] = datetime.now().isoformat()
def register_lineage(self, target_table: str, source_tables: List[str]):
"""注册数据血缘"""
self.lineage[target_table] = source_tables
def get_upstream(self, table_id: str) -> List[str]:
"""获取上游表"""
return self.lineage.get(table_id, [])
def get_downstream(self, table_id: str) -> List[str]:
"""获取下游表"""
downstream = []
for target, sources in self.lineage.items():
if table_id in sources:
downstream.append(target)
return downstream
def search_tables(self, keyword: str = None, tags: List[str] = None) -> List[Dict]:
"""搜索表"""
results = []
for table_id, table_info in self.tables.items():
match = True
if keyword:
if keyword.lower() not in table_info["description"].lower() and \
keyword.lower() not in table_id.lower():
match = False
if tags:
if not any(tag in table_info["tags"] for tag in tags):
match = False
if match:
results.append(table_info)
return results
class DataPipeline:
"""数据管道"""
def __init__(self, pipeline_id: str, name: str):
self.pipeline_id = pipeline_id
self.name = name
self.steps = []
self.schedule = None
self.status = "idle"
self.last_run = None
def add_step(self, step_type: str, config: Dict):
"""添加步骤"""
self.steps.append({
"step_type": step_type,
"config": config,
"order": len(self.steps)
})
def set_schedule(self, cron_expression: str):
"""设置调度计划"""
self.schedule = cron_expression
def execute(self, context: Dict = None):
"""执行管道"""
self.status = "running"
self.last_run = datetime.now().isoformat()
results = []
for step in self.steps:
try:
result = self._execute_step(step, context or {})
results.append({"step": step["order"], "status": "success", "result": result})
except Exception as e:
results.append({"step": step["order"], "status": "failed", "error": str(e)})
self.status = "failed"
return results
self.status = "success"
return results
def _execute_step(self, step: Dict, context: Dict):
"""执行单个步骤"""
step_type = step["step_type"]
config = step["config"]
if step_type == "extract":
return self._extract(config, context)
elif step_type == "transform":
return self._transform(config, context)
elif step_type == "load":
return self._load(config, context)
else:
raise ValueError(f"Unknown step type: {step_type}")
def _extract(self, config: Dict, context: Dict):
"""数据提取"""
return {"status": "extracted", "rows": 1000}
def _transform(self, config: Dict, context: Dict):
"""数据转换"""
return {"status": "transformed", "rows": 1000}
def _load(self, config: Dict, context: Dict):
"""数据加载"""
return {"status": "loaded", "rows": 1000}
1.3 服务层设计
服务层是AI中台对外提供服务的核心层，负责将模型层的能力封装为标准化的服务接口。
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from typing import Dict, List, Any, Optional
import uvicorn
import time
import uuid
app = FastAPI(title="AI中台服务网关")
class PredictionRequest(BaseModel):
model_name: str
model_version: str = "latest"
input_data: Dict[str, Any]
request_id: Optional[str] = None
timeout: int = 30
class PredictionResponse(BaseModel):
request_id: str
model_name: str
model_version: str
predictions: Any
latency_ms: float
metadata: Dict[str, Any] = {}
class ModelRegistry:
"""模型注册中心"""
def __init__(self):
self.models: Dict[str, Dict] = {}
self.deployments: Dict[str, Dict] = {}
def register_model(self, model_name: str, model_version: str,
model_config: Dict, metadata: Dict = None):
"""注册模型"""
key = f"{model_name}:{model_version}"
self.models[key] = {
"model_name": model_name,
"model_version": model_version,
"config": model_config,
"metadata": metadata or {},
"registered_at": datetime.now().isoformat(),
"status": "registered"
}
def get_model(self, model_name: str, model_version: str = "latest"):
"""获取模型信息"""
if model_version == "latest":
# 查找最新版本
versions = [
k for k in self.models.keys()
if k.startswith(f"{model_name}:")
]
if not versions:
return None
key = sorted(versions)[-1]
else:
key = f"{model_name}:{model_version}"
return self.models.get(key)
def deploy_model(self, model_name: str, model_version: str,
deployment_config: Dict):
"""部署模型"""
key = f"{model_name}:{model_version}"
if key not in self.models:
raise ValueError(f"Model {key} not found")
deployment_id = str(uuid.uuid4())[:8]
self.deployments[deployment_id] = {
"deployment_id": deployment_id,
"model_name": model_name,
"model_version": model_version,
"config": deployment_config,
"status": "deploying",
"deployed_at": datetime.now().isoformat()
}
# 更新模型状态
self.models[key]["status"] = "deployed"
return deployment_id
# 全局模型注册中心
model_registry = ModelRegistry()
@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
"""模型预测接口"""
start_time = time.time()
# 生成请求ID
request_id = request.request_id or str(uuid.uuid4())
# 获取模型
model_info = model_registry.get_model(request.model_name, request.model_version)
if not model_info:
raise HTTPException(status_code=404, detail=f"Model {request.model_name} not found")
# 执行预测（模拟）
predictions = {"result": "sample_prediction", "confidence": 0.95}
latency = (time.time() - start_time) * 1000
return PredictionResponse(
request_id=request_id,
model_name=request.model_name,
model_version=model_info["model_version"],
predictions=predictions,
latency_ms=round(latency, 2),
metadata={
"model_status": model_info["status"],
"timestamp": datetime.now().isoformat()
}
)
@app.get("/models")
async def list_models():
"""列出所有模型"""
return {"models": list(model_registry.models.values())}
@app.get("/models/{model_name}/versions")
async def list_model_versions(model_name: str):
"""列出模型的所有版本"""
versions = [
info for key, info in model_registry.models.items()
if info["model_name"] == model_name
]
return {"model_name": model_name, "versions": versions}
@app.post("/models/{model_name}/deploy")
async def deploy_model(model_name: str, model_version: str,
config: Dict = None):
"""部署模型"""
try:
deployment_id = model_registry.deploy_model(
model_name, model_version, config or {}
)
return {"deployment_id": deployment_id, "status": "deploying"}
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
第二章：特征平台（Feature Store）设计
2.1 特征平台概述
特征平台是AI中台的核心组件之一，负责特征的统一管理、存储、计算和服务。它解决了特征复用困难、线上线下特征不一致、特征计算效率低等问题。
2.2 特征定义与管理
from datetime import datetime, timedelta
from typing import Dict, List, Any, Optional, Union
from enum import Enum
import json
import hashlib
class FeatureType(Enum):
"""特征类型"""
NUMERICAL = "numerical"
CATEGORICAL = "categorical"
TEXT = "text"
EMBEDDING = "embedding"
IMAGE = "image"
TIME_SERIES = "time_series"
class FeatureStatus(Enum):
"""特征状态"""
DRAFT = "draft"
ACTIVE = "active"
DEPRECATED = "deprecated"
ARCHIVED = "archived"
class AggregationType(Enum):
"""聚合类型"""
COUNT = "count"
SUM = "sum"
AVG = "avg"
MIN = "min"
MAX = "max"
STDDEV = "stddev"
PERCENTILE = "percentile"
DISTINCT_COUNT = "distinct_count"
class FeatureDefinition:
"""特征定义"""
def __init__(self, feature_id: str, name: str, feature_type: FeatureType,
entity_type: str, description: str = "",
owner: str = "", tags: List[str] = None):
self.feature_id = feature_id
self.name = name
self.feature_type = feature_type
self.entity_type = entity_type
self.description = description
self.owner = owner
self.tags = tags or []
self.status = FeatureStatus.DRAFT
self.created_at = datetime.now().isoformat()
self.updated_at = self.created_at
self.version = 1
self.metadata = {}
# 特征来源
self.source_table = None
self.source_column = None
self.transformation = None
# 聚合配置（用于流式特征）
self.aggregation = None
self.time_window = None
# 数据质量
self.nullable = True
self.value_range = None
self.enum_values = None
def set_source(self, table: str, column: str, transformation: str = None):
"""设置特征来源"""
self.source_table = table
self.source_column = column
self.transformation = transformation
def set_aggregation(self, agg_type: AggregationType, time_window: str):
"""设置聚合配置"""
self.aggregation = agg_type.value
self.time_window = time_window
def set_quality_constraints(self, nullable: bool = True,
value_range: tuple = None,
enum_values: List = None):
"""设置数据质量约束"""
self.nullable = nullable
self.value_range = value_range
self.enum_values = enum_values
def to_dict(self):
return {
"feature_id": self.feature_id,
"name": self.name,
"feature_type": self.feature_type.value,
"entity_type": self.entity_type,
"description": self.description,
"owner": self.owner,
"tags": self.tags,
"status": self.status.value,
"created_at": self.created_at,
"updated_at": self.updated_at,
"version": self.version,
"source_table": self.source_table,
"source_column": self.source_column,
"transformation": self.transformation,
"aggregation": self.aggregation,
"time_window": self.time_window,
"nullable": self.nullable,
"value_range": self.value_range,
"enum_values": self.enum_values,
"metadata": self.metadata
}
class FeatureGroup:
"""特征组"""
def __init__(self, group_id: str, name: str, entity_type: str,
description: str = ""):
self.group_id = group_id
self.name = name
self.entity_type = entity_type
self.description = description
self.features: Dict[str, FeatureDefinition] = {}
self.created_at = datetime.now().isoformat()
def add_feature(self, feature: FeatureDefinition):
"""添加特征"""
if feature.entity_type != self.entity_type:
raise ValueError(f"Entity type mismatch: {feature.entity_type} != {self.entity_type}")
self.features[feature.feature_id] = feature
def get_feature(self, feature_id: str) -> Optional[FeatureDefinition]:
"""获取特征"""
return self.features.get(feature_id)
def list_features(self, status: FeatureStatus = None) -> List[FeatureDefinition]:
"""列出特征"""
features = list(self.features.values())
if status:
features = [f for f in features if f.status == status]
return features
class FeatureStore:
"""特征存储"""
def __init__(self, store_id: str, name: str, backend: str = "redis"):
self.store_id = store_id
self.name = name
self.backend = backend
self.feature_groups: Dict[str, FeatureGroup] = {}
self.online_store = {} # 在线存储（Redis等）
self.offline_store = {} # 离线存储（数据湖等）
def create_feature_group(self, group_id: str, name: str,
entity_type: str, description: str = "") -> FeatureGroup:
"""创建特征组"""
group = FeatureGroup(group_id, name, entity_type, description)
self.feature_groups[group_id] = group
return group
def ingest_features(self, group_id: str, features: List[Dict],
is_online: bool = True, is_offline: bool = True):
"""摄入特征数据"""
if group_id not in self.feature_groups:
raise ValueError(f"Feature group {group_id} not found")
group = self.feature_groups[group_id]
for feature_data in features:
entity_id = feature_data.get("entity_id")
feature_values = feature_data.get("features", {})
if is_online:
key = f"{group_id}:{entity_id}"
self.online_store[key] = {
"entity_id": entity_id,
"features": feature_values,
"updated_at": datetime.now().isoformat()
}
if is_offline:
if group_id not in self.offline_store:
self.offline_store[group_id] = []
self.offline_store[group_id].append({
"entity_id": entity_id,
"features": feature_values,
"timestamp": datetime.now().isoformat()
})
def get_online_features(self, group_id: str, entity_ids: List[str],
feature_ids: List[str] = None) -> Dict[str, Dict]:
"""获取在线特征"""
results = {}
for entity_id in entity_ids:
key = f"{group_id}:{entity_id}"
if key in self.online_store:
data = self.online_store[key]["features"]
if feature_ids:
data = {k: v for k, v in data.items() if k in feature_ids}
results[entity_id] = data
else:
results[entity_id] = {}
return results
def get_offline_features(self, group_id: str,
start_time: datetime = None,
end_time: datetime = None,
entity_ids: List[str] = None) -> List[Dict]:
"""获取离线特征"""
if group_id not in self.offline_store:
return []
records = self.offline_store[group_id]
# 过滤时间范围
if start_time:
records = [r for r in records if r["timestamp"] >= start_time.isoformat()]
if end_time:
records = [r for r in records if r["timestamp"] <= end_time.isoformat()]
# 过滤实体
if entity_ids:
records = [r for r in records if r["entity_id"] in entity_ids]
return records
def compute_features(self, group_id: str, feature_definitions: List[FeatureDefinition],
source_data: List[Dict]) -> List[Dict]:
"""计算特征"""
computed_features = []
for record in source_data:
entity_id = record.get("entity_id")
feature_values = {}
for feature_def in feature_definitions:
value = self._compute_single_feature(feature_def, record)
feature_values[feature_def.feature_id] = value
computed_features.append({
"entity_id": entity_id,
"features": feature_values
})
return computed_features
def _compute_single_feature(self, feature_def: FeatureDefinition,
record: Dict) -> Any:
"""计算单个特征"""
source_value = record.get(feature_def.source_column)
if source_value is None:
return None
if feature_def.transformation:
# 应用转换
return self._apply_transformation(source_value, feature_def.transformation)
return source_value
def _apply_transformation(self, value: Any, transformation: str) -> Any:
"""应用特征转换"""
if transformation == "log":
import math
return math.log(value + 1) if value > 0 else 0
elif transformation == "normalize":
# 简单的归一化
return value
elif transformation == "bucketize":
# 分桶
return int(value // 10) * 10
else:
return value
class FeatureServer:
"""特征服务"""
def __init__(self, feature_store: FeatureStore):
self.feature_store = feature_store
self.cache = {}
self.cache_ttl = 300 # 5分钟
def get_features_for_model(self, model_name: str, entity_ids: List[str]) -> Dict:
"""获取模型所需的特征"""
# 获取模型的特征配置
feature_config = self._get_model_feature_config(model_name)
results = {}
for group_id, feature_ids in feature_config.items():
group_features = self.feature_store.get_online_features(
group_id, entity_ids, feature_ids
)
for entity_id, features in group_features.items():
if entity_id not in results:
results[entity_id] = {}
results[entity_id].update(features)
return results
def _get_model_feature_config(self, model_name: str) -> Dict[str, List[str]]:
"""获取模型的特征配置（模拟）"""
return {
"user_features": ["age", "gender", "income"],
"item_features": ["category", "price", "rating"]
}
2.3 特征计算引擎
class FeatureComputeEngine:
"""特征计算引擎"""
def __init__(self):
self.jobs: Dict[str, Dict] = {}
def submit_batch_job(self, job_id: str, feature_definitions: List[FeatureDefinition],
source_config: Dict, schedule: str = None):
"""提交批量特征计算任务"""
job = {
"job_id": job_id,
"type": "batch",
"feature_definitions": [f.to_dict() for f in feature_definitions],
"source_config": source_config,
"schedule": schedule,
"status": "submitted",
"submitted_at": datetime.now().isoformat()
}
self.jobs[job_id] = job
return job
def submit_streaming_job(self, job_id: str, feature_definitions: List[FeatureDefinition],
source_config: Dict):
"""提交流式特征计算任务"""
job = {
"job_id": job_id,
"type": "streaming",
"feature_definitions": [f.to_dict() for f in feature_definitions],
"source_config": source_config,
"status": "running",
"started_at": datetime.now().isoformat()
}
self.jobs[job_id] = job
return job
def get_job_status(self, job_id: str) -> Dict:
"""获取任务状态"""
return self.jobs.get(job_id, {})
def cancel_job(self, job_id: str):
"""取消任务"""
if job_id in self.jobs:
self.jobs[job_id]["status"] = "cancelled"
第三章：模型训练平台
3.1 实验管理
import json
import os
from datetime import datetime
from typing import Dict, List, Any, Optional
from pathlib import Path
class Experiment:
"""实验定义"""
def __init__(self, experiment_id: str, name: str, description: str = "",
project: str = "", tags: List[str] = None):
self.experiment_id = experiment_id
self.name = name
self.description = description
self.project = project
self.tags = tags or []
self.created_at = datetime.now().isoformat()
self.updated_at = self.created_at
self.status = "created"
self.runs: List[ExperimentRun] = []
self.metadata = {}
def create_run(self, run_name: str = None) -> 'ExperimentRun':
"""创建实验运行"""
run_id = f"run_{len(self.runs) + 1:04d}"
run = ExperimentRun(run_id, run_name or f"Run {len(self.runs) + 1}", self.experiment_id)
self.runs.append(run)
return run
def get_best_run(self, metric: str, ascending: bool = True) -> Optional['ExperimentRun']:
"""获取最佳运行"""
if not self.runs:
return None
valid_runs = [r for r in self.runs if metric in r.metrics]
if not valid_runs:
return None
return min(valid_runs, key=lambda r: r.metrics[metric]) if ascending \
else max(valid_runs, key=lambda r: r.metrics[metric])
def to_dict(self):
return {
"experiment_id": self.experiment_id,
"name": self.name,
"description": self.description,
"project": self.project,
"tags": self.tags,
"created_at": self.created_at,
"status": self.status,
"run_count": len(self.runs)
}
class ExperimentRun:
"""实验运行"""
def __init__(self, run_id: str, name: str, experiment_id: str):
self.run_id = run_id
self.name = name
self.experiment_id = experiment_id
self.created_at = datetime.now().isoformat()
self.updated_at = self.created_at
self.status = "running"
self.start_time = datetime.now()
self.end_time = None
# 参数
self.params: Dict[str, Any] = {}
# 指标
self.metrics: Dict[str, float] = {}
# 工件
self.artifacts: List[Dict] = []
# 标签
self.tags: Dict[str, str] = {}
# 日志
self.logs: List[str] = []
def log_param(self, key: str, value: Any):
"""记录参数"""
self.params[key] = value
self.updated_at = datetime.now().isoformat()
def log_params(self, params: Dict[str, Any]):
"""记录多个参数"""
self.params.update(params)
self.updated_at = datetime.now().isoformat()
def log_metric(self, key: str, value: float, step: int = None):
"""记录指标"""
if step is not None:
metric_key = f"{key}_step_{step}"
self.metrics[metric_key] = value
self.metrics[key] = value
self.updated_at = datetime.now().isoformat()
def log_metrics(self, metrics: Dict[str, float]):
"""记录多个指标"""
self.metrics.update(metrics)
self.updated_at = datetime.now().isoformat()
def log_artifact(self, local_path: str, artifact_name: str = None):
"""记录工件"""
artifact = {
"name": artifact_name or Path(local_path).name,
"path": local_path,
"size_bytes": os.path.getsize(local_path) if os.path.exists(local_path) else 0,
"logged_at": datetime.now().isoformat()
}
self.artifacts.append(artifact)
def log_tag(self, key: str, value: str):
"""记录标签"""
self.tags[key] = value
def log_message(self, message: str):
"""记录日志"""
timestamp = datetime.now().isoformat()
self.logs.append(f"[{timestamp}] {message}")
def finish(self, status: str = "completed"):
"""完成运行"""
self.status = status
self.end_time = datetime.now()
self.updated_at = datetime.now().isoformat()
def get_duration(self) -> float:
"""获取运行时长（秒）"""
if self.end_time:
return (self.end_time - self.start_time).total_seconds()
return (datetime.now() - self.start_time).total_seconds()
def to_dict(self):
return {
"run_id": self.run_id,
"name": self.name,
"experiment_id": self.experiment_id,
"created_at": self.created_at,
"status": self.status,
"params": self.params,
"metrics": self.metrics,
"artifacts_count": len(self.artifacts),
"duration_seconds": self.get_duration()
}
class ExperimentTracker:
"""实验追踪器"""
def __init__(self, tracking_dir: str = "./mlruns"):
self.tracking_dir = Path(tracking_dir)
self.tracking_dir.mkdir(parents=True, exist_ok=True)
self.experiments: Dict[str, Experiment] = {}
def create_experiment(self, name: str, description: str = "",
project: str = "", tags: List[str] = None) -> Experiment:
"""创建实验"""
experiment_id = f"exp_{len(self.experiments) + 1:04d}"
experiment = Experiment(experiment_id, name, description, project, tags)
self.experiments[experiment_id] = experiment
# 创建实验目录
exp_dir = self.tracking_dir / experiment_id
exp_dir.mkdir(exist_ok=True)
# 保存实验信息
with open(exp_dir / "meta.json", 'w') as f:
json.dump(experiment.to_dict(), f, indent=2)
return experiment
def get_experiment(self, experiment_id: str) -> Optional[Experiment]:
"""获取实验"""
return self.experiments.get(experiment_id)
def list_experiments(self, project: str = None) -> List[Dict]:
"""列出实验"""
experiments = list(self.experiments.values())
if project:
experiments = [e for e in experiments if e.project == project]
return [e.to_dict() for e in experiments]
def compare_runs(self, run_ids: List[str]) -> Dict:
"""比较多个运行"""
runs = []
for exp in self.experiments.values():
for run in exp.runs:
if run.run_id in run_ids:
runs.append(run)
if not runs:
return {"error": "No runs found"}
# 比较参数
all_params = set()
for run in runs:
all_params.update(run.params.keys())
param_comparison = {}
for param in all_params:
param_comparison[param] = [run.params.get(param) for run in runs]
# 比较指标
all_metrics = set()
for run in runs:
all_metrics.update(run.metrics.keys())
metric_comparison = {}
for metric in all_metrics:
values = [run.metrics.get(metric) for run in runs]
metric_comparison[metric] = {
"values": values,
"min": min(v for v in values if v is not None),
"max": max(v for v in values if v is not None),
"best_run": runs[values.index(min(v for v in values if v is not None))].run_id
}
return {
"run_ids": run_ids,
"param_comparison": param_comparison,
"metric_comparison": metric_comparison
}
3.2 分布式训练
import torch
import torch.nn as nn
import torch.distributed as dist
from torch.nn.parallel import DistributedDataParallel as DDP
from torch.utils.data import DataLoader, DistributedSampler
class DistributedTrainingManager:
"""分布式训练管理器"""
def __init__(self, config: Dict):
self.config = config
self.backend = config.get("backend", "nccl")
self.world_size = config.get("world_size", 1)
self.rank = config.get("rank", 0)
self.local_rank = config.get("local_rank", 0)
def setup(self):
"""初始化分布式环境"""
if self.world_size > 1:
dist.init_process_group(
backend=self.backend,
init_method=self.config.get("init_method", "env://"),
world_size=self.world_size,
rank=self.rank
)
# 设置设备
if torch.cuda.is_available():
torch.cuda.set_device(self.local_rank)
self.device = torch.device(f"cuda:{self.local_rank}")
else:
self.device = torch.device("cpu")
def cleanup(self):
"""清理分布式环境"""
if self.world_size > 1:
dist.destroy_process_group()
def wrap_model(self, model: nn.Module) -> nn.Module:
"""包装模型为分布式模型"""
model = model.to(self.device)
if self.world_size > 1:
model = DDP(model, device_ids=[self.local_rank])
return model
def create_dataloader(self, dataset, batch_size: int, shuffle: bool = True):
"""创建分布式数据加载器"""
if self.world_size > 1:
sampler = DistributedSampler(dataset, shuffle=shuffle)
dataloader = DataLoader(
dataset, batch_size=batch_size, sampler=sampler,
num_workers=self.config.get("num_workers", 4),
pin_memory=True
)
else:
dataloader = DataLoader(
dataset, batch_size=batch_size, shuffle=shuffle,
num_workers=self.config.get("num_workers", 4)
)
return dataloader
def is_main_process(self) -> bool:
"""是否是主进程"""
return self.rank == 0
def barrier(self):
"""同步屏障"""
if self.world_size > 1:
dist.barrier()
def all_reduce(self, tensor, op=dist.ReduceOp.SUM):
"""全归约"""
if self.world_size > 1:
dist.all_reduce(tensor, op=op)
return tensor
class DistributedTrainer:
"""分布式训练器"""
def __init__(self, model, optimizer, criterion, training_manager: DistributedTrainingManager):
self.model = model
self.optimizer = optimizer
self.criterion = criterion
self.manager = training_manager
self.experiment_tracker = None
def set_experiment_tracker(self, tracker: ExperimentTracker):
"""设置实验追踪器"""
self.experiment_tracker = tracker
def train_epoch(self, dataloader, epoch: int, log_interval: int = 100):
"""训练一个epoch"""
self.model.train()
total_loss = 0
num_batches = 0
for batch_idx, (data, target) in enumerate(dataloader):
data = data.to(self.manager.device)
target = target.to(self.manager.device)
self.optimizer.zero_grad()
output = self.model(data)
loss = self.criterion(output, target)
loss.backward()
self.optimizer.step()
total_loss += loss.item()
num_batches += 1
if batch_idx % log_interval == 0 and self.manager.is_main_process():
print(f"Epoch {epoch}, Batch {batch_idx}, Loss: {loss.item():.4f}")
if self.experiment_tracker and self.experiment_tracker.current_run:
self.experiment_tracker.current_run.log_metric(
"batch_loss", loss.item(), step=epoch * len(dataloader) + batch_idx
)
avg_loss = total_loss / num_batches if num_batches > 0 else 0
# 同步损失
if self.manager.world_size > 1:
loss_tensor = torch.tensor([avg_loss]).to(self.manager.device)
self.manager.all_reduce(loss_tensor)
avg_loss = loss_tensor.item() / self.manager.world_size
return avg_loss
def evaluate(self, dataloader):
"""评估模型"""
self.model.eval()
total_loss = 0
correct = 0
total = 0
with torch.no_grad():
for data, target in dataloader:
data = data.to(self.manager.device)
target = target.to(self.manager.device)
output = self.model(data)
loss = self.criterion(output, target)
total_loss += loss.item()
_, predicted = output.max(1)
total += target.size(0)
correct += predicted.eq(target).sum().item()
avg_loss = total_loss / len(dataloader)
accuracy = correct / total
return {"loss": avg_loss, "accuracy": accuracy}
def train(self, train_dataset, val_dataset, epochs: int, batch_size: int):
"""完整训练流程"""
# 创建数据加载器
train_loader = self.manager.create_dataloader(train_dataset, batch_size)
val_loader = self.manager.create_dataloader(val_dataset, batch_size, shuffle=False)
# 包装模型
self.model = self.manager.wrap_model(self.model)
best_val_loss = float('inf')
for epoch in range(epochs):
# 训练
train_loss = self.train_epoch(train_loader, epoch)
# 验证
val_metrics = self.evaluate(val_loader)
if self.manager.is_main_process():
print(f"Epoch {epoch}: Train Loss: {train_loss:.4f}, "
f"Val Loss: {val_metrics['loss']:.4f}, "
f"Val Accuracy: {val_metrics['accuracy']:.4f}")
# 记录指标
if self.experiment_tracker and self.experiment_tracker.current_run:
self.experiment_tracker.current_run.log_metrics({
"train_loss": train_loss,
"val_loss": val_metrics["loss"],
"val_accuracy": val_metrics["accuracy"],
"epoch": epoch
})
# 保存最佳模型
if val_metrics["loss"] < best_val_loss:
best_val_loss = val_metrics["loss"]
self.save_checkpoint(f"best_model_epoch_{epoch}.pt")
# 同步
self.manager.barrier()
def save_checkpoint(self, filename: str):
"""保存检查点"""
checkpoint = {
"model_state_dict": self.model.module.state_dict() if hasattr(self.model, 'module') else self.model.state_dict(),
"optimizer_state_dict": self.optimizer.state_dict(),
}
torch.save(checkpoint, filename)
3.3 超参数优化
import random
import numpy as np
from typing import Dict, List, Callable, Any
from dataclasses import dataclass
from enum import Enum
class SearchSpaceType(Enum):
"""搜索空间类型"""
UNIFORM = "uniform"
LOG_UNIFORM = "log_uniform"
CHOICE = "choice"
INT_UNIFORM = "int_uniform"
@dataclass
class HyperParameter:
"""超参数定义"""
name: str
space_type: SearchSpaceType
low: float = None
high: float = None
choices: List[Any] = None
log_scale: bool = False
def sample(self) -> Any:
"""采样"""
if self.space_type == SearchSpaceType.UNIFORM:
return random.uniform(self.low, self.high)
elif self.space_type == SearchSpaceType.LOG_UNIFORM:
return np.exp(random.uniform(np.log(self.low), np.log(self.high)))
elif self.space_type == SearchSpaceType.INT_UNIFORM:
return random.randint(int(self.low), int(self.high))
elif self.space_type == SearchSpaceType.CHOICE:
return random.choice(self.choices)
else:
raise ValueError(f"Unknown space type: {self.space_type}")
class HyperparameterOptimizer:
"""超参数优化器"""
def __init__(self, objective_function: Callable, search_space: List[HyperParameter]):
self.objective = objective_function
self.search_space = search_space
self.trials: List[Dict] = []
self.best_trial = None
def random_search(self, n_trials: int = 50) -> Dict:
"""随机搜索"""
for i in range(n_trials):
# 采样参数
params = {}
for hp in self.search_space:
params[hp.name] = hp.sample()
# 评估
score = self.objective(params)
# 记录试验
trial = {
"trial_id": i,
"params": params,
"score": score,
"timestamp": datetime.now().isoformat()
}
self.trials.append(trial)
# 更新最佳
if self.best_trial is None or score < self.best_trial["score"]:
self.best_trial = trial
print(f"Trial {i}: score={score:.4f}, best={self.best_trial['score']:.4f}")
return self.best_trial
def grid_search(self, n_points: int = 3) -> Dict:
"""网格搜索"""
# 为每个参数生成网格点
grid_points = []
for hp in self.search_space:
if hp.space_type == SearchSpaceType.UNIFORM:
points = np.linspace(hp.low, hp.high, n_points).tolist()
elif hp.space_type == SearchSpaceType.INT_UNIFORM:
points = list(range(int(hp.low), int(hp.high) + 1,
max(1, (int(hp.high) - int(hp.low)) // (n_points - 1))))
elif hp.space_type == SearchSpaceType.CHOICE:
points = hp.choices[:n_points]
else:
points = [hp.sample() for _ in range(n_points)]
grid_points.append(points)
# 生成所有组合
import itertools
combinations = list(itertools.product(*grid_points))
for i, combo in enumerate(combinations):
params = {hp.name: val for hp, val in zip(self.search_space, combo)}
score = self.objective(params)
trial = {
"trial_id": i,
"params": params,
"score": score,
"timestamp": datetime.now().isoformat()
}
self.trials.append(trial)
if self.best_trial is None or score < self.best_trial["score"]:
self.best_trial = trial
return self.best_trial
def bayesian_search(self, n_trials: int = 50, n_initial: int = 10) -> Dict:
"""贝叶斯搜索（简化版）"""
# 初始随机探索
for i in range(n_initial):
params = {hp.name: hp.sample() for hp in self.search_space}
score = self.objective(params)
trial = {
"trial_id": i,
"params": params,
"score": score,
"timestamp": datetime.now().isoformat()
}
self.trials.append(trial)
if self.best_trial is None or score < self.best_trial["score"]:
self.best_trial = trial
# 基于历史的智能探索
for i in range(n_initial, n_trials):
# 简化的贝叶斯：基于历史最佳参数附近采样
if self.best_trial:
params = self._sample_near_best()
else:
params = {hp.name: hp.sample() for hp in self.search_space}
score = self.objective(params)
trial = {
"trial_id": i,
"params": params,
"score": score,
"timestamp": datetime.now().isoformat()
}
self.trials.append(trial)
if score < self.best_trial["score"]:
self.best_trial = trial
return self.best_trial
def _sample_near_best(self) -> Dict:
"""在最佳参数附近采样"""
params = {}
for hp in self.search_space:
best_value = self.best_trial["params"].get(hp.name)
if hp.space_type == SearchSpaceType.UNIFORM:
# 在最佳值附近添加噪声
noise = random.gauss(0, (hp.high - hp.low) * 0.1)
value = np.clip(best_value + noise, hp.low, hp.high)
elif hp.space_type == SearchSpaceType.INT_UNIFORM:
noise = random.randint(-1, 1)
value = int(np.clip(best_value + noise, hp.low, hp.high))
elif hp.space_type == SearchSpaceType.CHOICE:
# 有概率选择最佳值，有概率随机
if random.random() < 0.7:
value = best_value
else:
value = hp.sample()
else:
value = hp.sample()
params[hp.name] = value
return params
def get_results_summary(self) -> Dict:
"""获取结果摘要"""
if not self.trials:
return {"message": "No trials completed"}
scores = [t["score"] for t in self.trials]
return {
"total_trials": len(self.trials),
"best_score": self.best_trial["score"],
"best_params": self.best_trial["params"],
"score_stats": {
"mean": np.mean(scores),
"std": np.std(scores),
"min": np.min(scores),
"max": np.max(scores)
}
}
# 使用示例
def train_and_evaluate(params: Dict) -> float:
"""训练并评估模型（目标函数）"""
learning_rate = params["learning_rate"]
batch_size = int(params["batch_size"])
n_layers = int(params["n_layers"])
dropout = params["dropout"]
# 模拟训练和评估
# 实际应用中，这里会进行真正的模型训练
score = (learning_rate * 1000 + batch_size * 0.01 +
n_layers * 0.1 + dropout * 0.5 + random.gauss(0, 0.1))
return score
# 定义搜索空间
search_space = [
HyperParameter("learning_rate", SearchSpaceType.LOG_UNIFORM, low=1e-5, high=1e-2),
HyperParameter("batch_size", SearchSpaceType.CHOICE, choices=[16, 32, 64, 128]),
HyperParameter("n_layers", SearchSpaceType.INT_UNIFORM, low=2, high=8),
HyperParameter("dropout", SearchSpaceType.UNIFORM, low=0.1, high=0.5)
]
# 创建优化器
optimizer = HyperparameterOptimizer(train_and_evaluate, search_space)
# 执行随机搜索
best_trial = optimizer.random_search(n_trials=20)
print(f"Best trial: {best_trial}")
第四章：模型管理平台
4.1 模型注册中心
from datetime import datetime
from typing import Dict, List, Optional, Any
from enum import Enum
import json
import hashlib
class ModelFramework(Enum):
"""模型框架"""
PYTORCH = "pytorch"
TENSORFLOW = "tensorflow"
ONNX = "onnx"
SCIKIT_LEARN = "scikit_learn"
XGBOOST = "xgboost"
LIGHTGBM = "lightgbm"
CUSTOM = "custom"
class ModelStage(Enum):
"""模型阶段"""
DEVELOPMENT = "development"
STAGING = "staging"
PRODUCTION = "production"
ARCHIVED = "archived"
class ModelStatus(Enum):
"""模型状态"""
REGISTERED = "registered"
VALIDATING = "validating"
APPROVED = "approved"
REJECTED = "rejected"
DEPLOYED = "deployed"
class ModelVersion:
"""模型版本"""
def __init__(self, model_name: str, version: str, framework: ModelFramework,
description: str = "", tags: List[str] = None):
self.model_name = model_name
self.version = version
self.framework = framework
self.description = description
self.tags = tags or []
self.created_at = datetime.now().isoformat()
self.updated_at = self.created_at
# 模型信息
self.model_path = None
self.model_size_bytes = 0
self.model_hash = None
# 训练信息
self.training_dataset = None
self.training_params = {}
self.training_metrics = {}
# 评估信息
self.evaluation_metrics = {}
self.evaluation_dataset = None
# 部署信息
self.stage = ModelStage.DEVELOPMENT
self.status = ModelStatus.REGISTERED
self.deployments: List[Dict] = []
# 审批信息
self.approval_history: List[Dict] = []
# 元数据
self.metadata = {}
self.labels: Dict[str, str] = {}
def set_model_path(self, path: str):
"""设置模型路径"""
import os
self.model_path = path
if os.path.exists(path):
self.model_size_bytes = os.path.getsize(path)
# 计算模型哈希
with open(path, 'rb') as f:
self.model_hash = hashlib.md5(f.read()).hexdigest()
def set_training_info(self, dataset: str, params: Dict, metrics: Dict):
"""设置训练信息"""
self.training_dataset = dataset
self.training_params = params
self.training_metrics = metrics
def set_evaluation_metrics(self, metrics: Dict, dataset: str = None):
"""设置评估指标"""
self.evaluation_metrics = metrics
self.evaluation_dataset = dataset
def transition_stage(self, new_stage: ModelStage, approved_by: str, reason: str = ""):
"""转换阶段"""
old_stage = self.stage
self.stage = new_stage
self.updated_at = datetime.now().isoformat()
self.approval_history.append({
"from_stage": old_stage.value,
"to_stage": new_stage.value,
"approved_by": approved_by,
"reason": reason,
"timestamp": datetime.now().isoformat()
})
def add_deployment(self, deployment_info: Dict):
"""添加部署记录"""
self.deployments.append({
**deployment_info,
"deployed_at": datetime.now().isoformat()
})
self.status = ModelStatus.DEPLOYED
def to_dict(self):
return {
"model_name": self.model_name,
"version": self.version,
"framework": self.framework.value,
"description": self.description,
"tags": self.tags,
"created_at": self.created_at,
"model_path": self.model_path,
"model_size_bytes": self.model_size_bytes,
"model_hash": self.model_hash,
"training_dataset": self.training_dataset,
"training_params": self.training_params,
"training_metrics": self.training_metrics,
"evaluation_metrics": self.evaluation_metrics,
"stage": self.stage.value,
"status": self.status.value,
"deployment_count": len(self.deployments),
"approval_count": len(self.approval_history)
}
class ModelRegistry:
"""模型注册中心"""
def __init__(self, registry_path: str = "./model_registry"):
self.registry_path = Path(registry_path)
self.registry_path.mkdir(parents=True, exist_ok=True)
self.models: Dict[str, Dict[str, ModelVersion]] = {}
def register_model(self, model_version: ModelVersion) -> str:
"""注册模型"""
model_name = model_version.model_name
version = model_version.version
if model_name not in self.models:
self.models[model_name] = {}
if version in self.models[model_name]:
raise ValueError(f"Model {model_name} version {version} already exists")
self.models[model_name][version] = model_version
# 保存到文件
self._save_model_info(model_version)
return f"{model_name}:{version}"
def get_model(self, model_name: str, version: str = "latest") -> Optional[ModelVersion]:
"""获取模型"""
if model_name not in self.models:
return None
if version == "latest":
versions = sorted(self.models[model_name].keys())
if not versions:
return None
version = versions[-1]
return self.models[model_name].get(version)
def list_models(self, framework: ModelFramework = None,
stage: ModelStage = None) -> List[Dict]:
"""列出模型"""
result = []
for model_name, versions in self.models.items():
for version_name, model_version in versions.items():
if framework and model_version.framework != framework:
continue
if stage and model_version.stage != stage:
continue
result.append(model_version.to_dict())
return result
def list_versions(self, model_name: str) -> List[Dict]:
"""列出模型的所有版本"""
if model_name not in self.models:
return []
return [v.to_dict() for v in self.models[model_name].values()]
def delete_model(self, model_name: str, version: str):
"""删除模型"""
if model_name in self.models and version in self.models[model_name]:
del self.models[model_name][version]
def _save_model_info(self, model_version: ModelVersion):
"""保存模型信息"""
model_dir = self.registry_path / model_version.model_name / model_version.version
model_dir.mkdir(parents=True, exist_ok=True)
with open(model_dir / "meta.json", 'w') as f:
json.dump(model_version.to_dict(), f, indent=2)
class ModelApprovalWorkflow:
"""模型审批工作流"""
def __init__(self, model_registry: ModelRegistry):
self.registry = model_registry
self.pending_approvals: List[Dict] = []
def submit_for_approval(self, model_name: str, version: str,
submitted_by: str, description: str = "") -> Dict:
"""提交模型审批"""
model = self.registry.get_model(model_name, version)
if not model:
raise ValueError(f"Model {model_name}:{version} not found")
approval_request = {
"request_id": f"req_{len(self.pending_approvals) + 1:04d}",
"model_name": model_name,
"version": version,
"submitted_by": submitted_by,
"description": description,
"submitted_at": datetime.now().isoformat(),
"status": "pending",
"reviews": []
}
model.status = ModelStatus.VALIDATING
self.pending_approvals.append(approval_request)
return approval_request
def approve(self, request_id: str, reviewer: str, comments: str = "") -> Dict:
"""批准模型"""
request = self._get_request(request_id)
if not request:
raise ValueError(f"Request {request_id} not found")
request["status"] = "approved"
request["reviews"].append({
"reviewer": reviewer,
"action": "approve",
"comments": comments,
"timestamp": datetime.now().isoformat()
})
# 更新模型状态
model = self.registry.get_model(request["model_name"], request["version"])
if model:
model.status = ModelStatus.APPROVED
model.transition_stage(ModelStage.STAGING, reviewer, comments)
return request
def reject(self, request_id: str, reviewer: str, reason: str) -> Dict:
"""拒绝模型"""
request = self._get_request(request_id)
if not request:
raise ValueError(f"Request {request_id} not found")
request["status"] = "rejected"
request["reviews"].append({
"reviewer": reviewer,
"action": "reject",
"reason": reason,
"timestamp": datetime.now().isoformat()
})
# 更新模型状态
model = self.registry.get_model(request["model_name"], request["version"])
if model:
model.status = ModelStatus.REJECTED
return request
def _get_request(self, request_id: str) -> Optional[Dict]:
"""获取审批请求"""
for req in self.pending_approvals:
if req["request_id"] == request_id:
return req
return None
def list_pending(self) -> List[Dict]:
"""列出待审批请求"""
return [r for r in self.pending_approvals if r["status"] == "pending"]
4.2 模型版本管理
class ModelVersionManager:
"""模型版本管理器"""
def __init__(self, model_registry: ModelRegistry):
self.registry = model_registry
def create_version_from_run(self, experiment_run: ExperimentRun,
model_name: str, version: str,
framework: ModelFramework) -> ModelVersion:
"""从实验运行创建模型版本"""
model_version = ModelVersion(
model_name=model_name,
version=version,
framework=framework,
description=f"Created from experiment run {experiment_run.run_id}"
)
# 设置训练信息
model_version.set_training_info(
dataset=experiment_run.params.get("dataset", ""),
params=experiment_run.params,
metrics=experiment_run.metrics
)
# 设置评估指标
eval_metrics = {k: v for k, v in experiment_run.metrics.items()
if k.startswith("val_") or k.startswith("test_")}
model_version.set_evaluation_metrics(eval_metrics)
# 查找模型文件
for artifact in experiment_run.artifacts:
if artifact["name"].endswith((".pt", ".pth", ".onnx", ".pkl")):
model_version.set_model_path(artifact["path"])
break
# 注册模型
self.registry.register_model(model_version)
return model_version
def promote_model(self, model_name: str, version: str,
target_stage: ModelStage, approved_by: str) -> ModelVersion:
"""提升模型阶段"""
model = self.registry.get_model(model_name, version)
if not model:
raise ValueError(f"Model {model_name}:{version} not found")
# 验证阶段转换的合法性
valid_transitions = {
ModelStage.DEVELOPMENT: [ModelStage.STAGING],
ModelStage.STAGING: [ModelStage.PRODUCTION, ModelStage.ARCHIVED],
ModelStage.PRODUCTION: [ModelStage.ARCHIVED],
}
if target_stage not in valid_transitions.get(model.stage, []):
raise ValueError(f"Invalid transition from {model.stage} to {target_stage}")
model.transition_stage(target_stage, approved_by)
return model
def compare_versions(self, model_name: str, version1: str, version2: str) -> Dict:
"""比较两个模型版本"""
v1 = self.registry.get_model(model_name, version1)
v2 = self.registry.get_model(model_name, version2)
if not v1 or not v2:
raise ValueError("One or both versions not found")
# 比较训练参数
param_diff = {}
all_params = set(v1.training_params.keys()) | set(v2.training_params.keys())
for param in all_params:
val1 = v1.training_params.get(param)
val2 = v2.training_params.get(param)
if val1 != val2:
param_diff[param] = {"v1": val1, "v2": val2}
# 比较评估指标
metric_diff = {}
all_metrics = set(v1.evaluation_metrics.keys()) | set(v2.evaluation_metrics.keys())
for metric in all_metrics:
val1 = v1.evaluation_metrics.get(metric)
val2 = v2.evaluation_metrics.get(metric)
if val1 is not None and val2 is not None:
metric_diff[metric] = {
"v1": val1,
"v2": val2,
"diff": val2 - val1,
"improvement": ((val2 - val1) / val1 * 100) if val1 != 0 else 0
}
return {
"model_name": model_name,
"version1": version1,
"version2": version2,
"param_diff": param_diff,
"metric_diff": metric_diff,
"stage_v1": v1.stage.value,
"stage_v2": v2.stage.value
}
def rollback(self, model_name: str, target_version: str,
reason: str = "") -> ModelVersion:
"""回滚到指定版本"""
# 获取当前生产版本
current_prod = None
for v in self.registry.list_versions(model_name):
if v["stage"] == "production":
current_prod = v
break
if current_prod:
# 将当前生产版本归档
current_model = self.registry.get_model(model_name, current_prod["version"])
current_model.transition_stage(ModelStage.ARCHIVED, "system", f"Rollback: {reason}")
# 提升目标版本到生产
target_model = self.promote_model(model_name, target_version,
ModelStage.PRODUCTION, "system")
return target_model
第五章：模型服务平台
5.1 API网关
from fastapi import FastAPI, HTTPException, Depends, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
from typing import Dict, List, Any, Optional
import time
import uuid
from collections import defaultdict
import asyncio
app = FastAPI(title="AI模型服务平台")
# CORS配置
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
security = HTTPBearer()
class RateLimiter:
"""速率限制器"""
def __init__(self, requests_per_minute: int = 60):
self.requests_per_minute = requests_per_minute
self.requests: Dict[str, List[float]] = defaultdict(list)
def is_allowed(self, client_id: str) -> bool:
"""检查是否允许请求"""
now = time.time()
minute_ago = now - 60
# 清理过期记录
self.requests[client_id] = [
t for t in self.requests[client_id] if t > minute_ago
]
if len(self.requests[client_id]) >= self.requests_per_minute:
return False
self.requests[client_id].append(now)
return True
rate_limiter = RateLimiter()
class CircuitBreaker:
"""熔断器"""
def __init__(self, failure_threshold: int = 5, timeout: int = 60):
self.failure_threshold = failure_threshold
self.timeout = timeout
self.failures: Dict[str, List[float]] = defaultdict(list)
self.state: Dict[str, str] = defaultdict(lambda: "closed")
self.last_failure_time: Dict[str, float] = {}
def is_available(self, service_id: str) -> bool:
"""检查服务是否可用"""
if self.state[service_id] == "open":
# 检查是否超时
if time.time() - self.last_failure_time.get(service_id, 0) > self.timeout:
self.state[service_id] = "half-open"
return True
return False
return True
def record_success(self, service_id: str):
"""记录成功"""
if self.state[service_id] == "half-open":
self.state[service_id] = "closed"
self.failures[service_id] = []
def record_failure(self, service_id: str):
"""记录失败"""
now = time.time()
self.failures[service_id].append(now)
self.last_failure_time[service_id] = now
# 检查是否需要打开熔断器
recent_failures = [
t for t in self.failures[service_id] if t > now - 60
]
if len(recent_failures) >= self.failure_threshold:
self.state[service_id] = "open"
circuit_breaker = CircuitBreaker()
class ModelRouter:
"""模型路由器"""
def __init__(self):
self.routes: Dict[str, Dict] = {}
self.backends: Dict[str, List[str]] = {}
def add_route(self, model_name: str, backends: List[str],
weights: List[float] = None):
"""添加路由"""
self.routes[model_name] = {
"backends": backends,
"weights": weights or [1.0 / len(backends)] * len(backends),
"strategy": "weighted_round_robin"
}
self.backends[model_name] = backends
def get_backend(self, model_name: str) -> str:
"""获取后端地址"""
if model_name not in self.routes:
raise ValueError(f"No route for model {model_name}")
route = self.routes[model_name]
backends = route["backends"]
weights = route["weights"]
# 过滤可用的后端
available = [
(b, w) for b, w in zip(backends, weights)
if circuit_breaker.is_available(b)
]
if not available:
raise HTTPException(status_code=503, detail="No available backends")
# 加权随机选择
import random
total_weight = sum(w for _, w in available)
r = random.uniform(0, total_weight)
cumulative = 0
for backend, weight in available:
cumulative += weight
if r <= cumulative:
return backend
return available[0][0]
model_router = ModelRouter()
class PredictionRequest(BaseModel):
model_name: str
model_version: str = "latest"
input_data: Dict[str, Any]
request_id: Optional[str] = None
timeout: int = 30
parameters: Optional[Dict[str, Any]] = None
class PredictionResponse(BaseModel):
request_id: str
model_name: str
model_version: str
predictions: Any
latency_ms: float
metadata: Dict[str, Any] = {}
async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
"""验证token"""
# 简化的token验证
token = credentials.credentials
if not token:
raise HTTPException(status_code=401, detail="Invalid token")
return token
@app.post("/v1/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest,
token: str = Depends(verify_token)):
"""模型预测接口"""
start_time = time.time()
# 速率限制
client_id = token[:10] # 简化：使用token前10位作为客户端ID
if not rate_limiter.is_allowed(client_id):
raise HTTPException(status_code=429, detail="Rate limit exceeded")
# 生成请求ID
request_id = request.request_id or str(uuid.uuid4())
# 获取后端
backend = model_router.get_backend(request.model_name)
# 熔断检查
if not circuit_breaker.is_available(backend):
raise HTTPException(status_code=503, detail="Service unavailable")
try:
# 调用后端服务（模拟）
predictions = await call_model_backend(backend, request)
# 记录成功
circuit_breaker.record_success(backend)
latency = (time.time() - start_time) * 1000
return PredictionResponse(
request_id=request_id,
model_name=request.model_name,
model_version=request.model_version,
predictions=predictions,
latency_ms=round(latency, 2),
metadata={
"backend": backend,
"timestamp": datetime.now().isoformat()
}
)
except Exception as e:
# 记录失败
circuit_breaker.record_failure(backend)
raise HTTPException(status_code=500, detail=str(e))
async def call_model_backend(backend: str, request: PredictionRequest) -> Any:
"""调用模型后端"""
# 模拟后端调用
await asyncio.sleep(0.1)
return {"result": "prediction", "confidence": 0.95}
@app.get("/v1/models")
async def list_models():
"""列出可用模型"""
return {"models": list(model_router.routes.keys())}
@app.get("/v1/health")
async def health_check():
"""健康检查"""
return {"status": "healthy", "timestamp": datetime.now().isoformat()}
5.2 灰度发布与A/B测试
class CanaryDeployment:
"""灰度发布"""
def __init__(self):
self.deployments: Dict[str, Dict] = {}
def create_canary(self, model_name: str, stable_version: str,
canary_version: str, canary_percentage: float = 10) -> str:
"""创建灰度发布"""
deployment_id = f"canary_{str(uuid.uuid4())[:8]}"
self.deployments[deployment_id] = {
"deployment_id": deployment_id,
"model_name": model_name,
"stable_version": stable_version,
"canary_version": canary_version,
"canary_percentage": canary_percentage,
"status": "active",
"created_at": datetime.now().isoformat(),
"metrics": {
"stable": {"requests": 0, "errors": 0, "latency_sum": 0},
"canary": {"requests": 0, "errors": 0, "latency_sum": 0}
}
}
return deployment_id
def route_request(self, deployment_id: str) -> str:
"""路由请求"""
deployment = self.deployments.get(deployment_id)
if not deployment:
raise ValueError(f"Deployment {deployment_id} not found")
import random
if random.random() * 100 < deployment["canary_percentage"]:
return deployment["canary_version"]
return deployment["stable_version"]
def record_metric(self, deployment_id: str, version_type: str,
latency: float, is_error: bool = False):
"""记录指标"""
deployment = self.deployments.get(deployment_id)
if not deployment:
return
metrics = deployment["metrics"][version_type]
metrics["requests"] += 1
if is_error:
metrics["errors"] += 1
metrics["latency_sum"] += latency
def get_metrics(self, deployment_id: str) -> Dict:
"""获取指标"""
deployment = self.deployments.get(deployment_id)
if not deployment:
return {}
result = {}
for version_type in ["stable", "canary"]:
metrics = deployment["metrics"][version_type]
result[version_type] = {
"requests": metrics["requests"],
"errors": metrics["errors"],
"error_rate": metrics["errors"] / metrics["requests"] if metrics["requests"] > 0 else 0,
"avg_latency": metrics["latency_sum"] / metrics["requests"] if metrics["requests"] > 0 else 0
}
return result
def promote_canary(self, deployment_id: str):
"""提升金丝雀版本为稳定版本"""
deployment = self.deployments.get(deployment_id)
if not deployment:
raise ValueError(f"Deployment {deployment_id} not found")
deployment["status"] = "promoted"
deployment["promoted_at"] = datetime.now().isoformat()
def rollback_canary(self, deployment_id: str):
"""回滚金丝雀版本"""
deployment = self.deployments.get(deployment_id)
if not deployment:
raise ValueError(f"Deployment {deployment_id} not found")
deployment["status"] = "rolled_back"
deployment["rolled_back_at"] = datetime.now().isoformat()
class ABTestingEngine:
"""A/B测试引擎"""
def __init__(self):
self.experiments: Dict[str, Dict] = {}
def create_experiment(self, experiment_id: str, model_name: str,
variants: List[Dict], traffic_split: List[float],
metrics: List[str]) -> str:
"""创建A/B测试实验"""
self.experiments[experiment_id] = {
"experiment_id": experiment_id,
"model_name": model_name,
"variants": variants,
"traffic_split": traffic_split,
"target_metrics": metrics,
"status": "running",
"created_at": datetime.now().isoformat(),
"results": {v["name"]: {"metrics": defaultdict(list)} for v in variants}
}
return experiment_id
def assign_variant(self, experiment_id: str, user_id: str) -> str:
"""分配变体"""
experiment = self.experiments.get(experiment_id)
if not experiment:
raise ValueError(f"Experiment {experiment_id} not found")
# 基于用户ID的确定性分配
import hashlib
hash_value = int(hashlib.md5(user_id.encode()).hexdigest(), 16)
normalized = (hash_value % 1000) / 1000.0
cumulative = 0
for i, split in enumerate(experiment["traffic_split"]):
cumulative += split
if normalized < cumulative:
return experiment["variants"][i]["name"]
return experiment["variants"][-1]["name"]
def record_result(self, experiment_id: str, variant_name: str,
user_id: str, metrics: Dict[str, float]):
"""记录结果"""
experiment = self.experiments.get(experiment_id)
if not experiment:
return
for metric_name, value in metrics.items():
experiment["results"][variant_name]["metrics"][metric_name].append({
"user_id": user_id,
"value": value,
"timestamp": datetime.now().isoformat()
})
def analyze_results(self, experiment_id: str) -> Dict:
"""分析结果"""
experiment = self.experiments.get(experiment_id)
if not experiment:
return {}
analysis = {
"experiment_id": experiment_id,
"status": experiment["status"],
"variants": {}
}
for variant in experiment["variants"]:
variant_name = variant["name"]
results = experiment["results"][variant_name]
variant_analysis = {"metrics": {}}
for metric_name, values in results["metrics"].items():
if values:
metric_values = [v["value"] for v in values]
variant_analysis["metrics"][metric_name] = {
"count": len(metric_values),
"mean": np.mean(metric_values),
"std": np.std(metric_values),
"min": np.min(metric_values),
"max": np.max(metric_values)
}
analysis["variants"][variant_name] = variant_analysis
return analysis
def get_winner(self, experiment_id: str, primary_metric: str,
higher_is_better: bool = True) -> Optional[str]:
"""获取获胜变体"""
analysis = self.analyze_results(experiment_id)
best_variant = None
best_value = float('-inf') if higher_is_better else float('inf')
for variant_name, variant_data in analysis.get("variants", {}).items():
metric_data = variant_data["metrics"].get(primary_metric)
if metric_data:
value = metric_data["mean"]
if (higher_is_better and value > best_value) or \
(not higher_is_better and value < best_value):
best_value = value
best_variant = variant_name
return best_variant
第六章：MLOps流水线设计
6.1 流水线定义
from abc import ABC, abstractmethod
from typing import Dict, List, Any, Optional
from datetime import datetime
import logging
class PipelineStep(ABC):
"""流水线步骤基类"""
def __init__(self, name: str, description: str = ""):
self.name = name
self.description = description
self.logger = logging.getLogger(name)
self.dependencies: List[str] = []
self.outputs: List[str] = []
@abstractmethod
def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
"""执行步骤"""
pass
def validate(self, context: Dict[str, Any]) -> bool:
"""验证输入"""
return True
class DataValidationStep(PipelineStep):
"""数据验证步骤"""
def __init__(self, validation_rules: Dict):
super().__init__("data_validation", "验证数据质量")
self.rules = validation_rules
def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
data = context.get("data")
if data is None:
raise ValueError("No data provided")
results = {}
for rule_name, rule_func in self.rules.items():
results[rule_name] = rule_func(data)
context["validation_results"] = results
context["data_valid"] = all(results.values())
return context
class FeatureEngineeringStep(PipelineStep):
"""特征工程步骤"""
def __init__(self, feature_config: Dict):
super().__init__("feature_engineering", "特征工程")
self.config = feature_config
def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
data = context.get("data")
# 应用特征转换
features = {}
for feature_name, transform in self.config.items():
features[feature_name] = transform(data)
context["features"] = features
return context
class ModelTrainingStep(PipelineStep):
"""模型训练步骤"""
def __init__(self, model_config: Dict):
super().__init__("model_training", "模型训练")
self.config = model_config
def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
features = context.get("features")
labels = context.get("labels")
# 训练模型（模拟）
model = {"type": self.config.get("model_type"), "trained": True}
context["model"] = model
context["training_metrics"] = {"loss": 0.1, "accuracy": 0.95}
return context
class ModelEvaluationStep(PipelineStep):
"""模型评估步骤"""
def __init__(self, metrics_config: List[str]):
super().__init__("model_evaluation", "模型评估")
self.metrics = metrics_config
def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
model = context.get("model")
test_data = context.get("test_data")
# 计算评估指标（模拟）
evaluation_results = {
"accuracy": 0.95,
"precision": 0.93,
"recall": 0.91,
"f1": 0.92
}
context["evaluation_results"] = evaluation_results
context["model_passed_threshold"] = evaluation_results["accuracy"] > 0.9
return context
class ModelRegistrationStep(PipelineStep):
"""模型注册步骤"""
def __init__(self, registry: ModelRegistry):
super().__init__("model_registration", "模型注册")
self.registry = registry
def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
model = context.get("model")
metrics = context.get("evaluation_results")
if not context.get("model_passed_threshold"):
context["registration_skipped"] = True
return context
# 注册模型
model_version = ModelVersion(
model_name=context.get("model_name", "default_model"),
version=context.get("model_version", "1.0.0"),
framework=ModelFramework.PYTORCH
)
model_version.set_evaluation_metrics(metrics)
self.registry.register_model(model_version)
context["registered_model"] = model_version.to_dict()
return context
class MLOpsPipeline:
"""MLOps流水线"""
def __init__(self, pipeline_id: str, name: str):
self.pipeline_id = pipeline_id
self.name = name
self.steps: List[PipelineStep] = []
self.logger = logging.getLogger(f"pipeline.{pipeline_id}")
self.runs: List[Dict] = []
def add_step(self, step: PipelineStep) -> 'MLOpsPipeline':
"""添加步骤"""
self.steps.append(step)
return self
def execute(self, initial_context: Dict[str, Any] = None) -> Dict[str, Any]:
"""执行流水线"""
context = initial_context or {}
context["pipeline_id"] = self.pipeline_id
context["started_at"] = datetime.now().isoformat()
run_id = f"run_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
run_log = {
"run_id": run_id,
"pipeline_id": self.pipeline_id,
"started_at": context["started_at"],
"steps": [],
"status": "running"
}
self.logger.info(f"Starting pipeline {self.name}")
for i, step in enumerate(self.steps):
step_start = datetime.now()
try:
# 验证
if not step.validate(context):
raise ValueError(f"Validation failed for step {step.name}")
# 执行
context = step.execute(context)
step_duration = (datetime.now() - step_start).total_seconds()
run_log["steps"].append({
"step": step.name,
"status": "success",
"duration_seconds": step_duration
})
self.logger.info(f"Step {step.name} completed in {step_duration:.2f}s")
except Exception as e:
step_duration = (datetime.now() - step_start).total_seconds()
run_log["steps"].append({
"step": step.name,
"status": "failed",
"error": str(e),
"duration_seconds": step_duration
})
self.logger.error(f"Step {step.name} failed: {str(e)}")
run_log["status"] = "failed"
break
context["completed_at"] = datetime.now().isoformat()
run_log["completed_at"] = context["completed_at"]
if run_log["status"] == "running":
run_log["status"] = "success"
self.runs.append(run_log)
return context
def get_run_history(self) -> List[Dict]:
"""获取运行历史"""
return self.runs
def get_last_run(self) -> Optional[Dict]:
"""获取最近一次运行"""
return self.runs[-1] if self.runs else None
# 使用示例
pipeline = MLOpsPipeline("sentiment_analysis", "情感分析流水线")
pipeline.add_step(DataValidationStep({
"not_null": lambda d: len(d) > 0,
"schema": lambda d: "text" in d[0] if d else False
}))
pipeline.add_step(FeatureEngineeringStep({
"text_length": lambda d: [len(item["text"]) for item in d],
"word_count": lambda d: [len(item["text"].split()) for item in d]
}))
pipeline.add_step(ModelTrainingStep({"model_type": "transformer"}))
pipeline.add_step(ModelEvaluationStep(["accuracy", "f1"]))
6.2 流水线调度器
import threading
from typing import Callable
import schedule
class PipelineScheduler:
"""流水线调度器"""
def __init__(self):
self.pipelines: Dict[str, MLOpsPipeline] = {}
self.schedules: Dict[str, Dict] = {}
self.running = False
self.thread = None
def register_pipeline(self, pipeline: MLOpsPipeline,
schedule_config: Dict = None):
"""注册流水线"""
self.pipelines[pipeline.pipeline_id] = pipeline
if schedule_config:
self.schedules[pipeline.pipeline_id] = {
"cron": schedule_config.get("cron"),
"interval_minutes": schedule_config.get("interval_minutes"),
"trigger": schedule_config.get("trigger"),
"last_run": None,
"next_run": None
}
def trigger_pipeline(self, pipeline_id: str, context: Dict = None) -> Dict:
"""手动触发流水线"""
pipeline = self.pipelines.get(pipeline_id)
if not pipeline:
raise ValueError(f"Pipeline {pipeline_id} not found")
return pipeline.execute(context)
def start_scheduler(self):
"""启动调度器"""
self.running = True
self.thread = threading.Thread(target=self._run_scheduler)
self.thread.start()
def stop_scheduler(self):
"""停止调度器"""
self.running = False
if self.thread:
self.thread.join()
def _run_scheduler(self):
"""运行调度器"""
while self.running:
now = datetime.now()
for pipeline_id, schedule_config in self.schedules.items():
# 检查是否到了执行时间
if self._should_run(schedule_config, now):
try:
self.trigger_pipeline(pipeline_id)
schedule_config["last_run"] = now
except Exception as e:
logging.error(f"Scheduled pipeline {pipeline_id} failed: {e}")
import time
time.sleep(60) # 每分钟检查一次
def _should_run(self, schedule_config: Dict, now: datetime) -> bool:
"""判断是否应该运行"""
interval = schedule_config.get("interval_minutes")
if interval:
last_run = schedule_config.get("last_run")
if last_run is None:
return True
elapsed = (now - last_run).total_seconds() / 60
return elapsed >= interval
return False
第七章：监控与可观测性
7.1 模型漂移检测
from scipy import stats
import numpy as np
from typing import Dict, List, Optional
from datetime import datetime, timedelta
class DriftDetector:
"""漂移检测器"""
def __init__(self, reference_data: np.ndarray):
self.reference_data = reference_data
self.reference_stats = self._compute_stats(reference_data)
self.drift_history: List[Dict] = []
def _compute_stats(self, data: np.ndarray) -> Dict:
"""计算统计信息"""
return {
"mean": np.mean(data, axis=0) if data.ndim > 1 else np.mean(data),
"std": np.std(data, axis=0) if data.ndim > 1 else np.std(data),
"median": np.median(data, axis=0) if data.ndim > 1 else np.median(data),
"q25": np.percentile(data, 25, axis=0) if data.ndim > 1 else np.percentile(data, 25),
"q75": np.percentile(data, 75, axis=0) if data.ndim > 1 else np.percentile(data, 75)
}
def detect_drift(self, current_data: np.ndarray, method: str = "ks_test",
threshold: float = 0.05) -> Dict:
"""检测漂移"""
current_stats = self._compute_stats(current_data)
if method == "ks_test":
return self._ks_test(current_data, threshold)
elif method == "psi":
return self._psi_test(current_data, threshold)
elif method == "wasserstein":
return self._wasserstein_test(current_data, threshold)
else:
raise ValueError(f"Unknown method: {method}")
def _ks_test(self, current_data: np.ndarray, threshold: float) -> Dict:
"""KS检验"""
if self.reference_data.ndim > 1:
# 对每个特征分别检验
results = []
for i in range(self.reference_data.shape[1]):
stat, p_value = stats.ks_2samp(
self.reference_data[:, i], current_data[:, i]
)
results.append({
"feature": i,
"statistic": stat,
"p_value": p_value,
"drift_detected": p_value < threshold
})
drift_detected = any(r["drift_detected"] for r in results)
else:
stat, p_value = stats.ks_2samp(self.reference_data, current_data)
results = [{
"feature": 0,
"statistic": stat,
"p_value": p_value,
"drift_detected": p_value < threshold
}]
drift_detected = p_value < threshold
result = {
"method": "ks_test",
"drift_detected": drift_detected,
"details": results,
"timestamp": datetime.now().isoformat()
}
self.drift_history.append(result)
return result
def _psi_test(self, current_data: np.ndarray, threshold: float) -> Dict:
"""PSI检验（Population Stability Index）"""
# 将数据分桶
n_bins = 10
if self.reference_data.ndim > 1:
psi_values = []
for i in range(self.reference_data.shape[1]):
psi = self._calculate_psi(
self.reference_data[:, i], current_data[:, i], n_bins
)
psi_values.append(psi)
avg_psi = np.mean(psi_values)
drift_detected = avg_psi > threshold
else:
avg_psi = self._calculate_psi(self.reference_data, current_data, n_bins)
drift_detected = avg_psi > threshold
result = {
"method": "psi",
"drift_detected": drift_detected,
"psi_value": avg_psi,
"threshold": threshold,
"timestamp": datetime.now().isoformat()
}
self.drift_history.append(result)
return result
def _calculate_psi(self, expected: np.ndarray, actual: np.ndarray,
n_bins: int) -> float:
"""计算PSI"""
# 使用参考数据的分位数作为分桶边界
breakpoints = np.percentile(expected, np.linspace(0, 100, n_bins + 1))
breakpoints[0] = -np.inf
breakpoints[-1] = np.inf
expected_counts = np.histogram(expected, breakpoints)[0]
actual_counts = np.histogram(actual, breakpoints)[0]
# 避免除以0
expected_pct = (expected_counts + 1) / (len(expected) + n_bins)
actual_pct = (actual_counts + 1) / (len(actual) + n_bins)
psi = np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct))
return psi
def _wasserstein_test(self, current_data: np.ndarray, threshold: float) -> Dict:
"""Wasserstein距离检验"""
if self.reference_data.ndim > 1:
distances = []
for i in range(self.reference_data.shape[1]):
dist = stats.wasserstein_distance(
self.reference_data[:, i], current_data[:, i]
)
distances.append(dist)
avg_distance = np.mean(distances)
else:
avg_distance = stats.wasserstein_distance(
self.reference_data, current_data
)
drift_detected = avg_distance > threshold
result = {
"method": "wasserstein",
"drift_detected": drift_detected,
"distance": avg_distance,
"threshold": threshold,
"timestamp": datetime.now().isoformat()
}
self.drift_history.append(result)
return result
def get_drift_report(self) -> Dict:
"""获取漂移报告"""
if not self.drift_history:
return {"message": "No drift checks performed"}
recent_checks = self.drift_history[-100:] # 最近100次检查
drift_count = sum(1 for d in recent_checks if d["drift_detected"])
return {
"total_checks": len(self.drift_history),
"recent_checks": len(recent_checks),
"recent_drift_count": drift_count,
"drift_rate": drift_count / len(recent_checks) if recent_checks else 0,
"last_check": self.drift_history[-1],
"drift_history": recent_checks[-10:] # 最近10次
}
class ModelMonitor:
"""模型监控器"""
def __init__(self, model_name: str):
self.model_name = model_name
self.predictions: List[Dict] = []
self.performance_metrics: List[Dict] = []
self.drift_detectors: Dict[str, DriftDetector] = {}
def log_prediction(self, input_data: Dict, prediction: Any,
actual: Any = None, latency_ms: float = 0):
"""记录预测"""
self.predictions.append({
"input": input_data,
"prediction": prediction,
"actual": actual,
"latency_ms": latency_ms,
"timestamp": datetime.now().isoformat()
})
def calculate_performance_metrics(self, window_minutes: int = 60) -> Dict:
"""计算性能指标"""
cutoff = datetime.now() - timedelta(minutes=window_minutes)
recent_predictions = [
p for p in self.predictions
if p["timestamp"] >= cutoff.isoformat()
]
if not recent_predictions:
return {"message": "No predictions in window"}
# 计算延迟统计
latencies = [p["latency_ms"] for p in recent_predictions]
# 计算准确率（如果有实际值）
with_actual = [p for p in recent_predictions if p["actual"] is not None]
accuracy = None
if with_actual:
correct = sum(1 for p in with_actual if p["prediction"] == p["actual"])
accuracy = correct / len(with_actual)
return {
"window_minutes": window_minutes,
"total_predictions": len(recent_predictions),
"latency": {
"mean": np.mean(latencies),
"p50": np.percentile(latencies, 50),
"p95": np.percentile(latencies, 95),
"p99": np.percentile(latencies, 99),
"max": np.max(latencies)
},
"accuracy": accuracy,
"throughput": len(recent_predictions) / window_minutes * 60 # 每秒
}
def setup_drift_detection(self, feature_name: str, reference_data: np.ndarray):
"""设置漂移检测"""
self.drift_detectors[feature_name] = DriftDetector(reference_data)
def check_drift(self, feature_name: str, current_data: np.ndarray) -> Dict:
"""检查漂移"""
if feature_name not in self.drift_detectors:
raise ValueError(f"No drift detector for feature {feature_name}")
return self.drift_detectors[feature_name].detect_drift(current_data)
def get_monitoring_dashboard(self) -> Dict:
"""获取监控仪表板数据"""
return {
"model_name": self.model_name,
"performance": self.calculate_performance_metrics(),
"drift_status": {
name: detector.get_drift_report()
for name, detector in self.drift_detectors.items()
},
"prediction_count": len(self.predictions),
"last_prediction": self.predictions[-1] if self.predictions else None
}
7.2 告警系统
class AlertSeverity(Enum):
INFO = "info"
WARNING = "warning"
CRITICAL = "critical"
class Alert:
def __init__(self, alert_id: str, severity: AlertSeverity,
title: str, message: str, source: str):
self.alert_id = alert_id
self.severity = severity
self.title = title
self.message = message
self.source = source
self.created_at = datetime.now().isoformat()
self.status = "active"
self.acknowledged_by = None
self.resolved_at = None
class AlertManager:
"""告警管理器"""
def __init__(self):
self.alerts: List[Alert] = []
self.rules: List[Dict] = []
self.handlers: Dict[str, Callable] = {}
def add_rule(self, rule_id: str, condition: Callable,
severity: AlertSeverity, title: str, message: str):
"""添加告警规则"""
self.rules.append({
"rule_id": rule_id,
"condition": condition,
"severity": severity,
"title": title,
"message": message
})
def register_handler(self, severity: AlertSeverity, handler: Callable):
"""注册告警处理器"""
self.handlers[severity.value] = handler
def check_rules(self, context: Dict):
"""检查告警规则"""
for rule in self.rules:
try:
if rule["condition"](context):
self.create_alert(
severity=rule["severity"],
title=rule["title"],
message=rule["message"],
source=rule["rule_id"]
)
except Exception as e:
logging.error(f"Error checking rule {rule['rule_id']}: {e}")
def create_alert(self, severity: AlertSeverity, title: str,
message: str, source: str) -> Alert:
"""创建告警"""
alert_id = f"alert_{len(self.alerts) + 1:06d}"
alert = Alert(alert_id, severity, title, message, source)
self.alerts.append(alert)
# 调用处理器
handler = self.handlers.get(severity.value)
if handler:
try:
handler(alert)
except Exception as e:
logging.error(f"Error handling alert {alert_id}: {e}")
return alert
def acknowledge_alert(self, alert_id: str, user: str):
"""确认告警"""
for alert in self.alerts:
if alert.alert_id == alert_id:
alert.status = "acknowledged"
alert.acknowledged_by = user
break
def resolve_alert(self, alert_id: str):
"""解决告警"""
for alert in self.alerts:
if alert.alert_id == alert_id:
alert.status = "resolved"
alert.resolved_at = datetime.now().isoformat()
break
def get_active_alerts(self) -> List[Dict]:
"""获取活跃告警"""
active = [a for a in self.alerts if a.status == "active"]
return [
{
"alert_id": a.alert_id,
"severity": a.severity.value,
"title": a.title,
"message": a.message,
"source": a.source,
"created_at": a.created_at
}
for a in active
]
def get_alert_statistics(self) -> Dict:
"""获取告警统计"""
from collections import Counter
severity_counts = Counter(a.severity.value for a in self.alerts)
status_counts = Counter(a.status for a in self.alerts)
return {
"total_alerts": len(self.alerts),
"by_severity": dict(severity_counts),
"by_status": dict(status_counts),
"active_count": status_counts.get("active", 0)
}
第八章：成本管理与资源调度
8.1 资源管理
from enum import Enum
from typing import Dict, List, Optional
from datetime import datetime
class ResourceType(Enum):
CPU = "cpu"
GPU = "gpu"
MEMORY = "memory"
STORAGE = "storage"
class ResourcePool:
"""资源池"""
def __init__(self, pool_id: str, name: str):
self.pool_id = pool_id
self.name = name
self.resources: Dict[str, Dict] = {}
self.allocations: Dict[str, Dict] = {}
def add_resource(self, resource_id: str, resource_type: ResourceType,
total_capacity: float, unit: str):
"""添加资源"""
self.resources[resource_id] = {
"resource_id": resource_id,
"type": resource_type,
"total": total_capacity,
"used": 0,
"available": total_capacity,
"unit": unit
}
def allocate(self, resource_id: str, amount: float,
allocation_id: str) -> bool:
"""分配资源"""
resource = self.resources.get(resource_id)
if not resource:
raise ValueError(f"Resource {resource_id} not found")
if resource["available"] < amount:
return False
resource["used"] += amount
resource["available"] -= amount
self.allocations[allocation_id] = {
"resource_id": resource_id,
"amount": amount,
"allocated_at": datetime.now().isoformat()
}
return True
def release(self, allocation_id: str):
"""释放资源"""
allocation = self.allocations.get(allocation_id)
if not allocation:
return
resource = self.resources[allocation["resource_id"]]
resource["used"] -= allocation["amount"]
resource["available"] += allocation["amount"]
del self.allocations[allocation_id]
def get_utilization(self) -> Dict:
"""获取资源利用率"""
utilization = {}
for resource_id, resource in self.resources.items():
utilization[resource_id] = {
"type": resource["type"].value,
"total": resource["total"],
"used": resource["used"],
"available": resource["available"],
"utilization_rate": resource["used"] / resource["total"] if resource["total"] > 0 else 0
}
return utilization
class CostTracker:
"""成本追踪器"""
def __init__(self):
self.cost_records: List[Dict] = []
self.pricing: Dict[str, float] = {
"cpu_per_hour": 0.05,
"gpu_per_hour": 1.50,
"memory_per_gb_hour": 0.01,
"storage_per_gb_month": 0.02
}
def record_usage(self, resource_type: str, amount: float,
duration_hours: float, metadata: Dict = None):
"""记录资源使用"""
cost = self._calculate_cost(resource_type, amount, duration_hours)
record = {
"resource_type": resource_type,
"amount": amount,
"duration_hours": duration_hours,
"cost": cost,
"metadata": metadata or {},
"timestamp": datetime.now().isoformat()
}
self.cost_records.append(record)
return record
def _calculate_cost(self, resource_type: str, amount: float,
duration_hours: float) -> float:
"""计算成本"""
price_key = f"{resource_type}_per_hour"
if resource_type == "storage":
price_key = "storage_per_gb_month"
duration_hours = duration_hours / 24 / 30 # 转换为月
price = self.pricing.get(price_key, 0)
return amount * duration_hours * price
def get_cost_report(self, start_date: str = None,
end_date: str = None) -> Dict:
"""获取成本报告"""
records = self.cost_records
if start_date:
records = [r for r in records if r["timestamp"] >= start_date]
if end_date:
records = [r for r in records if r["timestamp"] <= end_date]
total_cost = sum(r["cost"] for r in records)
# 按资源类型汇总
cost_by_type = {}
for record in records:
rtype = record["resource_type"]
if rtype not in cost_by_type:
cost_by_type[rtype] = 0
cost_by_type[rtype] += record["cost"]
return {
"total_cost": total_cost,
"record_count": len(records),
"cost_by_type": cost_by_type,
"records": records[-100:] # 最近100条
}
class ResourceScheduler:
"""资源调度器"""
def __init__(self, resource_pool: ResourcePool):
self.resource_pool = resource_pool
self.queue: List[Dict] = []
self.running_jobs: Dict[str, Dict] = {}
def submit_job(self, job_id: str, resource_requirements: Dict,
priority: int = 0):
"""提交任务"""
job = {
"job_id": job_id,
"requirements": resource_requirements,
"priority": priority,
"status": "queued",
"submitted_at": datetime.now().isoformat()
}
self.queue.append(job)
self.queue.sort(key=lambda x: -x["priority"]) # 优先级高的排前面
return job
def schedule(self) -> List[str]:
"""调度任务"""
scheduled = []
for job in self.queue[:]:
if self._can_allocate(job["requirements"]):
# 分配资源
allocations = self._allocate_resources(job)
if allocations:
job["status"] = "running"
job["allocations"] = allocations
self.running_jobs[job["job_id"]] = job
self.queue.remove(job)
scheduled.append(job["job_id"])
return scheduled
def _can_allocate(self, requirements: Dict) -> bool:
"""检查是否可以分配资源"""
utilization = self.resource_pool.get_utilization()
for resource_type, amount in requirements.items():
# 查找可用资源
available = sum(
r["available"] for r in utilization.values()
if r["type"] == resource_type
)
if available < amount:
return False
return True
def _allocate_resources(self, job: Dict) -> Dict:
"""分配资源"""
allocations = {}
for resource_type, amount in job["requirements"].items():
# 查找合适的资源
for resource_id, resource in self.resource_pool.resources.items():
if resource["type"].value == resource_type and resource["available"] >= amount:
allocation_id = f"{job['job_id']}_{resource_type}"
if self.resource_pool.allocate(resource_id, amount, allocation_id):
allocations[resource_type] = {
"resource_id": resource_id,
"amount": amount,
"allocation_id": allocation_id
}
break
return allocations
def complete_job(self, job_id: str):
"""完成任务"""
job = self.running_jobs.get(job_id)
if not job:
return
# 释放资源
for resource_type, allocation in job.get("allocations", {}).items():
self.resource_pool.release(allocation["allocation_id"])
job["status"] = "completed"
job["completed_at"] = datetime.now().isoformat()
del self.running_jobs[job_id]
def get_status(self) -> Dict:
"""获取调度状态"""
return {
"queued_jobs": len(self.queue),
"running_jobs": len(self.running_jobs),
"resource_utilization": self.resource_pool.get_utilization()
}
第九章：实战案例——构建企业AI中台
9.1 项目概述
本实战案例将构建一个简化版的企业AI中台，包含以下核心组件：
- 特征平台
- 模型训练平台
- 模型管理平台
- 模型服务平台
- 监控系统
9.2 完整实现
"""
企业级AI中台实战案例
"""
import json
from datetime import datetime
from typing import Dict, List, Any
class EnterpriseAIMiddlePlatform:
"""企业AI中台"""
def __init__(self, config: Dict):
self.config = config
self.logger = logging.getLogger("AIPlatform")
# 初始化各组件
self.feature_store = FeatureStore("main_store", "主特征存储")
self.model_registry = ModelRegistry()
self.experiment_tracker = ExperimentTracker()
self.alert_manager = AlertManager()
self.cost_tracker = CostTracker()
# 资源管理
self.resource_pool = ResourcePool("main_pool", "主资源池")
self._init_resources()
# 设置告警规则
self._setup_alerts()
self.logger.info("AI中台初始化完成")
def _init_resources(self):
"""初始化资源"""
# 添加GPU资源
for i in range(4):
self.resource_pool.add_resource(
f"gpu_{i}", ResourceType.GPU, 16.0, "GB"
)
# 添加CPU资源
self.resource_pool.add_resource(
"cpu_cluster", ResourceType.CPU, 128.0, "cores"
)
# 添加内存资源
self.resource_pool.add_resource(
"memory_pool", ResourceType.MEMORY, 512.0, "GB"
)
def _setup_alerts(self):
"""设置告警规则"""
# GPU利用率过高告警
self.alert_manager.add_rule(
rule_id="high_gpu_usage",
condition=lambda ctx: ctx.get("gpu_utilization", 0) > 0.9,
severity=AlertSeverity.WARNING,
title="GPU利用率过高",
message="GPU利用率超过90%，建议扩容或优化任务调度"
)
# 模型延迟过高告警
self.alert_manager.add_rule(
rule_id="high_latency",
condition=lambda ctx: ctx.get("avg_latency_ms", 0) > 1000,
severity=AlertSeverity.CRITICAL,
title="模型推理延迟过高",
message="模型平均延迟超过1000ms，请检查模型和基础设施"
)
def create_feature_group(self, name: str, entity_type: str,
features: List[Dict]) -> FeatureGroup:
"""创建特征组"""
group = self.feature_store.create_feature_group(
group_id=name.lower().replace(" ", "_"),
name=name,
entity_type=entity_type
)
for feature_config in features:
feature = FeatureDefinition(
feature_id=feature_config["id"],
name=feature_config["name"],
feature_type=FeatureType(feature_config["type"]),
entity_type=entity_type,
description=feature_config.get("description", "")
)
group.add_feature(feature)
self.logger.info(f"创建特征组: {name}")
return group
def train_model(self, model_name: str, training_config: Dict) -> Dict:
"""训练模型"""
# 创建实验
experiment = self.experiment_tracker.create_experiment(
name=f"{model_name}_training",
description=f"Training {model_name}"
)
# 创建运行
run = experiment.create_run()
run.log_params(training_config)
# 模拟训练过程
self.logger.info(f"开始训练模型: {model_name}")
# 记录资源使用
self.cost_tracker.record_usage("gpu", 1.0, 2.0, {"model": model_name})
# 模拟训练结果
metrics = {
"train_loss": 0.15,
"val_loss": 0.18,
"accuracy": 0.92,
"f1_score": 0.90
}
run.log_metrics(metrics)
run.finish("completed")
self.logger.info(f"模型训练完成: {model_name}, 准确率: {metrics['accuracy']}")
return {
"experiment_id": experiment.experiment_id,
"run_id": run.run_id,
"metrics": metrics
}
def register_and_deploy_model(self, model_name: str, version: str,
training_result: Dict) -> Dict:
"""注册并部署模型"""
# 创建模型版本
model_version = ModelVersion(
model_name=model_name,
version=version,
framework=ModelFramework.PYTORCH
)
model_version.set_evaluation_metrics(training_result["metrics"])
# 注册模型
self.model_registry.register_model(model_version)
# 模拟部署
model_version.add_deployment({
"endpoint": f"/v1/models/{model_name}/predict",
"replicas": 2,
"gpu": 1
})
self.logger.info(f"模型已注册并部署: {model_name}:{version}")
return model_version.to_dict()
def predict(self, model_name: str, input_data: Dict) -> Dict:
"""模型预测"""
start_time = datetime.now()
# 获取模型
model = self.model_registry.get_model(model_name)
if not model:
raise ValueError(f"Model {model_name} not found")
# 模拟预测
prediction = {
"result": "positive",
"confidence": 0.85,
"probabilities": {"positive": 0.85, "negative": 0.15}
}
latency = (datetime.now() - start_time).total_seconds() * 1000
# 检查延迟告警
self.alert_manager.check_rules({"avg_latency_ms": latency})
return {
"model_name": model_name,
"model_version": model.version,
"prediction": prediction,
"latency_ms": round(latency, 2)
}
def get_platform_status(self) -> Dict:
"""获取平台状态"""
return {
"features": {
"group_count": len(self.feature_store.feature_groups),
"total_features": sum(
len(g.features) for g in self.feature_store.feature_groups.values()
)
},
"models": {
"total_models": len(self.model_registry.models),
"total_versions": sum(
len(versions) for versions in self.model_registry.models.values()
)
},
"resources": self.resource_pool.get_utilization(),
"costs": self.cost_tracker.get_cost_report(),
"alerts": self.alert_manager.get_alert_statistics(),
"timestamp": datetime.now().isoformat()
}
# 使用示例
if __name__ == "__main__":
config = {
"feature_store_backend": "redis",
"model_registry_path": "./models",
"gpu_count": 4
}
# 初始化AI中台
platform = EnterpriseAIMiddlePlatform(config)
# 创建特征组
user_features = platform.create_feature_group(
name="用户特征",
entity_type="user",
features=[
{"id": "age", "name": "年龄", "type": "numerical"},
{"id": "gender", "name": "性别", "type": "categorical"},
{"id": "income", "name": "收入", "type": "numerical"}
]
)
# 训练模型
training_result = platform.train_model(
model_name="user_churn_model",
training_config={
"learning_rate": 0.001,
"batch_size": 64,
"epochs": 100
}
)
# 注册并部署模型
model_info = platform.register_and_deploy_model(
model_name="user_churn_model",
version="1.0.0",
training_result=training_result
)
# 进行预测
prediction_result = platform.predict(
model_name="user_churn_model",
input_data={"age": 30, "gender": "male", "income": 50000}
)
# 获取平台状态
status = platform.get_platform_status()
print("\n" + "=" * 50)
print("AI中台状态:")
print(json.dumps(status, indent=2, ensure_ascii=False))
print("\n预测结果:")
print(json.dumps(prediction_result, indent=2, ensure_ascii=False))
最佳实践
1. 架构设计原则
- 模块化设计：每个组件独立开发、部署和扩展
- 标准化接口：使用统一的API规范和数据格式
- 松耦合：组件之间通过消息队列或API通信
- 可扩展性：支持水平扩展和垂直扩展
2. 数据管理实践
- 数据质量第一：建立完善的数据质量监控体系
- 特征复用：建立统一的特征平台，避免重复计算
- 数据血缘：追踪数据的来源和转换过程
- 数据安全：实施数据加密、脱敏和访问控制
3. 模型管理实践
- 版本控制：对所有模型进行版本管理
- 实验追踪：记录每次训练的参数、指标和工件
- 审批流程：模型上线前必须经过审批
- 灰度发布：新模型先小流量验证再全量上线
4. 运维监控实践
- 全面监控：监控模型性能、数据漂移、系统资源
- 告警机制：建立完善的告警规则和通知机制
- 日志管理：统一收集和分析日志
- 故障恢复：制定故障恢复预案
5. 成本优化实践
- 资源调度：智能调度GPU/CPU资源
- 弹性伸缩：根据负载自动扩缩容
- 成本分析：定期分析各业务线的AI成本
- 效率优化：优化模型推理效率，降低单次预测成本
常见问题
Q1: 如何评估AI中台的ROI？
A: 评估AI中台的ROI可以从以下维度考虑：
- 效率提升：AI项目交付周期缩短比例
- 成本节约：减少的重复建设和资源浪费
- 质量提升：模型性能和稳定性的改善
- 业务价值：AI能力对业务指标的贡献
Q2: AI中台需要哪些团队？
A: AI中台通常需要以下团队：
- 平台开发团队：负责中台基础设施开发
- 数据工程团队：负责数据管道和特征平台
- 算法团队：负责模型研发和优化
- SRE团队：负责平台运维和稳定性
- 产品经理：负责需求管理和优先级排序
Q3: 如何选择技术栈？
A: 技术栈选择考虑因素：
- 数据层：Spark、Flink、Kafka、Redis、Delta Lake
- 模型层：PyTorch、TensorFlow、MLflow、Kubeflow
- 服务层：FastAPI、Triton、TorchServe、TensorFlow Serving
- 基础设施：Kubernetes、Docker、Prometheus、Grafana
Q4: 如何处理模型版本管理？
A: 模型版本管理最佳实践：
- 语义化版本：使用主版本.次版本.修订号的格式
- 模型注册中心：统一管理所有模型版本
- 元数据记录：记录训练参数、数据集、评估指标
- 工件存储：存储模型文件、配置文件等
Q5: 如何保证AI服务的稳定性？
A: 稳定性保障措施：
- 冗余部署：多副本部署，避免单点故障
- 熔断降级：实现熔断器和降级策略
- 限流控制：实施请求限流和背压机制
- 健康检查：定期检查服务健康状态
- 自动恢复：实现自动重启和故障转移
总结
本教程全面介绍了企业级AI中台的架构设计和实现，包括：
- 整体架构：四层架构模型（数据层、模型层、服务层、应用层）
- 特征平台：特征定义、存储、计算和服务
- 模型训练平台：实验管理、分布式训练、超参数优化
- 模型管理平台：模型注册、版本管理、审批流程
- 模型服务平台：API网关、灰度发布、A/B测试
- MLOps流水线：流水线定义、调度、监控
- 监控与可观测性：漂移检测、性能监控、告警系统
- 成本管理：资源调度、成本追踪、优化策略
- 实战案例：完整的AI中台实现
通过本教程，你应该能够：
- 理解AI中台的核心概念和架构设计
- 掌握各组件的技术实现方案
- 构建一套完整的AI中台系统
- 实施有效的运维监控和成本管理
AI中台是企业AI能力建设的关键基础设施。通过构建完善的AI中台，企业可以实现AI能力的快速迭代、高效复用和规模化落地，从而在AI时代获得竞争优势。
