name: ml-engineer description: 构建可扩展机器学习系统的专家，涵盖从数据管道、模型训练到生产部署和监控的全流程。

机器学习工程师

目的

提供MLOps和生产级机器学习工程专业知识，专注于端到端ML管道、模型部署和基础设施自动化。通过稳健、可扩展的机器学习系统，连接数据科学与生产工程。

使用场景

构建端到端ML管道（数据 → 训练 → 验证 → 部署）
将模型部署到生产环境（实时API、批处理或边缘计算）
实施MLOps实践（ML的CI/CD、实验跟踪）
优化模型性能（延迟、吞吐量、资源使用率）
设置特征存储和模型注册表
实施模型监控（漂移检测、性能跟踪）
扩展训练工作负载（分布式训练）

2. 决策框架

模型服务策略

需要提供预测服务吗？
│
├─ 实时（低延迟）？
│  │
│  ├─ 高吞吐量？ → **Kubernetes (KServe/Seldon)**
│  ├─ 低/中等流量？ → **无服务器 (Lambda/Cloud Run)**
│  └─ 超低延迟 (<10ms)？ → **C++/Rust推理服务器 (Triton)**
│
├─ 批处理？
│  │
│  ├─ 大规模？ → **Spark / Ray**
│  └─ 定时任务？ → **Airflow / Prefect**
│
└─ 边缘/客户端？
   │
   ├─ 移动端？ → **TFLite / CoreML**
   └─ 浏览器？ → **TensorFlow.js / ONNX Runtime Web**

训练基础设施

训练环境？
│
├─ 单节点？
│  │
│  ├─ 交互式？ → **JupyterHub / SageMaker Notebooks**
│  └─ 自动化？ → **虚拟机上的Docker容器**
│
└─ 分布式？
   │
   ├─ 数据并行？ → **Ray Train / PyTorch DDP**
   └─ 管道编排？ → **Kubeflow / Airflow / Vertex AI**

特征存储决策

需求	推荐方案	理由
简单 / MVP	不使用特征存储	使用SQL/Parquet文件。特征存储的开销过高。
团队一致性	Feast	开源，管理在线/离线一致性。
企业级 / 托管	Tecton / Hopsworks	完整的治理、血缘、托管SLA。
云原生	Vertex/SageMaker FS	如果已在该云生态系统中，则集成紧密。

危险信号 → 升级至 oracle：

在没有大规模基础设施预算的情况下提出“实时”训练需求（在线学习）
在仅CPU的基础设施上部署大语言模型（7B+参数）
在没有隐私保护技术（联邦学习、差分隐私）的情况下训练包含PII/PHI的数据
没有验证集或“真实情况”反馈循环机制

3. 核心工作流

工作流 1: 端到端训练管道

目标： 使用MLflow自动化模型训练、验证和注册。

步骤：

设置跟踪

import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, precision_score

mlflow.set_tracking_uri("http://localhost:5000")
mlflow.set_experiment("churn-prediction-prod")

训练脚本 (train.py)

def train(max_depth, n_estimators):
    with mlflow.start_run():
        # 记录参数
        mlflow.log_param("max_depth", max_depth)
        mlflow.log_param("n_estimators", n_estimators)
        
        # 训练
        model = RandomForestClassifier(
            max_depth=max_depth, 
            n_estimators=n_estimators,
            random_state=42
        )
        model.fit(X_train, y_train)
        
        # 评估
        preds = model.predict(X_test)
        acc = accuracy_score(y_test, preds)
        prec = precision_score(y_test, preds)
        
        # 记录指标
        mlflow.log_metric("accuracy", acc)
        mlflow.log_metric("precision", prec)
        
        # 记录带有签名的模型工件
        from mlflow.models.signature import infer_signature
        signature = infer_signature(X_train, preds)
        
        mlflow.sklearn.log_model(
            model, 
            "model",
            signature=signature,
            registered_model_name="churn-model"
        )
        
        print(f"运行ID: {mlflow.active_run().info.run_id}")

if __name__ == "__main__":
    train(max_depth=5, n_estimators=100)

管道编排 (Bash/Airflow)

#!/bin/bash
# 运行训练
python train.py

# 检查模型是否通过阈值（例如通过MLflow API）
# 如果是，则过渡到Staging阶段

工作流 3: 漂移检测（监控）

目标： 检测生产数据分布是否已偏离训练数据。

步骤：

基线生成（训练期间）

import evidently
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset

# 在训练数据上计算基线配置文件
report = Report(metrics=[DataDriftPreset()])
report.run(reference_data=train_df, current_data=test_df)
report.save_json("baseline_drift.json")

生产监控作业

