使用 MLflow 跟踪基于 Git 的应用程序版本

本指南演示了当您的应用程序代码位于 Git 或类似的版本控制系统时，如何跟踪您的 GenAI 应用程序版本。在此工作流程中，MLflow LoggedModel 充当元数据中心，将每个概念性应用程序版本链接到其特定的外部代码（例如 Git 提交）和配置。此 LoggedModel 随后可以与 MLflow 实体（如跟踪和评估运行）关联。

mlflow.set_active_model(name=...) 是版本跟踪的关键：调用此函数将您的应用程序跟踪链接到 LoggedModel。如果名称不存在，将自动创建一个新的 LoggedModel。

为什么基于 Git 的版本控制适用于 GenAI

基于 Git 的版本控制将您的版本控制系统转变为一个强大的应用程序生命周期管理工具。每次提交都成为一个潜在的应用程序版本，并内置完整的代码历史记录和更改跟踪。

基于提交的版本控制

使用 Git 提交哈希作为唯一的版本标识符。每次提交都代表一个完整的应用程序状态，具有完全的可复现性。

基于分支的开发

利用 Git 分支进行并行开发。特性分支成为可以系统合并的隔离版本流。

自动元数据捕获

MLflow 在运行期间自动捕获 Git 提交、分支和仓库 URL。无需手动版本跟踪。

无缝集成

与您现有的 Git 工作流程自然配合。无需更改开发过程或额外工具。

MLflow 如何捕获 Git 上下文

当您在 Git 仓库中记录模型或开始运行时，MLflow 会自动捕获 Git 元数据。这以透明的方式发生——无需额外配置。

自动 Git 检测

MLflow 检测 Git 仓库并在运行期间自动捕获提交哈希、分支名称和仓库 URL。

版本上下文链接

使用 mlflow.set_active_model() 建立版本上下文，然后所有后续跟踪都将自动链接到该基于 Git 的版本。

统一元数据中心

每个版本都成为一个中心参考，将 Git 提交、配置、跟踪和评估结果链接到一个版本化实体中。

先决条件

安装 MLflow 和所需的包

pip install --upgrade "mlflow>=3.1" openai

设置您的 OpenAI API 密钥

export OPENAI_API_KEY="your-api-key-here"

按照入门指南创建 MLflow 实验。

步骤 1：创建一个示例应用程序

以下代码创建了一个简单的应用程序，用于提示 LLM 进行响应

import mlflow
import openai

# Enable MLflow's autologging to instrument your application with Tracing
mlflow.openai.autolog()

# Set up OpenAI client
client = openai.OpenAI()


# Use the trace decorator to capture the application's entry point
@mlflow.trace
def my_app(input: str):
    """Customer support agent application"""
    # This call is automatically instrumented by `mlflow.openai.autolog()`
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "You are a helpful customer support agent."},
            {"role": "user", "content": input},
        ],
        temperature=0.7,
        max_tokens=150,
    )
    return response.choices[0].message.content


# Test the application
result = my_app(input="What is MLflow?")
print(result)

步骤 2：向应用程序代码添加版本跟踪

LoggedModel 版本充当应用程序特定版本的中央记录（元数据中心）。它不需要存储应用程序代码本身。相反，它指向您的代码管理位置（例如 Git 提交哈希）。

使用 mlflow.set_active_model() 声明您当前正在使用的 LoggedModel，或创建一个新的。此函数返回一个 ActiveModel 对象，其中包含 model_id，这对于后续操作很有用。

注意

以下代码使用当前的 Git 提交哈希作为模型的名称，因此您的模型版本仅在您提交时才增加。要为代码库中的每次更改创建一个新的 LoggedModel，请参阅帮助函数，它会为代码库中的任何更改创建唯一的 LoggedModel，即使未提交到 Git。

将以下代码插入到步骤 1 中应用程序的顶部。在您的应用程序中，您必须在执行应用程序代码之前调用 set_active_model()。

# Keep original imports
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}'"
)

步骤 3：（可选）记录参数

您可以使用 mlflow.log_model_params() 直接将定义此应用程序版本的关键配置参数记录到 LoggedModel 中。这对于记录与此代码版本相关的 LLM 名称、温度设置或检索策略等内容非常有用。

在步骤 2 的代码下方添加以下代码

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。在跟踪选项卡中，您可以查看生成每个跟踪的应用程序版本。在版本选项卡中，您可以查看每个 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}")

用于计算任何文件更改的唯一哈希的帮助函数

下面的帮助函数根据仓库的状态自动为每个 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")

后续步骤

既然您已经了解了使用 MLflow 进行基于 Git 的应用程序版本控制的基础知识，您可以探索这些相关主题

比较应用程序版本

学习使用基于跟踪的比较来评估不同版本的系统方法

了解更多 →

版本跟踪快速入门

通过 MLflow 中的版本跟踪动手指南快速入门

了解更多 →