Metrics & Judges

Synthesis judge

DspyGeneralSynthesisJudge(lm, enable_reasoning_traces=False, confidence_threshold=0.7, signature=None, retry_temperatures=None)

Bases: SynthesisJudgeInterface

Enhanced DSPy module for evaluating GeneralSynthesisOntology extraction quality against source synthesis text.

Implements a two-level fallback chain for robust structured output

1. Strict json_schema(native for Claude/Gemini; extra_body for OpenRouter)
2. json_object mode (valid JSON, prompt-guided schema compliance)

Within each strategy, temperature is escalated on validation failures. API-level format errors (400/unsupported) skip immediately to the next strategy without wasting temperature retries.

Initialize the unified synthesis judge.

Parameters:

Name	Type	Description	Default
signature	type[Signature] | None	DSPy signature for evaluation	None
lm	LM	Language model for evaluation	required
enable_reasoning_traces	bool	Whether to include detailed reasoning	False
confidence_threshold	float	Minimum confidence threshold for reliable	0.7
retry_temperatures	list[float] | None	Temperatures to try per strategy on content	None

Source code in src/llm_synthesis/metrics/judge/general_synthesis_judge.py

def __init__(
    self,
    lm: dspy.LM,
    enable_reasoning_traces: bool = False,
    confidence_threshold: float = 0.7,
    signature: type[dspy.Signature] | None = None,
    retry_temperatures: list[float] | None = None,
):
    """
    Initialize the unified synthesis judge.

    Args:
        signature: DSPy signature for evaluation
        lm: Language model for evaluation
        enable_reasoning_traces: Whether to include detailed reasoning
        traces
        confidence_threshold: Minimum confidence threshold for reliable
        evaluations
        retry_temperatures: Temperatures to try per strategy on content
        validation failures. Defaults to [0.0, 0.3, 0.5].
    """
    self._validate_signature(signature)
    self.signature = signature
    self.enable_reasoning_traces = enable_reasoning_traces
    self.confidence_threshold = confidence_threshold
    self.retry_temperatures = retry_temperatures or [0.0, 0.3, 0.5]

    # Store a clean base LM — strip any pre-existing response_format /
    # extra_body so that _build_format_strategies() has full control.
    _clean_keys = {"response_format", "extra_body"}
    if _clean_keys & set(lm.kwargs):
        self.lm = copy.copy(lm)
        self.lm.kwargs = {
            k: v for k, v in lm.kwargs.items() if k not in _clean_keys
        }
    else:
        self.lm = lm

    super().__init__()

Methods:

forward(input)

Evaluate extracted GeneralSynthesisOntology against source text.

Tries each format strategy in order. Within a strategy, retries at escalating temperatures on content-validation failures. API-level format errors skip immediately to the next strategy.

Parameters:

Name	Type	Description	Default
input	tuple[str, str] | tuple[str, str, str]	Tuple of (source_text, extracted_ontology_json) or (source_text, extracted_ontology_json, target_material)	required

Returns:

Type	Description
GeneralSynthesisEvaluation	Comprehensive evaluation of the ontology extraction

Source code in src/llm_synthesis/metrics/judge/general_synthesis_judge.py

def forward(
    self, input: tuple[str, str] | tuple[str, str, str]
) -> GeneralSynthesisEvaluation:
    """
    Evaluate extracted GeneralSynthesisOntology against source text.

    Tries each format strategy in order.  Within a strategy, retries at
    escalating temperatures on content-validation failures.  API-level
    format errors skip immediately to the next strategy.

    Args:
        input: Tuple of (source_text, extracted_ontology_json) or
               (source_text, extracted_ontology_json, target_material)

    Returns:
        Comprehensive evaluation of the ontology extraction
    """
    if len(input) == 2:
        source_text, extracted_ontology_json = input
        target_material = self._extract_target_from_json(
            extracted_ontology_json
        )
    else:
        source_text, extracted_ontology_json, target_material = input

    self._validate_inputs(source_text, extracted_ontology_json)

    strategies = self._build_format_strategies()
    last_exc: Exception | None = None

    for s_idx, strategy_kwargs in enumerate(strategies):
        strategy_label = (
            "json_schema" if s_idx == 0 else "json_object-fallback"
        )
        for t_idx, temp in enumerate(self.retry_temperatures):
            lm = self._lm_with_overrides(
                {**strategy_kwargs, "temperature": temp}
            )
            try:
                with dspy.settings.context(
                    lm=lm, adapter=dspy.adapters.JSONAdapter()
                ):
                    prediction = dspy.Predict(self.signature)(
                        source_text=source_text,
                        extracted_ontology_json=extracted_ontology_json,
                        target_material=target_material,
                    )
                evaluation = prediction.evaluation
                evaluation = self._post_process_evaluation(evaluation)
                if s_idx > 0 or t_idx > 0:
                    log.info(
                        "Judge succeeded: strategy=%s, temperature=%.1f",
                        strategy_label,
                        temp,
                    )
                return evaluation
            except Exception as e:
                # Recovery: model returned complete JSON but without the
                # DSPy-expected {"evaluation": {...}} wrapper key.
                recovered = self._try_recover_bare_json(e)
                if recovered is not None:
                    recovered = self._post_process_evaluation(recovered)
                    log.info(
                        "Judge: recovered bare JSON response"
                        " (strategy=%s, temp=%.1f)",
                        strategy_label,
                        temp,
                    )
                    return recovered

                last_exc = e
                if self._is_api_format_error(e):
                    log.warning(
                        "Judge: format unsupported (%s): %r"
                        " — falling back to next strategy",
                        strategy_label,
                        e,
                    )
                    break  # skip remaining temperatures, try next strategy
                elif t_idx < len(self.retry_temperatures) - 1:
                    log.warning(
                        "Judge: validation failure"
                        " (strategy=%s, temp=%.1f): %r"
                        " — retrying at temp=%.1f",
                        strategy_label,
                        temp,
                        e,
                        self.retry_temperatures[t_idx + 1],
                    )
                else:
                    log.warning(
                        "Judge: all temperatures exhausted"
                        " for strategy=%s — trying next strategy",
                        strategy_label,
                    )

    raise last_exc  # type: ignore[misc]

