AIRegulation/siemens_ragas
7
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
465845b2a1667f9322960040ca7f5153adb215e4
siemens_ragas/docs/superpowers/plans/2026-07-01-chinese-judge-prompt.md
T

48 KiB
Raw Blame History

中文评判 Prompt 适配 Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 让 RAGAS 的 6 个 LLM 评判指标可切换为中文 prompt，通过场景 YAML judge_language: zh 与 score API 可选字段 judge_language 控制，默认 en 完全向后兼容。

Architecture: 用 RAGAS 原生 BasePrompt.adapt("chinese", llm, adapt_instruction=True) 一次性生成中文 prompt，序列化为提交入库的缓存 JSON。运行时一个共享本地化器把缓存覆盖到指标实例的 prompt 属性上，两条路径（YAML 场景 factory.build_metric_pipeline / score API inline_scorer）共用它。缺失或漂移的缓存优雅降级回英文，绝不中断评分。

Tech Stack: Python 3.12、RAGAS 0.4.3（ragas.metrics.collections）、Pydantic v2、pydantic-settings、FastAPI、pytest。

Global Constraints

  • Python 3.12+，PEP 8，4 空格缩进；公共函数加类型注解；每个函数加函数注释，非显然逻辑加行内注释（AGENTS.md）。
  • 测试用 pytest，放在 tests/，命名 test_*.py；确定性、Mock 外部调用，不依赖在线模型 API（AGENTS.md）。
  • 运行单测命令：python -m pytest tests/<file>::<test> -v（本机 python = C:\software\Python312\python.exe，pytest 9.0.3）。
  • 提交信息用简短祈使句（如 Add ...），每次提交聚焦一个逻辑变更；提交尾部加 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>。
  • 需本地化的指标→prompt 属性映射（已核实 RAGAS 0.4.3 源码，权威，勿改）：
    • faithfulness: statement_generator_prompt, nli_statement_prompt
    • answer_relevancy: prompt
    • context_recall: prompt
    • context_precision: prompt
    • noise_sensitivity: 函数式 to_string()，无 instruction/examples，无法 adapt，跳过
    • factual_correctness: prompt, nli_prompt
    • semantic_similarity: 无（跳过）
  • 语言优先级（两条路径一致）：显式值（YAML/请求）> settings.ragas_judge_language（默认 "en"）；仅规范化后等于 "zh" 触发本地化，其余按英文。
  • 缓存目录：configs/judge_prompts/<language>/<metric>__<attr>.json（如 configs/judge_prompts/zh/faithfulness__nli_statement_prompt.json）。
  • 设计文档：docs/superpowers/specs/2026-07-01-chinese-judge-prompt-design.md。

File Structure

