📚 课程目录第1章设计AI训练流程的端到端架构

1. 设计AI训练流程的端到端架构

任务目标

设计一套从数据采集到模型部署的端到端AI训练流程架构，明确各阶段的核心组件、数据流转关系与交互接口，为后续案例中的具体实现提供架构基准。

架构总览

本架构采用七阶段流水线设计：数据采集 → 数据清洗 → 特征工程 → 模型训练 → 模型评估 → 模型部署 → 监控反馈。每个阶段通过标准化接口（数据格式、模型文件、评估报告）解耦，支持独立迭代与替换。

核心组件与交互定义

阶段	核心组件	输入/输出	关键交互
① 数据采集	采集器（Kafka/Flume）、调度器（Airflow）	输入：业务源；输出：原始数据（RawData）	向数据湖写入原始数据
② 数据清洗	清洗任务（Spark/Pandas）、质量校验器	输入：RawData；输出：CleanData	从数据湖读取，向仓库写入
③ 特征工程	特征处理器、特征存储（Feast）	输入：CleanData；输出：FeatureSet	特征与训练/推理共用
④ 模型训练	训练框架（PyTorch/TF）、实验跟踪（MLflow）	输入：FeatureSet；输出：ModelArtifact	读取特征，注册模型
⑤ 模型评估	评估流水线、指标计算器	输入：ModelArtifact+测试集；输出：EvalReport	从注册中心拉取模型
⑥ 模型部署	模型服务（Triton/Seldon）、发布系统	输入：EvalReport通过的模型；输出：在线服务	拉取模型制品，部署至生产
⑦ 监控反馈	监控器（Prometheus）、漂移检测器	输入：线上日志；输出：告警/触发信号	检测到漂移后触发①重新采集

架构设计关键代码（最小可运行验证）

以下代码定义各阶段数据契约与流水线编排逻辑，可直接运行验证架构的完整链路。

# pipeline_architecture.py
"""端到端AI训练流程架构的最小可运行定义与验证。"""
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
from enum import Enum
import json

class StageStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    SUCCEEDED = "succeeded"
    FAILED = "failed"

@dataclass
class DataContract:
    """阶段间传递的统一数据契约。"""
    schema_version: str = "1.0"
    format: str = "parquet"
    metadata: Dict[str, Any] = field(default_factory=dict)

@dataclass
class RawData(DataContract):
    """阶段①输出：原始数据。"""
    source: str = ""
    record_count: int = 0

@dataclass
class CleanData(DataContract):
    """阶段②输出：清洗后数据。"""
    dirty_count: int = 0
    clean_count: int = 0

@dataclass
class FeatureSet(DataContract):
    """阶段③输出：特征集。"""
    feature_names: List[str] = field(default_factory=list)
    feature_count: int = 0

@dataclass
class ModelArtifact(DataContract):
    """阶段④输出：模型制品。"""
    model_id: str = ""
    model_path: str = ""
    framework: str = ""
    version: str = ""

@dataclass
class EvalReport(DataContract):
    """阶段⑤输出：评估报告。"""
    model_id: str = ""
    metrics: Dict[str, float] = field(default_factory=dict)
    passed: bool = False

class PipelineStage:
    """流水线阶段基类。"""
    def __init__(self, name: str):
        self.name = name
        self.status = StageStatus.PENDING

    def execute(self, input_data: Any) -> Any:
        raise NotImplementedError

class DataCollectionStage(PipelineStage):
    """① 数据采集。"""
    def __init__(self):
        super().__init__("data_collection")

    def execute(self, input_data: Dict[str, Any]) -> RawData:
        self.status = StageStatus.RUNNING
        # 模拟从业务库/API采集
        raw = RawData(
            source=input_data.get("source", "mock_db"),
            record_count=input_data.get("records", 1000),
            metadata={"collected_at": "2025-01-01T00:00:00Z"}
        )
        self.status = StageStatus.SUCCEEDED
        return raw

class DataCleaningStage(PipelineStage):
    """② 数据清洗。"""
    def __init__(self):
        super().__init__("data_cleaning")

    def execute(self, raw: RawData) -> CleanData:
        self.status = StageStatus.RUNNING
        # 模拟清洗：假设10%脏数据
        dirty = int(raw.record_count * 0.1)
        clean = CleanData(
            dirty_count=dirty,
            clean_count=raw.record_count - dirty,
            metadata={"cleaning_rules": ["dedup", "null_fill"]}
        )
        self.status = StageStatus.SUCCEEDED
        return clean

class FeatureEngineeringStage(PipelineStage):
    """③ 特征工程。"""
    def __init__(self):
        super().__init__("feature_engineering")

    def execute(self, clean: CleanData) -> FeatureSet:
        self.status = StageStatus.RUNNING
        # 模拟特征生成
        features = ["feat_a", "feat_b", "feat_c"]
        fs = FeatureSet(
            feature_names=features,
            feature_count=len(features),
            metadata={"clean_count_used": clean.clean_count}
        )
        self.status = StageStatus.SUCCEEDED
        return fs