GeneralSynthesisEvaluation

Bases: BaseModel

Complete evaluation of GeneralSynthesisOntology extraction quality.

GeneralSynthesisEvaluationScore

Bases: BaseModel

Evaluation scores for GeneralSynthesisOntology extraction quality. Scores are on a scale of 1.0 (poor) to 5.0 (excellent) with 0.5 increments.

make_general_synthesis_judge_signature(signature_name='GeneralSynthesisJudgeSignature', instructions=None, source_text_description='Original synthesis text for ontology extraction evaluation.', extracted_ontology_description='JSON representation of extracted GeneralSynthesisOntology.', target_material_description='Target material for synthesis context.', evaluation_description='Comprehensive evaluation of ontology extraction quality. CRITICAL: populate ALL fields — reasoning, confidence_level, all seven *_score and *_reasoning pairs inside scores, and scores.overall_reasoning. Omitting any field is invalid.')

Create a DSPy signature for GeneralSynthesisOntology evaluation.

Parameters:

Name	Type	Description	Default
signature_name	str	Name of the signature class	'GeneralSynthesisJudgeSignature'
instructions	str | None	Custom instructions for the evaluation	None
source_text_description	str	Description for source text input	'Original synthesis text for ontology extraction evaluation.'
extracted_ontology_description	str	Description for ontology JSON input	'JSON representation of extracted GeneralSynthesisOntology.'
target_material_description	str	Description for target material input	'Target material for synthesis context.'
evaluation_description	str	Description for evaluation output	'Comprehensive evaluation of ontology extraction quality. CRITICAL: populate ALL fields — reasoning, confidence_level, all seven *_score and *_reasoning pairs inside scores, and scores.overall_reasoning. Omitting any field is invalid.'

Returns:

Type	Description
type[Signature]	DSPy signature class for ontology evaluation

Source code in src/llm_synthesis/metrics/judge/general_synthesis_judge.py

def make_general_synthesis_judge_signature(
    signature_name: str = "GeneralSynthesisJudgeSignature",
    instructions: str | None = None,
    source_text_description: str = (
        "Original synthesis text for ontology extraction evaluation."
    ),
    extracted_ontology_description: str = (
        "JSON representation of extracted GeneralSynthesisOntology."
    ),
    target_material_description: str = (
        "Target material for synthesis context."
    ),
    evaluation_description: str = (
        "Comprehensive evaluation of ontology extraction quality. "
        "CRITICAL: populate ALL fields — reasoning, confidence_level, "
        "all seven *_score and *_reasoning pairs inside scores, and "
        "scores.overall_reasoning. Omitting any field is invalid."
    ),
) -> type[dspy.Signature]:
    """
    Create a DSPy signature for GeneralSynthesisOntology evaluation.

    Args:
        signature_name: Name of the signature class
        instructions: Custom instructions for the evaluation
        source_text_description: Description for source text input
        extracted_ontology_description: Description for ontology JSON input
        target_material_description: Description for target material input
        evaluation_description: Description for evaluation output

    Returns:
        DSPy signature class for ontology evaluation
    """
    if instructions is None:
        instructions = (
            "You are an expert in materials science and data extraction. "
            "Evaluate how well the GeneralSynthesisOntology extraction "
            "captures all synthesis information from the source text. "
            "Assess completeness, accuracy, and semantic preservation "
            "across all ontology components."
        )

    signature = {
        "source_text": (
            str,
            dspy.InputField(description=source_text_description),
        ),
        "extracted_ontology_json": (
            str,
            dspy.InputField(description=extracted_ontology_description),
        ),
        "target_material": (
            str,
            dspy.InputField(description=target_material_description),
        ),
        "evaluation": (
            GeneralSynthesisEvaluation,
            dspy.OutputField(description=evaluation_description),
        ),
    }

    return dspy.make_signature(
        signature_name=signature_name,
        instructions=instructions,
        signature=signature,
    )

