大模型实验管道化：SageMaker Pipelines与MLflow协同实践

笥課鸴煕

235人浏览 · 2026-07-01 11:46:49

笥課鸴煕 · 2026-07-01 11:46:49 发布

1. 项目概述：为什么企业级大模型实验需要“管道化”与“可追溯性”

最近半年，我带的三个客户团队——一家金融风控中台、一家医疗知识图谱初创公司、一个省级政务AI平台——不约而同地卡在同一个地方：不是模型训不出来，而是训出来的模型根本没法比、不敢上线、复现不了。有人用A卡跑出一个87.3%的F1值，换B卡重跑一次掉到84.1%；有人在开发环境微调完模型，部署到生产环境后指标直接崩掉12个百分点；还有人三个月后想回溯某次关键实验的超参组合，发现连当时用的是哪个数据切片都查不到。这些问题背后，暴露的不是算法能力短板，而是工程化底座的严重缺失。

这正是本文要解决的核心问题： 如何让大语言模型的实验过程，从“笔记本里随手敲几行代码”的临时行为，变成可调度、可对比、可审计、可复现的企业级研发流水线 。关键词不是“训练更快”，而是“实验更稳”；不是“参数调得更细”，而是“每次改动都有据可查”。SageMaker Pipelines 和 MLflow 的组合，恰恰是 AWS 生态中目前最成熟、最贴近真实产研节奏的一套解法。它不鼓吹“一键炼丹”，而是把实验拆解成原子化的步骤——数据准备、预处理、PEFT 微调、评估、模型注册——每个步骤独立封装、版本可控、输入输出明确；再通过 MLflow 的 Tracking Server 统一记录所有元数据：谁在什么时间、用哪版代码、哪份数据、哪些超参，跑出了什么指标、生成了什么模型文件。这不是锦上添花的工具链，而是大模型项目从“能跑通”迈向“敢上线”的必经门槛。

我见过太多团队在模型效果上花了90%精力，却在实验管理上只投入10%资源，结果就是：模型迭代越快，技术债越厚；实验次数越多，决策依据越模糊。本文接下来要讲的，不是教你怎么写一个LoRA层，而是告诉你怎么让每一次LoRA微调都成为可归档、可回滚、可横向拉齐的标准化动作。它适合两类人：一是正在搭建AI平台的工程师，你需要知道这套架构如何落地；二是业务侧的算法负责人，你需要理解为什么必须把实验流程“管道化”，否则你的团队永远在重复造轮子、填坑、救火。

2. 整体设计思路：为什么选 SageMaker Pipelines + MLflow 而非其他方案

2.1 架构选型背后的三重现实约束

很多团队第一反应是：“我们已经有Kubeflow了，为什么还要学SageMaker Pipelines？”或者“MLflow和Weights & Biases（W&B）比，是不是功能弱？”这类问题背后，其实是对“技术选型”本质的误解——它从来不是比谁功能多，而是看谁最贴合你当前阶段的 组织约束、运维成本和交付节奏 。我帮客户做技术选型时，会强制问清三个问题：

第一，你的数据主权和网络边界在哪里？
如果核心训练数据必须留在VPC内，且不能出公网，那么W&B这种SaaS服务天然被排除。MLflow Tracking Server 可以完全私有化部署在EKS或EC2上，SageMaker Pipelines 的所有计算节点默认运行在用户VPC内，数据不出域。而Kubeflow虽然也能私有化，但其Argo Workflows的权限模型复杂，IAM策略配置稍有不慎就会导致Pipeline卡在Pending状态，调试成本极高。

第二，你的团队是否具备全栈K8s运维能力？
Kubeflow的强项是灵活性，代价是深度依赖K8s原语（CRD、Operator、RBAC）。我曾协助一个医疗客户迁移Kubeflow Pipeline，光是配置一个能访问S3的Worker Pod，就花了两周时间反复调试ServiceAccount和IRSA（IAM Roles for Service Accounts）的绑定关系。而SageMaker Pipelines 将底层K8s细节全部封装，你只需定义ProcessingJob、TrainingJob、ModelStep等高层抽象，SageMaker自动为你创建、调度、销毁EC2实例，并处理好网络、存储、权限的联动。对一个只有2名MLOps工程师的团队，这是决定性的效率差。

