Melacak versi aplikasi berbasis Git dengan MLflow

Panduan ini menunjukkan cara melacak versi aplikasi GenAI Anda saat kode aplikasi Anda berada di Git atau sistem kontrol versi serupa. Dalam alur kerja ini, MLflow LoggedModel bertindak sebagai hub metadata, menautkan setiap versi aplikasi konseptual ke kode eksternal tertentu (seperti commit Git), dan konfigurasi. Ini LoggedModel kemudian dapat dikaitkan dengan entitas MLflow seperti jejak dan evaluasi yang sedang berjalan.

mlflow.set_active_model(name=...) adalah kunci untuk pelacakan versi: memanggil fungsi ini menautkan jejak aplikasi Anda ke LoggedModel. Jika name tidak ada, LoggedModel akan dibuat secara otomatis.

Panduan ini mencakup hal-hal berikut:

• Lacak versi aplikasi Anda menggunakan LoggedModels.
• Hubungkan evaluasi tautan ke LoggedModel.

Petunjuk / Saran

Databricks menyarankan menggunakan LoggedModels bersamaan dengan registri prompt MLflow. Jika Anda menggunakan registri prompt, setiap versi prompt secara otomatis dikaitkan dengan LoggedModel Anda. Lihat melacak versi prompt bersama versi aplikasi.

Prasyarat

1. Menginstal MLflow dan paket yang diperlukan

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

2. Buat eksperimen MLflow dengan mengikuti panduan mulai cepat untuk menyiapkan lingkungan Anda.

Langkah 1: Membuat aplikasi sampel

Kode berikut membuat aplikasi sederhana yang mengajukan pertanyaan kepada LLM untuk respons.

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)

Langkah 2: Menambahkan pelacakan versi ke kode aplikasi Anda

Versi LoggedModel berfungsi sebagai catatan pusat (hub metadata) untuk versi tertentu dari aplikasi Anda. Tidak perlu menyimpan kode aplikasi itu sendiri. Sebaliknya, ini menunjuk ke tempat kode Anda dikelola (seperti hash commit Git).

Gunakan mlflow.set_active_model() untuk mendeklarasikan LoggedModel yang sedang Anda kerjakan, atau untuk membuat yang baru. Fungsi ini mengembalikan objek yang ActiveModel berisi model_id yang berguna untuk operasi berikutnya.

Petunjuk / Saran

Dalam produksi, Anda dapat mengatur variabel MLFLOW_ACTIVE_MODEL_ID lingkungan alih-alih memanggil set_active_model(). Lihat pelacakan versi dalam panduan produksi.

Nota

Kode berikut menggunakan hash commit Git saat ini sebagai nama model, sehingga versi model Anda hanya meningkat ketika Anda melakukan commit. Untuk membuat LoggedModel baru untuk setiap perubahan dalam basis kode Anda, lihat fungsi pembantu yang membuat LoggedModel unik untuk setiap perubahan di basis kode Anda, bahkan jika tidak berkomitmen pada Git.

Sisipkan kode berikut di bagian atas aplikasi Anda dari langkah 1. Di aplikasi, Anda harus memanggil 'set_active_model() SEBELUM menjalankan kode aplikasi.

# 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
### ...

Langkah 3: Parameter rekaman (Opsional)

Anda dapat mencatat parameter konfigurasi kunci yang menentukan versi aplikasi Anda ini langsung ke LoggedModel penggunaan mlflow.log_model_params(). Ini berguna untuk merekam hal-hal seperti nama LLM, pengaturan suhu, atau strategi pengambilan yang terkait dengan versi kode ini.

Tambahkan kode berikut di bawah kode dari langkah 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)

Langkah 4: Jalankan aplikasi

1. Panggil aplikasi untuk melihat bagaimana LoggedModel dibuat dan dilacak.

# 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. Untuk mensimulasikan perubahan tanpa menerapkan, tambahkan baris berikut untuk membuat model baru yang dicatat secara manual.

# 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)

Langkah 5: Lihat jejak yang ditautkan ke LoggedModel

Gunakan antarmuka pengguna

Masuk ke UI Eksperimen MLflow. Di tab Jejak , Anda dapat melihat versi aplikasi yang menghasilkan setiap jejak (perhatikan, jejak pertama tidak akan memiliki versi yang terpasang karena kami memanggil aplikasi tanpa memanggil set_active_model() terlebih dahulu). Di tab Versi , Anda dapat melihat masing-masing LoggedModel bersama parameter dan jejak tertautnya.

Menggunakan SDK

Anda dapat menggunakan search_traces() untuk mengkueri jejak dari LoggedModel:

import mlflow

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

Anda dapat menggunakan get_logged_model() untuk mendapatkan detail dari 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}")

Langkah 6: Menautkan hasil evaluasi ke LoggedModel

Untuk mengevaluasi aplikasi Anda dan menautkan hasil ke versi ini LoggedModel , lihat Menautkan Hasil Evaluasi dan Jejak ke Versi Aplikasi. Panduan ini mencakup cara menggunakan mlflow.genai.evaluate() untuk menilai performa aplikasi Anda dan secara otomatis mengaitkan metrik, tabel evaluasi, dan jejak dengan versi spesifik LoggedModel Anda.

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())

Lihat hasilnya di tab Versi dan Evaluasi di UI Eksperimen MLflow:

Fungsi pembantu untuk menghitung hash unik untuk setiap perubahan file

Fungsi pembantu di bawah ini secara otomatis menghasilkan nama untuk setiap LoggedModel berdasarkan status repositori Anda. Untuk menggunakan fungsi ini, panggil set_active_model(name=get_current_git_hash()).

get_current_git_hash() menghasilkan pengidentifikasi deterministik yang unik untuk status repositori Git saat ini dengan mengembalikan hash penerapan HEAD (untuk repositori bersih) atau kombinasi hash HEAD dan hash perubahan yang tidak dilakukan (untuk repositori kotor). Ini memastikan bahwa berbagai status repositori selalu menghasilkan pengidentifikasi yang berbeda, sehingga setiap perubahan kode menghasilkan baru 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")

Langkah Selanjutnya

• Jika Anda perlu menggabungkan kode dengan LoggedModel, lihat Kode paket untuk Databricks Model Serving.