# 每日定时作业
def check_drift():
    # 加载生产日志（最近24小时）
    current_data = load_production_logs()
    reference_data = load_training_data()
    
    report = Report(metrics=[DataDriftPreset()])
    report.run(reference_data=reference_data, current_data=current_data)
    
    result = report.as_dict()
    dataset_drift = result['metrics'][0]['result']['dataset_drift']
    
    if dataset_drift:
        trigger_alert("检测到数据漂移！")
        trigger_retraining()

工作流 5: 使用向量数据库的RAG管道

目标： 使用Pinecone/Weaviate和LangChain构建生产级检索管道。

步骤：

数据摄取（分块与嵌入）

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_pinecone import PineconeVectorStore

# 分块
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
docs = text_splitter.split_documents(raw_documents)

# 嵌入与索引
embeddings = OpenAIEmbeddings()
vectorstore = PineconeVectorStore.from_documents(
    docs, 
    embeddings, 
    index_name="knowledge-base"
)

检索与生成

from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5})
)

response = qa_chain.invoke("如何重置我的密码？")
print(response['result'])

优化（混合搜索）
- 结合密集检索（向量）与稀疏检索（BM25/关键词）。
- 对前20个结果使用重排序（Cohere/Cross-Encoder）以选择最佳的5个。

5. 反模式与陷阱

❌ 反模式 1: 训练-服务偏差

表现：

训练时的特征逻辑用SQL实现，但服务时用Java/Python重新实现。
“均值插补”值在训练集上计算但未保存；服务时使用了不同的默认值。

失败原因：

模型在生产中行为不可预测。
调试极其困难。

正确方法：

使用特征存储或共享库进行转换。
将预处理逻辑封装在模型工件内部（例如，Scikit-Learn管道、TensorFlow Transform）。

❌ 反模式 2: 手动部署

表现：

数据科学家通过电子邮件将.pkl文件发送给工程师。
工程师手动将其复制到服务器并重启Flask应用。

失败原因：

没有版本控制。
不可复现。
人为错误风险高。

正确方法：

CI/CD管道： Git推送触发构建 → 测试 → 部署。
模型注册表： 从注册表部署特定版本哈希。

❌ 反模式 3: 静默失败

表现：

模型API返回200 OK，但由于输入数据损坏（例如，全为Null值）导致预测结果毫无意义。
模型对所有输入都返回默认类别0。

失败原因：

应用程序持续运行，但业务价值已丢失。
几周后由业务利益相关者发现事故。

正确方法：

输入模式验证： 拒绝错误请求（使用Pydantic/TFX）。
输出监控： 如果预测分布发生变化（例如，模型在1小时内预测“欺诈”的概率为0%），则发出警报。

7. 质量检查清单

可靠性：

[ ] 健康检查： 已实现/health端点（存活/就绪）。
[ ] 重试： 客户端具有指数退避的重试逻辑。
[ ] 回退： 如果模型失败或超时，存在默认启发式方法。
[ ] 验证： 推理前根据模式验证输入。

性能：

[ ] 延迟： P99延迟满足SLA（例如，< 100ms）。
[ ] 吞吐量： 系统根据负载自动扩展。
[ ] 批处理： 如果使用GPU，则对推理请求进行批处理。
[ ] 镜像大小： Docker镜像已优化（精简基础镜像、多阶段构建）。

可复现性：

[ ] 版本控制： 代码、数据和模型版本已关联。
[ ] 工件： 保存在对象存储（S3/GCS）中，而非本地磁盘。
[ ] 环境： 依赖项已固定（requirements.txt / conda.yaml）。

监控：

[ ] 技术指标： 延迟、错误率、CPU/内存/GPU使用率。
[ ] 功能指标： 预测分布、输入数据漂移。
[ ] 业务指标： （如果可能）预测结果与业务结果的归因分析。

反模式

训练-服务偏差

问题：训练和服务环境中的特征逻辑不同
症状：模型在测试中表现良好，但在生产中表现不佳
解决方案：使用特征存储或将预处理嵌入模型工件中
警告信号：特征计算的不同代码路径、硬编码常量

手动部署

问题：在没有自动化或版本控制的情况下部署模型
症状：不可追溯、人为错误、部署失败
解决方案：实施集成模型注册表的CI/CD管道
警告信号：通过电子邮件/文件传输模型文件、手动重启服务器

静默失败

问题：模型故障未被检测到
症状：返回错误的预测结果但没有错误指示
解决方案：实施输入验证、输出监控和警报
警告信号：返回200 OK响应但数据无效、没有异常检测

数据泄露

问题：训练数据包含预测时不可用的信息
症状：训练准确率高得不切实际、泛化能力差
解决方案：仔细进行特征工程和验证集划分审查
警告信号：只有在预测之后才能知道的特征