第三，你的实验规模是否需要“跨账户、跨区域”的统一视图？
大型企业常有多个业务线、多个AWS账户。MLflow的Tracking Server支持跨账户ARN引用（ arn:aws:sagemaker:us-east-1:123456789012:mlflow-tracking-server/prod-mlflow ），所有账户的实验日志可汇聚到一个中心化Server；SageMaker Pipelines 的Pipeline Definition本身是JSON格式，可存入CodeCommit并触发跨账户的CI/CD。而W&B的Team Workspace虽支持多项目，但无法与AWS IAM策略深度集成，审计日志难以对接企业SIEM系统。

2.2 SageMaker Pipelines 与 MLflow 的职责切分：谁管“流程”，谁管“痕迹”

很多人混淆两者的定位，以为Pipelines是“执行引擎”，MLflow是“日志系统”，其实远不止于此。它们的协同逻辑，本质上是 将“实验”这个抽象概念，拆解为“可调度的动作”和“可查询的证据”两个正交维度 ：

SageMaker Pipelines 负责“动” ：它定义的是 因果链 ——“当数据集版本更新时，自动触发微调；当微调完成且验证集指标达标，自动触发A/B测试；当A/B测试胜出率>95%，自动将模型推送到生产Endpoint”。它的核心产出物是PipelineExecution，一个带有明确开始/结束时间戳、各Step状态（Succeeded/Failed/Executing）、输入输出Artifact S3路径的不可变对象。你可以把它理解为“实验的DNA序列”，精确到每一个碱基（Step）的执行顺序和依赖关系。
MLflow 负责“静” ：它记录的是快照 ——在Pipeline的某个Step（比如TrainingStep）内部，代码调用 mlflow.log_metric("eval_f1", 0.873) 时，MLflow会将这个数值、当时的 mlflow.get_run().info.run_id 、 mlflow.get_run().data.params （所有超参字典）、甚至 mlflow.log_artifact("model/pytorch_model.bin") 的二进制哈希值，一并写入Backend Store（可以是MySQL或S3）。它的核心产出物是Run，一个带有完整上下文的“实验切片”。你可以把它理解为“实验的高清照片”，清晰展示那一刻的所有变量状态。

二者结合，就形成了“时空双维度”的实验治理：Pipelines告诉你 这件事是怎么发生的（How） ，MLflow告诉你 那一刻具体是什么样子（What） 。比如，当你发现某次PipelineExecution的评估指标异常下跌，你可以：

在SageMaker Console里点开该Execution，定位到失败的EvaluationStep；
查看该Step的CloudWatch Logs，确认是代码报错还是资源超限；
找到该Step关联的MLflow Run ID（通常在Step的OutputParameters里显式传递）；
进入MLflow UI，输入Run ID，查看该次评估所用的具体模型版本、数据切片、随机种子、甚至 git commit hash （如果启用了 mlflow.set_tag("mlflow.source.git.commit", commit_hash) ）。

这种“流程可追踪、痕迹可回溯”的能力，是单靠任何一方都无法提供的。这也是为什么我在所有客户项目中，都坚持将二者作为一对“黄金搭档”来部署，而非二选一。

2.3 为什么放弃Hugging Face Hub作为模型注册中心？

原文提到“Hugging Face Token: Access datasets and models”，这容易引发一个危险的实践误区：把HF Hub当作生产环境的模型注册中心。我必须强调： HF Hub是极佳的模型发现与共享平台，但绝非企业级模型注册中心 。原因有三：

