エージェントを Model Serving から Databricks Apps に移行する

既存のエージェントをモデル サービス エンドポイントから Databricks Apps に移行します。

Databricks では、モデル サービスよりも次の利点があるため 、Databricks Apps でエージェントを作成 することをお勧めします。

  • 迅速な反復: ローカル デバッグとログとエージェントの動作に対する完全な透過性を使用して、数秒でエージェント コードとデプロイ構成を反復処理します。
  • Git ベースのバージョン管理と CI/CD: Git を使用してモジュール式 Python エージェント コードをパッケージ化およびバージョン管理し、宣言型オートメーション バンドルを使用してデプロイします。
  • AI コーディング アシスタントのサポート: AI コーディング アシスタントを使用して、エージェントをローカルで開発および移行します。
  • スケーラブルな非同期エージェント: ネイティブの Python 非同期パターンを使用して非同期エージェントを構築し、コンカレンシーを高めます。
  • 柔軟なサーバーのカスタマイズ: 任意のフレームワークまたはスタックを使用し、カスタム ルートとミドルウェアを追加し、LLM エンドポイントとツールに対するユーザーとエージェントの認証を構成します。
  • MLflow トレース: MLflow Git ベースのログ記録モデルとリアルタイム トレースを使用して、エージェントの動作を監視します。
  • 組み込みのチャット UI: 会話エージェント テンプレートには、ストリーミング、認証、永続的な履歴を含むすぐに使用できるチャット インターフェイスが含まれています。

Requirements

移行テンプレートを複製する

移行テンプレートには、Databricks Apps でエージェントを開発およびデプロイするためのスキャフォールディングと、AI コーディング アシスタントに各移行手順の実行方法を教えるエージェント スキル ファイルが用意されています。

テンプレートを複製し、フォルダーに移動します。

git clone https://github.com/databricks/app-templates.git
cd app-templates/agent-migration-from-model-serving

テンプレート フォルダーには次のものが含まれます。

  • AGENTS.md: 移行ワークフローを記述する AI コーディング アシスタントの手順
  • skills/: アシスタントによって順番に実行される各移行ステップのスキル ファイル
  • agent_server/: @invoke() および @stream() ハンドラー用のプレースホルダー コードを含む対象の Databricks Apps エージェントのスキャフォールディング
  • databricks.yml: プレースホルダー リソース宣言を含む宣言型オートメーション バンドル構成テンプレート

AI 支援型の移行は、このテンプレートを使用するための推奨される方法です。 AI コーディング アシスタントは、 AGENTS.md とスキル ファイルを読み取り、コードと構成の変更を自動的に処理します。

  1. Cursor、GitHub Copilot、Claude などの AI コーディング アシスタントでテンプレート フォルダーを開きます。
  2. エンドポイント名を指定して、移行を実行するようにアシスタントに依頼します。
"Migrate my Model Serving endpoint `my-agent-endpoint` to a Databricks App"
  1. アシスタントは移行計画を生成し、各ステップを実行します。

エージェントを Model Serving から Databricks Apps に移行するためのステップ バイ ステップの TODO リストを表示する AI コーディング アシスタントのスクリーンショット。

手動移行

Databricks では、AI コーディング アシスタントを使用して移行を実行することをお勧めします。 AI コーディング アシスタントなしで移行する場合は、次の大まかな手順でプロセスについて説明します。

Important

これらの手順は概要であり、ステートフル エージェント、非同期と同期のトレードオフ、Unity カタログ成果物アクセス、複雑なリソース構成など、すべての移行シナリオについては説明しません。

AI コーディング アシスタントを使用して移行を支援したり、テンプレートの migrate-from-model-serving スキル を参照して詳細を確認したりできます。

ステップ 1. エージェント成果物のダウンロード

  1. エンドポイントからモデル名とバージョンを取得します。
databricks serving-endpoints get <endpoint-name> --output json
  1. 応答で served_entities[0].entity_name (モデル名) と entity_version を見つけて、成果物をダウンロードします。
