MLflow を使用して Git ベースのアプリケーションのバージョンを追跡する

このガイドでは、アプリのコードが Git または同様のバージョン管理システムに存在する場合に、GenAI アプリケーションのバージョンを追跡する方法を示します。このワークフローでは、MLflow LoggedModel メタデータ ハブ として機能し、各概念アプリケーション バージョンを特定の外部コード (Git コミットなど) 構成にリンクします。この LoggedModel は、トレースや評価実行などの MLflow エンティティに関連付けることができます。

mlflow.set_active_model(name=...)はバージョン追跡の鍵です: この関数を呼び出すと、アプリケーションのトレースがLoggedModelにリンクされます。nameが存在しない場合は、新しいLoggedModelが自動的に作成されます。

このガイドでは、次の内容について説明します。

• LoggedModelsを使用してアプリケーションのバージョンを追跡します。
• 評価の実行を LoggedModelにリンクします。

ヒント

Databricks では、 LoggedModels を MLflow のプロンプト レジストリと共に使用することを提案しています。プロンプト・レジストリを使用する場合、各プロンプトのバージョンは自動的に LoggedModelに関連付けられます。「プロンプトのバージョンをアプリケーションのバージョンと共に追跡する」を参照してください。

前提 条件

1. MLflow と必要なパッケージをインストールする

   Bash

   pip install --upgrade "mlflow[databricks]>=3.1.0" openai

2. MLflow エクスペリメントを作成するには、環境のセットアップに関するクイックスタートに従ってください。

ステップ 1: サンプル アプリケーションを作成する

次のコードは、LLM に応答を求める単純なアプリケーションを作成します。

Python

import mlflow
from openai import OpenAI

# Enable MLflow's autologging to instrument your application with Tracing
mlflow.openai.autolog()

# Connect to a Databricks LLM via OpenAI using the same credentials as MLflow
# Alternatively, you can use your own OpenAI credentials here
mlflow_creds = mlflow.utils.databricks_utils.get_databricks_host_creds()
client = OpenAI(
    api_key=mlflow_creds.token,
    base_url=f"{mlflow_creds.host}/serving-endpoints"
)

# Use the trace decorator to capture the application's entry point
@mlflow.trace
def my_app(input: str):
    # This call is automatically instrumented by `mlflow.openai.autolog()`
    response = client.chat.completions.create(
    model="databricks-claude-sonnet-4",  # This example uses a Databricks hosted LLM - you can replace this with any AI Gateway or Model Serving endpoint. If you provide your own OpenAI credentials, replace with a valid OpenAI model e.g., gpt-4o, etc.
    messages=[
        {
        "role": "system",
        "content": "You are a helpful assistant.",
        },
        {
        "role": "user",
        "content": input,
        },
    ],
    )
    return response.choices[0].message.content

result = my_app(input="What is MLflow?")
print(result)

ステップ 2: アプリのコードにバージョン追跡を追加する

LoggedModelバージョンは、アプリケーションの特定のバージョンの中央レコード (メタデータ ハブ) として機能します。アプリケーション コード自体を格納する必要はありません。代わりに、コードが管理されている場所 (Git コミット ハッシュなど) を指します。

mlflow.set_active_model() を使用して、現在使用しているLoggedModelを宣言するか、新しいを作成します。この関数は、後続の操作に役立つmodel_idを含むActiveModelオブジェクトを返します。

ヒント

本番運用では、set_active_model()を呼び出す代わりに、環境変数をMLFLOW_ACTIVE_MODEL_IDに設定することができます。本番運用ガイドのバージョン追跡を参照してください。

注記

次のコードでは、現在の Git コミット ハッシュをモデルの名前として使用しているため、コミット時にのみモデル バージョンがインクリメントされます。コード ベースのすべての変更に対して新しい LoggedModel を作成するには、Git にコミットされていない場合でも、コード ベースの任意の変更に対して一意の LoggedModel を作成する ヘルパー関数 を参照してください。

手順 1 のコードをアプリケーションの上部に挿入します。 あなたのアプリケーションでは、アプリのコードを実行する前に set_active_model() を呼び出す必要があります 。

Python

# Keep original imports
### NEW CODE
import subprocess

# Define your application and its version identifier
app_name = "customer_support_agent"

# Get current git commit hash for versioning
try:
    git_commit = (
        subprocess.check_output(["git", "rev-parse", "HEAD"])
        .decode("ascii")
        .strip()[:8]
    )
    version_identifier = f"git-{git_commit}"
