Pular para o conteúdo principal

Monitoramento da produção

info

Beta

Este recurso está em fase beta.

Visão geral​

MLflow permite que você execute automaticamente pontuadores em seus traços de aplicativos GenAI de produção para monitorar continuamente a qualidade. É possível programar qualquer avaliador (incluindo métricas personalizadas e juízes integrados/personalizados LLM ) para avaliar automaticamente uma amostra do tráfego de produção.

Principais benefícios:

  • Avaliação automatizada da qualidade sem intervenção manual.
  • Amostragem flexível para equilibrar a cobertura com o custo computacional.
  • Avaliação consistente usando os mesmos pontuadores do desenvolvimento.
  • Monitoramento contínuo com execução periódica em segundo plano.
nota

MLflow O monitoramento da produção 3 é compatível com os registros de traços do site MLflow 2.

Pré-requisitos​

Antes de configurar o monitoramento de qualidade, certifique-se de que possui:

  1. MLflow Experimento: Um experimento de " MLflow " em que os traços estão sendo registrados. Se não for especificado, o experimento ativo é usado.
  2. Aplicação de produção instrumentada : seu aplicativo GenAI deve estar registrando rastreamentos usando o MLflow Tracing. Consulte o guia Rastreamento de produção.
  3. Pontuadores definidos:pontuadores testados que funcionam com o formato de rastreamento do seu aplicativo.
dica

Se você usou seu aplicativo de produção como predict_fn em mlflow.genai.evaluate() durante o desenvolvimento, seus marcadores provavelmente já são compatíveis.

Iniciar o monitoramento da produção​

Esta seção inclui um exemplo de código que mostra como criar os diferentes tipos de marcadores.

nota

A qualquer momento, no máximo 20 avaliadores podem estar associados a um experimento para monitoramento contínuo da qualidade.

Use marcadores predefinidos​

O MLflow oferece vários avaliadores predefinidos que podem ser utilizados imediatamente para monitoramento.

Python
from mlflow.genai.scorers import Safety, ScorerSamplingConfig

# Register the scorer with a name and start monitoring
safety_scorer = Safety().register(name="my_safety_scorer")  # name must be unique to experiment
safety_scorer = safety_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.7))

Por default, cada juiz usa um modelo LLM especialmente ajustado e hospedado Databricks , projetado para realizar avaliações de qualidade do GenAI. Você pode alterar o modelo do juiz para usar um endpoint do modelo de atendimento Databricks usando o argumento model na definição do pontuador. O modelo deve ser especificado no formato databricks:/<databricks-serving-endpoint-name>.

Python
safety_scorer = Safety(model="databricks:/databricks-gpt-oss-20b").register(name="my_custom_safety_scorer")

Utilize avaliadores de LLM baseados em diretrizes​

Os avaliadores de LLM baseados em diretrizes podem avaliar entradas e saídas utilizando critérios de aprovação/reprovação em linguagem natural.

Python
from mlflow.genai.scorers import Guidelines

# Create and register the guidelines scorer
english_scorer = Guidelines(
  name="english",
  guidelines=["The response must be in English"]
).register(name="is_english")  # name must be unique to experiment

# Start monitoring with the specified sample rate
english_scorer = english_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.7))

Assim como os pontuadores predefinidos, você pode alterar o modelo do juiz para usar um endpoint do modelo de apoio Databricks .

Python
english_scorer = Guidelines(
  name="english",
  guidelines=["The response must be in English"],
  model="databricks:/databricks-gpt-oss-20b",
).register(name="custom_is_english")

Use pontuadores baseados em prompts​

Para obter mais flexibilidade do que os avaliadores LLM baseados em diretrizes, é possível utilizar avaliadores baseados em prompts, que permitem uma avaliação de qualidade em vários níveis com categorias de escolha personalizáveis (por exemplo, excelente/bom/insuficiente) e pontuação numérica opcional.

Python
from mlflow.genai.scorers import scorer, ScorerSamplingConfig