DATABRICKS_CONFIG_PROFILE=<profile> uv run --no-project \
  --with "mlflow[databricks]>=2.15.0" \
  python3 << 'EOF'
import mlflow
mlflow.set_tracking_uri("databricks")
mlflow.artifacts.download_artifacts(
    artifact_uri="models:/<model-name>/<version>",
    dst_path="./original_mlflow_model"
)
EOF

ダウンロードしたフォルダーには次のものが含まれます。

  • MLmodel — 元のエージェントのリソース宣言
  • code/ — エージェントの Python ソース ファイル
  • artifacts/ — オプションの構成ファイルとプロンプト
  • input_example.json — テストのサンプル要求

ステップ 2. エージェント コードを移行する

code/ から agent_server/ にすべてのPython ファイルをコピーし、artifacts/ から agent_server/artifacts/ に成果物をコピーします。

ファイルを移動した後、新しいフォルダー構造を反映するように、相対インポートとハードコーディングされたファイル パスを更新します。 次に、手順 3 で示したパターンを使用するように agent_server/agent.py 書き直します。

ステップ 3. エージェント コードの変換

モデル サービスでは、エージェントはクラスベースの ResponsesAgentpredict() メソッドと predict_stream() メソッドを使用します。 Databricks Apps では、MLflow AgentServer は、 @invoke()@stream()で装飾されたモジュール レベルの関数を提供します。

移行する場合は、次のいずれかのパターンを選択します。

  • Async (推奨): 複数の要求を同時に処理するために、Python async defawait を使用します。 1 つの要求が LLM 応答を待機している間、サーバーは他の要求を処理します。
  • Sync: Model Serving エージェントからの同期Python パターンを保持します。 移行を最小限に抑える場合、またはコードが同期のみのライブラリに依存している場合は、これを選択します。

モデルサービング (前)

元のクラス ベースのエージェント構造。

from mlflow.pyfunc import ResponsesAgent, ResponsesAgentRequest, ResponsesAgentResponse

class MyAgent(ResponsesAgent):
  def predict(self, request: ResponsesAgentRequest, params=None) -> ResponsesAgentResponse:
    # Synchronous implementation
    ...
    return ResponsesAgentResponse(output=outputs)

  def predict_stream(self, request: ResponsesAgentRequest, params=None):
    # Synchronous generator
    for chunk in ...:
      yield ResponsesAgentStreamEvent(...)

プライマリ エージェント ロジックは、 streaming()に存在します。 non_streaming()関数は、その出力を収集し、1 つの応答として返します。

from mlflow.genai.agent_server import invoke, stream
from mlflow.types.responses import (
  ResponsesAgentRequest,
  ResponsesAgentResponse,
  ResponsesAgentStreamEvent,
)

@invoke()
async def non_streaming(request: ResponsesAgentRequest) -> ResponsesAgentResponse:
  # Async implementation - typically calls streaming() and collects results
  outputs = [
    event.item
    async for event in streaming(request)
    if event.type == "response.output_item.done"
  ]
  return ResponsesAgentResponse(output=outputs)

@stream()
async def streaming(request: ResponsesAgentRequest) -> AsyncGenerator[ResponsesAgentStreamEvent, None]:
  # Async generator
  async for event in ...:
    yield event

アプリ - 同期

最小限の構造変更で、装飾されたモジュール レベルの関数にクラス メソッドを抽出します。

from mlflow.genai.agent_server import invoke, stream
from mlflow.types.responses import (
  ResponsesAgentRequest,
  ResponsesAgentResponse,
  ResponsesAgentStreamEvent,
)

@invoke()
def non_streaming(request: ResponsesAgentRequest) -> ResponsesAgentResponse:
  # Same sync logic from original predict(), extracted from the class
  ...
  return ResponsesAgentResponse(output=outputs)

@stream()
def streaming(request: ResponsesAgentRequest):
  # Same sync generator from original predict_stream(), extracted from the class
  for chunk in ...:
    yield ResponsesAgentStreamEvent(...)

