Criação de marcadores personalizados

Os marcadores personalizados oferecem a máxima flexibilidade para definir com precisão como a qualidade do seu aplicativo GenAI é medida. Os pontuadores personalizados oferecem flexibilidade para definir métricas de avaliação adaptadas ao seu caso de uso comercial específico, seja com base em heurística simples, lógica avançada ou avaliações programáticas.

Use pontuadores personalizados para os seguintes cenários:

1. Definir uma métrica de avaliação heurística ou baseada em código personalizada.
2. Personalize como os dados do rastreamento do seu aplicativo são mapeados para os juízes LLM baseados em pesquisas da Databricks nos avaliadores LLM predefinidos.
3. Criando um LLM juiz do com um texto de prompt personalizado usando o LLM scorers artigos baseado em prompt.
4. Utilizar seu próprio modelo LLM (em vez de um modelo LLM judge hospedado pela Databricks) para avaliação.
5. Qualquer outro caso de uso em que você precise de mais flexibilidade e controle do que os fornecidos pelas abstrações predefinidas.

nota

Consulte a página de conceito do avaliador ou os documentos da API para obter uma referência detalhada sobre as interfaces personalizadas do avaliador.

Visão geral do uso

Os avaliadores personalizados são escritos em Python e oferecem controle total para avaliar quaisquer dados dos rastreamentos do seu aplicativo. Um único avaliador personalizado funciona tanto no evaluate(...) harness para avaliação off-line quanto para monitoramento da produção. Depois de definir um marcador personalizado, você pode usá-lo exatamente como você usaria um marcador predefinido.

Os seguintes tipos de saídas são suportados:

• Strings de aprovação/reprovação: os valores das strings "yes" or "no" são renderizados como "Pass" ou "Fail" na interface do usuário.
• Valor numérico: valores ordinais: números inteiros ou flutuantes.
• Boolean Valor: True ou False.
• Objeto de feedback: retorne um objetoFeedback com pontuação, justificativa e metadados adicionais

Como entrada, os marcadores personalizados têm acesso a:

• O rastreamento completo do MLflow, incluindo períodos, atributos e saídas. O rastreamento é passado para o marcador personalizado como uma classe mlflow.entities.trace instanciada.
• O dicionário inputs, derivado do conjunto de dados de entrada ou dos pós-processamentos do MLflow do seu rastreamento.
• O valor outputs, derivado da entrada dataset ou do rastreamento. Se predict_fn for fornecido, o valor outputs será o retorno do predict_fn.
• O dicionário expectations, derivado do campo expectations na entrada dataset, ou avaliações associadas ao rastreamento.

O decorador @scorer permite que os usuários definam métricas de avaliação personalizadas que podem ser passadas para o mlflow.genai.evaluate() usando o argumento scorers ou usado para o monitoramento da produção.

A função scorer é chamada com argumentos nomeados com base na assinatura abaixo. Todos os argumentos nomeados são opcionais para que você possa usar qualquer combinação. Por exemplo, você pode definir um marcador que tenha apenas inputs e trace como argumentos e omitir outputs e expectations:

Python

from mlflow.genai.scorers import scorer
from typing import Optional, Any
from mlflow.entities import Feedback

@scorer
def my_custom_scorer(
  *,  # evaluate(...) harness will always call your scorer with named arguments
  inputs: Optional[dict[str, Any]],  # The agent's raw input, parsed from the Trace or dataset, as a Python dict
  outputs: Optional[Any],  # The agent's raw output, parsed from the Trace or
  expectations: Optional[dict[str, Any]],  # The expectations passed to evaluate(data=...), as a Python dict
  trace: Optional[mlflow.entities.Trace] # The app's resulting Trace containing spans and other metadata
) -> int | float | bool | str | Feedback | list[Feedback]

Abordagem personalizada de desenvolvimento de marcadores

À medida que desenvolve métricas, o senhor precisa iterar rapidamente as métricas sem ter que executar o aplicativo sempre que fizer uma alteração no marcador. Para fazer isso, recomendamos as seguintes etapas:

Etapa 1: Defina suas métricas iniciais, o aplicativo e os dados de avaliação

Python

import mlflow
from mlflow.entities import Trace
from mlflow.genai.scorers import scorer
from typing import Any