新增：

  • rag_eval/metrics/judge_prompts.py — 本地化器：映射表、LocalizationReport、prompt_source_hash、缓存加载（内存 memo）、apply_localized_prompt、localize_pipeline_prompts、reset_cache。
  • scripts/build_judge_prompt_cache.py — 一次性引导脚本，用 adapt() 生成缓存 JSON。
  • configs/judge_prompts/zh/*.json — 提交入库的中文 prompt 缓存（Task 8 生成）。
  • tests/test_judge_prompt_localizer.py — 本地化器单测。
  • tests/test_judge_prompt_cache_builder.py — 引导脚本单测（mock LLM）。
  • tests/test_judge_language_config.py — settings/Scenario/ScoreRequest 配置单测。

修改：

  • rag_eval/settings.py — 新增 ragas_judge_language。
  • rag_eval/config/schema.py — ScenarioModel.judge_language。
  • rag_eval/config/loader.py — 透传 judge_language。
  • rag_eval/shared/models.py — Scenario.judge_language 字段。
  • rag_eval/metrics/factory.py — 新增 build_metric_registry()；build_metric_pipeline() 接入本地化。
  • webapp/models.py — ScoreRequest.judge_language。
  • webapp/services/inline_scorer.py — score()/_build_metric_instances() +参数+本地化，复用 build_metric_registry。
  • webapp/api/score.py、webapp/services/score_job_manager.py、webapp/services/session_score_manager.py — 透传 judge_language。
  • scenarios/siemens_build/siemens-pdf-question-bank-online.yaml（或现有 siemens 评估场景）+ 一个 offline 示例 — 增加 judge_language: zh。
  • README.md — 简述机制与重新生成缓存的命令。

Task 1: 配置管道（settings + Scenario 字段）

新增 judge_language 配置面，暂不产生行为，只让值能被读取与校验。

Files:

  • Modify: rag_eval/settings.py:24
  • Modify: rag_eval/config/schema.py:44-59
  • Modify: rag_eval/config/loader.py:48-67
  • Modify: rag_eval/shared/models.py:66-81
  • Test: tests/test_judge_language_config.py

Interfaces:

  • Produces:

    • EvaluationSettings().ragas_judge_language: str（默认 "en"，env RAGAS_JUDGE_LANGUAGE）
    • ScenarioModel.judge_language: Literal["en", "zh"] | None（默认 None）
    • Scenario.judge_language: str | None（默认 None）

  • Step 1: 写失败测试

创建 tests/test_judge_language_config.py：

"""Tests for judge_language plumbing across settings, scenario schema, and loader."""

from pathlib import Path

import pytest

from rag_eval.settings import EvaluationSettings
from rag_eval.config.loader import load_scenario


def test_settings_default_judge_language_is_en():
    """ragas_judge_language defaults to 'en' when the env var is absent."""
    settings = EvaluationSettings(_env_file=None)
    assert settings.ragas_judge_language == "en"


def _write_scenario(tmp_path: Path, extra: str) -> Path:
    """Write a minimal valid offline scenario YAML plus the given extra line(s)."""
    dataset = tmp_path / "data.csv"
    dataset.write_text("sample_id,question,answer,contexts,ground_truth\n", encoding="utf-8")
    text = (
        "scenario_name: t\n"
        "mode: offline\n"
        f"dataset: {dataset.name}\n"
        "judge_model: gpt-5\n"
        "embedding_model: text-embedding-3-small\n"
        "metrics: [faithfulness]\n"
        "output_dir: out\n"
        f"{extra}"
    )
    path = tmp_path / "s.yaml"
    path.write_text(text, encoding="utf-8")
    return path


def test_scenario_loads_judge_language_zh(tmp_path):
    """A scenario may declare judge_language: zh and it lands on the dataclass."""
    path = _write_scenario(tmp_path, "judge_language: zh\n")
    scenario = load_scenario(path)
    assert scenario.judge_language == "zh"


def test_scenario_defaults_judge_language_none(tmp_path):
    """Omitting judge_language leaves it None so the factory can apply the settings default."""
    path = _write_scenario(tmp_path, "")
    scenario = load_scenario(path)
    assert scenario.judge_language is None


def test_scenario_rejects_invalid_judge_language(tmp_path):
    """An unsupported judge_language value is rejected at schema validation."""
    path = _write_scenario(tmp_path, "judge_language: fr\n")
    with pytest.raises(Exception):
        load_scenario(path)
  • Step 2: 运行测试确认失败

Run: python -m pytest tests/test_judge_language_config.py -v Expected: FAIL（AttributeError: ragas_judge_language 或 Scenario 无 judge_language）。

  • Step 3: 实现配置字段

在 rag_eval/settings.py 第 24 行 ragas_judge_model 之后新增：

    ragas_judge_language: str = Field(default="en", alias="RAGAS_JUDGE_LANGUAGE")

在 rag_eval/config/schema.py 的 ScenarioModel 中（embedding_model 字段之后）新增：

    judge_language: Literal["en", "zh"] | None = None

（Literal 已在该文件顶部 from typing import Any, Literal 导入，无需改导入。）

在 rag_eval/config/loader.py 的 Scenario(...) 构造里（embedding_model=model.embedding_model, 之后）新增：

        judge_language=model.judge_language,

在 rag_eval/shared/models.py 的 Scenario dataclass 中（doc_weights 之后，作为带默认值字段）新增：

    judge_language: str | None = None
  • Step 4: 运行测试确认通过

Run: python -m pytest tests/test_judge_language_config.py -v Expected: 4 passed。

  • Step 5: 提交
git add rag_eval/settings.py rag_eval/config/schema.py rag_eval/config/loader.py rag_eval/shared/models.py tests/test_judge_language_config.py
git commit -m "Add judge_language config plumbing (settings + scenario)" -m "Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>"

Task 2: ScoreRequest 增加 judge_language 字段

让 score 系列 API 能接收可选 judge_language；SessionScoreRequest 继承自动获得。

Files:

  • Modify: webapp/models.py:472-475
  • Test: tests/test_judge_language_config.py（追加）

Interfaces:

  • Produces: ScoreRequest.judge_language: str | None（默认 None），SessionScoreRequest 继承。

  • Step 1: 写失败测试

在 tests/test_judge_language_config.py 末尾追加：

def test_score_request_judge_language_defaults_none():
    """ScoreRequest exposes an optional judge_language defaulting to None."""
    from webapp.models import ScoreRequest

    req = ScoreRequest(question="q", answer="a")
    assert req.judge_language is None
    req_zh = ScoreRequest(question="q", answer="a", judge_language="zh")
    assert req_zh.judge_language == "zh"


def test_session_score_request_inherits_judge_language():
    """SessionScoreRequest inherits the judge_language field from ScoreRequest."""
    from webapp.models import SessionScoreRequest

    req = SessionScoreRequest(session_id="s1", question="q", answer="a", judge_language="zh")
    assert req.judge_language == "zh"
  • Step 2: 运行测试确认失败

Run: python -m pytest tests/test_judge_language_config.py -k judge_language -v Expected: FAIL（ScoreRequest 无 judge_language）。

  • Step 3: 实现字段

在 webapp/models.py 的 ScoreRequest 中，embedding_model 字段（第 472-475 行）之后新增：

    judge_language: str | None = Field(
        default=None,
        description="评判 prompt 语言；'zh' 启用中文评判，为 null 时使用 RAGAS_JUDGE_LANGUAGE（默认 en）。",
    )
  • Step 4: 运行测试确认通过

Run: python -m pytest tests/test_judge_language_config.py -k judge_language -v Expected: 2 passed（若 SessionScoreRequest 必填字段名不同，按其真实必填字段调整测试构造参数后再跑）。

  • Step 5: 提交
git add webapp/models.py tests/test_judge_language_config.py
git commit -m "Add optional judge_language field to ScoreRequest" -m "Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>"

Task 3: 指标 registry 复用（DRY 重构）

抽出 build_metric_registry()，供 factory、inline_scorer、引导脚本共用，消除三处重复。

Files:

  • Modify: rag_eval/metrics/factory.py:114-131
  • Modify: webapp/services/inline_scorer.py:34-45
  • Test: tests/test_metric_registry.py

Interfaces:

  • Produces: rag_eval.metrics.factory.build_metric_registry(llm: Any, embeddings: Any) -> dict[str, Any]，返回全部 7 个指标实例。

  • Step 1: 写失败测试

创建 tests/test_metric_registry.py：

"""Tests for the shared metric registry factory."""

from rag_eval.metrics.factory import build_metric_registry


def test_build_metric_registry_has_all_seven_metrics():
    """The registry exposes every supported metric keyed by its canonical name."""
    registry = build_metric_registry(llm=object(), embeddings=object())
    assert set(registry) == {
        "faithfulness", "answer_relevancy", "context_recall", "context_precision",
        "noise_sensitivity", "factual_correctness", "semantic_similarity",
    }
  • Step 2: 运行测试确认失败

Run: python -m pytest tests/test_metric_registry.py -v Expected: FAIL（ImportError: cannot import name 'build_metric_registry'）。

  • Step 3: 实现 registry 工厂并复用

在 rag_eval/metrics/factory.py 新增函数（放在 build_metric_pipeline 之前）：

def build_metric_registry(llm: Any, embeddings: Any) -> dict[str, Any]:
    """Instantiate the full set of supported RAGAS metrics keyed by canonical name.

    Shared by the scenario pipeline, the inline scorer, and the prompt-cache
    bootstrap so the metric set is defined in exactly one place.
    """
    return {
        "faithfulness": Faithfulness(llm=llm),
        "answer_relevancy": AnswerRelevancy(llm=llm, embeddings=embeddings),
        "context_recall": ContextRecall(llm=llm),
        "context_precision": ContextPrecision(llm=llm),
        "noise_sensitivity": NoiseSensitivity(llm=llm),
        "factual_correctness": FactualCorrectness(llm=llm),
        "semantic_similarity": SemanticSimilarity(embeddings=embeddings),
    }

把 build_metric_pipeline 中内联的 registry = { ... }（第 114-127 行）替换为：

    registry = build_metric_registry(llm, embeddings)

在 webapp/services/inline_scorer.py 把 _build_metric_instances（第 34-45 行）改为复用：

def _build_metric_instances(metrics: list[str], llm: Any, embeddings: Any) -> dict[str, Any]:
    """Instantiate only the RAGAS metric objects requested."""
    from rag_eval.metrics.factory import build_metric_registry

    registry = build_metric_registry(llm, embeddings)
    return {name: registry[name] for name in metrics if name in registry}

并删除该文件顶部不再使用的 from ragas.metrics.collections import (...) 导入块（第 23-31 行）。

  • Step 4: 运行测试确认通过

Run: python -m pytest tests/test_metric_registry.py tests/test_score_api.py -v Expected: registry 测试 passed；既有 score API 测试保持 passed（若个别用例需真实 LLM 则会 skip/无关，重点是不新增失败）。

  • Step 5: 提交
git add rag_eval/metrics/factory.py webapp/services/inline_scorer.py tests/test_metric_registry.py
git commit -m "Extract shared build_metric_registry factory" -m "Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>"

Task 4: 本地化器核心 judge_prompts.py

纯逻辑模块：加载缓存、覆盖 prompt、漂移检测、内存 memo、优雅降级。用假 prompt 对象测试，不触碰 RAGAS。

Files:

  • Create: rag_eval/metrics/judge_prompts.py
  • Test: tests/test_judge_prompt_localizer.py

Interfaces:

  • Produces:

    • METRIC_PROMPT_ATTRS: dict[str, tuple[str, ...]]
    • CACHE_ROOT: Path
    • LocalizationReport(language, applied, skipped, warnings)（dataclass）
    • prompt_source_hash(prompt) -> str
    • apply_localized_prompt(prompt, data: dict) -> None
    • localize_pipeline_prompts(registry: dict[str, Any], language: str) -> LocalizationReport
    • reset_cache() -> None

  • Step 1: 写失败测试

创建 tests/test_judge_prompt_localizer.py：

"""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 <root>/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
  • Step 2: 运行测试确认失败

Run: python -m pytest tests/test_judge_prompt_localizer.py -v Expected: FAIL（ModuleNotFoundError / 函数缺失）。

  • Step 3: 实现本地化器

创建 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
  • Step 4: 运行测试确认通过

Run: python -m pytest tests/test_judge_prompt_localizer.py -v Expected: 6 passed。

  • Step 5: 提交
git add rag_eval/metrics/judge_prompts.py tests/test_judge_prompt_localizer.py
git commit -m "Add judge-prompt localizer with graceful fallback and drift detection" -m "Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>"

Task 5: 接入 factory 与 inline_scorer

让两条运行时路径按语言实际调用本地化器。

Files:

  • Modify: rag_eval/metrics/factory.py:96-131
  • Modify: webapp/services/inline_scorer.py:80-114
  • Test: tests/test_judge_language_wiring.py

Interfaces:

  • Consumes: build_metric_registry（Task 3）、localize_pipeline_prompts（Task 4）、Scenario.judge_language（Task 1）、settings.ragas_judge_language（Task 1）。

  • Produces:

    • build_metric_pipeline(scenario, settings, llm=None, embeddings=None) 在语言解析为 zh 时本地化。
    • InlineScorer.score(..., judge_language: str = "en")；_build_metric_instances(metrics, llm, embeddings, judge_language="en")。

  • Step 1: 写失败测试

创建 tests/test_judge_language_wiring.py：

"""Tests that the factory and inline scorer invoke the localizer per language."""

import rag_eval.metrics.factory as factory
import webapp.services.inline_scorer as inline_mod


def test_build_pipeline_localizes_when_zh(monkeypatch):
    """build_metric_pipeline calls localize_pipeline_prompts with the resolved language."""
    calls = []
    monkeypatch.setattr(
        factory, "localize_pipeline_prompts",
        lambda registry, language: calls.append(language),
    )
    from rag_eval.shared.models import DatasetConfig, Scenario
    from rag_eval.settings import EvaluationSettings
    from pathlib import Path

    scenario = Scenario(
        scenario_name="t", mode="offline",
        dataset=DatasetConfig(path=Path("x.csv")),
        judge_model="gpt-5", embedding_model="text-embedding-3-small",
        metrics=["faithfulness"], output_dir=Path("out"),
        judge_language="zh",
    )
    factory.build_metric_pipeline(scenario, EvaluationSettings(_env_file=None),
                                  llm=object(), embeddings=object())
    assert calls == ["zh"]


def test_build_pipeline_falls_back_to_settings_default(monkeypatch):
    """When scenario.judge_language is None the settings default is used."""
    calls = []
    monkeypatch.setattr(
        factory, "localize_pipeline_prompts",
        lambda registry, language: calls.append(language),
    )
    from rag_eval.shared.models import DatasetConfig, Scenario
    from rag_eval.settings import EvaluationSettings
    from pathlib import Path

    scenario = Scenario(
        scenario_name="t", mode="offline",
        dataset=DatasetConfig(path=Path("x.csv")),
        judge_model="gpt-5", embedding_model="text-embedding-3-small",
        metrics=["faithfulness"], output_dir=Path("out"),
        judge_language=None,
    )
    settings = EvaluationSettings(_env_file=None)
    settings.ragas_judge_language = "zh"
    factory.build_metric_pipeline(scenario, settings, llm=object(), embeddings=object())
    assert calls == ["zh"]


def test_inline_score_threads_judge_language(monkeypatch):
    """InlineScorer.score forwards judge_language into _build_metric_instances."""
    seen = {}
    monkeypatch.setattr(inline_mod.InlineScorer, "_get_models",
                        lambda self, j, e, s: (object(), object()))
    monkeypatch.setattr(
        inline_mod, "_build_metric_instances",
        lambda metrics, llm, embeddings, judge_language="en": seen.setdefault("lang", judge_language) or {},
    )

    class _Pipe:
        def __init__(self, *a, **k):
            pass

    monkeypatch.setattr(inline_mod, "MetricPipeline", _Pipe)
    monkeypatch.setattr(inline_mod.asyncio, "run", lambda coro: type("R", (), {"metrics": {}})())

    scorer = inline_mod.InlineScorer()
    scorer.score(question="q", answer="a", contexts=[], ground_truth=None,
                 metrics=["faithfulness"], judge_model="gpt-5",
                 embedding_model="e", settings=object(), judge_language="zh")
    assert seen["lang"] == "zh"
  • Step 2: 运行测试确认失败

Run: python -m pytest tests/test_judge_language_wiring.py -v Expected: FAIL（factory 未导入/调用 localizer；score() 无 judge_language 参数）。

  • Step 3: 实现接入

在 rag_eval/metrics/factory.py 顶部（第 27 行 from .pipeline import MetricPipeline 之后）新增导入：

from .judge_prompts import localize_pipeline_prompts

把 build_metric_pipeline 结尾（构建 registry 之后、return MetricPipeline(...) 处）改为先切片再本地化：

    registry = build_metric_registry(llm, embeddings)
    selected = {name: registry[name] for name in scenario.metrics}
    language = (scenario.judge_language or settings.ragas_judge_language or "en")
    localize_pipeline_prompts(selected, language)
    return MetricPipeline(
        metrics=selected,
        metric_timeout_seconds=settings.ragas_metric_timeout_seconds,
    )

在 webapp/services/inline_scorer.py：把 _build_metric_instances 增加 judge_language 形参并本地化：

def _build_metric_instances(
    metrics: list[str], llm: Any, embeddings: Any, judge_language: str = "en"
) -> dict[str, Any]:
    """Instantiate only the RAGAS metric objects requested, localized if needed."""
    from rag_eval.metrics.factory import build_metric_registry
    from rag_eval.metrics.judge_prompts import localize_pipeline_prompts

    registry = build_metric_registry(llm, embeddings)
    selected = {name: registry[name] for name in metrics if name in registry}
    localize_pipeline_prompts(selected, judge_language)
    return selected

把 InlineScorer.score 签名（第 80-90 行）末尾加参数并在调用处透传：

    def score(
        self,
        question: str,
        answer: str,
        contexts: list[str],
        ground_truth: str | None,
        metrics: list[str],
        judge_model: str,
        embedding_model: str,
        settings: EvaluationSettings,
        judge_language: str = "en",
    ) -> dict[str, float | None]:
        """Score one sample synchronously and return {metric_name: score | None}."""
        llm, embeddings = self._get_models(judge_model, embedding_model, settings)
        metric_instances = _build_metric_instances(metrics, llm, embeddings, judge_language)

（其余函数体不变。）

  • Step 4: 运行测试确认通过

Run: python -m pytest tests/test_judge_language_wiring.py -v Expected: 3 passed。

  • Step 5: 提交
git add rag_eval/metrics/factory.py webapp/services/inline_scorer.py tests/test_judge_language_wiring.py
git commit -m "Wire judge-prompt localization into factory and inline scorer" -m "Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>"

Task 6: 三个 score 端点透传 judge_language

Files:

  • Modify: webapp/api/score.py:117-142
  • Modify: webapp/services/score_job_manager.py:120-141
  • Modify: webapp/services/session_score_manager.py:205-226
  • Test: tests/test_score_endpoint_language.py

Interfaces:

  • Consumes: ScoreRequest.judge_language（Task 2）、settings.ragas_judge_language、InlineScorer.score(..., judge_language=...)（Task 5）。

  • Step 1: 写失败测试

创建 tests/test_score_endpoint_language.py：

"""The /api/score route forwards the resolved judge_language to the scorer."""

from fastapi.testclient import TestClient

import webapp.api.score as score_mod
from webapp.server import create_app


def test_score_route_forwards_judge_language(monkeypatch):
    """A request with judge_language='zh' reaches inline_scorer.score."""
    captured = {}

    def fake_score(**kwargs):
        captured.update(kwargs)
        return {"faithfulness": 0.9}

    monkeypatch.setattr(score_mod.inline_scorer, "score", lambda **kw: fake_score(**kw))

    client = TestClient(create_app())
    resp = client.post("/api/score", json={
        "question": "q", "answer": "a", "contexts": "c",
        "ground_truth": "g", "metrics": ["faithfulness"], "judge_language": "zh",
    })
    assert resp.status_code == 200
    assert captured.get("judge_language") == "zh"


def test_score_route_defaults_language_from_settings(monkeypatch):
    """Omitting judge_language falls back to settings.ragas_judge_language."""
    captured = {}
    monkeypatch.setattr(score_mod.inline_scorer, "score",
                        lambda **kw: captured.update(kw) or {"faithfulness": 0.9})

    client = TestClient(create_app())
    resp = client.post("/api/score", json={
        "question": "q", "answer": "a", "contexts": "c",
        "ground_truth": "g", "metrics": ["faithfulness"],
    })
    assert resp.status_code == 200
    assert captured.get("judge_language") == "en"

（若 create_app 的导入路径不同，先 grep -n "def create_app" webapp/server.py 校正 import。）

  • Step 2: 运行测试确认失败

Run: python -m pytest tests/test_score_endpoint_language.py -v Expected: FAIL（score() 未收到 judge_language）。

  • Step 3: 实现透传

webapp/api/score.py：在 judge_model = request.judge_model or settings.ragas_judge_model（第 117 行）附近新增解析，并在 inline_scorer.score(...) 调用（第 133-142 行）加参数：

    judge_model = request.judge_model or settings.ragas_judge_model
    embedding_model = request.embedding_model or settings.ragas_embedding_model
    judge_language = request.judge_language or settings.ragas_judge_language

        raw_scores = inline_scorer.score(
            question=request.question,
            answer=request.answer,
            contexts=request.contexts_as_list(),
            ground_truth=request.ground_truth,
            metrics=effective,
            judge_model=judge_model,
            embedding_model=embedding_model,
            settings=settings,
            judge_language=judge_language,
        )

webapp/services/score_job_manager.py：在第 121-122 行之后新增，并在第 132-141 行的 inline_scorer.score(...) 加同名参数：

        judge_language = request.judge_language or settings.ragas_judge_language

                    judge_language=judge_language,

webapp/services/session_score_manager.py：在第 206-207 行之后新增，并在第 217-226 行的 inline_scorer.score(...) 加同名参数：

        judge_language = request.judge_language or settings.ragas_judge_language

                    judge_language=judge_language,
  • Step 4: 运行测试确认通过

Run: python -m pytest tests/test_score_endpoint_language.py -v Expected: 2 passed。

  • Step 5: 提交
git add webapp/api/score.py webapp/services/score_job_manager.py webapp/services/session_score_manager.py tests/test_score_endpoint_language.py
git commit -m "Forward judge_language through score, async, and session endpoints" -m "Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>"

Task 7: 引导脚本 build_judge_prompt_cache.py

用 RAGAS 原生 adapt() 生成缓存 JSON。用 mock LLM/prompt 测试，不触发真实网络。

Files:

  • Create: scripts/build_judge_prompt_cache.py
  • Test: tests/test_judge_prompt_cache_builder.py

Interfaces:

  • Consumes: build_metric_registry（Task 3）、METRIC_PROMPT_ATTRS/CACHE_ROOT/prompt_source_hash（Task 4）、build_models（既有）。

  • Produces:

    • serialize_prompt(adapted, source_hash, metric, attr, language, ragas_version) -> dict
    • async build_cache(language, judge_model, embedding_model, settings) -> list[Path]

  • Step 1: 写失败测试

创建 tests/test_judge_prompt_cache_builder.py：

"""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:
    def __init__(self):
        self.instruction = "中文指令。"
        self.language = "chinese"
        self.examples = [(_In(question="问题"), _Out(statements=["陈述"]))]


class _FakePrompt:
    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 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) into CACHE_ROOT/<language>/."""
    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 = sum(len(a) for a in builder.METRIC_PROMPT_ATTRS.values())
    assert len(written) == expected
    sample = tmp_path / "zh" / "context_recall__prompt.json"
    assert sample.exists()
    data = json.loads(sample.read_text(encoding="utf-8"))
    assert data["instruction"] == "中文指令。"
  • Step 2: 运行测试确认失败

Run: python -m pytest tests/test_judge_prompt_cache_builder.py -v Expected: FAIL（模块不存在）。

  • Step 3: 实现脚本

创建 scripts/build_judge_prompt_cache.py：

"""One-off bootstrap: generate committed Chinese judge-prompt cache via RAGAS adapt().

Usage:
    python -m scripts.build_judge_prompt_cache --language zh

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/<language>/<metric>__<attr>.json. Re-run after a RAGAS
upgrade to refresh the cache. Requires a working judge LLM.
"""

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")

# judge_language short code -> RAGAS adapt() natural-language name.
_ADAPT_LANGUAGE = {"zh": "chinese"}


def serialize_prompt(
    adapted: Any, source_hash: str, metric: str, attr: str, language: str, ragas_version: str
) -> dict:
    """Serialize an adapted prompt 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: EvaluationSettings
) -> list[Path]:
    """Generate and write the full prompt cache for one language; return written paths."""
    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, attrs in METRIC_PROMPT_ATTRS.items():
        for attr in attrs:
            prompt = getattr(registry[metric], attr)
            source_hash = prompt_source_hash(prompt)
            adapted = await prompt.adapt(target, llm, adapt_instruction=True)
            data = serialize_prompt(adapted, source_hash, metric, attr, language, ragas.__version__)
            path = out_dir / f"{metric}__{attr}.json"
            path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
            written.append(path)
            logger.info("wrote %s", path)
    return written


def main() -> None:
    """CLI entry point: parse args, resolve models, and build the cache."""
    parser = argparse.ArgumentParser(description="Build the Chinese judge-prompt cache.")
    parser.add_argument("--language", default="zh")
    parser.add_argument("--judge-model", default=None)
    parser.add_argument("--embedding-model", default=None)
    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 files written", len(paths))


if __name__ == "__main__":
    main()

同时确保 scripts/ 可作为包导入：若 scripts/__init__.py 不存在则创建空文件（测试用 import scripts.build_judge_prompt_cache 需要）。

  • Step 4: 运行测试确认通过

Run: python -m pytest tests/test_judge_prompt_cache_builder.py -v Expected: 2 passed。

  • Step 5: 提交
git add scripts/build_judge_prompt_cache.py scripts/__init__.py tests/test_judge_prompt_cache_builder.py
git commit -m "Add judge-prompt cache bootstrap script" -m "Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>"

Task 8: 生成中文缓存 + 示例场景 + 文档

产出真实中文缓存、暴露配置示例、补文档。这是让功能端到端可用的收尾任务。

Files:

  • Create: configs/judge_prompts/zh/*.json（10 个文件）

  • Modify: 一个 siemens 评估场景 YAML + 一个 offline 示例 YAML（增加 judge_language: zh）

  • Modify: README.md

  • Step 1: 生成中文缓存

有可用评判 LLM 时运行（生产做法）：

python -m scripts.build_judge_prompt_cache --language zh

预期在 configs/judge_prompts/zh/ 生成 10 个文件： faithfulness__statement_generator_prompt.json、faithfulness__nli_statement_prompt.json、answer_relevancy__prompt.json、context_recall__prompt.json、context_precision__prompt.json、noise_sensitivity__statement_prompt.json、noise_sensitivity__faithfulness_prompt.json、factual_correctness__prompt.json、factual_correctness__nli_prompt.json（注意 answer_relevancy 与部分指标各一个）。

无 LLM 访问时的兜底：按 Task 4 缓存 schema 手工撰写等价中文 JSON（instruction 译为中文、examples 内字符串译为中文、source_hash 用当前英文源经 prompt_source_hash 计算填入；结构字段名/枚举/数字保持不变）。功能与测试不依赖翻译质量，可后续用脚本刷新。

  • Step 2: 验证缓存被正确加载

写一个临时验证（跑完删除或保留为集成测试 tests/test_zh_cache_integration.py）：

"""Integration: committed zh cache loads onto a real metric instance."""

from rag_eval.metrics.judge_prompts import CACHE_ROOT, localize_pipeline_prompts


def test_zh_cache_files_present():
    """All 10 expected zh cache files are committed."""
    expected = [
        "faithfulness__statement_generator_prompt.json",
        "faithfulness__nli_statement_prompt.json",
        "answer_relevancy__prompt.json",
        "context_recall__prompt.json",
        "context_precision__prompt.json",
        "noise_sensitivity__statement_prompt.json",
        "noise_sensitivity__faithfulness_prompt.json",
        "factual_correctness__prompt.json",
        "factual_correctness__nli_prompt.json",
    ]
    for name in expected:
        assert (CACHE_ROOT / "zh" / name).exists(), f"missing {name}"

Run: python -m pytest tests/test_zh_cache_integration.py -v Expected: PASS（确认 9 个必需文件在位；answer_relevancy 只有 1 个 prompt，总计与实际 METRIC_PROMPT_ATTRS 展开数一致）。

注：METRIC_PROMPT_ATTRS 展开后文件总数 = 2+1+1+1+2+2 = 9。若脚本因某指标含多 prompt 而不同，以 METRIC_PROMPT_ATTRS 为准同步该测试清单。

  • Step 3: 场景 YAML 增加 judge_language

在一个 siemens 评估场景与一个 offline 示例的顶层加入：

judge_language: zh

（用 grep -rl "mode: offline" scenarios/ 找到目标文件；仅改评估场景，勿改 dataset_build 场景。）

  • Step 4: 补文档

在 README.md 增加一节「中文评判 Prompt 适配」，说明：

  • 场景 judge_language: zh 与 score API judge_language 字段、RAGAS_JUDGE_LANGUAGE 全局默认。

  • 重新生成缓存命令：python -m scripts.build_judge_prompt_cache --language zh。

  • RAGAS 升级后需重跑脚本（漂移检测会在日志告警）。

  • Step 5: 运行相关测试并提交

Run: python -m pytest tests/test_judge_prompt_localizer.py tests/test_zh_cache_integration.py -v Expected: 全部 passed。

git add configs/judge_prompts/zh/*.json scenarios/ README.md tests/test_zh_cache_integration.py
git commit -m "Add committed zh judge-prompt cache and enable it in sample scenarios" -m "Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>"

最终回归

  • 全量单测：python -m pytest tests/ -q
    • Expected: 本计划新增测试全部通过；不引入新的失败。已知的 6 个历史失败（见设计背景，与本功能无关）保持原状，勿在本计划内处理。

Self-Review 记录

  • Spec 覆盖：§4.4 配置面→Task1/2；§4.3 本地化器→Task4；§4.5 集成→Task5/6；§4.1 引导脚本→Task7；§4.2 缓存文件→Task8；§5 错误处理/漂移→Task4（missing/stale/apply-fail 测试）；§6 测试→各 Task；§2 指标映射→Task4 常量；§8 兼容性→默认 en no-op（Task4/5 覆盖）。
  • 占位符扫描：无 TBD/TODO；所有步骤含真实代码与命令。
  • 类型一致：localize_pipeline_prompts(registry, language)、build_metric_registry(llm, embeddings)、InlineScorer.score(..., judge_language="en")、serialize_prompt(...)、build_cache(...) 在定义与调用处签名一致。
Reference in New Issue View Git Blame Copy Permalink