Sledování verzí aplikací založených na Gitu pomocí MLflow

Tato příručka ukazuje, jak sledovat verze aplikace GenAI, když se kód aplikace nachází v Gitu nebo podobném systému správy verzí. V tomto pracovním postupu funguje MLflow LoggedModel jako centrum metadat, které propojuje každou koncepční verzi aplikace s konkrétním externím kódem (například potvrzením Gitu), konfigurací. To LoggedModel pak může být přidruženo k entitám MLflow, jako jsou sledování a vyhodnocovací běhy.

Klíčem mlflow.set_active_model(name=...) ke sledování verzí je volání této funkce propojení trasování vaší aplikace na LoggedModel. name Pokud neexistuje, vytvoří se automaticky novýLoggedModel.

Tento průvodce popisuje následující:

• Sledujte verze aplikace pomocí LoggedModels.
• Proveďte vyhodnocení odkazů k vašemu LoggedModel.

Návod

Databricks navrhuje použití LoggedModels spolu s registrem příkazového řádku MLflow. Pokud používáte registr výzvy, každá verze výzvy je automaticky přidružena k vašemu LoggedModel. Podívejte se jak sledovat verze výzev spolu s verzemi aplikací.

Požadavky

1. Instalace MLflow a požadovaných balíčků

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

2. Postupujte podle rychlého startu pro nastavení prostředí a vytvořte experiment MLflow.

Krok 1: Vytvoření ukázkové aplikace

Následující kód vytvoří jednoduchou aplikaci, která vyzve LLM k zadání odpovědi.

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)

Krok 2: Přidání sledování verzí do kódu aplikace

Verze LoggedModel slouží jako centrální záznam (centrum metadat) pro konkrétní verzi vaší aplikace. Nemusí ukládat samotný kód aplikace. Místo toho odkazuje na místo, kde je váš kód spravovaný (například hash potvrzení Gitu).

Slouží mlflow.set_active_model() k deklarování, se LoggedModel kterým právě pracujete, nebo k vytvoření nového. Tato funkce vrátí ActiveModel objekt, který obsahuje model_id a je užitečný pro následné operace.

Návod

V produkčním prostředí můžete místo volání MLFLOW_ACTIVE_MODEL_IDnastavit proměnnou set_active_model() prostředí . Podívejte se na sledování verzí v provozní příručce.

Poznámka:

Následující kód používá aktuální hash Gitu jako název modelu, takže se verze modelu zvyšuje pouze při potvrzení změn. Pokud chcete vytvořit nový LoggedModel pro každou změnu v základu kódu, podívejte se na pomocnou funkci , která vytvoří jedinečnou loggedModel pro všechny změny v základu kódu, a to i v případě, že není potvrzena do Gitu.

Do horní části aplikace vložte následující kód z kroku 1. V aplikaci musíte před spuštěním kódu aplikace volat 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
### ...

Krok 3: (Volitelné) Parametry záznamu

Klíčové konfigurační parametry, které definují tuto verzi vaší aplikace, můžete protokolovat přímo do LoggedModel pomocí mlflow.log_model_params(). To je užitečné pro zaznamenávání věcí, jako jsou názvy LLM, nastavení teploty nebo strategie načítání, které jsou svázané s touto verzí kódu.

Pod kód z kroku 3 přidejte následující kód:

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)

Krok 4: Spuštění aplikace

1. Zavolejte aplikaci, abyste viděli, jak je loggedModel vytvořen a sledován.

# 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. Pokud chcete simulovat změnu bez potvrzení, přidejte následující řádky pro ruční vytvoření nového protokolovaného modelu.

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

Krok 5: Zobrazení sledů propojených s protokolovaným modelem

Použití uživatelského rozhraní

Přejděte do uživatelského rozhraní experimentu MLflow. Na kartě Trasování uvidíte verzi aplikace, která generovala každé trasování (všimněte si, že první trasování nebude mít připojenou verzi, protože jsme volali aplikaci bez volání set_active_model() první). Na kartě Verze můžete zobrazit každou LoggedModel vedle svých parametrů a spojených trasování.

Použití sady SDK

Můžete pomocí search_traces() dotazovat na stopy z LoggedModel.

import mlflow

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

Můžete použít get_logged_model() k získání podrobností o 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}")

Krok 6: Propojení výsledků vyhodnocení s protokolovaným modelem

Pokud chcete vyhodnotit aplikaci a propojit výsledky s touto LoggedModel verzí, podívejte se na Propojení výsledků vyhodnocení a trasování s verzemi aplikací. Tato příručka popisuje, jak pomocí mlflow.genai.evaluate() posoudit výkon vaší aplikace a automaticky přidružit metriky, hodnotící tabulky a trasování k vaší konkrétní verzi 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())

Výsledky můžete zobrazit na kartách Verze a Vyhodnocení v uživatelském rozhraní experimentu MLflow:

Pomocná funkce pro výpočet jedinečné hodnoty hash pro jakoukoli změnu souboru

Následující pomocná funkce automaticky vygeneruje název každého LoggedModel na základě stavu vašeho úložiště. Chcete-li použít tuto funkci, zavolejte set_active_model(name=get_current_git_hash()).

get_current_git_hash() Generuje jedinečný deterministický identifikátor pro aktuální stav úložiště Git vrácením hodnoty hash potvrzení HEAD (pro čistá úložiště) nebo kombinací hodnoty hash HEAD a hodnoty hash nepotvrzených změn (pro špinavá úložiště). Zajišťuje, aby různé stavy úložiště vždy vytvářely různé identifikátory, takže každá změna kódu vede k novému 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")

Další kroky

• Pokud potřebujete zabalit kód pomocí LoggedModel, přečtěte si Jak zabalit kód pro obsluhu modelu v Databricks.