@mlflow.trace
def my_app(input_field_name: str):
    return {'output': input_field_name+'_output'}

@scorer
def my_metric() -> int:
    # placeholder return value
    return 1

eval_set = [{'inputs': {'input_field_name': 'test'}}]

Etapa 2: gere rastreamentos do seu aplicativo usando evaluate()

Python

eval_results = mlflow.genai.evaluate(
    data=eval_set,
    predict_fn=my_app,
    scorers=[dummy_metric]
)

Etapa 3: consultar e armazenar os traços resultantes

Python

generated_traces = mlflow.search_traces(run_id=eval_results.run_id)

Etapa 4: Passe os traços resultantes como entrada para evaluate() enquanto o senhor itera em suas métricas

A função search_traces retorna um objeto Pandas DataFrame de traços, que pode ser passado diretamente para evaluate() como um dataset e de entrada. Isso permite que você itere rapidamente suas métricas sem precisar reexecutar seu aplicativo.

Python

@scorer
def my_metric(outputs: Any):
    # Implement the actual metric logic here.
    return outputs == "test_output"

# Note the lack of a predict_fn parameter
mlflow.genai.evaluate(
    data=generated_traces,
    scorers=[my_metric],
)

Exemplos de pontuadores personalizados

Esta seção apresenta diferentes abordagens para criar pontuadores personalizados.

Pré-requisito: criar um aplicativo de amostra e obter uma cópia local dos rastreamentos

Em todas as abordagens, usamos o aplicativo de amostra abaixo e uma cópia dos rastreamentos (extraídos usando a abordagem acima).

1. Inicialize um cliente OpenAI para se conectar a LLMs hospedados pela Databricks ou LLMs hospedados pela OpenAI.

Databricks-hosted LLMs

Use o MLflow para obter um cliente OpenAI que se conecta aos LLMs hospedados pela Databricks. Selecione um modelo dentre os modelos de base disponíveis.

Python

import mlflow
from databricks.sdk import WorkspaceClient

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

# Set up MLflow tracking to Databricks
mlflow.set_tracking_uri("databricks")
mlflow.set_experiment("/Shared/docs-demo")

# Create an OpenAI client that is connected to Databricks-hosted LLMs
w = WorkspaceClient()
client = w.serving_endpoints.get_open_ai_client()

# Select an LLM
model_name = "databricks-claude-sonnet-4"

OpenAI-hosted LLMs

Use o SDK nativo do OpenAI para se conectar a modelos hospedados pelo OpenAI. Selecione um modelo dentre os modelos OpenAI disponíveis.

Python

import mlflow
import os
import openai

# Ensure your OPENAI_API_KEY is set in your environment
# os.environ["OPENAI_API_KEY"] = "<YOUR_API_KEY>" # Uncomment and set if not globally configured

# Enable auto-tracing for OpenAI
mlflow.openai.autolog()

# Set up MLflow tracking to Databricks
mlflow.set_tracking_uri("databricks")
mlflow.set_experiment("/Shared/docs-demo")

# Create an OpenAI client connected to OpenAI SDKs
client = openai.OpenAI()

# Select an LLM
model_name = "gpt-4o-mini"

2. Gere traços usando um marcador simples.

Python

from typing import Any
from mlflow.entities import Trace
from mlflow.genai.scorers import scorer

@mlflow.trace
def sample_app(messages: list[dict[str, str]]):
    # 1. Prepare messages for the LLM
    messages_for_llm = [
        {"role": "system", "content": "You are a helpful assistant."},
        *messages,
    ]

    # 2. Call LLM to generate a response
    response = client.chat.completions.create(
        model= model_name,  # Select a model
        messages=messages_for_llm,
    )
    return response.choices[0].message.content


# Create a list of messages for the LLM to generate a response
eval_dataset = [
    {
        "inputs": {
            "messages": [
                {"role": "user", "content": "How much does a microwave cost?"},
            ]
        },
    },
    {
        "inputs": {
            "messages": [
                {
                    "role": "user",
                    "content": "Can I return the microwave I bought 2 months ago?",
                },
            ]
        },
    },
    {
        "inputs": {
            "messages": [
                {
                    "role": "user",
                    "content": "I'm having trouble with my account.  I can't log in.",
                },
                {
                    "role": "assistant",
                    "content": "I'm sorry to hear that you're having trouble with your account.  Are you using our website or mobile app?",
                },
                {"role": "user", "content": "Website"},
            ]
        },
    },
]