Linking judge

DspyLinkingJudge(lm, enable_reasoning_traces=False, confidence_threshold=0.7, signature=None)

Bases: LinkingJudgeInterface

DSPy module for evaluating synthesis-to-performance linking quality.

The judge receives

1. The full paper text (source of truth).
2. The extracted synthesis ontologies (JSON list).
3. The extracted plot data (JSON list).
4. The linking output mapping syntheses to plot series (JSON).

It produces a LinkingEvaluation with four criterion scores (1-5 in 0.5 increments), nine failure-mode flags, and supporting reasoning.

Source code in src/llm_synthesis/metrics/judge/linking_judge.py

def __init__(
    self,
    lm: dspy.LM,
    enable_reasoning_traces: bool = False,
    confidence_threshold: float = 0.7,
    signature: type[dspy.Signature] | None = None,
):
    self._validate_signature(signature)
    self.signature = signature
    self.lm = lm
    self.enable_reasoning_traces = enable_reasoning_traces
    self.confidence_threshold = confidence_threshold
    super().__init__()

Methods:

forward(input)

Evaluate linking output against the paper and extracted data.

Parameters:

Name	Type	Description	Default
input	tuple[str, str, str, str]	Tuple of (source_text, synthesis_json, plot_data_json, linking_output_json)	required

Returns:

Type	Description
LinkingEvaluation	A LinkingEvaluation instance.

Source code in src/llm_synthesis/metrics/judge/linking_judge.py

def forward(
    self,
    input: tuple[str, str, str, str],
) -> LinkingEvaluation:
    """Evaluate linking output against the paper and extracted data.

    Args:
        input: Tuple of
            (source_text, synthesis_json, plot_data_json,
             linking_output_json)

    Returns:
        A ``LinkingEvaluation`` instance.
    """
    source_text, synthesis_json, plot_data_json, linking_output_json = input

    self._validate_inputs(
        source_text, synthesis_json, plot_data_json, linking_output_json
    )

    with dspy.settings.context(
        lm=self.lm, adapter=dspy.adapters.JSONAdapter()
    ):
        prediction = dspy.Predict(self.signature)(
            source_text=source_text,
            synthesis_json=synthesis_json,
            plot_data_json=plot_data_json,
            linking_output_json=linking_output_json,
        )

        evaluation = prediction.evaluation
        evaluation = self._post_process_evaluation(evaluation)
        return evaluation

LinkingEvaluation

Bases: BaseModel

Complete evaluation of synthesis-to-performance linking quality.

make_linking_judge_signature(signature_name='LinkingJudgeSignature', instructions=None, source_text_description='Full paper text for linking evaluation.', synthesis_json_description='JSON list of extracted synthesis ontologies.', plot_data_json_description='JSON list of extracted plot data with series and coordinates.', linking_output_json_description='JSON linking output mapping syntheses to plot series.', evaluation_description='Comprehensive evaluation of linking quality.')

Factory for creating a customised LinkingJudge DSPy signature.

Follows the same pattern as make_general_synthesis_judge_signature.

Source code in src/llm_synthesis/metrics/judge/linking_judge.py

def make_linking_judge_signature(
    signature_name: str = "LinkingJudgeSignature",
    instructions: str | None = None,
    source_text_description: str = ("Full paper text for linking evaluation."),
    synthesis_json_description: str = (
        "JSON list of extracted synthesis ontologies."
    ),
    plot_data_json_description: str = (
        "JSON list of extracted plot data with series and coordinates."
    ),
    linking_output_json_description: str = (
        "JSON linking output mapping syntheses to plot series."
    ),
    evaluation_description: str = (
        "Comprehensive evaluation of linking quality."
    ),
) -> type[dspy.Signature]:
    """Factory for creating a customised LinkingJudge DSPy signature.

    Follows the same pattern as
    ``make_general_synthesis_judge_signature``.
    """
    if instructions is None:
        instructions = (
            "You are an expert materials scientist evaluating how well "
            "an automated algorithm has linked extracted synthesis "
            "procedures to performance data from plots. Assess "
            "correctness, completeness, and structural quality of the "
            "linking output against the original paper text. Flag "
            "specific failure modes when detected."
        )

    signature = {
        "source_text": (
            str,
            dspy.InputField(description=source_text_description),
        ),
        "synthesis_json": (
            str,
            dspy.InputField(description=synthesis_json_description),
        ),
        "plot_data_json": (
            str,
            dspy.InputField(description=plot_data_json_description),
        ),
        "linking_output_json": (
            str,
            dspy.InputField(description=linking_output_json_description),
        ),
        "evaluation": (
            LinkingEvaluation,
            dspy.OutputField(description=evaluation_description),
        ),
    }

    return dspy.make_signature(
        signature_name=signature_name,
        instructions=instructions,
        signature=signature,
    )