class ModelTrainingStage(PipelineStage):
    """④ 模型训练。"""
    def __init__(self):
        super().__init__("model_training")

    def execute(self, fs: FeatureSet) -> ModelArtifact:
        self.status = StageStatus.RUNNING
        artifact = ModelArtifact(
            model_id="model_001",
            model_path="/artifacts/model_001.pkl",
            framework="sklearn",
            version="1.0.0",
            metadata={"feature_count_used": fs.feature_count}
        )
        self.status = StageStatus.SUCCEEDED
        return artifact

class ModelEvaluationStage(PipelineStage):
    """⑤ 模型评估。"""
    def __init__(self):
        super().__init__("model_evaluation")

    def execute(self, artifact: ModelArtifact) -> EvalReport:
        self.status = StageStatus.RUNNING
        # 模拟评估：假设准确率0.85，阈值0.8
        report = EvalReport(
            model_id=artifact.model_id,
            metrics={"accuracy": 0.85, "f1": 0.82},
            passed=True,
            metadata={"threshold": 0.8}
        )
        self.status = StageStatus.SUCCEEDED
        return report

class ModelDeploymentStage(PipelineStage):
    """⑥ 模型部署。"""
    def __init__(self):
        super().__init__("model_deployment")

    def execute(self, report: EvalReport) -> Dict[str, str]:
        self.status = StageStatus.RUNNING
        if not report.passed:
            self.status = StageStatus.FAILED
            raise ValueError(f"模型{report.model_id}未通过评估，拒绝部署")
        deploy_info = {
            "model_id": report.model_id,
            "endpoint": "http://serving.example.com/v1/models/model_001",
            "status": "deployed"
        }
        self.status = StageStatus.SUCCEEDED
        return deploy_info

class MonitoringStage(PipelineStage):
    """⑦ 监控反馈。"""
    def __init__(self):
        super().__init__("monitoring")

    def execute(self, deploy_info: Dict[str, str]) -> Dict[str, str]:
        self.status = StageStatus.RUNNING
        # 模拟监控：检测数据漂移
        drift_detected = False  # 实际中基于统计检验
        result = {
            "model_id": deploy_info["model_id"],
            "drift_detected": drift_detected,
            "action": "none" if not drift_detected else "retrain"
        }
        self.status = StageStatus.SUCCEEDED
        return result

class TrainingPipeline:
    """端到端流水线编排器。"""
    def __init__(self):
        self.stages = [
            DataCollectionStage(),
            DataCleaningStage(),
            FeatureEngineeringStage(),
            ModelTrainingStage(),
            ModelEvaluationStage(),
            ModelDeploymentStage(),
            MonitoringStage()
        ]

    def run(self, initial_input: Dict[str, Any]) -> Dict[str, Any]:
        """执行完整流水线，返回各阶段产物。"""
        results = {}
        current_data: Any = initial_input

        for stage in self.stages:
            print(f"[{stage.name}] 执行中...")
            current_data = stage.execute(current_data)
            results[stage.name] = current_data
            print(f"  → 状态: {stage.status.value}")

        return results

def main():
    """最小验证入口。"""
    pipeline = TrainingPipeline()
    init_data = {"source": "user_behavior_log", "records": 5000}

    print("=" * 50)
    print("启动端到端AI训练流水线")
    print("=" * 50)

    results = pipeline.run(init_data)

    print("\n" + "=" * 50)
    print("流水线执行完成，各阶段产物摘要：")
    print("=" * 50)

    for stage_name, artifact in results.items():
        if isinstance(artifact, DataContract):
            meta = artifact.metadata
            print(f"  {stage_name}: {type(artifact).__name__} | 元数据: {json.dumps(meta, ensure_ascii=False)}")
        else:
            print(f"  {stage_name}: {artifact}")

    # 验证部署成功
    deploy_info = results["model_deployment"]
    assert deploy_info["status"] == "deployed", "模型部署失败"
    print("\n✓ 验证通过：模型已成功部署，端到端架构闭环成立")

if __name__ == "__main__":
    main()

运行与验证

运行方式：将上述代码保存为 pipeline_architecture.py，执行 python pipeline_architecture.py（Python 3.8+，无需第三方依赖）。
预期输出：控制台依次打印7个阶段的执行状态，最终输出各阶段产物摘要，并以 ✓ 验证通过 结束。
通过条件：程序无异常退出，且最后一行断言通过（模型部署状态为 deployed），即证明端到端架构中数据从采集到部署的完整链路可正常流转。

架构设计要点

数据契约统一：所有阶段间传递对象继承自 DataContract，保证schema版本与格式一致，便于替换任意阶段实现。
状态可观测：每个阶段维护 StageStatus，便于接入调度系统（如Airflow）进行任务编排与重试。
反馈闭环：监控阶段输出 drift_detected 标志，实际系统中该信号会触发数据采集阶段重新执行，形成持续训练闭环。
扩展性：新增阶段只需继承 PipelineStage 并实现 execute，在 TrainingPipeline.stages 列表中注册即可。