权限粒度太粗 ：HF Hub的Organization级别权限，只能控制“谁能看到Repo”，无法控制“谁能下载模型权重”、“谁能覆盖已有版本”。在金融或医疗场景，模型发布必须经过QA、合规、安全三道人工审批，而HF Hub没有审批工作流（Approval Workflow）机制。
审计能力缺失 ：HF Hub不提供详细的下载日志（Who downloaded which version at when），也无法与企业AD/LDAP集成实现SSO登录审计。而SageMaker Model Registry天然支持CloudTrail事件捕获，每一次 CreateModelPackage 、 UpdateModelPackage 操作都会生成结构化日志，可直接接入Splunk或Datadog。
生命周期管理薄弱 ：HF Hub的版本（tag）是纯字符串，没有状态机（Draft/Approved/Deprecated）。而SageMaker Model Package支持自定义Status字段，你可以定义 Staging 、 Production 、 Deprecated 三种状态，并通过Lambda函数监听 ModelPackageStatusChanged 事件，自动触发下游动作（如Status变为 Production 时，自动更新API Gateway的路由权重）。

因此，在我的架构设计中，HF Hub仅用于 上游模型获取 （ load_pretrained_model("meta-llama/Llama-2-7b-hf") ），而 下游模型注册与发布 ，必须走SageMaker Model Registry。两者分工明确：HF Hub是“模型超市”，Model Registry是“模型银行”。

3. 核心细节解析：从零搭建可复现实验流水线的七步实操

3.1 前置条件检查：五个不可妥协的硬性要求

在敲下第一行代码前，必须确保以下五项基础设置100%到位。我见过太多团队因其中一项疏漏，导致后续数天陷入无意义的Debug。这不是“建议”，而是“红线”。

SageMaker Execution Role 必须包含 AmazonSageMakerFullAccess + 自定义内联策略
仅附加 AmazonSageMakerFullAccess 是不够的。该托管策略默认不包含对S3特定前缀的 PutObject 权限（例如你的训练数据存放在 s3://my-bucket/data/raw/ ），也不包含对ECR的 GetAuthorizationToken 权限（如果你要用自定义Docker镜像）。必须添加内联策略：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:PutObject",
                "s3:ListBucket"
            ],
            "Resource": [
                "arn:aws:s3:::my-bucket/*",
                "arn:aws:s3:::my-bucket"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "ecr:GetAuthorizationToken",
                "ecr:BatchCheckLayerAvailability",
                "ecr:GetDownloadUrlForLayer",
                "ecr:BatchGetImage"
            ],
            "Resource": "*"
        }
    ]
}

提示：策略中的S3 Resource ARN必须精确到你的实际桶名和前缀，不能写 * 。这是为了满足最小权限原则，也是后续Pipeline Step能正确读写数据的前提。

MLflow Tracking Server 必须启用 --backend-store-uri 和 --default-artifact-root
启动命令不能只写 mlflow server --host 0.0.0.0 --port 5000 。必须指定持久化后端：
```
mlflow server \
  --backend-store-uri mysql+pymysql://mlflow:password@mlflow-db.cluster-xxx.us-east-1.rds.amazonaws.com/mlflow \
  --default-artifact-root s3://my-bucket/mlflow-artifacts/ \
  --host 0.0.0.0 \
  --port 5000
```
关键点在于： --backend-store-uri 指向MySQL（推荐RDS Aurora MySQL，保证高可用）， --default-artifact-root 指向S3。如果只用 file:/mlflow ，重启Server后所有实验记录将丢失；如果 artifact-root 指向本地路径，Pipeline中的不同Step（运行在不同EC2实例上）将无法共享模型文件。
Hugging Face Token 必须以Secrets Manager方式注入，而非硬编码
原文示例中 from datasets import load_dataset 看似简单，但若数据集是Private Repo（如 my-org/private-dataset ）， load_dataset() 会尝试读取 ~/.huggingface/token 。在SageMaker Processing Job中，这个路径不存在。正确做法是：
- 将HF Token存入AWS Secrets Manager，命名为 /mlflow/hf-token ；
- 在ProcessingJob的 Environment 参数中，通过 SecretsManager 方式注入：
```
processing_job = ProcessingJob(
    # ... other args
    environment={
        "HF_TOKEN": "secretsmanager:/mlflow/hf-token"
    }
)
```
- 在Processing Job的Python脚本中，直接读取环境变量： os.environ["HF_TOKEN"] 。这样既安全，又避免了Token泄露风险。
SageMaker Studio Domain 必须启用 DefaultUserSettings 中的 SecurityGroups
这是极易被忽略的网络陷阱。SageMaker Studio的Notebook Instance默认运行在VPC的Public Subnet，而你的MLflow Tracking Server很可能部署在Private Subnet。若未显式配置Security Group放行 Ingress 规则（端口5000，源为Studio的Security Group），Notebook将无法连接MLflow Server，报错 ConnectionRefusedError 。解决方案是在Studio Domain的 DefaultUserSettings 中，指定一个已配置好Ingress规则的Security Group。
所有Pipeline Step 的 RoleArn 必须与Pipeline Execution Role 一致
初学者常犯错误：为TrainingStep单独创建一个Role，认为“更安全”。但SageMaker Pipelines要求所有Step必须使用同一个Execution Role。因为Pipeline的Input/Output Artifact传递，依赖于该Role对S3路径的统一读写权限。若Step A用Role-A写入 s3://bucket/output/model/ ，Step B用Role-B去读，必然失败。所以，务必确保所有Step的 role_arn 参数，都指向你在第1步中创建的那个Role。