@scorer
def formality(inputs, outputs, trace):
    # Must be imported inline within the scorer function body
    from mlflow.genai.judges.databricks import custom_prompt_judge
    from mlflow.entities.assessment import DEFAULT_FEEDBACK_NAME

    formality_prompt = """
    You will look at the response and determine the formality of the response.

    <request>{{request}}</request>
    <response>{{response}}</response>

    You must choose one of the following categories.

    [[formal]]: The response is very formal.
    [[semi_formal]]: The response is somewhat formal. The response is somewhat formal if the response mentions friendship, etc.
    [[not_formal]]: The response is not formal.
    """

    my_prompt_judge = custom_prompt_judge(
        name="formality",
        prompt_template=formality_prompt,
        numeric_values={
            "formal": 1,
            "semi_formal": 0.5,
            "not_formal": 0,
        },
        model="databricks:/databricks-gpt-oss-20b",  # optional
    )

    result = my_prompt_judge(request=inputs, response=inputs)
    if hasattr(result, "name"):
        result.name = DEFAULT_FEEDBACK_NAME
    return result

# Register the custom scorer and start monitoring
formality_scorer = formality.register(name="my_formality_scorer")  # name must be unique to experiment
formality_scorer = formality_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.1))

Use funções personalizadas de pontuação​

Para obter flexibilidade máxima, incluindo a opção de dispensar a pontuação baseada em LLM, é possível definir e utilizar uma função de pontuação personalizada para monitoramento.

Ao definir pontuadores personalizados, não use dicas de tipo que precisem ser importadas na assinatura da função. Se o corpo da função de pontuação utiliza pacotes que precisam ser importados, importe esses pacotes inline dentro da função para uma serialização adequada.

Alguns pacotes estão disponíveis em default sem a necessidade de uma importação inline. Estes incluem databricks-agents, mlflow-skinny, openai e todos os pacotes incluídos na versão 2 do ambiente sem servidor.

Python
from mlflow.genai.scorers import scorer, ScorerSamplingConfig


# Custom metric: Check if response mentions Databricks
@scorer
def mentions_databricks(outputs):
    """Check if the response mentions Databricks"""
    return "databricks" in str(outputs.get("response", "")).lower()

# Custom metric: Response length check
@scorer(aggregations=["mean", "min", "max"])
def response_length(outputs):
    """Measure response length in characters"""
    return len(str(outputs.get("response", "")))

# Custom metric with multiple inputs
@scorer
def response_relevance_score(inputs, outputs):
    """Score relevance based on keyword matching"""
    query = str(inputs.get("query", "")).lower()
    response = str(outputs.get("response", "")).lower()

    # Simple keyword matching (replace with your logic)
    query_words = set(query.split())
    response_words = set(response.split())

    if not query_words:
        return 0.0

    overlap = len(query_words & response_words)
    return overlap / len(query_words)

# Register and start monitoring custom scorers
databricks_scorer = mentions_databricks.register(name="databricks_mentions")
databricks_scorer = databricks_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.5))

length_scorer = response_length.register(name="response_length")
length_scorer = length_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=1.0))

relevance_scorer = response_relevance_score.register(name="response_relevance_score")  # name must be unique to experiment
relevance_scorer = relevance_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=1.0))

Configuração de vários marcadores​

Para uma configuração abrangente do monitoramento, é possível registrar e iniciar vários avaliadores individualmente.

Python
from mlflow.genai.scorers import Safety, RelevanceToQuery, ScorerSamplingConfig

# Configure multiple scorers for comprehensive monitoring
safety_scorer = Safety().register(name="safety_check")  # name must be unique within an MLflow experiment
safety_scorer = safety_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=1.0))  # Check all traces

relevance_scorer = RelevanceToQuery().register(name="relevance_check")
relevance_scorer = relevance_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.5))  # Sample 50%

length_scorer = response_length.register(name="length_analysis")
length_scorer = length_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.3))

gerenciar pontuadores programados​

Listar os marcadores atuais​

Para view todos os avaliadores registrados para o seu experimento:

