diff --git a/rag_eval/metrics/judge_prompts.py b/rag_eval/metrics/judge_prompts.py new file mode 100644 index 0000000..c2845b5 --- /dev/null +++ b/rag_eval/metrics/judge_prompts.py @@ -0,0 +1,152 @@ +"""Localize RAGAS collections judge prompts to a target language (e.g. Chinese). + +Loads committed, pre-translated prompt cache files from +configs/judge_prompts//__.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 diff --git a/tests/test_judge_prompt_localizer.py b/tests/test_judge_prompt_localizer.py new file mode 100644 index 0000000..3024368 --- /dev/null +++ b/tests/test_judge_prompt_localizer.py @@ -0,0 +1,121 @@ +"""Tests for the judge-prompt localizer (no RAGAS dependency; uses fake prompts).""" + +import json + +from pydantic import BaseModel + +from rag_eval.metrics import judge_prompts as jp + + +class _In(BaseModel): + question: str + + +class _Out(BaseModel): + statements: list[str] + + +class _FakePrompt: + """Minimal stand-in for a RAGAS BasePrompt with overridable attributes.""" + + def __init__(self): + self.input_model = _In + self.output_model = _Out + self.instruction = "English instruction." + self.examples = [(_In(question="q"), _Out(statements=["s"]))] + self.language = "english" + + +class _FakeMetric: + def __init__(self): + self.prompt = _FakePrompt() + + +def _cache_dict(prompt): + """Build a valid cache dict for the given fake prompt.""" + return { + "metric": "context_recall", + "prompt_attr": "prompt", + "language": "chinese", + "ragas_version": "0.4.3", + "source_hash": jp.prompt_source_hash(prompt), + "instruction": "中文指令。", + "examples": [{"input": {"question": "问题"}, "output": {"statements": ["陈述"]}}], + } + + +def setup_function(_): + jp.reset_cache() + + +def test_apply_localized_prompt_overrides_instruction_and_examples(): + """apply_localized_prompt swaps instruction/examples and rebuilds example models.""" + prompt = _FakePrompt() + jp.apply_localized_prompt(prompt, _cache_dict(prompt)) + assert prompt.instruction == "中文指令。" + assert prompt.examples[0][0].question == "问题" + assert prompt.examples[0][1].statements == ["陈述"] + assert prompt.language == "chinese" + + +def test_localize_english_is_noop(): + """language='en' leaves the registry untouched.""" + metric = _FakeMetric() + report = jp.localize_pipeline_prompts({"context_recall": metric}, "en") + assert report.applied == [] + assert metric.prompt.instruction == "English instruction." + + +def test_localize_applies_from_cache_file(tmp_path, monkeypatch): + """localize reads /zh/context_recall__prompt.json and applies it.""" + metric = _FakeMetric() + root = tmp_path / "configs" / "judge_prompts" + (root / "zh").mkdir(parents=True) + (root / "zh" / "context_recall__prompt.json").write_text( + json.dumps(_cache_dict(metric.prompt), ensure_ascii=False), encoding="utf-8" + ) + monkeypatch.setattr(jp, "CACHE_ROOT", root) + report = jp.localize_pipeline_prompts({"context_recall": metric}, "zh") + assert "context_recall.prompt" in report.applied + assert metric.prompt.instruction == "中文指令。" + + +def test_localize_missing_cache_keeps_english(tmp_path, monkeypatch): + """A missing cache file degrades gracefully to the English prompt with a warning.""" + metric = _FakeMetric() + monkeypatch.setattr(jp, "CACHE_ROOT", tmp_path / "empty") + report = jp.localize_pipeline_prompts({"context_recall": metric}, "zh") + assert metric.prompt.instruction == "English instruction." + assert "context_recall.prompt" in report.skipped + assert report.warnings + + +def test_localize_stale_hash_warns_but_applies(tmp_path, monkeypatch): + """A source_hash mismatch still applies Chinese but records a stale warning.""" + metric = _FakeMetric() + data = _cache_dict(metric.prompt) + data["source_hash"] = "deadbeef" + root = tmp_path / "configs" / "judge_prompts" + (root / "zh").mkdir(parents=True) + (root / "zh" / "context_recall__prompt.json").write_text( + json.dumps(data, ensure_ascii=False), encoding="utf-8" + ) + monkeypatch.setattr(jp, "CACHE_ROOT", root) + report = jp.localize_pipeline_prompts({"context_recall": metric}, "zh") + assert metric.prompt.instruction == "中文指令。" + assert any("stale" in w for w in report.warnings) + + +def test_cache_memoized(tmp_path, monkeypatch): + """A second localize call does not re-read the file (in-memory memo).""" + metric = _FakeMetric() + root = tmp_path / "configs" / "judge_prompts" + (root / "zh").mkdir(parents=True) + path = root / "zh" / "context_recall__prompt.json" + path.write_text(json.dumps(_cache_dict(metric.prompt), ensure_ascii=False), encoding="utf-8") + monkeypatch.setattr(jp, "CACHE_ROOT", root) + jp.localize_pipeline_prompts({"context_recall": _FakeMetric()}, "zh") + path.unlink() # delete file; memo should still serve the parsed data + metric2 = _FakeMetric() + report = jp.localize_pipeline_prompts({"context_recall": metric2}, "zh") + assert "context_recall.prompt" in report.applied