ml-engineer

机器学习工程师

用途

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

适用场景

• 构建端到端机器学习流水线（数据 → 训练 → 验证 → 部署）
• 将模型部署到生产环境（实时 API、批处理或边缘设备）
• 实施 MLOps 实践（机器学习的 CI/CD、实验跟踪）
• 优化模型性能（延迟、吞吐量、资源使用）
• 搭建特征库和模型注册表
• 实施模型监控（漂移检测、性能跟踪）
• 扩展训练工作负载（分布式训练）

---

---

2. 决策框架

模型服务策略

Need to serve predictions?
│
├─ Real-time (Low Latency)?
│  │
│  ├─ High Throughput? → **Kubernetes (KServe/Seldon)**
│  ├─ Low/Medium Traffic? → **Serverless (Lambda/Cloud Run)**
│  └─ Ultra-low latency (<10ms)? → **C++/Rust Inference Server (Triton)**
│
├─ Batch Processing?
│  │
│  ├─ Large Scale? → **Spark / Ray**
│  └─ Scheduled Jobs? → **Airflow / Prefect**
│
└─ Edge / Client-side?
   │
   ├─ Mobile? → **TFLite / CoreML**
   └─ Browser? → **TensorFlow.js / ONNX Runtime Web**

训练基础设施

Training Environment?
│
├─ Single Node?
│  │
│  ├─ Interactive? → **JupyterHub / SageMaker Notebooks**
│  └─ Automated? → **Docker Container on VM**
│
└─ Distributed?
   │
   ├─ Data Parallelism? → **Ray Train / PyTorch DDP**
   └─ Pipeline orchestration? → **Kubeflow / Airflow / Vertex AI**

特征库决策

需求	建议	理由
简单 / MVP	无特征库	使用 SQL/Parquet 文件。特征库的开销过高。
团队一致性	Feast	开源，可管理在线与离线的一致性。
企业级 / 托管	Tecton / Hopsworks	提供完整的治理、血缘追踪和托管 SLA。
云原生	Vertex/SageMaker FS	如果已在相应云生态系统中，则可紧密集成。

危险信号 → 升级给 oracle：

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

---

---

3. 核心工作流

工作流 1：端到端训练流水线

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

步骤：

1. 设置跟踪

   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")
   

2. 训练脚本（train.py）

   def train(max_depth, n_estimators):
       with mlflow.start_run():
           # Log params
           mlflow.log_param("max_depth", max_depth)
           mlflow.log_param("n_estimators", n_estimators)
           
           # Train
           model = RandomForestClassifier(
               max_depth=max_depth, 
               n_estimators=n_estimators,
               random_state=42
           )
           model.fit(X_train, y_train)
           
           # Evaluate
           preds = model.predict(X_test)
           acc = accuracy_score(y_test, preds)
           prec = precision_score(y_test, preds)
           
           # Log metrics
           mlflow.log_metric("accuracy", acc)
           mlflow.log_metric("precision", prec)
           
           # Log model artifact with signature
           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"Run ID: {mlflow.active_run().info.run_id}")
   
   if __name__ == "__main__":
       train(max_depth=5, n_estimators=100)
   

3. 流水线编排（Bash/Airflow）

   #!/bin/bash
   # Run training
   python train.py
   
   # Check if model passed threshold (e.g. via MLflow API)
   # If yes, transition to Staging
   

---

---

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

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

步骤：

1. 基线生成（训练期间）

   import evidently
   from evidently.report import Report
   from evidently.metric_preset import DataDriftPreset
   
   # Calculate baseline profile on training data
   report = Report(metrics=[DataDriftPreset()])
   report.run(reference_data=train_df, current_data=test_df)
   report.save_json("baseline_drift.json")
   

2. 生产监控任务

   # Scheduled daily job
   def check_drift():
       # Load production logs (last 24h)
       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("Data Drift Detected!")
           trigger_retraining()
   

---

---

工作流 5：使用向量数据库构建 RAG 流水线

**目标：**使用 Pinecone/Weaviate 和 LangChain 构建生产级检索流水线。

步骤：

1. 数据摄取（分块与嵌入）

   from langchain.text_splitter import RecursiveCharacterTextSplitter
   from langchain_openai import OpenAIEmbeddings
   from langchain_pinecone import PineconeVectorStore
   
   # Chunking
   text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
   docs = text_splitter.split_documents(raw_documents)
   
   # Embedding & Indexing
   embeddings = OpenAIEmbeddings()
   vectorstore = PineconeVectorStore.from_documents(
       docs, 
       embeddings, 
       index_name="knowledge-base"
   )
   

2. 检索与生成

   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("How do I reset my password?")
   print(response['result'])
   

3. 优化（混合搜索）

   • 将稠密检索（向量）与稀疏检索（BM25/关键词）相结合。
   • 对排名前 20 的结果使用重排序（Cohere/Cross-Encoder），选出最佳的 5 个。

---

---

5. 反模式与注意事项

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

表现：

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

失败原因：

• 模型在生产环境中的行为难以预测。
• 调试极其困难。

正确做法：

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

❌ 反模式 2：手动部署

表现：

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

失败原因：

• 没有版本控制。
• 无法复现。
• 人为错误风险高。

正确做法：

• **CI/CD 流水线：**Git push 触发构建 → 测试 → 部署。
• **模型注册表：**从注册表部署指定版本哈希。

❌ 反模式 3：静默失败

表现形式：

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

失败原因：

• 应用程序仍在运行，但业务价值已流失。
• 数周后，业务相关方才发现问题。

正确做法：

• 输入模式验证： 拒绝无效请求（Pydantic/TFX）。
• 输出监控： 如果预测分布发生变化，则发出警报（例如，模型连续 1 小时预测“Fraud”的比例为 0%）。

---

---

7. 质量检查清单

可靠性：

• 健康检查： 已实现 /health 端点（存活状态/就绪状态）。
• 重试： 客户端具备指数退避重试逻辑。
• 回退： 模型失败或超时时，有默认启发式方案可用。
• 验证： 推理前根据模式验证输入。

性能：

• 延迟： P99 延迟符合 SLA（例如，< 100ms）。
• 吞吐量： 系统会根据负载自动扩缩容。
• 批处理： 使用 GPU 时，对推理请求进行批处理。
• 镜像大小： 优化 Docker 镜像（精简基础镜像、多阶段构建）。

可复现性：

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

监控：

• 技术指标： 延迟、错误率、CPU/内存/GPU 使用情况。
• 功能指标： 预测分布、输入数据漂移。
• 业务指标： （如可行）将预测与结果相关联。

反模式

训练与服务偏差

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

手动部署

• 问题：部署模型时没有自动化或版本控制
• 症状：无法追溯、出现人为错误、部署失败
• 解决方案：实施集成模型注册表的 CI/CD 流水线
• 警示信号：通过电子邮件/文件传输模型文件、手动重启服务器

静默失败

• 问题：模型故障未被检测到
• 症状：返回错误预测，却没有错误提示
• 解决方案：实施输入验证、输出监控和警报机制
• 警示信号：返回 200 OK，但数据毫无意义；没有异常检测

数据泄漏

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