Python
from mlflow.genai.scorers import list_scorers

# List all registered scorers
scorers = list_scorers()
for scorer in scorers:
    print(f"Name: {scorer._server_name}")
    print(f"Sample rate: {scorer.sample_rate}")
    print(f"Filter: {scorer.filter_string}")
    print("---")

Atualizar um marcador​

Para modificar as configurações existentes do marcador:

Python
from mlflow.genai.scorers import get_scorer

# Get existing scorer and update its configuration (immutable operation)
safety_scorer = get_scorer(name="safety_monitor")
updated_scorer = safety_scorer.update(sampling_config=ScorerSamplingConfig(sample_rate=0.8))  # Increased from 0.5

# Note: The original scorer remains unchanged; update() returns a new scorer instance
print(f"Original sample rate: {safety_scorer.sample_rate}")  # Original rate
print(f"Updated sample rate: {updated_scorer.sample_rate}")   # New rate

Pare e exclua marcadores​

Para interromper o monitoramento ou remover um avaliador completamente:

Python
from mlflow.genai.scorers import get_scorer, delete_scorer

# Get existing scorer
databricks_scorer = get_scorer(name="databricks_mentions")

# Stop monitoring (sets sample_rate to 0, keeps scorer registered)
stopped_scorer = databricks_scorer.stop()
print(f"Sample rate after stop: {stopped_scorer.sample_rate}")  # 0

# Remove scorer entirely from the server
delete_scorer(name=databricks_scorer.name)

# Or restart monitoring from a stopped scorer
restarted_scorer = stopped_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.5))

Avaliar traços históricos (preenchimento de métricas)​

É possível aplicar retroativamente métricas novas ou atualizadas a rastreamentos históricos.

Preenchimento básico de métricas utilizando taxas de amostragem atuais​

Python
from databricks.agents.scorers import backfill_scorers

safety_scorer = Safety()
safety_scorer.register(name="safety_check")
safety_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.5))

#custom scorer
@scorer(aggregations=["mean", "min", "max"])
def response_length(outputs):
    """Measure response length in characters"""
    return len(outputs)

response_length.register(name="response_length")
response_length.start(sampling_config=ScorerSamplingConfig(sample_rate=0.5))

# Use existing sample rates for specified scorers
job_id = backfill_scorers(
    scorers=["safety_check", "response_length"]
)

Preenchimento de métricas utilizando taxas de amostragem e intervalos de tempo personalizados​

Python
from databricks.agents.scorers import backfill_scorers, BackfillScorerConfig
from datetime import datetime
from mlflow.genai.scorers import Safety, Correctness

safety_scorer = Safety()
safety_scorer.register(name="safety_check")
safety_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.5))

#custom scorer
@scorer(aggregations=["mean", "min", "max"])
def response_length(outputs):
    """Measure response length in characters"""
    return len(outputs)

response_length.register(name="response_length")
response_length.start(sampling_config=ScorerSamplingConfig(sample_rate=0.5))

# Define custom sample rates for backfill
custom_scorers = [
    BackfillScorerConfig(scorer=safety_scorer, sample_rate=0.8),
    BackfillScorerConfig(scorer=response_length, sample_rate=0.9)
]

job_id = backfill_scorers(
    experiment_id=YOUR_EXPERIMENT_ID,
    scorers=custom_scorers,
    start_time=datetime(2024, 6, 1),
    end_time=datetime(2024, 6, 30)
)

Preenchimento de dados recentes​

Python
from datetime import datetime, timedelta

# Backfill last week's data with higher sample rates
one_week_ago = datetime.now() - timedelta(days=7)

job_id = backfill_scorers(
    scorers=[
        BackfillScorerConfig(scorer=safety_scorer, sample_rate=0.8),
        BackfillScorerConfig(scorer=response_length, sample_rate=0.9)
    ],
    start_time=one_week_ago
)

visualizar resultados​