@scorer
def dummy_metric():
    # This scorer is just to help generate initial traces.
    return 1


# Generate initial traces by running the sample_app.
# The results, including traces, are logged to the MLflow experiment defined above.
initial_eval_results = mlflow.genai.evaluate(
    data=eval_dataset, predict_fn=sample_app, scorers=[dummy_metric]
)

generated_traces = mlflow.search_traces(run_id=initial_eval_results.run_id)

Depois de executar o código acima, o senhor deve ter três traços em seu experimento.

Exemplo 1: Acesse dados do rastreamento

Acesse o objeto completo do MLflow Trace para usar vários detalhes (períodos, entradas, saídas, atributos, tempo) para o cálculo de métricas refinadas.

nota

O generated_traces da seção de pré-requisitos será usado como dados de entrada para esses exemplos.

Esse marcador verifica se o tempo total de execução do rastreamento está dentro de uma faixa aceitável.

Python

import mlflow
from mlflow.genai.scorers import scorer
from mlflow.entities import Trace, Feedback, SpanType

@scorer
def llm_response_time_good(trace: Trace) -> Feedback:
    # Search particular span type from the trace
    llm_span = trace.search_spans(span_type=SpanType.CHAT_MODEL)[0]

    response_time = (llm_span.end_time_ns - llm_span.start_time_ns) / 1e9 # second
    max_duration = 5.0
    if response_time <= max_duration:
        return Feedback(
            value="yes",
            rationale=f"LLM response time {response_time:.2f}s is within the {max_duration}s limit."
        )
    else:
        return Feedback(
            value="no",
            rationale=f"LLM response time {response_time:.2f}s exceeds the {max_duration}s limit."
        )

# Evaluate the scorer using the pre-generated traces from the prerequisite code block.
span_check_eval_results = mlflow.genai.evaluate(
    data=generated_traces,
    scorers=[llm_response_time_good]
)

Exemplo 2: Envolver um juiz LLM predefinido

Crie um avaliador personalizado que envolva os juízes LLM predefinidos do MLflow. Use isso para pré-processar dados de rastreamento para o juiz ou pós-processar seu feedback.

Este exemplo demonstra como agrupar o juiz is_context_relevant que avalia se o contexto fornecido é relevante para a consulta, para avaliar se a resposta do assistente é relevante para a consulta do usuário.

Python

import mlflow
from mlflow.entities import Trace, Feedback
from mlflow.genai.judges import is_context_relevant
from mlflow.genai.scorers import scorer
from typing import Any

# Assume `generated_traces` is available from the prerequisite code block.

@scorer
def is_message_relevant(inputs: dict[str, Any], outputs: str) -> Feedback:
    # The `inputs` field for `sample_app` is a dictionary like: {"messages": [{"role": ..., "content": ...}, ...]}
    # We need to extract the content of the last user message to pass to the relevance judge.

    last_user_message_content = None
    if "messages" in inputs and isinstance(inputs["messages"], list):
        for message in reversed(inputs["messages"]):
            if message.get("role") == "user" and "content" in message:
                last_user_message_content = message["content"]
                break

    if not last_user_message_content:
        raise Exception("Could not extract the last user message from inputs to evaluate relevance.")

    # Call the `relevance_to_query judge. It will return a Feedback object.
    return is_context_relevant(
        request=last_user_message_content,
        context={"response": outputs},
    )

# Evaluate the custom relevance scorer
custom_relevance_eval_results = mlflow.genai.evaluate(
    data=generated_traces,
    scorers=[is_message_relevant]
)

Exemplo 3: Uso expectations

Quando mlflow.genai.evaluate() é chamado com um argumento data que é uma lista de dicionários ou um Pandas DataFrame, cada linha pode conter um expectations key. O valor associado a este key é transmitido diretamente para o seu avaliador personalizado.

Este exemplo também mostra o uso de um marcador personalizado junto com um marcador predefinido (neste exemplo, o marcador de segurança).

Python

import mlflow
from mlflow.entities import Feedback
from mlflow.genai.scorers import scorer, Safety
from typing import Any, List, Optional, Union

