Alinhar juízes com humanos

O alinhamento de juízes ensina os juízes do LLM a corresponderem aos padrões de avaliação humana por meio de feedback sistemático. Esse processo transforma avaliadores genéricos em especialistas específicos da área, que entendem seus critérios de qualidade exclusivos, melhorando a concordância com as avaliações humanas em 30 a 50% em comparação com os avaliadores de referência.

O mesmo fluxo de trabalho de alinhamento se aplica tanto aos juízes integrados (como RelevanceToQuery, Safety ou Correctness) quanto aos juízes personalizados criados com make_judge(). Use o alinhamento com juízes integrados para adaptar seus critérios genéricos ao seu domínio, ou com juízes personalizados para refinar a lógica de avaliação especializada.

O alinhamento dos juízes segue um fluxo de trabalho de três etapas:

1. Gerar Avaliações Iniciais : utilize um juiz integrado ou personalizado para avaliar rastreamentos e estabelecer uma linha de base.
2. Coletar feedback humano : especialistas do domínio revisam e corrigem as avaliações dos juízes.
3. Alinhamento e Implantação : invocar o método align() do juiz para criar um novo juiz que esteja mais alinhado com o feedback humano.

O sistema oferece suporte aos otimizadores que estão disponíveis no pacote mlflow.genai.judges.optimizers.

Requisitos

• MLflow 3.4.0 ou acima para usar o recurso de alinhamento de juiz

  Python

  %pip install --upgrade "mlflow[databricks]>=3.4.0" databricks_openai dspy
  dbutils.library.restartPython()

• Um juiz para alinhar. Isso pode ser um juiz integrado (por exemplo, RelevanceToQuery ou Correctness) ou um juiz personalizado criado com make_judge().

• O nome da avaliação de feedback humano deve corresponder exatamente ao atributo name do avaliador. Para juízes integrados, este é o nome default em snake_case (por exemplo, relevance_to_query para RelevanceToQuery) a menos que você o substitua passando name= ao instanciar a classe. Para juízes personalizados, é o name que foi passado para make_judge() (por exemplo, product_quality).

• O alinhamento não é compatível para juízes de nível de sessão (interações múltiplas), como ConversationCompleteness.

Passo 1: configurar o juiz e gerar rastreamentos

Configurar o juiz inicial e gerar rastreamentos com avaliações. Você pode alcançar um alinhamento razoável com pelo menos 10 rastreamentos, mas 50-100 rastreamentos produzem resultados melhores.

Built-in judge

Instancie um juiz integrado diretamente. Juízes integrados expõem um atributo name (o default é uma string snake_case como relevance_to_query) que será usado ao registrar feedback humano no Passo 2.

Python

from mlflow.genai.scorers import RelevanceToQuery
import mlflow

# Create or set an MLflow experiment for alignment.
# Use a workspace path such as /Shared/<name> or /Users/<your-email>/<name>.
experiment = mlflow.set_experiment("/Shared/relevance-alignment")
experiment_id = experiment.experiment_id

# Use a built-in judge
initial_judge = RelevanceToQuery()

Custom judge

Crie um juiz personalizado com make_judge(). O argumento name é o mesmo nome que será usado ao registrar feedback humano no passo 2.

Python

from mlflow.genai.judges import make_judge
import mlflow

# Create or set an MLflow experiment for alignment.
# Use a workspace path such as /Shared/<name> or /Users/<your-email>/<name>.
experiment = mlflow.set_experiment("/Shared/product-quality-alignment")
experiment_id = experiment.experiment_id

# Create initial judge with template-based evaluation
initial_judge = make_judge(
    name="product_quality",
    instructions=(
        "Evaluate if the product description in {{ outputs }} "
        "is accurate and helpful for the query in {{ inputs }}. "
        "Rate as: excellent, good, fair, or poor"
    ),
    model="databricks:/databricks-gpt-oss-120b",
)

Defina a lógica do seu aplicativo. O exemplo a seguir usa um modelo de base hospedado pelo Databricks para gerar uma descrição do produto a partir de uma consulta. Substitua isto pelo seu próprio código de aplicativo:

Python

import mlflow
from databricks_openai import DatabricksOpenAI

# Enable automatic tracing of OpenAI calls
mlflow.openai.autolog()

# Create an OpenAI client connected to Databricks-hosted LLMs
client = DatabricksOpenAI()
model_name = "databricks-claude-sonnet-4"


