Rastreie versões de aplicativos baseados em Git com o MLflow

Este guia demonstra como rastrear versões do seu aplicativo GenAI quando o código do aplicativo reside no Git ou em um sistema de controle de versão semelhante. Nesse fluxo de trabalho, um MLflow LoggedModel atua como um hub de metadados , vinculando cada versão conceitual do aplicativo ao seu código externo específico (como um commit do Git), configurações. Esse LoggedModel pode então ser associado a MLflow entidades como traços e execução de avaliações.

O mlflow.set_active_model(name=...) é key para acompanhamento da versão: chamar essa função vincula os traços do seu aplicativo a um LoggedModel. Se o name não existir, um novo LoggedModel será criado automaticamente.

Este guia aborda o seguinte:

• Acompanhe as versões do seu aplicativo usando LoggedModels.
• Vincule a execução da avaliação ao seu site LoggedModel.

dica

A Databricks sugere o uso do site LoggedModels juntamente com o registro imediato do MLflow. Se você usar o registro de solicitações, a versão de cada solicitação será automaticamente associada à sua LoggedModel. Veja as versões do prompt de rastreamento junto com as versões do aplicativo.

Pré-requisitos

1. Instale o site MLflow e o pacote necessário

   Bash

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

2. Crie um experimento MLflow seguindo o início rápido de configuração do ambiente.

Etapa 1: criar um aplicativo de amostra

O código a seguir cria um aplicativo simples que solicita uma resposta a um 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)

Etapa 2: Adicione o acompanhamento da versão ao código do seu aplicativo

A versão LoggedModel serve como um registro central (hub de metadados) para uma versão específica do seu aplicativo. Ele não precisa armazenar o código do aplicativo em si. Em vez disso, ele aponta para onde seu código está gerenciando (como um hash Git commit ).

Use mlflow.set_active_model() para declarar o LoggedModel com o qual você está trabalhando atualmente ou para criar um novo. Essa função retorna um objeto ActiveModel contendo o model_id, que é útil para operações subsequentes.

dica

Na produção, o senhor pode definir a variável de ambiente MLFLOW_ACTIVE_MODEL_ID em vez de chamar set_active_model(). Veja o guia de acompanhamento da versão em produção.

nota

O código a seguir usa o hash do commit do Git atual como o nome do modelo, de modo que a versão do modelo só aumenta quando o senhor faz o commit. Para criar um novo LoggedModel para cada alteração em sua base de código, consulte a função auxiliar que cria um LoggedModel exclusivo para qualquer alteração em sua base de código, mesmo que não tenha sido confirmada no Git.

Insira o código a seguir na parte superior do seu aplicativo a partir da etapa 1. Em seu aplicativo, você deve chamar `set_active_model () ANTES de executar o código do seu aplicativo.

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

Etapa 3: (Opcional) Parâmetros de registro

O senhor pode log key parâmetros de configuração que definem essa versão do seu aplicativo diretamente para o LoggedModel usando mlflow.log_model_params(). Isso é útil para registrar coisas como nomes de LLM, configurações de temperatura ou estratégias de recuperação que estejam vinculadas a essa versão do código.

Adicione o seguinte código abaixo do código da etapa 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)

Etapa 4: execução do aplicativo

1. Ligue para o aplicativo para ver como o LoggedModel é criado e rastreado.

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. Para simular uma alteração sem confirmação, adicione as seguintes linhas para criar manualmente um novo modelo registrado.

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)

Etapa 5: visualizar os rastreamentos vinculados ao LoggedModel

Use a interface do usuário

Acesse a UI do experimento MLflow. Em Traces tab, o senhor pode ver a versão do aplicativo que gerou cada rastreamento (observe que o primeiro rastreamento não terá uma versão anexada, pois chamamos o aplicativo sem chamar set_active_model() primeiro). Em Versions (Versões ) tab, o senhor pode ver cada LoggedModel juntamente com seus parâmetros e traços vinculados.

Usar o SDK

Você pode usar search_traces() para consultar traços de um LoggedModel:

Python

import mlflow

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

Você pode usar get_logged_model() para obter detalhes do 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}")

Etapa 6: Vincular os resultados da avaliação ao LoggedModel

Para avaliar seu aplicativo e vincular os resultados a essa versão LoggedModel, consulte Vincular resultados de avaliação e rastreamentos às versões do aplicativo. Este guia aborda como usar o site mlflow.genai.evaluate() para avaliar o desempenho do seu aplicativo e associar automaticamente as métricas, as tabelas de avaliação e os rastreamentos à versão específica do site 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())

Veja os resultados na guia Versions and Evaluations ( Versões e avaliações ) na UI do experimento MLflow:

Função auxiliar para compute um hash exclusivo para qualquer alteração de arquivo

A função auxiliar abaixo gera automaticamente um nome para cada LoggedModel com base no status do seu repositório. Para usar essa função, chame set_active_model(name=get_current_git_hash()).

get_current_git_hash() gera um identificador único e determinístico para o estado atual de um repositório Git, retornando o hash de commit do HEAD (para repositórios limpos) ou uma combinação do hash do HEAD e um hash de alterações não confirmadas (para repositórios sujos). Isso garante que diferentes estados do repositório sempre produzam identificadores diferentes, de modo que cada alteração de código resulte em um novo 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")

Próximas etapas

• Se o senhor precisar agrupar o código com o LoggedModel, consulte o código do pacote para Databricks servindo modelo.