expectations_eval_dataset_list = [
    {
        "inputs": {"messages": [{"role": "user", "content": "What is 2+2?"}]},
        "expectations": {
            "expected_response": "2+2 equals 4.",
            "expected_keywords": ["4", "four", "equals"],
        }
    },
    {
        "inputs": {"messages": [{"role": "user", "content": "Describe MLflow in one sentence."}]},
        "expectations": {
            "expected_response": "MLflow is an open-source platform to streamline machine learning development, including tracking experiments, packaging code into reproducible runs, and sharing and deploying models.",
            "expected_keywords": ["mlflow", "open-source", "platform", "machine learning"],
        }
    },
    {
        "inputs": {"messages": [{"role": "user", "content": "Say hello."}]},
        "expectations": {
            "expected_response": "Hello there!",
            # No keywords needed for this one, but the field can be omitted or empty
        }
    }
]

Exemplo 3.1: Correspondência exata com a resposta esperada

Esse marcador verifica se a resposta do assistente corresponde exatamente à expected_response fornecida no expectations.

Python

@scorer
def exact_match(outputs: str, expectations: dict[str, Any]) -> bool:
    # Scorer can return primitive value like bool, int, float, str, etc.
    return outputs == expectations["expected_response"]

exact_match_eval_results = mlflow.genai.evaluate(
    data=expectations_eval_dataset_list,
    predict_fn=sample_app, # sample_app is from the prerequisite section
    scorers=[exact_match, Safety()]  # You can include any number of scorers
)

Exemplo 3.2: Verificação de palavras-chave a partir das expectativas

Esse marcador verifica se todos os expected_keywords do expectations estão presentes na resposta do assistente.

Python

@scorer
def keyword_presence_scorer(outputs: str, expectations: dict[str, Any]) -> Feedback:
    expected_keywords = expectations.get("expected_keywords")
    print(expected_keywords)
    if expected_keywords is None:
        return Feedback(
            score=None, # Undetermined, as no keywords were expected
            rationale="No 'expected_keywords' provided in expectations."
        )

    missing_keywords = []
    for keyword in expected_keywords:
        if keyword.lower() not in outputs.lower():
            missing_keywords.append(keyword)

    if not missing_keywords:
        return Feedback(value="yes", rationale="All expected keywords are present in the response.")
    else:
        return Feedback(value="no", rationale=f"Missing keywords: {', '.join(missing_keywords)}.")

keyword_presence_eval_results = mlflow.genai.evaluate(
    data=eval_dataset,
    predict_fn=sample_app, # sample_app is from the prerequisite section
    scorers=[keyword_presence_scorer]
)

Exemplo 4: Retornar vários objetos de feedback

Um único pontuador pode retornar uma lista de objetos Feedback, permitindo que um pontuador avalie várias facetas de qualidade (por exemplo, PII, sentimento, concisão) simultaneamente. Cada objeto Feedback deve, idealmente, possuir um nome de métrica exclusivo ( name ), que se tornará o nome da métrica nos resultados; caso contrário, eles poderão sobrescrever-se mutuamente se os nomes forem gerados automaticamente e entrarem em conflito. Se um nome não for fornecido, o MLflow tentará gerar um com base no nome da função do avaliador e em um índice.

Este exemplo demonstra um marcador que retorna dois comentários distintos para cada rastreamento:

1. is_not_empty_check: Um booleano que indica se o conteúdo da resposta é não vazio.
2. response_char_length: um valor numérico para o tamanho do caractere da resposta.

Python

import mlflow
from mlflow.genai.scorers import scorer
from mlflow.entities import Feedback, Trace # Ensure Feedback and Trace are imported
from typing import Any, Optional

# Assume `generated_traces` is available from the prerequisite code block.

@scorer
def comprehensive_response_checker(outputs: str) -> list[Feedback]:
    feedbacks = []
    # 1. Check if the response is not empty
    feedbacks.append(
        Feedback(name="is_not_empty_check", value="yes" if outputs != "" else "no")
    )
    # 2. Calculate response character length
    char_length = len(outputs)
    feedbacks.append(Feedback(name="response_char_length", value=char_length))
    return feedbacks

multi_feedback_eval_results = mlflow.genai.evaluate(
    data=generated_traces,
    scorers=[comprehensive_response_checker]
)

O resultado terá duas colunas: is_not_empty_check e response_char_length como avaliações.

