このガイドでは、アプリのコードが Git または同様のバージョン管理システムに存在する場合に、GenAI アプリケーションのバージョンを追跡する方法について説明します。 このワークフローでは、MLflow LoggedModel は メタデータ ハブとして機能し、各概念アプリケーション バージョンを特定の外部コード (Git コミットなど) の構成にリンクします。 この LoggedModel は、トレースや評価実行などの MLflow エンティティに関連付けることができます。
mlflow.set_active_model(name=...)はバージョン追跡の鍵となります。この関数を呼び出すと、アプリケーションのトレースがLoggedModelにリンクされます。
nameが存在しない場合は、新しいLoggedModelが自動的に作成されます。
学習内容:
- を使用してアプリケーションのバージョンを追跡する
LoggedModels - 評価の実行を
LoggedModelに関連付ける
ヒント
MLflow のプロンプト レジストリと共に LoggedModels を使用することをお勧めします。 プロンプト レジストリを使用すると、各プロンプトのバージョンが自動的に LoggedModelに関連付けられます。
アプリケーションのバージョンと共にプロンプトバージョンの追跡を参照してください。
[前提条件]
MLflow と必要なパッケージをインストールする
pip install --upgrade "mlflow[databricks]>=3.1.0" openai環境のセットアップのクイックスタートに従って、MLflow 実験を作成します。
手順 1: サンプル アプリケーションを作成する
LLM に応答を求める単純なアプリケーションを次に示します。
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を宣言したり、まだ存在しない場合は新しいを作成したりします。 この関数は、後続の操作に役立つActiveModelを含むmodel_id オブジェクトを返します。
ヒント
運用環境では、MLFLOW_ACTIVE_MODEL_IDを呼び出す代わりに、環境変数set_active_model()を設定できます。 詳細については、 運用ガイドのバージョン追跡 を参照してください。
注
次のコードでは、現在の 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
### ...
手順 4: (必要に応じて) パラメーターを記録する
必要に応じて、このバージョンのアプリケーションを定義するキー構成パラメーターを、LoggedModelを使用してmlflow.log_model_params()に直接ログに記録できます。 これは、このコード バージョンに関連付けられている 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)
手順 5:アプリケーションの実行
- 次に、アプリケーションを呼び出して、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)
手順 6: 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 の バージョン と 評価 タブで結果を表示します。
ファイル変更の一意のハッシュを計算するヘルパー関数
次のヘルパー関数は、リポジトリの状態に基づいて、各 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とバンドルする必要があるシナリオの場合。