3.2 数据准备Step：为什么必须用Processing Job而非直接在Training Job里加载？

原文中 load_dataset("HuggingFaceH4/no_robots", split="train") 一行代码，看似简洁，但在企业级Pipeline中，这是严重的反模式。原因在于： 数据加载与模型训练必须解耦 。我强制要求所有客户的数据准备，都通过独立的 ProcessingJob 完成，理由如下：

可复用性 ：同一份原始数据，可能被用于微调、蒸馏、强化学习等多个Pipeline。若把 load_dataset 写死在Training Job里，每次新增任务都要复制粘贴、修改代码，违背DRY原则。而Processing Job的输出是标准S3路径（如 s3://bucket/processed-data/train.parquet ），可被任意Pipeline的任意Step作为Input引用。
可审计性 ：Processing Job的代码（ preprocess.py ）和输入参数（ --max_length 512 , --seed 42 ）会被完整记录在PipelineExecution中。当发现某次训练数据质量异常，你可以精准回溯到是哪个Processing Job的哪个参数导致了截断错误，而不是在千行训练脚本里大海捞针。
资源隔离性 ：数据预处理（如tokenization、shuffling）通常是CPU密集型，而模型训练是GPU密集型。若混在同一Job里，要么GPU空转等CPU，要么CPU被GPU抢占。分离后，Processing Job可选用c5.4xlarge（高CPU），Training Job选用p4d.24xlarge（高GPU），成本优化立竿见影。

实操中， preprocess.py 的核心逻辑应遵循“三不原则”：

不写死路径 ：所有S3路径通过 argparse 传入，如 --input-data-s3-uri 、 --output-data-s3-uri ；
不硬编码超参 ： max_length 、 stride 等通过 --max-length 等参数传入；
不跳过错误 ：对每条样本执行 try...except ，将失败样本写入 error_log.txt 并上传S3，确保数据问题可追溯。

# preprocess.py 示例片段
import argparse
import pandas as pd
from datasets import load_dataset

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--input-data-s3-uri", type=str, required=True)
    parser.add_argument("--output-data-s3-uri", type=str, required=True)
    parser.add_argument("--max-length", type=int, default=512)
    args = parser.parse_args()

    # 加载数据（此处可替换为任何数据源：S3 CSV、Redshift Query、API Pull）
    dataset = load_dataset("HuggingFaceH4/no_robots", split="train")
    
    # 预处理逻辑（示例：过滤过短文本）
    df = pd.DataFrame(dataset)
    df = df[df["text"].str.len() > 100]  # 确保文本长度
    
    # 保存为Parquet（高效、压缩、Schema明确）
    output_path = f"{args.output_data_s3_uri}/train.parquet"
    df.to_parquet(output_path, index=False)