Exemplo 5: Utilize seu próprio LLM para um juiz

Integre um LLM personalizado ou hospedado externamente em um avaliador. O pontuador lida com chamadas de API, formatação de entrada/saída e gera Feedback a partir da resposta do seu LLM, oferecendo controle total sobre o processo de avaliação.

O senhor também pode definir o campo source no objeto Feedback para indicar que a fonte da avaliação é um juiz do LLM.

Python

import mlflow
import json
from mlflow.genai.scorers import scorer
from mlflow.entities import AssessmentSource, AssessmentSourceType, Feedback
from typing import Any, Optional


# Assume `generated_traces` is available from the prerequisite code block.
# Assume `client` (OpenAI SDK client configured for Databricks) is available from the prerequisite block.
# client = OpenAI(...)

# Define the prompts for the Judge LLM.
judge_system_prompt = """
You are an impartial AI assistant responsible for evaluating the quality of a response generated by another AI model.
Your evaluation should be based on the original user query and the AI's response.
Provide a quality score as an integer from 1 to 5 (1=Poor, 2=Fair, 3=Good, 4=Very Good, 5=Excellent).
Also, provide a brief rationale for your score.

Your output MUST be a single valid JSON object with two keys: "score" (an integer) and "rationale" (a string).
Example:
{"score": 4, "rationale": "The response was mostly accurate and helpful, addressing the user's query directly."}
"""
judge_user_prompt = """
Please evaluate the AI's Response below based on the Original User Query.

Original User Query:
```{user_query}```

AI's Response:
```{llm_response_from_app}```

Provide your evaluation strictly as a JSON object with "score" and "rationale" keys.
"""

@scorer
def answer_quality(inputs: dict[str, Any], outputs: str) -> Feedback:
    user_query = inputs["messages"][-1]["content"]

    # Call the Judge LLM using the OpenAI SDK client.
    judge_llm_response_obj = client.chat.completions.create(
        model="databricks-claude-3-7-sonnet",  # This example uses Databricks hosted Claude. If you provide your own OpenAI credentials, replace with a valid OpenAI model e.g., gpt-4o-mini, etc.
        messages=[
            {"role": "system", "content": judge_system_prompt},
            {"role": "user", "content": judge_user_prompt.format(user_query=user_query, llm_response_from_app=outputs)},
        ],
        max_tokens=200,  # Max tokens for the judge's rationale
        temperature=0.0, # For more deterministic judging
    )
    judge_llm_output_text = judge_llm_response_obj.choices[0].message.content

    # Parse the Judge LLM's JSON output.
    judge_eval_json = json.loads(judge_llm_output_text)
    parsed_score = int(judge_eval_json["score"])
    parsed_rationale = judge_eval_json["rationale"]

    return Feedback(
        value=parsed_score,
        rationale=parsed_rationale,
        # Set the source of the assessment to indicate the LLM judge used to generate the feedback
        source=AssessmentSource(
            source_type=AssessmentSourceType.LLM_JUDGE,
            source_id="claude-3-7-sonnet",
        )
    )


# Evaluate the scorer using the pre-generated traces.
custom_llm_judge_eval_results = mlflow.genai.evaluate(
    data=generated_traces,
    scorers=[answer_quality]
)

Ao abrir o rastreamento na interface do usuário e clicar na avaliação " answer_quality ", você pode ver os metadados do juiz, como justificativa, data e hora, nome do modelo do juiz e assim por diante. Se a avaliação do juiz não estiver correta, você pode anular a pontuação clicando no botão Edit.

A nova avaliação substitui a avaliação original do juiz. A história foi editada e preservada para referência futura.

Próximas etapas

Continue sua jornada com as seguintes ações e o tutorial.

• Avalie com avaliadores LLM personalizados - Crie avaliações semânticas utilizando LLMs.
• Execução dos avaliadores na produção - Implante seus avaliadores para monitoramento contínuo.
• Crie um conjunto de dados de avaliação - Crie dados de teste para seus avaliadores.

Guia de referência

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

• Marcadores - Mergulhe profundamente em como os marcadores trabalham e em sua arquitetura.
• Evaluation Harness - Entenda como mlflow.genai.evaluate() usa seus marcadores.
• LLM juízes - Aprenda os fundamentos da avaliação baseada n AI.