def generate_product_description(query: str) -> str:
    response = client.chat.completions.create(
        model=model_name,
        messages=[
            {
                "role": "system",
                "content": "You write concise, accurate product descriptions.",
            },
            {"role": "user", "content": query},
        ],
    )
    return response.choices[0].message.content

Gerar rastreamentos e executar o juiz. Use o atributo name do avaliador (por exemplo, relevance_to_query para o avaliador integrado acima, ou product_quality para o avaliador personalizado acima) como o feedback name:

Python

# Generate traces for alignment (minimum 10, recommended 50+)
for i in range(50):
    query = f"Tell me about product {i}"
    description = generate_product_description(query)

    # Retrieve the ID of the most recent finished trace
    trace_id = mlflow.get_last_active_trace_id()
    trace = mlflow.get_trace(trace_id)

    # Generate judge assessment
    judge_result = initial_judge(trace=trace)

    # Log judge feedback to the trace using the judge's name
    mlflow.log_feedback(
        trace_id=trace_id,
        name=initial_judge.name,
        value=judge_result.value,
        rationale=judge_result.rationale,
    )

a etapa 2: Coletar feedback humano

Reúna feedback humano para ensinar ao avaliador seus padrões de qualidade. Escolha uma das seguintes abordagens:

Databricks UI review

Coletar feedback humano quando:

• É preciso que especialistas do domínio revisem as saídas.
• O objetivo é refinar iterativamente os critérios de feedback.
• Está-se trabalhando com um dataset menor (< 100 exemplos).

Utilize a interface do usuário do MLflow para revisar manualmente e fornecer feedback:

1. Navegue até seu experimento MLflow no workspace do Databricks.
2. Clique na **tab Rastreios** para ver os rastreamentos.
3. Revise cada rastreamento e a sua avaliação do juiz.
4. Adicionar feedback humano usando a interface de feedback da UI.
5. Certifique-se de que o nome do feedback corresponda exatamente ao atributo name do seu juiz (por exemplo, relevance_to_query para uma instância RelevanceToQuery integrada ou product_quality para o juiz personalizado acima).

Programmatic feedback

Utilize o feedback programático quando:

• Existem rótulos de verdade fundamental pré-existentes.
• Ao trabalhar com grandes conjuntos de dados (mais de 100 exemplos).
• É preciso coletar feedback reproduzível.

Se você já possui rótulos de referência, log os programaticamente:

Python

from mlflow.entities import AssessmentSource, AssessmentSourceType

# Your ground truth data
ground_truth_data = [
    {"trace_id": "<trace_id_1>", "label": "excellent", "rationale": "Comprehensive and accurate description"},
    {"trace_id": "<trace_id_2>", "label": "poor", "rationale": "Missing key product features"},
    {"trace_id": "<trace_id_3>", "label": "good", "rationale": "Accurate but could be more detailed"},
    # ... more ground truth labels
]

# Log human feedback for each trace
for item in ground_truth_data:
    mlflow.log_feedback(
        trace_id=item["trace_id"],
        name=initial_judge.name,  # Must match judge name (built-in or custom)
        value=item["label"],
        rationale=item.get("rationale", ""),
        source=AssessmentSource(
            source_type=AssessmentSourceType.HUMAN,
            source_id="ground_truth_dataset"
        ),
    )

Melhores práticas para coleta de feedback

• Revisores diversos : Inclua vários especialistas da área para capturar perspectivas variadas.
• Exemplos equilibrados : Inclua pelo menos 30% de exemplos negativos (classificações ruins/regulares).
• Justificativas claras : Forneça explicações detalhadas para as classificações.
• Exemplos representativos : Abrangem casos extremos e cenários comuns.

o passo 3: Alinhar e registrar o juiz

Quando houver feedback humano suficiente, alinhe o juiz. O mesmo método align() é usado tanto para juízes integrados quanto para juízes personalizados.

Default optimizer (recommended)

Ao chamar align() sem especificar um otimizador, o otimizador MemAlign é usado automaticamente:

Python

# Retrieve traces with both judge and human assessments
traces_for_alignment = mlflow.search_traces(
    experiment_ids=[experiment_id],
    max_results=100,
    return_type="list"
)

if len(traces_for_alignment) >= 10:
    # Align the judge based on human feedback using the default optimizer
    aligned_judge = initial_judge.align(traces_for_alignment)

    # Register the aligned judge for production use.
    # Use a new name to distinguish it from the original judge.
    aligned_judge.register(
        experiment_id=experiment_id,
        name=f"{initial_judge.name}_aligned",
        tags={"alignment_date": "2025-10-23", "num_traces": str(len(traces_for_alignment))}
    )

    print(f"Successfully aligned judge using {len(traces_for_alignment)} traces")