ステップ 4. アプリのセットアップ

  1. 依存関係をインストールします。 これにより、 pyproject.toml の依存関係が解決され、再現可能なインストール用にピン留めする uv.lock ファイルが作成されます。

    uv sync
    
  2. クイック スタート スクリプトを実行して認証を構成し、MLflow 実験を作成し、 .env ファイルを生成します。

    uv run quickstart
    

デプロイ時に Databricks Apps によって同じピン留めされた依存関係がインストールされるように、生成された uv.lock ファイルをコミットします。

ステップ 5. ローカルでテストする

デプロイする前に、アプリ サーバーを起動し、エージェントが正しく応答することを確認します。

curl を使用して元の input_example.json でテストし、エージェントが期待どおりに応答した後にデプロイします。

ステップ 6. リソースの構成

モデル サービス エージェントは、 MLmodel ファイル内のリソースを宣言します。 Databricks Apps エージェントは、宣言型オートメーション バンドルを使用して、 databricks.yml 構成ファイル内のリソースを宣言します。

エージェントの認証を参照してください。

リソース宣言を同等の宣言型オートメーション バンドル形式にマップします。

MLmodel リソースの種類 databricks.yml 相当 許可
serving_endpoint serving_endpoint CAN_QUERY
lakebase database CAN_CONNECT_AND_CREATE
vector_search_index uc_securable (保護可能タイプ: TABLE) SELECT
function uc_securable (保護可能タイプ: FUNCTION) EXECUTE
table uc_securable (保護可能タイプ: TABLE) SELECT または MODIFY
uc_connection uc_securable (保護可能タイプ: CONNECTION) USE_CONNECTION
sql_warehouse sql_warehouse CAN_USE
genie_space genie_space CAN_RUN

ステップ 7. 宣言型オートメーション バンドルを使用してエージェントをデプロイする

宣言型オートメーション バンドルを使用して、エージェントを Databricks Apps にデプロイします。

展開する前に、フォルダー構造が次のようであるかどうかを確認します。

<working-directory>/
├── original_mlflow_model/    # Downloaded artifacts from Model Serving
│   ├── MLmodel
│   ├── code/
│   │   └── agent.py
│   ├── input_example.json
│   └── requirements.txt
│
└── <app-name>/               # New Databricks App (ready to deploy)
    ├── agent_server/
    │   ├── agent.py          # Migrated agent code
    │   └── ...
    ├── app.yaml
    ├── databricks.yml        # Bundle config with resources
    ├── pyproject.toml        # Python dependencies (uv)
    ├── uv.lock               # Pinned dependencies for reproducible installs
    └── ...

Note

Azure Databricksでは、Python依存関係管理にuv (pyproject.toml + uv.lock) することをお勧めします。これによって、より高速なインストールと再現可能なビルドが提供されます。 アプリに pyproject.tomluv.lock が含まれており、 requirements.txtがない場合、Databricks Apps は依存関係をインストールするために uv を使用します。 requirements.txt は引き続きサポートされます。存在する場合は常に優先され、Databricks Apps では代わりに pip が使用されます。 Databricks Apps のベスト プラクティスと、uvを使用したPythonの依存関係の定義に関するページを参照してください。

2つのことを行う必要があります: uv.lockをコミットし、pyproject.tomlrequires-python = ">=3.12,<3.13"を使ってインタープリターを固定します。 Databricks Appsはpyproject.tomluv.lockの両方が存在する場合にのみuvを使用しuv.lock、パッケージのバージョンをロックするがPythonバージョンは選択しません。 インタプリタピンがない場合、Databricks Appsのビルドイメージは、一部のエージェント依存関係に対してプリビルドホイールのない新しいPython(例:3.14)を選んでしまいます。 その後、ソースからパッケージをビルドすることに戻り、ビルドは失敗します。

  1. バンドル構成を検証します。

    databricks bundle validate
    
  2. ワークスペースにバンドルをデプロイします (bundle deploy はファイルをアップロードしますが、アプリは起動しません)。

    databricks bundle deploy
    
  3. アプリを起動します。

    databricks bundle run <app-resource-name>
    

その他のリソース

エージェントを移行した後、次を参照してください。