ml-engineer

機械学習エンジニア

目的

エンドツーエンドの ML パイプライン、モデルのデプロイ、インフラストラクチャの自動化を専門とする、MLOps および本番環境向け ML エンジニアリングの専門知識を提供します。堅牢でスケーラブルな機械学習システムを通じて、データサイエンスと本番環境向けエンジニアリングをつなぎます。

使用する場面

• エンドツーエンドの ML パイプライン（データ → トレーニング → 検証 → デプロイ）を構築する
• モデルを本番環境にデプロイする（リアルタイム API、バッチ、またはエッジ）
• MLOps の実践（ML の 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 ファイルを使用します。FS のオーバーヘッドが大きすぎます。
チーム内の一貫性	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
   
   # トレーニングデータのベースラインプロファイルを計算
   report = Report(metrics=[DataDriftPreset()])
   report.run(reference_data=train_df, current_data=test_df)
   report.save_json("baseline_drift.json")
   

2. 本番環境のモニタリングジョブ

   # 毎日実行するスケジュール済みジョブ
   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("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
   
   # チャンク分割
   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"
   )
   

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／クロスエンコーダー）を適用し、最適な5件を選択します。

---

---

5. アンチパターンと注意点

❌ アンチパターン 1：トレーニングとサービングの不整合

具体例：

• トレーニングでは特徴量の処理を SQL で実装し、サービングでは Java／Python で再実装しています。
• 「平均値補完」の値をトレーニングセットで計算しても保存せず、サービングでは別のデフォルト値を使用しています。

問題点：

• 本番環境でモデルの動作が予測不能になります。
• デバッグが非常に困難です。

推奨される方法：

• 変換にはフィーチャーストアまたは共有ライブラリを使用します。
• 前処理ロジックをモデルアーティファクトの内部に含めます（例：Scikit-Learn Pipeline、TensorFlow Transform）。

❌ アンチパターン 2：手動デプロイ

具体例：

• データサイエンティストが .pkl ファイルをエンジニアにメールで送ります。
• エンジニアが手動でサーバーにコピーし、flask アプリを再起動します。

問題点：

• バージョン管理がありません。
• 再現性がありません。
• 人為的ミスのリスクが高くなります。

推奨される方法：

• CI/CD パイプライン： Git へのプッシュをきっかけに、ビルド → テスト → デプロイを実行します。
• モデルレジストリ： レジストリから特定のバージョンハッシュをデプロイします。

❌ アンチパターン 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 の応答、異常検知がない

データリーク

• 問題：トレーニングデータに、予測時点では利用できない情報が含まれています
• 症状：トレーニング精度が非現実的に高く、汎化性能が低くなります
• 解決策：特徴量エンジニアリングを慎重に行い、検証用データの分割方法を見直します
• 警告の兆候：予測後にしか得られない特徴量