目次 › 第3章 › 3-8

第3章 アプリケーション開発 / 想定学習時間:30〜40分 / 最終確認:2026年8月

3-8. MLflowでのチェーン/エージェント管理(pyfunc・ResponsesAgent)

1. なぜチェーンをMLflowモデルにするのか

3-1〜3-7 で構築してきたチェーン(RAGパイプラインやガードレール付きの処理列)は、このままではノートブック上のPythonコードにすぎません。これを本番にデプロイし、バージョン管理し、チームに配布するためには、標準化されたパッケージ形式に変換する必要があります。それが MLflowモデルとしてのログ(logging)です。

MLflowモデルとしてログすると、次が手に入ります。

• 再現性 — コード・依存ライブラリ・設定が1つのアーティファクトにまとまる
• バージョン管理とガバナンス — Unity Catalog にモデルとして登録できる(4-1)
• デプロイ可能性 — Model Serving のエンドポイントとしてそのままデプロイできる(4-2)
• 統一インターフェース — どんな中身でも predict() という共通の呼び出し方になる

つまりこの節は、「開発したチェーン」を「デプロイ可能な資産」に変える、第3章と第4章をつなぐ結節点です。なお、現行の MLflow 3 は GenAI 向けに Tracing・評価(mlflow.genai.evaluate)・プロンプトレジストリを統合した世代で、チェーン/エージェントのログから評価・運用までを一貫して支える基盤になっています。

2. 実装インターフェースの選択:PythonModel と ResponsesAgent

チェーン/エージェントをMLflowモデルにするときの実装インターフェースは、主に2つから選びます。

従来の PythonModel ラップや ChatAgent インターフェースで作られたモデルも引き続き動作しますが、新規のエージェント/チェーン開発では ResponsesAgent が第一候補です。入出力が業界標準の形式に揃うため、クライアント・評価ツール・トレーシングとの相互運用性が高くなります。

2-1. mlflow.pyfunc.PythonModel によるラップ(汎用)

最も汎用的な方法が、mlflow.pyfunc.PythonModel を継承したクラスを作り、predict() メソッドにチェーンの呼び出しを実装する方法です。pyfunc(Python function)フレーバーは「Pythonで書けるものなら何でもモデルにできる」万能の入れ物で、独自のガードレールや複数ステージを含む複雑なチェーンもラップできます。

import mlflow
import mlflow.pyfunc

class RAGChainModel(mlflow.pyfunc.PythonModel):
    def load_context(self, context):
        # エンドポイント名などの初期化(サービング起動時に1回実行される)
        from databricks_langchain import ChatDatabricks
        self.llm = ChatDatabricks(
            endpoint="databricks-meta-llama-3-3-70b-instruct"
        )

    def predict(self, context, model_input):
        # model_input から質問を取り出してチェーンを実行
        question = model_input["question"][0]
        docs = self.retrieve(question)      # 検索ステージ
        answer = self.generate(question, docs)  # 生成ステージ
        return [answer]

with mlflow.start_run():
    mlflow.pyfunc.log_model(
        name="rag_chain",
        python_model=RAGChainModel(),
        pip_requirements=["mlflow", "databricks-langchain"],
        input_example={"question": ["RAGとは何ですか?"]},
    )

2-2. ResponsesAgent による実装(エージェントの推奨)

対話型のエージェント/チェーンを作るなら、ResponsesAgent を継承します。predict() が受け取るリクエストと返すレスポンスが OpenAI Responses API 互換の形式に統一されている点が特徴です。

from mlflow.pyfunc import ResponsesAgent
from mlflow.types.responses import (
    ResponsesAgentRequest,
    ResponsesAgentResponse,
)

class SimpleRAGAgent(ResponsesAgent):
    def predict(self, request: ResponsesAgentRequest) -> ResponsesAgentResponse:
        # Responses API 互換のリクエストから最新のユーザー入力を取り出す
        question = request.input[-1].content
        docs = self.retrieve(question)          # 検索ステージ
        answer = self.generate(question, docs)  # 生成ステージ
        return ResponsesAgentResponse(
            output=[self.create_text_output_item(text=answer, id="msg-1")]
        )

3. LangChainフレーバーと models from code

インターフェースの選択とは別に、「どうやってログするか」にも選択肢があります。

models from code(コードベースロギング)は、チェーンオブジェクトをシリアライズする代わりに「チェーンを定義するコードファイル」を保存し、ロード時にそのコードを実行してチェーンを再構築します。複雑なGenAIチェーンはオブジェクトの直列化が壊れやすいため、MLflow 3 ではこれが推奨ロギング方式です。ResponsesAgent もこの方式でログするのが基本で、エージェントを定義したファイル内で mlflow.models.set_model(SimpleRAGAgent()) のようにモデルを指定し、log_model にはオブジェクトではなくそのファイルのパスを渡します。

4. モデルシグネチャと input_example

モデルシグネチャ(model signature)は、モデルの入力・出力のスキーマ(名前と型)を定義するメタデータです。GenAIのチェーンでは特に重要で、次の役割を持ちます。

• Model Serving でのバリデーション — エンドポイントに届いたリクエストがスキーマに適合するかの検証に使われる。不正な形式のリクエストを早期に弾ける
• 利用者への契約の明示 — このモデルに何を渡せば何が返るかがカタログ上で明確になる

シグネチャは input_example(入力例)を渡すと自動推論されます。input_example はドキュメントとしても機能し、ロードしたモデルの動作確認にも使えるため、GenAIモデルでは必ず指定する習慣をつけましょう。なお、ResponsesAgent は入出力形式が標準化されているため、シグネチャが自動的に設定されるのも実務上の利点です。

5. 依存関係の記録とデプロイへの流れ

モデルをログする際には、実行に必要なライブラリを pip_requirements(または conda 環境)として記録します。サービング環境はノートブックの環境をそのまま引き継がないため、依存関係の記録漏れはデプロイ後のロードエラーの代表的な原因です。databricks-langchain のようなチェーンが使うライブラリは漏れなく列挙します。

ログしたチェーンのその後の流れは次のとおりです(第4章の予告)。

また、MLflow 3 には MLflow Tracing が統合されており、チェーン実行の各ステップ(検索・LLM呼び出し・パース)の入出力とレイテンシをトレースとして自動記録できます。mlflow.genai.evaluate による評価やプロンプトレジストリと合わせて、デバッグ・品質分析の基盤になるもので、詳しくは 5-6 で扱います。

練習問題