orchestrata_ultimate.harness.evaluators.llm_judge

Reference › orchestrata_ultimate › harness › evaluators › llm_judge

LLMJudgeEvaluator, registered as llm_judge (SPEC-102 US-005, REQ-102-070 to 073), scores subjective quality with an LLM. _build_user_message builds the prompt from the output and the evaluator's config. evaluate calls call_llm from llm.client at temperature 0.0 under the harness-judge volk alias U:HJ; the model is the model param, else the configured reasoning model, else a built-in default. _parse_response strips optional markdown fences and requires a JSON object with an integer score from 0 to 100 and an optional feedback; anything else raises ValueError. The judge passes when the score is at least the evaluator's threshold, and the reported score is divided by 100 so it is comparable with the deterministic evaluators. A failed LLM call or an unparseable response is returned as a failing result with score 0.0 and the reason, not raised. See the evaluator registry.

At a glance

Class diagram

Dependency diagram

API

Classes

LLMJudgeEvaluator

Bases: Evaluator · Decorators: register_evaluator, traces(spec='102', story='US-005', reqs=['REQ-102-070', 'REQ-102-072', 'REQ-102-073']) · Line 83

Evaluator that uses an LLM (via call_llm) to score subjective quality.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/harness/evaluators/llm_judge.py

"""LLM-as-judge evaluator using the project's call_llm client."""

from __future__ import annotations

import json
import logging
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.harness.evaluators import register_evaluator
from orchestrata_ultimate.harness.models import (
    Evaluator,
    EvaluatorConfig,
    EvaluatorResult,
    HarnessContext,
)
from orchestrata_ultimate.llm.client import call_llm

_logger = logging.getLogger(__name__)

_SYSTEM_PROMPT = """\
You are an expert evaluator of AI agent outputs.
Score the provided output on a scale of 0 to 100 based on the rubric.
Respond with ONLY a JSON object in this exact format:
{"score": <int>, "feedback": "<string>"}
Do not include any other text, markdown, or explanation."""


@traces(spec="102", story="US-005", reqs=["REQ-102-071", "REQ-102-124"])
def _build_user_message(output: Any, config: EvaluatorConfig) -> str:
    """Construct the evaluation prompt from output and evaluator config."""
    rubric = config.params.get("rubric", "Evaluate the quality of the output.")
    expected = config.params.get("expected_behavior", "")

    parts = [
        "## Rubric",
        rubric,
        "",
        "## Expected Behavior",
        expected if expected else "No specific expectations provided.",
        "",
        "## Actual Output",
        json.dumps(output, indent=2, default=str),
        "",
        "Provide your evaluation as JSON only.",
    ]
    return "\n".join(parts)


@traces(spec="102", story="US-005", reqs=["REQ-102-073"])
def _parse_response(text: str) -> tuple[int, str]:
    """Parse the LLM response text into (score, feedback).

    Raises:
        ValueError: If the response cannot be parsed.
    """
    text = text.strip()
    # Some models wrap JSON in markdown fences
    if text.startswith("```json"):
        text = text.removeprefix("```json").removesuffix("```").strip()
    elif text.startswith("```"):
        text = text.removeprefix("```").removesuffix("```").strip()

    data = json.loads(text)
    if not isinstance(data, dict):
        raise ValueError(f"Expected JSON object, got {type(data).__name__}")

    score = data.get("score")
    feedback = data.get("feedback", "")

    if score is None:
        raise ValueError("Missing 'score' field in response")
    if not isinstance(score, int):
        raise ValueError(f"'score' must be an integer, got {type(score).__name__}")
    if not (0 <= score <= 100):
        raise ValueError(f"'score' must be between 0 and 100, got {score}")

    return score, str(feedback)


@register_evaluator
@traces(spec="102", story="US-005", reqs=["REQ-102-070", "REQ-102-072", "REQ-102-073"])
class LLMJudgeEvaluator(Evaluator):
    """Evaluator that uses an LLM (via call_llm) to score subjective quality."""

    name = "llm_judge"

    @traces(spec="102", story="US-005", reqs=["REQ-102-074"])
    async def evaluate(
        self,
        output: Any,
        context: HarnessContext,
        config: EvaluatorConfig,
    ) -> EvaluatorResult:
        """Evaluate scenario output using the project's LLM client.

        Uses :func:`orchestrata_ultimate.llm.client.call_llm` for the
        actual provider dispatch, retries, and error handling.
        """
        user_message = _build_user_message(output, config)
        model = (
            config.params.get("model")
            or context.config.reasoning_llm
            or "deepseek-v4-pro"
        )

        try:
            response_text = await call_llm(
                system_prompt=_SYSTEM_PROMPT,
                user_message=user_message,
                model=model,
                temperature=0.0,
                volk="U:HJ",  # Harness Judge volk alias
            )
        except Exception as exc:
            _logger.warning("LLM judge call failed: %s", exc)
            return EvaluatorResult(
                name=self.name,
                passed=False,
                score=0.0,
                details=f"LLM call failed: {exc}",
            )

        try:
            score, feedback = _parse_response(response_text)
        except Exception as exc:
            _logger.warning("LLM judge parse failed: %s", exc)
            return EvaluatorResult(
                name=self.name,
                passed=False,
                score=0.0,
                details=f"Failed to parse LLM response: {exc}\nRaw: {response_text[:500]}",
            )

        threshold = config.threshold
        passed = score >= threshold
        return EvaluatorResult(
            name=self.name,
            passed=passed,
            score=score / 100.0,
            details=feedback,
        )