else:
    print(f"Insufficient traces for alignment. Found {len(traces_for_alignment)}, need at least 10")

Explicit optimizer

Python

from mlflow.genai.judges.optimizers import MemAlignOptimizer

# Retrieve traces with both judge and human assessments
traces_for_alignment = mlflow.search_traces(
    experiment_ids=[experiment_id], max_results=15, return_type="list"
)

# Align the judge using human corrections (minimum 10 traces recommended)
if len(traces_for_alignment) >= 10:
    # Explicitly specify optimizer with custom model configuration
    optimizer = MemAlignOptimizer(model="databricks:/databricks-gpt-oss-120b")
    aligned_judge = initial_judge.align(traces_for_alignment, optimizer)

    # Register the aligned judge
    aligned_judge.register(experiment_id=experiment_id)
    print("Judge aligned successfully with human feedback")
else:
    print(f"Need at least 10 traces for alignment, have {len(traces_for_alignment)}")

Ativar registro detalhado

Para monitorar o processo de alinhamento, ative o log de depuração para o otimizador:

Python

import logging

# Enable detailed logging
logging.getLogger("mlflow.genai.judges.optimizers.memalign").setLevel(logging.DEBUG)

# Run alignment with verbose output
aligned_judge = initial_judge.align(traces_for_alignment)

Validar alinhamento

Confirme se o alinhamento melhorou o juiz:

Python

def test_alignment_improvement(
    original_judge, aligned_judge, test_traces: list
) -> dict:
    """Compare judge performance before and after alignment."""

    original_correct = 0
    aligned_correct = 0

    for trace in test_traces:
        # Get human ground truth from trace assessments
        feedbacks = trace.search_assessments(type="feedback")
        human_feedback = next(
            (f for f in feedbacks if f.source.source_type == "HUMAN"), None
        )

        if not human_feedback:
            continue

        # Get judge evaluations
        # Judges can evaluate entire traces instead of individual inputs/outputs
        original_eval = original_judge(trace=trace)
        aligned_eval = aligned_judge(trace=trace)

        # Check agreement with human
        if original_eval.value == human_feedback.value:
            original_correct += 1
        if aligned_eval.value == human_feedback.value:
            aligned_correct += 1

    total = len(test_traces)
    return {
        "original_accuracy": original_correct / total,
        "aligned_accuracy": aligned_correct / total,
        "improvement": (aligned_correct - original_correct) / total,
    }

Crie otimizadores de alinhamento personalizados

Para estratégias de alinhamento especializadas, estenda a classe base AlignmentOptimizer :

Python

from mlflow.genai.judges.base import AlignmentOptimizer, Judge
from mlflow.entities.trace import Trace

class MyCustomOptimizer(AlignmentOptimizer):
    """Custom optimizer implementation for judge alignment."""

    def __init__(self, model: str = None, **kwargs):
        """Initialize your optimizer with custom parameters."""
        self.model = model
        # Add any custom initialization logic

    def align(self, judge: Judge, traces: list[Trace]) -> Judge:
        """
        Implement your alignment algorithm.

        Args:
            judge: The judge to be optimized
            traces: List of traces containing human feedback

        Returns:
            A new Judge instance with improved alignment
        """
        # Your custom alignment logic here
        # 1. Extract feedback from traces
        # 2. Analyze disagreements between judge and human
        # 3. Generate improved instructions
        # 4. Return new judge with better alignment

        # Example: Return judge with modified instructions
        from mlflow.genai.judges import make_judge

        improved_instructions = self._optimize_instructions(judge.instructions, traces)

        return make_judge(
            name=judge.name,
            instructions=improved_instructions,
            model=judge.model,
        )

    def _optimize_instructions(self, instructions: str, traces: list[Trace]) -> str:
        """Your custom optimization logic."""
        # Implement your optimization strategy
        pass

# Create your custom optimizer
custom_optimizer = MyCustomOptimizer(model="your-model")

# Use it for alignment
aligned_judge = initial_judge.align(traces_with_feedback, custom_optimizer)

Limitações

• O alinhamento de juízes não suporta a avaliação baseada no agente ou na expectativa.

Próximos passos

• Saiba mais sobre monitoramento de produção para implantar juízes alinhados em escala.
• Consulte avaliadores baseados em código para métricas determinísticas complementares.
• Saiba mais sobre a criação de juízes personalizados neste blog da Databricks.