diff --git a/scripts/__init__.py b/scripts/__init__.py new file mode 100644 index 0000000..af1e95b --- /dev/null +++ b/scripts/__init__.py @@ -0,0 +1 @@ +# scripts package — exposes bootstrap and maintenance scripts as importable modules. diff --git a/scripts/build_judge_prompt_cache.py b/scripts/build_judge_prompt_cache.py new file mode 100644 index 0000000..4b52c52 --- /dev/null +++ b/scripts/build_judge_prompt_cache.py @@ -0,0 +1,119 @@ +"""One-off bootstrap: generate committed Chinese judge-prompt cache via RAGAS adapt(). + +Usage: + python -m scripts.build_judge_prompt_cache --language zh [--judge-model ] + +Runs BasePrompt.adapt("chinese", llm, adapt_instruction=True) for every judge +prompt of every LLM-scored metric and writes the result to +configs/judge_prompts//__.json. + +Re-run after a RAGAS upgrade to refresh the cache (the localizer will warn +via 'stale cache' log messages when the source hash changes). +Requires a working judge LLM configured in settings / LLM profiles. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import logging +from pathlib import Path +from typing import Any + +from rag_eval.compat import ensure_ragas_import_compat +from rag_eval.metrics.factory import build_metric_registry, build_models +from rag_eval.metrics.judge_prompts import CACHE_ROOT, METRIC_PROMPT_ATTRS, prompt_source_hash +from rag_eval.settings import EvaluationSettings + +ensure_ragas_import_compat() + +logger = logging.getLogger("scripts.build_judge_prompt_cache") + +# Map from judge_language short code to the natural-language name used by RAGAS adapt(). +_ADAPT_LANGUAGE: dict[str, str] = { + "zh": "chinese", + "en": "english", +} + + +def serialize_prompt( + adapted: Any, + source_hash: str, + metric: str, + attr: str, + language: str, + ragas_version: str, +) -> dict: + """Serialize an adapted prompt instance into the committed cache JSON schema.""" + return { + "metric": metric, + "prompt_attr": attr, + "language": getattr(adapted, "language", language), + "ragas_version": ragas_version, + "source_hash": source_hash, + "instruction": adapted.instruction, + "examples": [ + {"input": inp.model_dump(), "output": out.model_dump()} + for inp, out in adapted.examples + ], + } + + +async def build_cache( + language: str, + judge_model: str, + embedding_model: str, + settings: Any, +) -> list[Path]: + """Generate and write the full prompt cache for one language; return written paths. + + Each prompt is adapted once, serialized, and written atomically (temp file rename + is not used — if the script crashes mid-way, re-running overwrites partial files). + """ + import ragas + + target = _ADAPT_LANGUAGE.get(language, language) + llm, _embeddings = build_models(judge_model, embedding_model, settings) + registry = build_metric_registry(llm, _embeddings) + out_dir = CACHE_ROOT / language + out_dir.mkdir(parents=True, exist_ok=True) + + written: list[Path] = [] + for metric_name, attrs in METRIC_PROMPT_ATTRS.items(): + for attr in attrs: + metric = registry[metric_name] + prompt = getattr(metric, attr) + src_hash = prompt_source_hash(prompt) + logger.info("adapting %s.%s …", metric_name, attr) + adapted = await prompt.adapt(target, llm, adapt_instruction=True) + data = serialize_prompt(adapted, src_hash, metric_name, attr, language, ragas.__version__) + path = out_dir / f"{metric_name}__{attr}.json" + path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") + written.append(path) + logger.info(" written → %s", path.name) + + return written + + +def main() -> None: + """CLI entry point: parse args, resolve models, and build the cache.""" + parser = argparse.ArgumentParser( + description="Build the RAGAS judge-prompt translation cache for a target language." + ) + parser.add_argument("--language", default="zh", help="Target language code (default: zh)") + parser.add_argument("--judge-model", default=None, help="Override judge model from settings") + parser.add_argument("--embedding-model", default=None, help="Override embedding model") + args = parser.parse_args() + + logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") + settings = EvaluationSettings() + judge_model = args.judge_model or settings.ragas_judge_model + embedding_model = args.embedding_model or settings.ragas_embedding_model + + paths = asyncio.run(build_cache(args.language, judge_model, embedding_model, settings)) + logger.info("Done: %d cache files written to configs/judge_prompts/%s/", len(paths), args.language) + + +if __name__ == "__main__": + main() diff --git a/tests/test_judge_prompt_cache_builder.py b/tests/test_judge_prompt_cache_builder.py new file mode 100644 index 0000000..0d9454a --- /dev/null +++ b/tests/test_judge_prompt_cache_builder.py @@ -0,0 +1,80 @@ +"""Tests for the prompt-cache bootstrap script (mocked adapt/LLM).""" + +import asyncio +import json + +from pydantic import BaseModel + +import scripts.build_judge_prompt_cache as builder + + +class _In(BaseModel): + question: str + + +class _Out(BaseModel): + statements: list[str] + + +class _FakeAdapted: + """Fake adapted prompt returned by prompt.adapt().""" + + def __init__(self): + self.instruction = "中文指令。" + self.language = "chinese" + self.examples = [(_In(question="问题"), _Out(statements=["陈述"]))] + + +class _FakePrompt: + """Fake prompt with async adapt() that returns _FakeAdapted.""" + + def __init__(self): + self.input_model = _In + self.output_model = _Out + self.instruction = "English." + self.examples = [(_In(question="q"), _Out(statements=["s"]))] + self.language = "english" + + async def adapt(self, target_language, llm, adapt_instruction=False): + assert target_language == "chinese" + assert adapt_instruction is True + return _FakeAdapted() + + +def test_serialize_prompt_shape(): + """serialize_prompt emits the expected committed cache schema.""" + data = builder.serialize_prompt( + _FakeAdapted(), "hash123", "context_recall", "prompt", "zh", "0.4.3" + ) + assert data["metric"] == "context_recall" + assert data["prompt_attr"] == "prompt" + assert data["language"] == "chinese" + assert data["source_hash"] == "hash123" + assert data["instruction"] == "中文指令。" + assert data["examples"] == [ + {"input": {"question": "问题"}, "output": {"statements": ["陈述"]}} + ] + + +def test_build_cache_writes_all_files(tmp_path, monkeypatch): + """build_cache writes one JSON per (metric, attr) under CACHE_ROOT//.""" + # Build a fake registry where every prompt attr is a _FakePrompt. + fake_registry = {name: type("M", (), {})() for name in builder.METRIC_PROMPT_ATTRS} + for name, attrs in builder.METRIC_PROMPT_ATTRS.items(): + for attr in attrs: + setattr(fake_registry[name], attr, _FakePrompt()) + + monkeypatch.setattr(builder, "build_models", lambda j, e, s: (object(), object())) + monkeypatch.setattr(builder, "build_metric_registry", lambda llm, emb: fake_registry) + monkeypatch.setattr(builder, "CACHE_ROOT", tmp_path) + + written = asyncio.run(builder.build_cache("zh", "gpt-5", "emb", object())) + + expected_count = sum(len(a) for a in builder.METRIC_PROMPT_ATTRS.values()) + assert len(written) == expected_count + + sample = tmp_path / "zh" / "context_recall__prompt.json" + assert sample.exists() + data = json.loads(sample.read_text(encoding="utf-8")) + assert data["instruction"] == "中文指令。" + assert data["metric"] == "context_recall"