LLM のジャッジを人間のフィードバックに合わせる

ジャッジアライメントは、体系的なフィードバックを通じて人間の評価基準に一致するようにLLMのジャッジに教えます。このプロセスにより、汎用エバリュエーターは、独自の品質基準を理解するドメイン固有の専門家に変換され、ベースラインジャッジと比較して人間の評価との合意が30〜50%向上します。

同じアラインメントワークフローは、組み込みのジャッジ (RelevanceToQuery、Safety、Correctnessなど) と、で作成されたmake_judge()の両方に適用されます。組み込みのジャッジとの連携を使用して、一般的な基準をドメインに適応させるか、カスタムジャッジと連携して特殊な評価ロジックを調整します。

ジャッジアライメントは、次の 3 つのステップのワークフローに従います。

初期評価を生成する: 組み込みまたはカスタムのジャッジを使用してトレースを評価し、ベースラインを確立します。
人間のフィードバックを収集する: ドメインの専門家は、評価をレビューし、正しく判断します。
調整してデプロイ: ジャッジの align() メソッドを呼び出して、人間のフィードバックにより整合した新しいジャッジを作成します。

システムは、パッケージ mlflow.genai.judges.optimizersで使用可能なオプティマイザーをサポートします。

Requirements

MLflow 3.4.0以上がジャッジアライメント機能を使用するために必要です。

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

調整するジャッジ。これには、組み込みのジャッジ (RelevanceToQueryやCorrectnessなど) や、で作成されたmake_judge()を指定できます。
ヒューマンフィードバック評価名は、ジャッジの name 属性と正確に一致している必要があります。組み込みジャッジの場合、クラスをインスタンス化する際に name= を渡してオーバーライドしない限り、これが、既定の snake_case 名 (RelevanceToQuery の relevance_to_query など) です。カスタムジャッジの場合、make_judge() に渡した name です (例えば、product_quality)。
ConversationCompletenessなどのセッションレベル (複数ターン) のジャッジでは、アライメントはサポートされていません。

手順 1: ジャッジを設定し、トレースを生成する

初期ジャッジを設定し、評価付きのトレースを生成します。少なくとも 10 個のトレースとの適切なアラインメントを実現できますが、50 から 100 個のトレースを使用すると、より優れた結果が得られます。

組み込みジャッジ

組み込みのジャッジを直接インスタンス化します。組み込みのジャッジは、手順 2 で人間のフィードバックをログに記録するときに使用する name 属性 (既定値は relevance_to_query などのsnake_case文字列) を公開します。

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

カスタムジャッジ

make_judge()でカスタムジャッジを作成します。 name引数は、手順 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",
)

アプリケーションロジックを定義します。次の例では、 Databricks でホストされる基盤モデルを使用して、クエリから製品の説明を生成します。これを独自のアプリケーションコードに置き換えます。

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

トレースを生成し、判定を実行します。フィードバックnameとして、ジャッジのrelevance_to_query属性 (たとえば、上記の組み込みのジャッジのproduct_quality、または上記のカスタムジャッジの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,
    )

手順 2: 人間のフィードバックを収集する

人間のフィードバックを収集して、あなたの品質基準を判事に教えます。次の方法から選択します。

Databricks UI レビュー

次の場合に人間のフィードバックを収集します。

出力を確認するには、ドメインの専門家が必要です。
フィードバック条件を繰り返し調整する必要があります。
小さいデータセットを使用しています (100 個の例< )。

MLflow UI を使用して、手動で確認し、フィードバックを提供します。

Databricks ワークスペースで MLflow 実験に移動します。
トレースを表示するには、[ トレース ] タブをクリックします。
各トレースとそのジャッジ評価を確認します。
UI のフィードバックインターフェイスを使用して、人間のフィードバックを追加します。
フィードバック名がジャッジのname属性と正確に一致していることを確認します (たとえば、組み込みのrelevance_to_query インスタンスのRelevanceToQueryや、上記のカスタムジャッジのproduct_qualityなど)。

プログラムによるフィードバック

プログラムによるフィードバックは、次の場合に使用します。

既存のグラウンドトゥルースラベルがあります。
大規模なデータセット (100 以上の例) を使用しています。
再現可能なフィードバック収集が必要です。

既存の正解ラベルがある場合は、プログラムを使用してログに記録してください。

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

フィードバック収集のベストプラクティス

多様なレビュー担当者: 複数のドメインエキスパートを含め、さまざまな視点を把握する
バランスの取れた例: 少なくとも 30% 否定的な例を含める (不適切/公正な評価)
明確な根拠: 評価の詳細な説明を提供する
代表的な事例: エッジケースと一般的なシナリオをカバーする

手順 3: ジャッジを調整して登録する

十分な人間のフィードバックを得たら、判事を揃えます。組み込みジャッジとカスタムジャッジの両方に同じ align() メソッドが使用されます。

既定のオプティマイザー (推奨)

オプティマイザーを指定せずに align() を呼び出すと、MemAlign オプティマイザーが自動的に使用されます。

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

明示的なオプティマイザー

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

詳細なログ記録を有効にする

アラインメントプロセスを監視するには、オプティマイザーのデバッグログを有効にします。

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)

配置の検証

位置合わせによって判定が改善されたことを検証します。


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

カスタムアラインメントオプティマイザーを作成する

特殊なアラインメント戦略の場合は、 AlignmentOptimizer 基底クラスを拡張します。

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)

制限事項

ジャッジアライメントは、エージェントベースまたは期待ベースの評価をサポートしていません。

次のステップ

調整されたジャッジを大規模にデプロイするための運用監視について説明します。
補完的な決定論的メトリックについては、コードベースのスコアラーを参照してください。
カスタマイズされたジャッジの構築の詳細については、この Databricks ブログを参照してください。

フィードバック

このページはお役に立ちましたか?

Last updated on 2026-06-24