Figure extraction metric

FigureExtractionMetric

Bases: LinePlotExtractionMetric

Methods:

__call__(preds, refs, error_metric='rmse')

Compute average RMSE or MAE across all matching series. For each series, it uses normalized-to-axis-sclae nearest-neighbor matching to find the closest points in the ground truth data to the extracted points from the LLM output. And then computes the error metric (RMSE or MAE) based on these matches.

Source code in src/llm_synthesis/metrics/figure_extraction/figure_extraction_metric.py

def __call__(
    self,
    preds: ExtractedLinePlotData,
    refs: ExtractedLinePlotData,
    error_metric: Literal["rmse", "mae"] = "rmse",
) -> float:
    """
    Compute average RMSE or MAE across all matching series.
    For each series, it uses normalized-to-axis-sclae nearest-neighbor
    matching to find the closest points
    in the ground truth data to the extracted points from the LLM output.
    And then computes the error metric (RMSE or MAE) based on these matches.
    """
    extracted = preds.name_to_coordinates
    ground_truth = refs.name_to_coordinates

    missing_keys = set(ground_truth) - set(extracted)
    if missing_keys:
        logging.info(f"Series missing in LLM output: {missing_keys}.")

    common_keys = set(extracted) & set(ground_truth)
    if not common_keys:
        logging.warning(
            "No common series names found between ground truth \
            and LLM output."
        )
        return None

    x_scale, y_scale = self.compute_scale(ground_truth)

    error_function = (
        self.pointwise_rmse
        if error_metric == "rmse"
        else self.pointwise_mae
    )

    errors = [
        error_function(extracted[k], ground_truth[k], x_scale, y_scale)
        for k in common_keys
    ]

    return sum(errors) / len(errors)

compute_scale(ground_truth) staticmethod

Compute normalization scales for x and y.

Source code in src/llm_synthesis/metrics/figure_extraction/figure_extraction_metric.py

@staticmethod
def compute_scale(
    ground_truth: dict[str, list[tuple[float, float]]],
) -> tuple[float, float]:
    """Compute normalization scales for x and y."""
    all_x = [x for coords in ground_truth.values() for x, _ in coords]
    all_y = [y for coords in ground_truth.values() for _, y in coords]
    x_scale = max(all_x) - min(all_x) or 1e-8
    y_scale = max(all_y) - min(all_y) or 1e-8
    return x_scale, y_scale

pointwise_rmse(extracted_coords, gt_coords, x_scale, y_scale) staticmethod

Compute RMSE using nearest-neighbor matching for one series.

Source code in src/llm_synthesis/metrics/figure_extraction/figure_extraction_metric.py

@staticmethod
def pointwise_rmse(
    extracted_coords: list[tuple[float, float]],
    gt_coords: list[tuple[float, float]],
    x_scale: float,
    y_scale: float,
) -> float:
    """Compute RMSE using nearest-neighbor matching for one series."""
    if not extracted_coords:
        return 0.0

    total_sq_error = sum(
        min(
            ((gt_x - ex_x) / x_scale) ** 2 + ((gt_y - ex_y) / y_scale) ** 2
            for gt_x, gt_y in gt_coords
        )
        for ex_x, ex_y in extracted_coords
    )

    return (total_sq_error / len(extracted_coords)) ** 0.5

pointwise_mae(extracted_coords, gt_coords, x_scale, y_scale) staticmethod

Compute MAE using nearest-neighbor matching for one series.

Source code in src/llm_synthesis/metrics/figure_extraction/figure_extraction_metric.py

@staticmethod
def pointwise_mae(
    extracted_coords: list[tuple[float, float]],
    gt_coords: list[tuple[float, float]],
    x_scale: float,
    y_scale: float,
) -> float:
    """Compute MAE using nearest-neighbor matching for one series."""
    if not extracted_coords:
        return 0.0

    total_abs_error = sum(
        min(
            (
                ((ex_x - gt_x) / x_scale) ** 2
                + ((ex_y - gt_y) / y_scale) ** 2
            )
            ** 0.5
            for gt_x, gt_y in gt_coords
        )
        for ex_x, ex_y in extracted_coords
    )

    return total_abs_error / len(extracted_coords)