1261 lines
48 KiB
Markdown
1261 lines
48 KiB
Markdown
# 中文评判 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`：
|
||
|
||
```python
|
||
"""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` 之后新增：
|
||
|
||
```python
|
||
ragas_judge_language: str = Field(default="en", alias="RAGAS_JUDGE_LANGUAGE")
|
||
```
|
||
|
||
在 `rag_eval/config/schema.py` 的 `ScenarioModel` 中（`embedding_model` 字段之后）新增：
|
||
|
||
```python
|
||
judge_language: Literal["en", "zh"] | None = None
|
||
```
|
||
|
||
（`Literal` 已在该文件顶部 `from typing import Any, Literal` 导入，无需改导入。）
|
||
|
||
在 `rag_eval/config/loader.py` 的 `Scenario(...)` 构造里（`embedding_model=model.embedding_model,` 之后）新增：
|
||
|
||
```python
|
||
judge_language=model.judge_language,
|
||
```
|
||
|
||
在 `rag_eval/shared/models.py` 的 `Scenario` dataclass 中（`doc_weights` 之后，作为带默认值字段）新增：
|
||
|
||
```python
|
||
judge_language: str | None = None
|
||
```
|
||
|
||
- [ ] **Step 4: 运行测试确认通过**
|
||
|
||
Run: `python -m pytest tests/test_judge_language_config.py -v`
|
||
Expected: 4 passed。
|
||
|
||
- [ ] **Step 5: 提交**
|
||
|
||
```bash
|
||
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` 末尾追加：
|
||
|
||
```python
|
||
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 行）之后新增：
|
||
|
||
```python
|
||
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: 提交**
|
||
|
||
```bash
|
||
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`：
|
||
|
||
```python
|
||
"""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` 之前）：
|
||
|
||
```python
|
||
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 行）替换为：
|
||
|
||
```python
|
||
registry = build_metric_registry(llm, embeddings)
|
||
```
|
||
|
||
在 `webapp/services/inline_scorer.py` 把 `_build_metric_instances`（第 34-45 行）改为复用：
|
||
|
||
```python
|
||
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: 提交**
|
||
|
||
```bash
|
||
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`：
|
||
|
||
```python
|
||
"""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`：
|
||
|
||
```python
|
||
"""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: 提交**
|
||
|
||
```bash
|
||
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`：
|
||
|
||
```python
|
||
"""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` 之后）新增导入：
|
||
|
||
```python
|
||
from .judge_prompts import localize_pipeline_prompts
|
||
```
|
||
|
||
把 `build_metric_pipeline` 结尾（构建 registry 之后、`return MetricPipeline(...)` 处）改为先切片再本地化：
|
||
|
||
```python
|
||
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` 形参并本地化：
|
||
|
||
```python
|
||
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 行）末尾加参数并在调用处透传：
|
||
|
||
```python
|
||
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: 提交**
|
||
|
||
```bash
|
||
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`：
|
||
|
||
```python
|
||
"""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 行）加参数：
|
||
|
||
```python
|
||
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
|
||
```
|
||
|
||
```python
|
||
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(...)` 加同名参数：
|
||
|
||
```python
|
||
judge_language = request.judge_language or settings.ragas_judge_language
|
||
```
|
||
```python
|
||
judge_language=judge_language,
|
||
```
|
||
|
||
`webapp/services/session_score_manager.py`：在第 206-207 行之后新增，并在第 217-226 行的 `inline_scorer.score(...)` 加同名参数：
|
||
|
||
```python
|
||
judge_language = request.judge_language or settings.ragas_judge_language
|
||
```
|
||
```python
|
||
judge_language=judge_language,
|
||
```
|
||
|
||
- [ ] **Step 4: 运行测试确认通过**
|
||
|
||
Run: `python -m pytest tests/test_score_endpoint_language.py -v`
|
||
Expected: 2 passed。
|
||
|
||
- [ ] **Step 5: 提交**
|
||
|
||
```bash
|
||
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`：
|
||
|
||
```python
|
||
"""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`：
|
||
|
||
```python
|
||
"""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: 提交**
|
||
|
||
```bash
|
||
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 时运行（生产做法）：
|
||
|
||
```bash
|
||
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`）：
|
||
|
||
```python
|
||
"""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 示例的顶层加入：
|
||
|
||
```yaml
|
||
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。
|
||
|
||
```bash
|
||
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(...)` 在定义与调用处签名一致。
|