if __name__ == "__main__":
    main()

3.3 PEFT微调Step：Trainer的隐藏参数与LoRA配置的黄金比例

原文中 Trainer(...) 一笔带过，但这恰恰是Pipeline中最易出错、也最需精细调控的环节。我将分享三个实战中总结出的“非文档参数”，它们直接影响训练稳定性与最终效果：

dataloader_num_workers 必须设为0
在SageMaker Training Job的分布式环境中，PyTorch DataLoader的 num_workers>0 会导致 fork 进程与主进程的CUDA上下文冲突，报错 RuntimeError: unable to open shared memory object 。这是SageMaker特有的坑，官方文档并未强调。解决方案是强制设为0，并在 TrainingArguments 中添加：
```
training_args = TrainingArguments(
    # ... other args
    dataloader_num_workers=0,
    # 关键：禁用多进程数据加载
)
```
per_device_train_batch_size 与 gradient_accumulation_steps 的乘积，必须等于目标全局Batch Size
不要被 per_device 迷惑。假设你用1台p4d.24xlarge（8张A100），目标全局Batch Size为128，则：
- per_device_train_batch_size = 128 / 8 = 16
- gradient_accumulation_steps = 1 （此时每步更新1次）
- 若想降低显存占用，可设 per_device=8 ， gradient_accumulation_steps=2 ，效果等价。
实操心得：我通常先用 per_device=16, accumulation=1 快速验证流程，再调优为 per_device=8, accumulation=2 以支持更大模型。切忌盲目增大 accumulation ，它会线性增加训练时间，且可能因梯度噪声累积导致收敛变慢。
LoRA的 r （rank）与 lora_alpha 的比值，应严格保持为 lora_alpha / r = 2
这是Hugging Face PEFT库的隐式约定。 lora_alpha 是缩放因子， r 是低秩矩阵的秩。当 lora_alpha / r != 2 时，LoRA层的实际缩放效果会偏离预期，导致微调不稳定。例如：
- 推荐配置： r=8, lora_alpha=16 （比值2）
- 危险配置： r=8, lora_alpha=32 （比值4，实际缩放翻倍，易过拟合）
- 危险配置： r=16, lora_alpha=16 （比值1，实际缩放减半，学习不足）
在Pipeline中，这些参数必须作为 TrainingStep 的 inputs 显式传入，而非写死在脚本里：
```
training_step = TrainingStep(
    name="FineTuneLLM",
    step_args=estimator.fit(
        inputs={
            "training_dataset": TrainingInput(
                s3_data=processing_step.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri,
                content_type="text/csv"
            )
        },
        # 通过hyperparameters传递LoRA参数
        hyperparameters={
            "r": "8",
            "lora_alpha": "16",
            "lora_dropout": "0.1"
        }
    )
)
```

3.4 评估Step：超越Accuracy的多维指标体系构建

原文仅提及“evaluation”，但企业级评估绝不能只看一个 eval_accuracy 。我为客户设计的标准评估框架，包含四个维度，每个维度对应一个独立的Processing Job：

维度	指标示例	计算方式	为什么重要
基础性能	`eval_loss` , `eval_f1` , `eval_perplexity`	Hugging Face `Trainer.evaluate()` 原生输出	衡量模型在标准任务上的基本能力
鲁棒性	`robustness_score` （对抗样本准确率下降率）	使用TextAttack生成对抗样本，对比原始准确率	检测模型是否“死记硬背”，在真实噪声数据上是否可靠
公平性	`demographic_parity_difference` （不同群体预测率差异）	对数据按敏感属性（如性别、地域）分组，计算预测正例率差异	规避模型偏见，满足合规要求（如欧盟AI Act）
效率	`inference_latency_p95` （95分位响应延迟）	在相同硬件上，对1000个样本批量推理，统计延迟分布	决定能否上线，直接影响用户体验

关键实现技巧： 所有评估指标必须以JSON格式写入S3，并由MLflow自动解析 。在评估脚本 evaluate.py 末尾：

