siemens_ragas/rag_eval/metrics/judge_prompts.py

"""Localize RAGAS collections judge prompts to a target language (e.g. Chinese).

Loads committed, pre-translated prompt cache files from
configs/judge_prompts/<language>/<metric>__<attr>.json and overrides each
metric's prompt instance attributes in place. Missing, corrupt, or schema-drifted
cache entries degrade gracefully to the built-in English prompt so scoring never
breaks.
"""

from __future__ import annotations

import hashlib
import json
import logging
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

logger = logging.getLogger("rag_eval.metrics.judge_prompts")

_REPO_ROOT = Path(__file__).resolve().parents[2]
CACHE_ROOT = _REPO_ROOT / "configs" / "judge_prompts"

# Metric name -> prompt instance attribute names holding a BasePrompt.
# Verified against RAGAS 0.4.3 collections source; semantic_similarity has none.
METRIC_PROMPT_ATTRS: dict[str, tuple[str, ...]] = {
    "faithfulness": ("statement_generator_prompt", "nli_statement_prompt"),
    "answer_relevancy": ("prompt",),
    "context_recall": ("prompt",),
    "context_precision": ("prompt",),
    "noise_sensitivity": ("statement_prompt", "faithfulness_prompt"),
    "factual_correctness": ("prompt", "nli_prompt"),
}

# In-memory memoization of parsed cache files, keyed by (language, metric, attr).
_MEMO: dict[tuple[str, str, str], dict | None] = {}


@dataclass
class LocalizationReport:
    """Outcome of a localize_pipeline_prompts() call, for logging and tests."""

    language: str
    applied: list[str] = field(default_factory=list)
    skipped: list[str] = field(default_factory=list)
    warnings: list[str] = field(default_factory=list)


def reset_cache() -> None:
    """Clear the in-memory parsed-cache memo (used by tests)."""
    _MEMO.clear()


def prompt_source_hash(prompt: Any) -> str:
    """Return a stable sha256 of a prompt's English instruction + examples."""
    examples = [
        {"input": inp.model_dump(), "output": out.model_dump()}
        for inp, out in getattr(prompt, "examples", [])
    ]
    payload = json.dumps(
        {"instruction": getattr(prompt, "instruction", ""), "examples": examples},
        ensure_ascii=False,
        sort_keys=True,
    )
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


def _cache_path(language: str, metric: str, attr: str) -> Path:
    """Resolve the cache file path for one (language, metric, attr) triple."""
    return CACHE_ROOT / language / f"{metric}__{attr}.json"


def _load_cache_file(language: str, metric: str, attr: str) -> dict | None:
    """Load and memoize a cache file; return None if absent or unreadable."""
    key = (language, metric, attr)
    if key in _MEMO:
        return _MEMO[key]
    path = _cache_path(language, metric, attr)
    data: dict | None
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except FileNotFoundError:
        data = None
    except (OSError, json.JSONDecodeError) as exc:  # corrupt file: degrade to English
        logger.warning("[judge_prompts] cache read failed %s: %s", path, exc)
        data = None
    _MEMO[key] = data
    return data


def apply_localized_prompt(prompt: Any, data: dict) -> None:
    """Override a live prompt's instruction/examples/language from a cache dict.

    Examples are rebuilt using the live prompt's input/output models, so an
    upstream schema change raises here and is caught by the caller (which then
    keeps the English prompt).
    """
    examples = [
        (prompt.input_model(**ex["input"]), prompt.output_model(**ex["output"]))
        for ex in data.get("examples", [])
    ]
    prompt.instruction = data["instruction"]
    prompt.examples = examples
    prompt.language = data.get("language", "chinese")


def localize_pipeline_prompts(registry: dict[str, Any], language: str) -> LocalizationReport:
    """Override judge prompts in `registry` with cached `language` translations.

    `registry` maps metric name -> RAGAS metric instance. Only metrics in
    METRIC_PROMPT_ATTRS are touched; unknown metrics and semantic_similarity are
    left untouched. English ("en"/"english"/empty) is a no-op.
    """
    report = LocalizationReport(language=language)
    normalized = (language or "en").strip().lower()
    if normalized in ("", "en", "english"):
        return report

    for metric_name, attrs in METRIC_PROMPT_ATTRS.items():
        metric = registry.get(metric_name)
        if metric is None:
            continue
        for attr in attrs:
            tag = f"{metric_name}.{attr}"
            prompt = getattr(metric, attr, None)
            if prompt is None:
                report.skipped.append(tag)
                continue
            data = _load_cache_file(normalized, metric_name, attr)
            if data is None:
                report.skipped.append(tag)
                report.warnings.append(f"missing cache for {tag}")
                continue
            # Drift detection: warn if the English source changed since caching.
            if data.get("source_hash") and data["source_hash"] != prompt_source_hash(prompt):
                report.warnings.append(f"stale cache for {tag} (regenerate)")
            try:
                apply_localized_prompt(prompt, data)
                report.applied.append(tag)
            except Exception as exc:  # noqa: BLE001  schema drift -> keep English
                report.warnings.append(f"apply failed for {tag}: {exc}; kept english")

    if report.warnings:
        logger.warning(
            "[judge_prompts] language=%s applied=%d skipped=%d warnings=%s",
            normalized, len(report.applied), len(report.skipped), report.warnings,
        )
    else:
        logger.info(
            "[judge_prompts] language=%s applied=%d", normalized, len(report.applied)
        )
    return report