Alinear los evaluadores de LLM con la retroalimentación humana

La alineación de evaluadores enseña a los evaluadores de LLM a ajustarse a los estándares de evaluación humanos mediante retroalimentación sistemática. Este proceso transforma los evaluadores genéricos en expertos específicos del dominio que comprenden sus criterios de calidad únicos, mejorando el acuerdo con las evaluaciones humanas en un 30 al 50 % en comparación con los jueces de línea base.

El mismo flujo de trabajo de alineación se aplica a los jueces integrados (como RelevanceToQuery, Safetyo Correctness) y a los jueces personalizados creados con make_judge(). Use la alineación con los jueces integrados para adaptar sus criterios genéricos al dominio o con jueces personalizados para refinar la lógica de evaluación especializada.

La alineación del juez sigue un flujo de trabajo de tres pasos:

  1. Generar evaluaciones iniciales: Use un evaluador integrado o personalizado para evaluar trazas y establecer una referencia.
  2. Recopilar comentarios humanos: los expertos en dominio revisan y corrigen las evaluaciones de los jueces.
  3. Alinear y desplegar: Invoque el método align() del juez para crear un nuevo juez más alineado con la retroalimentación humana.

El sistema admite los optimizadores que están disponibles en el paquete mlflow.genai.judges.optimizers.

Requisitos

  • MLflow 3.4.0 o superior para utilizar las funciones de alineación de evaluadores

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

  • Un evaluador que se va a alinear. Puede ser un juez integrado (por ejemplo, RelevanceToQuery o ) o Correctnessun juez personalizado creado con make_judge().

  • El nombre de la evaluación de retroalimentación humana debe coincidir exactamente con el atributo name del evaluador. Para los evaluadores integrados, se trata del nombre predeterminado en snake_case (por ejemplo, relevance_to_query para RelevanceToQuery), a menos que lo sobrescribas pasando name= al instanciar la clase. En el caso de los evaluadores personalizados, es el name que has pasado a make_judge() (por ejemplo, product_quality).

  • La alineación no es compatible con los evaluadores a nivel de sesión (de varios turnos), como ConversationCompleteness.

Paso 1: Configurar el juez y generar trazas

Configura tu evaluador inicial y genera trazas con evaluaciones. Se puede lograr una alineación razonable con al menos 10 trazas, pero entre 50 y 100 trazas ofrecen mejores resultados.

Evaluador integrado

Instancia directamente un evaluador integrado. Los jueces integrados exponen un atributo name (el valor predeterminado es una cadena en snake_case como relevance_to_query) que usarás al registrar retroalimentación humana en el Paso 2.

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()

Juez personalizado

Cree un juez personalizado con make_judge(). El name argumento es el mismo nombre que usará al registrar comentarios humanos en el paso 2.

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 la lógica de la aplicación. En el ejemplo siguiente se usa un modelo de base hospedado en Databricks para generar una descripción del producto a partir de una consulta. Reemplace esto por su propio código de aplicación:

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

Genera trazas y ejecuta el evaluador. Utiliza el atributo name del evaluador (por ejemplo, relevance_to_query para el evaluador integrado anterior, o product_quality para el evaluador personalizado anterior) como retroalimentación name:

# 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,
    )

Paso 2: Recopilar comentarios humanos

Recopile comentarios humanos para enseñar al juez sus estándares de calidad. Elija entre los métodos siguientes:

Revisión de la interfaz de usuario de Databricks

Recolectar comentarios humanos cuando:

  • Necesitas expertos en la materia para revisar los resultados.
  • Quieres refinar iterativamente los criterios de retroalimentación.
  • Está trabajando con un conjunto de datos más pequeño (< 100 ejemplos).

Use la interfaz de usuario de MLflow para revisar y proporcionar comentarios manualmente:

  1. Accede a tu experimento de MLflow en el espacio de trabajo de Databricks.
  2. Haga clic en la pestaña Seguimientos para ver los seguimientos.
  3. Revisa cada traza y la evaluación correspondiente del evaluador.
  4. Agregue comentarios humanos mediante la interfaz de comentarios de la interfaz de usuario.
  5. Asegúrese de que el nombre de la retroalimentación coincida exactamente con el atributo name de su juez (por ejemplo, relevance_to_query para una instancia predefinida RelevanceToQuery o product_quality para el juez personalizado anterior).

Retroalimentación programática

Use comentarios mediante programación cuando:

  • Dispones de etiquetas de referencia preexistentes.
  • Está trabajando con grandes conjuntos de datos (más de 100 ejemplos).
  • Necesitas una recopilación de comentarios reproducible.

Si dispone de etiquetas de referencia existentes, regístrelas mediante programación:

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"
        ),
    )

Procedimientos recomendados para la recopilación de comentarios

  • Diversos revisores: incluir varios expertos en dominios para capturar perspectivas variadas
  • Ejemplos equilibrados: incluya al menos 30% ejemplos negativos (clasificaciones deficientes o justas)
  • Justificaciones claras: Proporcione explicaciones detalladas para las clasificaciones
  • Ejemplos representativos: Cubrir casos perimetrales y escenarios comunes

Paso 3: Alinear y registrar al juez

Una vez que tenga suficientes comentarios humanos, alinee al juez. Se utiliza el mismo método align() tanto para evaluadores integrados como personalizados.

Cuando se llama align() sin especificar un optimizador, el optimizador MemAlign se usa automáticamente:

# 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")

Optimizador explícito

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)}")

Habilitación del registro detallado

Para supervisar el proceso de alineación, habilite el registro de depuración para el optimizador:

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)

Validación de la alineación

Compruebe que la alineación ha mejorado el evaluador:


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,
    }

Creación de optimizadores de alineación personalizados

Para estrategias de alineación especializadas, extienda la AlignmentOptimizer clase base:

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)

Limitaciones

  • La alineación del juez no admite la evaluación basada en agentes ni basada en expectativas.

Pasos siguientes

  • Last updated on