# 计算所有指标，存入字典
metrics = {
    "eval_f1": f1_score,
    "robustness_score": 1 - (adv_acc / clean_acc),
    "demographic_parity_diff": abs(group_a_rate - group_b_rate),
    "inference_latency_p95": np.percentile(latencies, 95)
}

# 写入S3（Pipeline Input/Output标准路径）
with open("/opt/ml/processing/output/metrics.json", "w") as f:
    json.dump(metrics, f)

# 同时记录到MLflow（确保与Training Run关联）
mlflow.log_metrics(metrics)

然后在Pipeline中，通过 MetricsSource 将该JSON文件注册为Step的Output：

evaluation_step = ProcessingStep(
    name="EvaluateModel",
    processor=processor,
    inputs=[
        ProcessingInput(
            source=training_step.properties.ModelArtifacts.ModelPackageName,
            destination="/opt/ml/processing/model"
        ),
        ProcessingInput(
            source=processing_step.properties.ProcessingOutputConfig.Outputs["test"].S3Output.S3Uri,
            destination="/opt/ml/processing/test"
        )
    ],
    outputs=[
        ProcessingOutput(
            output_name="metrics",
            source="/opt/ml/processing/output/",
            destination=f"s3://my-bucket/evaluation/{execution_id}/metrics.json"
        )
    ]
)

这样，当Pipeline执行完毕，你不仅能在SageMaker Console看到 evaluation_step 的输出S3路径，还能在MLflow UI中，点击对应的Run，直接看到所有四维指标的可视化图表。这才是真正可行动的评估结果。

4. 实操过程详解：从Pipeline定义到端到端执行的完整链路

4.1 Pipeline定义：用Python SDK编写声明式流水线

SageMaker Pipelines 的核心是 Pipeline 类，它接受一个 steps 列表，每个Step是一个 Step 子类实例。整个定义过程是纯Python，无需YAML或JSON手写，极大降低了学习门槛。以下是本文案例的完整Pipeline定义（已脱敏，保留核心逻辑）：

from sagemaker.workflow.pipeline import Pipeline
from sagemaker.workflow.steps import ProcessingStep, TrainingStep, CreateModelStep, RegisterModelStep
from sagemaker.workflow.parameters import ParameterString, ParameterInteger
from sagemaker.sklearn.processing import SKLearnProcessor
from sagemaker.huggingface import HuggingFace

# 1. 定义可配置参数（Pipeline可复用的关键）
instance_type = ParameterString(name="InstanceType", default_value="ml.p3.2xlarge")
model_id = ParameterString(name="ModelId", default_value="google/flan-t5-base")
lora_r = ParameterInteger(name="LoRARank", default_value=8)
lora_alpha = ParameterInteger(name="LoRAAlpha", default_value=16)

# 2. 数据预处理Step（使用SKLearnProcessor，轻量高效）
sklearn_processor = SKLearnProcessor(
    framework_version="0.23-1",
    role=role,
    instance_type="ml.m5.xlarge",  # CPU实例，节省成本
    instance_count=1,
    env={"HF_TOKEN": "secretsmanager:/mlflow/hf-token"}  # 注入HF Token
)

processing_step = ProcessingStep(
    name="PreprocessData",
    processor=sklearn_processor,
    inputs=[
        ProcessingInput(
            source="s3://my-bucket/raw-data/",  # 原始数据位置
            destination="/opt/ml/processing/input/"
        )
    ],
    outputs=[
        ProcessingOutput(
            output_name="train",
            source="/opt/ml/processing/output/train/",
            destination=f"s3://my-bucket/processed-data/{execution_id}/train/"
        ),
        ProcessingOutput(
            output_name="test",
            source="/opt/ml/processing/output/test/",
            destination=f"s3://my-bucket/processed-data/{execution_id}/test/"
        )
    ],
    code="preprocess.py"  # 上传的预处理脚本
)

