ジャッジを人間に合わせる

ジャッジの調整では、体系的なフィードバックを通じて、LLM ジャッジに人間の評価基準に合わせることを教えます。このプロセスにより、一般的な評価者が、お客様独自の品質基準を理解しているドメイン固有の専門家に変わり、基準となるジャッジと比較して、人間による評価との一致率が 30 ～ 50 パーセント向上します。

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

ジャッジの調整は、次の 3 つのステップのワークフローに従います。

初期評価を生成する ：組み込みまたはカスタムジャッジを使用してトレースを評価し、ベースラインを確立します。
人間のフィードバックを収集する ：ドメイン専門家がジャッジの評価をレビューして修正する。
連携と展開 : 人間のフィードバックにさらに合わせた新しいジャッジを作成するには、ジャッジのalign()メソッドを呼び出してください。

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

要件

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

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

合わせるジャッジこれは、組み込みジャッジ (たとえば、RelevanceToQuery Correctnessまたは)、またはを使用して作成された make_judge()カスタムジャッジです。
人間によるフィードバック評価名は、ジャッジのname属性と完全に一致する必要があります。組み込みのジャッジの場合、これはデフォルトのsnake_case名です（たとえば、RelevanceToQueryのrelevance_to_query）。ただし、クラスをインスタンス化するときにname=を渡してオーバーライドしない限りです。カスタム判定の場合、make_judge()に渡されたnameです（例: product_quality）。
ConversationCompleteness などのセッションレベル（複数ターン）のジャッジでは、アラインメントはサポートされていません。

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

最初のジャッジを作成し、評価を含むトレースを生成します。少なくとも 10 個のトレースで妥当なアライメントが可能ですが、50 ～ 100 個のトレースでより良い結果が得られます。

Built-in judge
Custom judge

組み込みジャッジを直接インスタンス化してください。組み込みジャッジは、ステップ2で人間のフィードバックをログに記録する際に使用するname属性（デフォルトはrelevance_to_queryのようなsnake_case文字列です）を公開します。

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

make_judge() を使用してカスタムジャッジを作成するname引数は、ステップ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",
)

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

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

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

ステップ 2: 人間のフィードバックを収集する

人間からのフィードバックを収集して、ジャッジに品質基準を伝えます。次のアプローチから選択してください。

Databricks UI review
Programmatic feedback

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

ドメインエキスパートが出力をレビューする必要があります。
フィードバック基準を反復的に洗練する必要があります。
小規模なデータセット（100例未満）を使用しています。

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

Databricks ワークスペース内の MLflow エクスペリメントに移動します。
トレース タブをクリックして、トレースを表示します。
各トレースとそのジャッジの評価を確認する
UIのフィードバックインターフェースを使用して、人間によるフィードバックを追加してください。
フィードバック名が、ジャッジのname属性と完全に一致していることを確認してください（たとえば、組み込みのRelevanceToQueryインスタンスの場合はrelevance_to_query、または上記のカスタムジャッジの場合はproduct_quality）。

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

既存のグラウンドトゥルースラベルがあります。
大規模なデータセット（100個以上の例）を扱っています。
再現性のあるフィードバック収集が必要です。

既存のグラウンドトゥルースラベルがある場合は、プログラムでログに記録します。

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

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

多様なレビュー担当者 : 多様な視点を捉えるために複数の分野の専門家を含める
バランスの取れた例 : 少なくとも 30% の否定的な例 (悪い/普通の評価) を含める
明確な根拠 : 評価の詳細な説明を提供する
代表的なサンプル : エッジケースと一般的なシナリオをカバー

ステップ 3: ジャッジを調整して登録する

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

Default optimizer (recommended)
Explicit optimizer

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

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={&quot;alignment_date&quot;: &quot;2025-10-23&quot;, &quot;num_traces&quot;: 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")

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

詳細ログを有効にする

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

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)

アライメントを検証する

整合により判定が改善されたことを確認します。

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

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

特殊なアライメント戦略の場合は、 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)

制限事項

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

次のステップ

本番運用モニタリングについて学び、調整されたジャッジを大規模に配置します。
補完的な決定論的メトリクスについては、コードベースのスコアラーを参照してください。
カスタマイズされたジャッジの構築について詳しくは、Databricksブログをご覧ください。

要件​

ステップ 1: ジャッジを設定し、トレースを生成する​

ステップ 2: 人間のフィードバックを収集する​

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

ステップ 3: ジャッジを調整して登録する​

詳細ログを有効にする​

アライメントを検証する​

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

制限事項​

次のステップ​

要件