Após programar os avaliadores, aguarde 15 a 20 minutos para o processamento inicial. Então:

  1. Navegue até o seu experimento MLflow.
  2. Abra o Traces tab para visualizar as avaliações anexadas aos traços.
  3. Utilize os painéis de monitoramento para acompanhar as tendências de qualidade.

Melhores práticas​

Estratégia de amostragem​

Equilibre a cobertura com o custo, conforme mostrado nestes exemplos:

Python
# High-priority scorers: higher sampling
safety_scorer = Safety().register(name="safety")
safety_scorer = safety_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=1.0))  # 100% coverage for critical safety

# Expensive scorers: lower sampling
complex_scorer = ComplexCustomScorer().register(name="complex_analysis")
complex_scorer = complex_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.05))  # 5% for expensive operations

Design de marcador personalizado​

Mantenha os marcadores personalizados independentes, conforme mostrado no exemplo a seguir:

Python
@scorer
def well_designed_scorer(inputs, outputs):
    # ✅ All imports inside the function
    import re
    import json

    # ✅ Handle missing data gracefully
    response = outputs.get("response", "")
    if not response:
        return 0.0

    # ✅ Return consistent types
    return float(len(response) > 100)

Solução de problemas​

Os marcadores não estão correndo​

Se os marcadores não estiverem em execução, verifique o seguinte:

  1. Verifique o experimento : certifique-se de que os rastreamentos são registros do experimento, e não de execuções individuais.
  2. Taxa de amostragem : com baixas taxas de amostragem, pode levar algum tempo para ver os resultados.

Problemas de serialização​

Ao criar um marcador personalizado, inclua as importações na definição da função.

Python
# ❌ Avoid external dependencies
import external_library  # Outside function

@scorer
def bad_scorer(outputs):
    return external_library.process(outputs)

# ✅ Include imports in the function definition
@scorer
def good_scorer(outputs):
    import json  # Inside function
    return len(json.dumps(outputs))

# ❌ Avoid using type hints in scorer function signature that requires imports
from typing import List

@scorer
def scorer_with_bad_types(outputs: List[str]):
    return False

problemas de preenchimento de métricas​

" Marcador programado 'X' não encontrado no experimento "

  • Certifique-se de que o nome do marcador corresponda a um marcador registrado em seu experimento
  • Verifique os marcadores disponíveis usando o método list_scorers

Rastros de arquivo​

O senhor pode salvar traços e suas avaliações associadas em uma tabela Delta do Unity Catalog para armazenamento de longo prazo e análise avançada. Isso é útil para criar painéis personalizados, realizar análises detalhadas dos dados de rastreamento e manter um registro duradouro do comportamento do seu aplicativo.

nota

O senhor deve ter as permissões necessárias para gravar na tabela Delta do Unity Catalog especificada. A tabela de destino será criada se ainda não existir.

Se a tabela já existir, os traços serão anexados a ela.

Habilitar rastreamentos de arquivamento​

Para começar a arquivar rastreamentos para um experimento, use a função enable_databricks_trace_archival. O senhor deve especificar o nome completo da tabela Delta de destino, incluindo o catálogo e o esquema. Se você não fornecer um experiment_id, o arquivamento de rastreamentos será habilitado para o experimento atualmente ativo.

Python
from mlflow.tracing.archival import enable_databricks_trace_archival

# Archive traces from a specific experiment to a Unity Catalog Delta table
enable_databricks_trace_archival(
    delta_table_fullname="my_catalog.my_schema.archived_traces",
    experiment_id="YOUR_EXPERIMENT_ID",
)

Desativar rastreamentos de arquivamento​

Você pode parar de arquivar rastreamentos para um experimento a qualquer momento usando a função disable_databricks_trace_archival.

Python
from mlflow.tracing.archival import disable_databricks_trace_archival

# Stop archiving traces for the specified experiment
disable_databricks_trace_archival(experiment_id="YOUR_EXPERIMENT_ID")

Próximos passos​

Continue sua jornada com o tutorial a seguir.

Guia de referência​

Explore a documentação detalhada sobre os conceitos e recursos mencionados neste guia.

Última atualização em