# 3. PEFT微调Step（使用HuggingFace Estimator）
huggingface_estimator = HuggingFace(
    entry_point="train.py",  # 训练脚本
    source_dir="./src",      # 包含train.py和requirements.txt的目录
    instance_type=instance_type,
    instance_count=1,
    base_job_name="llm-finetune",
    role=role,
    transformers_version="4.26",
    pytorch_version="1.13",
    py_version="py39",
    hyperparameters={
        "model_id": model_id,
        "r": lora_r,
        "lora_alpha": lora_alpha,
        "per_device_train_batch_size": "8",
        "gradient_accumulation_steps": "2"
    }
)

training_step = TrainingStep(
    name="FineTuneLLM",
    step_args=huggingface_estimator.fit(
        inputs={
            "training_dataset": TrainingInput(
                s3_data=processing_step.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri,
                content_type="text/csv"
            ),
            "test_dataset": TrainingInput(
                s3_data=processing_step.properties.ProcessingOutputConfig.Outputs["test"].S3Output.S3Uri,
                content_type="text/csv"
            )
        }
    )
)

# 4. 创建模型Step（为后续评估和部署准备）
create_model_step = CreateModelStep(
    name="CreateModel",
    model=Model(
        image_uri=huggingface_estimator.image_uri,
        model_data=training_step.properties.ModelArtifacts.S3ModelArtifacts,
        role=role,
        predictor_cls=HuggingFacePredictor
    ),
    inputs=CreateModelInput(
        instance_type="ml.m5.xlarge"
    )
)

# 5. 评估Step（独立Processing Job）
evaluation_processor = SKLearnProcessor(
    framework_version="0.23-1",
    role=role,
    instance_type="ml.m5.xlarge",
    instance_count=1
)

evaluation_step = ProcessingStep(
    name="EvaluateModel",
    processor=evaluation_processor,
    inputs=[
        ProcessingInput(
            source=training_step.properties.ModelArtifacts.S3ModelArtifacts,
            destination="/opt/ml/processing/model"
        ),
        ProcessingInput(
            source=processing_step.properties.ProcessingOutputConfig.Outputs["test"].S3Output.S3Uri,
            destination="/opt/ml/processing/test"
        )
    ],
    outputs=[
        ProcessingOutput(
            output_name="metrics",
            source="/opt/ml/processing/output/",
            destination=f"s3://my-bucket/evaluation/{execution_id}/metrics.json"
        )
    ],
    code="evaluate.py"
)

# 6. 模型注册Step（进入SageMaker Model Registry）
register_model_step = RegisterModelStep(
    name="RegisterModel",
    estimator=huggingface_estimator,
    model_data=training_step.properties.ModelArtifacts.S3ModelArtifacts,
    content_types=["text/plain"],
    response_types=["text/plain"],
    inference_instances=["ml.m5.xlarge", "ml.c5.2xlarge"],
    transform_instances=["ml.m5.xlarge"],
    model_package_group_name="LLM-FineTune-Package-Group",
    approval_status="PendingManualApproval"  # 强制人工审批
)

# 7. 组装Pipeline
pipeline = Pipeline(
    name="LLM-FineTune-Pipeline",
    parameters=[
        instance_type,
        model_id,
        lora_r,
        lora_alpha
    ],
    steps=[
        processing_step,
        training_step,
        create_model_step,
        evaluation_step,
        register_model_step
    ],
    sagemaker_session=sagemaker_session
)

# 8. 提交Pipeline（生成可执行的JSON定义）
pipeline.upsert(role_arn=role)
print(f"Pipeline ARN: {pipeline.arn}")

这段代码的价值，远不止于“能跑通”。它体现了企业级Pipeline的三大特征：

参数化 ：所有可变因素（实例类型、模型ID、LoRA参数）都定义为 Parameter ，Pipeline可被不同团队、不同场景复用；
模块化 ：每个Step职责单一， PreprocessData 只管数据， FineTuneLLM 只管训练， EvaluateModel 只管评估，便于独立调试和替换；
可审计 ： upsert() 生成的JSON定义，会自动存入S3，每一次变更都有Git Commit记录，满足SOX审计要求。

