MLflow を使用して Git ベースのアプリケーションのバージョンを追跡する
このガイドでは、アプリのコードが Git または同様のバージョン管理システムに存在する場合に、生成AI アプリケーションのバージョンを追跡する方法を示します。このワークフローでは、MLflowのLoggedModelが メタデータ ハブ として機能し、各概念アプリケーション バージョンを特定の外部コード (Git コミットなど) 構成にリンクします。この LoggedModel は、トレースや評価ランなどの MLflow エンティティに関連付けることができます。
mlflow.set_active_model(name=...)はバージョン追跡の鍵です:この関数を呼び出すと、アプリケーションのトレースがLoggedModelにリンクされます。nameが存在しない場合は、新しいLoggedModelが自動的に作成されます。
このガイドでは、次の内容について説明します。
LoggedModelsを使用してアプリケーションのバージョンを追跡します。- 評価ランを
LoggedModelにリンクします。
Databricks では、 LoggedModels を MLflow のプロンプト レジストリと共に使用することを提案しています。プロンプトレジストリを使用する場合、各プロンプトのバージョンは自動的に LoggedModelに関連付けられます。プロンプトのバージョンをアプリケーションのバージョンと共に追跡するを参照してください。
前提 条件
-
MLflow と必要なパッケージをインストールする
Bashpip install --upgrade "mlflow[databricks]>=3.1.0" openai -
MLflow エクスペリメントを作成するには、環境のセットアップに関するクイックスタートに従ってください。
ステップ 1: サンプル アプリケーションを作成する
次のコードは、LLM に応答を求める単純なアプリケーションを作成します。
- OpenAI クライアントを初期化して、Databricks でホストされている LLM または OpenAI でホストされている LLM に接続します。
- Databricks-hosted LLMs
- OpenAI-hosted LLMs
MLflow を使用して、Databricks でホストされている LLM に接続する OpenAI クライアントを取得します。利用可能な基盤モデルからモデルを選択します。
import mlflow
from databricks.sdk import WorkspaceClient
# Enable MLflow's autologging to instrument your application with Tracing
mlflow.openai.autolog()
# Set up MLflow tracking to Databricks
mlflow.set_tracking_uri("databricks")
mlflow.set_experiment("/Shared/docs-demo")
# Create an OpenAI client that is connected to Databricks-hosted LLMs
w = WorkspaceClient()
client = w.serving_endpoints.get_open_ai_client()
# Select an LLM
model_name = "databricks-claude-sonnet-4"
ネイティブの OpenAI SDK を使用して、OpenAI でホストされるモデルに接続します。利用可能なOpenAIモデルからモデルを選択します。
import mlflow
import os
import openai
# Ensure your OPENAI_API_KEY is set in your environment
# os.environ["OPENAI_API_KEY"] = "<YOUR_API_KEY>" # Uncomment and set if not globally configured
# Enable auto-tracing for OpenAI
mlflow.openai.autolog()
# Set up MLflow tracking to Databricks
mlflow.set_tracking_uri("databricks")
mlflow.set_experiment("/Shared/docs-demo")
# Create an OpenAI client connected to OpenAI SDKs
client = openai.OpenAI()
# Select an LLM
model_name = "gpt-4o-mini"
-
サンプル アプリケーションを作成します。
Python# 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=model_name, # 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() を呼び出す必要があります 。
# 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: (省略可能) パラメーターの記録
このバージョンのアプリケーションを定義する主要な設定パラメーターをmlflow.log_model_params()を使用してLoggedModelに直接記録できます。これは、このコードバージョンに関連付けられたLLM名、温度設定、取得戦略などを記録するのに役立ちます。
ステップ 3 のコードの下に次のコードを追加します。
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: アプリケーションを実行する
- アプリケーションを呼び出して、LoggedModel がどのように作成および追跡されるかを確認します。
# 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)
- コミットせずに変更をシミュレートするには、次の行を追加して、新しい記録済みモデルを手動で作成します。
# 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からのトレースを照会できます。
import mlflow
traces = mlflow.search_traces(
filter_string=f"metadata.`mlflow.modelId` = '{active_model_info.model_id}'"
)
print(traces)
get_logged_model()を使用して、LoggedModelの詳細を取得できます。
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 バージョンに自動的に関連付ける方法について説明します。
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 のバージョンと 評価 タブに結果を表示します。
![[バージョンと評価] タブ](https://assets.docs.databricks.com/_static/images/mlflow3-genai/new-images/version-eval.gif)
任意のファイル変更に対して一意のハッシュをコンピュートするヘルパー関数
以下のヘルパー関数は、リポジトリのステータスに基づいて、各 LoggedModel の名前を自動的に生成します。この関数を使用するには、 set_active_model(name=get_current_git_hash())を呼び出します。
get_current_git_hash() クリーンなリポジトリの場合はHEADコミットハッシュを、ダーティなリポジトリの場合はHEADハッシュとコミットされていない変更のハッシュの組み合わせを返すことで、Gitリポジトリの現在の状態に対する一意で決定論的な識別子を生成します。これにより、リポジトリの異なる状態は常に異なる識別子が生成されるため、コードを変更するたびに新しい LoggedModelが発生します。
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モデルサービングのコードをパッケージ化するを参照してください。