except subprocess.CalledProcessError:
    version_identifier = "local-dev"  # Fallback if not in a git repo
logged_model_name = f"{app_name}-{version_identifier}"

# Set the active model context
active_model_info = mlflow.set_active_model(name=logged_model_name)
print(
    f"Active LoggedModel: '{active_model_info.name}', Model ID: '{active_model_info.model_id}'"
)

### END NEW CODE

### ORIGINAL CODE BELOW
### ...

ステップ 3: (省略可能) Record パラメーター

このバージョンのアプリケーションを定義する主要な設定パラメーターは、 mlflow.log_model_params()を使用してLoggedModelに直接ログに記録できます。これは、このコード バージョンに関連付けられている LLM 名、温度設定、取得戦略などを記録する場合に便利です。

ステップ 3 のコードの下に次のコードを追加します。

Python

app_params = {
    "llm": "gpt-4o-mini",
    "temperature": 0.7,
    "retrieval_strategy": "vector_search_v3",
}

# Log params
mlflow.log_model_params(model_id=active_model_info.model_id, params=app_params)

ステップ 4: アプリケーションを実行する

1. アプリケーションを呼び出して、LoggedModel がどのように作成および追跡されるかを確認します。

Python

# These 2 invocations will be linked to the same LoggedModel
result = my_app(input="What is MLflow?")
print(result)

result = my_app(input="What is Databricks?")
print(result)

2. コミットせずに変更をシミュレートするには、次の行を追加して、新しい記録済みモデルを手動で作成します。

Python

# Set the active model context
active_model_info = mlflow.set_active_model(name="new-name-set-manually")
print(
    f"Active LoggedModel: '{active_model_info.name}', Model ID: '{active_model_info.model_id}'"
)

app_params = {
    "llm": "gpt-4o",
    "temperature": 0.7,
    "retrieval_strategy": "vector_search_v4",
}

# Log params
mlflow.log_model_params(model_id=active_model_info.model_id, params=app_params)

# This will create a new LoggedModel
result = my_app(input="What is GenAI?")
print(result)

ステップ 5: LoggedModel にリンクされたトレースを表示する

UI を使用する

MLflow エクスペリメント UI に移動します。 [トレース ] タブでは、各トレースを生成したアプリのバージョンを確認できます (最初にset_active_model()を呼び出さずにアプリを呼び出したため、最初のトレースにはバージョンがアタッチされていないことに注意してください)。 [バージョン] タブでは、各LoggedModelをそのパラメーターとリンクされたトレースと共に表示できます。

SDK を使用する

search_traces() を使用して、LoggedModelからのトレースをクエリできます。

Python

import mlflow

traces = mlflow.search_traces(
    filter_string=f"metadata.`mlflow.modelId` = '{active_model_info.model_id}'"
)
print(traces)

get_logged_model()を使用して、LoggedModelの詳細を取得できます。

Python

import mlflow
import datetime
# Get LoggedModel metadata
logged_model = mlflow.get_logged_model(model_id=active_model_info.model_id)

# Inspect basic properties
print(f"\n=== LoggedModel Information ===")
print(f"Model ID: {logged_model.model_id}")
print(f"Name: {logged_model.name}")
print(f"Experiment ID: {logged_model.experiment_id}")
print(f"Status: {logged_model.status}")
print(f"Model Type: {logged_model.model_type}")
creation_time = datetime.datetime.fromtimestamp(logged_model.creation_timestamp / 1000)
print(f"Created at: {creation_time}")

# Access the parameters
print(f"\n=== Model Parameters ===")
for param_name, param_value in logged_model.params.items():
    print(f"{param_name}: {param_value}")

# Access tags if any were set
if logged_model.tags:
    print(f"\n=== Model Tags ===")
    for tag_key, tag_value in logged_model.tags.items():
        print(f"{tag_key}: {tag_value}")

ステップ 6: 評価結果を LoggedModel にリンクする

アプリケーションを評価し、結果をこの LoggedModel バージョンにリンクするには、 評価結果とトレースをアプリのバージョンにリンクするをご覧ください。このガイドでは、 mlflow.genai.evaluate() を使用してアプリケーションのパフォーマンスを評価し、メトリクス、評価テーブル、トレースを特定の LoggedModel バージョンに自動的に関連付ける方法について説明します。

Python

import mlflow
from mlflow.genai import scorers

eval_dataset = [
    {
        "inputs": {"input": "What is the most common aggregate function in SQL?"},
    }
]