4.2 Pipeline执行：如何触发、监控与中断

Pipeline定义完成后，执行分为三步：触发、监控、干预。这不是简单的“点一下按钮”，而是一套完整的运营流程。

触发方式 ：

手动触发 ：在SageMaker Studio的Pipeline UI中，点击“Start pipeline execution”，填写参数值（如 ModelId=meta-llama/Llama-2-7b-hf ）；
自动触发（推荐） ：将Pipeline与EventBridge Rule绑定。例如，当S3中 my-bucket/raw-data/ 有新文件上传时，触发Pipeline：
```
{
    "source": ["aws.s3"],
    "detail-type": ["Object Created"],
    "detail": {
        "bucket": {"name": ["my-bucket"]},
        "object": {"key": [{"prefix": "raw-data/"}]}
    }
}
```
EventBridge Rule的Target设为SageMaker Pipeline的 StartPipelineExecution API。这样，数据一就位，实验自动开始，彻底消除人为延迟。

监控要点 ：

Step状态 ：在Pipeline Execution详情页，重点关注各Step的 Status （Succeeded/Failed/Executing）和 Duration 。若某Step长时间处于 Executing ，立即查看其CloudWatch Logs；
资源消耗 ：在CloudWatch中，创建Dashboard监控 SageMaker/ProcessingJob 和 SageMaker/TrainingJob 的 CPUUtilization 、 GPUUtilization 、 DiskReadOps 。若 GPUUtilization 持续低于30%，说明数据加载瓶颈，需检查 dataloader_num_workers 或S3吞吐；
MLflow同步 ：打开MLflow UI，搜索 experiment_name ，确认是否有新的Run生成，且 params 、 metrics 、 artifacts 完整。若Run存在但 artifacts 为空，检查Training Job的 mlflow.log_artifact() 调用是否正确。

中断与重试 ：

安全中断 ：若发现Pipeline执行错误（如数据路径错误），不要直接Cancel Execution。正确做法是：在Step详情页，点击“Stop execution”，这会优雅终止当前Step，保留已生成的中间产物（如预处理后的数据），便于调试；
选择性重试 ：Pipeline Execution支持“Rerun from failed step”。若 EvaluateModel 失败，可右键点击该Step，选择“Rerun”，SageMaker会自动复用前面Step（ PreprocessData 、 FineTuneLLM ）的输出，跳过耗时的重新训练，极大提升调试效率。

4.3 MLflow集成：不只是记录，而是构建实验知识图谱

MLflow的威力，远不止于 log_metric 和 log_param 。在本项目中，我将其升级为“实验知识图谱”的构建引擎，通过四个高级特性，将零散的实验记录，编织成可推理、可搜索的知识网络。

特性一： mlflow.set_tags() 构建多维标签体系
除了默认的 mlflow.user 、 mlflow.source.name ，我强制添加业务标签：

mlflow.set_tags({
    "team": "finance-risk",           # 所属业务团队
    "use_case": "fraud-detection",   # 具体应用场景
    "data_version": "v20240501",     # 数据集版本（来自Processing Job输出）
    "code_commit": "a1b2c3d4"        # Git Commit Hash（自动化注入）
})

这样，在MLflow UI的Search栏，可输入 tags.team = 'finance-risk' and tags.use_case = 'fraud-detection' ，瞬间筛选出所有相关实验，无需翻阅数百个Run。

特性二： mlflow.log_dict() 记录复杂结构化参数
LoRA配置、Tokenizer参数等，不是简单字符串，而是嵌套字典。用 log_dict 可保持结构：

lora_config = {
    "r": 8,
    "alpha": 16,
    "dropout": 0.1,
    "target_modules": ["q_proj", "v_proj"]
}
mlflow.log_dict(lora_config, "lora_config")

在UI中， lora_config 会以可折叠的JSON树形展示，点击即可展开查看每个字段，比平铺的 param.lora_r=8 直观百倍。

特性三： mlflow.log_figure() 可视化评估结果
评估Step生成的混淆矩阵、ROC曲线，不应只存图片