mlflow.genai.evaluate(data=eval_dataset, predict_fn=my_app, model_id=active_model_info.model_id, scorers=scorers.get_all_scorers())

MLflowエクスペリメント UI のバージョンと 評価 タブに結果を表示します。

任意のファイル変更に対して一意のハッシュをコンピュートするヘルパー関数

以下のヘルパー関数は、リポジトリのステータスに基づいて、各 LoggedModel の名前を自動的に生成します。この関数を使用するには、 set_active_model(name=get_current_git_hash())を呼び出します。

get_current_git_hash() は、HEAD コミット ハッシュ (クリーン リポジトリの場合) または HEAD ハッシュとコミットされていない変更のハッシュの組み合わせ (ダーティ リポジトリの場合) を返すことで、Git リポジトリの現在の状態に対する一意で決定論的な識別子を生成します。これにより、リポジトリの異なる状態は常に異なる識別子が生成されるため、コードを変更するたびに新しい LoggedModelが発生します。

Python

import subprocess
import hashlib
import os

def get_current_git_hash():
    """
    Get a deterministic hash representing the current git state.
    For clean repositories, returns the HEAD commit hash.
    For dirty repositories, returns a combination of HEAD + hash of changes.
    """
    try:
        # Get the git repository root
        result = subprocess.run(
            ["git", "rev-parse", "--show-toplevel"],
            capture_output=True, text=True, check=True
        )
        git_root = result.stdout.strip()

        # Get the current HEAD commit hash
        result = subprocess.run(
            ["git", "rev-parse", "HEAD"], capture_output=True, text=True, check=True
        )
        head_hash = result.stdout.strip()

        # Check if repository is dirty
        result = subprocess.run(
            ["git", "status", "--porcelain"], capture_output=True, text=True, check=True
        )

        if not result.stdout.strip():
            # Repository is clean, return HEAD hash
            return head_hash

        # Repository is dirty, create deterministic hash of changes
        # Collect all types of changes
        changes_parts = []

        # 1. Get staged changes
        result = subprocess.run(
            ["git", "diff", "--cached"], capture_output=True, text=True, check=True
        )
        if result.stdout:
            changes_parts.append(("STAGED", result.stdout))

        # 2. Get unstaged changes to tracked files
        result = subprocess.run(
            ["git", "diff"], capture_output=True, text=True, check=True
        )
        if result.stdout:
            changes_parts.append(("UNSTAGED", result.stdout))

        # 3. Get all untracked/modified files from status
        result = subprocess.run(
            ["git", "status", "--porcelain", "-uall"],
            capture_output=True, text=True, check=True
        )

        # Parse status output to handle all file states
        status_lines = result.stdout.strip().split('\n') if result.stdout.strip() else []
        file_contents = []

        for line in status_lines:
            if len(line) >= 3:
                status_code = line[:2]
                filepath = line[3:]  # Don't strip - filepath starts exactly at position 3

                # For any modified or untracked file, include its current content
                if '?' in status_code or 'M' in status_code or 'A' in status_code:
                    try:
                        # Use absolute path relative to git root
                        abs_filepath = os.path.join(git_root, filepath)
                        with open(abs_filepath, 'rb') as f:
                            # Read as binary to avoid encoding issues
                            content = f.read()
                            # Create a hash of the file content
                            file_hash = hashlib.sha256(content).hexdigest()
                            file_contents.append(f"{filepath}:{file_hash}")
                    except (IOError, OSError):
                        file_contents.append(f"{filepath}:unreadable")

        # Sort file contents for deterministic ordering
        file_contents.sort()

        # Combine all changes
        all_changes_parts = []

        # Add diff outputs
        for change_type, content in changes_parts:
            all_changes_parts.append(f"{change_type}:\n{content}")

        # Add file content hashes
        if file_contents:
            all_changes_parts.append("FILES:\n" + "\n".join(file_contents))

        # Create final hash
        all_changes = "\n".join(all_changes_parts)
        content_to_hash = f"{head_hash}\n{all_changes}"
        changes_hash = hashlib.sha256(content_to_hash.encode()).hexdigest()

        # Return HEAD hash + first 8 chars of changes hash
        return f"{head_hash[:32]}-dirty-{changes_hash[:8]}"

    except subprocess.CalledProcessError as e:
        raise RuntimeError(f"Git command failed: {e}")
    except FileNotFoundError:
        raise RuntimeError("Git is not installed or not in PATH")

次のステップ

• LoggedModelにコードをバンドルする必要がある場合は、Databricksモデルサービングのコードをパッケージ化するを参照してください。