jp170na/dp-zp-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Rag_answer_evaluation_podpora

Browse Source
This commit is contained in:
Ján Pták 2026-08-16 17:36:21 +02:00
parent cd3047511c
commit 7dab5a25ce
4 changed files with 2099 additions and 797 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1171
evaluation/evaluate_rag_answers.py
View File

File diff suppressed because it is too large Load Diff

315
evaluation/rag_answer_data.py Normal file
View File

@ -0,0 +1,315 @@
from __future__ import annotations
import hashlib
import json
from pathlib import Path
from typing import Any
from evaluation.retrieval_runner import filter_questions_by_split
DEFAULT_DELAY = 0.5
ALLOWED_OVERRIDE_FIELDS = {
"question",
"expected_answer_contains",
"expected_source_urls",
"should_answer",
"note",
}
def apply_question_overrides(
questions: list[dict[str, Any]],
path: Path,
) -> tuple[list[dict[str, Any]], list[str]]:
if not path.exists():
raise FileNotFoundError(
f"RAG answer override súbor neexistuje: {path}"
)
with path.open("r", encoding="utf-8") as file:
overrides = json.load(file)
if not isinstance(overrides, dict):
raise ValueError(
"rag_answer_overrides.json musí obsahovať JSON objekt."
)
known_ids = {str(item["id"]) for item in questions}
unknown_ids = set(overrides.keys()) - known_ids
if unknown_ids:
raise ValueError(
"Override súbor obsahuje neznáme question ID: "
+ ", ".join(sorted(unknown_ids))
)
result: list[dict[str, Any]] = []
applied_ids: list[str] = []
for question in questions:
item = dict(question)
question_id = str(item["id"])
override = overrides.get(question_id)
if override is None:
result.append(item)
continue
if not isinstance(override, dict):
raise ValueError(
f"{question_id}: override musí byť JSON objekt."
)
unsupported_fields = (
set(override.keys())
- ALLOWED_OVERRIDE_FIELDS
)
if unsupported_fields:
raise ValueError(
f"{question_id}: nepovolené override polia: "
+ ", ".join(
sorted(unsupported_fields)
)
)
if "question" in override:
overridden_question = override["question"]
if (
not isinstance(
overridden_question,
str,
)
or not overridden_question.strip()
):
raise ValueError(
f"{question_id}: "
"override question musí byť "
"neprázdny reťazec."
)
item.update(override)
result.append(item)
applied_ids.append(question_id)
return result, applied_ids
def validate_string_list(
question_id: str,
field_name: str,
value: Any,
) -> None:
if not isinstance(value, list):
raise ValueError(
f"{question_id}: "
f"{field_name} musí byť JSON pole."
)
for index, item in enumerate(value):
if not isinstance(item, str):
raise ValueError(
f"{question_id}: "
f"{field_name}[{index}] "
"musí byť reťazec."
)
def validate_answer_questions(
questions: list[dict[str, Any]],
) -> None:
seen_ids: set[str] = set()
for index, question in enumerate(
questions,
start=1,
):
if not isinstance(question, dict):
raise ValueError(
f"Otázka #{index} "
"musí byť JSON objekt."
)
question_id = question.get("id")
if (
not isinstance(question_id, str)
or not question_id.strip()
):
raise ValueError(
f"Otázka #{index}: "
"id musí byť neprázdny reťazec."
)
if question_id in seen_ids:
raise ValueError(
f"Duplicitné question ID: {question_id}"
)
seen_ids.add(question_id)
split = question.get("split")
if split not in {
"dev",
"test",
}:
raise ValueError(
f"{question_id}: "
"split musí byť "
"'dev' alebo 'test'."
)
question_text = question.get(
"question"
)
if (
not isinstance(
question_text,
str,
)
or not question_text.strip()
):
raise ValueError(
f"{question_id}: "
"question musí byť "
"neprázdny reťazec."
)
validate_string_list(
question_id,
"expected_documents",
question.get(
"expected_documents"
),
)
validate_string_list(
question_id,
"expected_source_urls",
question.get(
"expected_source_urls"
),
)
validate_string_list(
question_id,
"expected_answer_contains",
question.get(
"expected_answer_contains"
),
)
if not isinstance(
question.get(
"should_answer"
),
bool,
):
raise ValueError(
f"{question_id}: "
"should_answer musí byť boolean."
)
for field_name in (
"category",
"difficulty",
):
value = question.get(
field_name
)
if (
value is not None
and not isinstance(
value,
str,
)
):
raise ValueError(
f"{question_id}: "
f"{field_name} musí byť "
"reťazec."
)
note = question.get("note")
if (
note is not None
and not isinstance(
note,
str,
)
):
raise ValueError(
f"{question_id}: "
"note musí byť reťazec "
"alebo null."
)
def select_questions(
questions: list[dict[str, Any]],
*,
split: str,
question_ids: list[str],
limit: int | None,
) -> list[dict[str, Any]]:
selected = filter_questions_by_split(
questions,
split,
)
if question_ids:
wanted = set(question_ids)
selected = [
item
for item in selected
if item.get("id") in wanted
]
if limit is not None:
selected = selected[:limit]
return selected
def result_prefix(
*,
split: str,
limit: int | None,
question_ids: list[str],
) -> str:
prefix = f"rag_answers_{split}"
if question_ids:
return prefix + "_selected"
if limit is not None:
return prefix + f"_limit{limit}"
return prefix
def questions_fingerprint(
questions: list[dict[str, Any]],
) -> str:
canonical = json.dumps(
questions,
ensure_ascii=False,
sort_keys=True,
separators=(
",",
":",
),
allow_nan=False,
).encode("utf-8")
return hashlib.sha256(
canonical
).hexdigest()

830
evaluation/rag_answer_state.py Normal file
View File

@ -0,0 +1,830 @@
from __future__ import annotations
import argparse
import csv
import json
import os
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from evaluation.rag_answer_data import (
questions_fingerprint,
)
from evaluation.rag_metrics import (
summarize_results,
)
from evaluation.rag_runner import (
LOCAL_RAG_URL,
OPENWEBUI_URL,
)
PARTIAL_SCHEMA_VERSION = 2
RESUME_COMPATIBILITY_KEYS = (
"schema_version",
"questions_file",
"overrides_file",
"questions_fingerprint",
"selected_question_ids",
"applied_override_ids",
"split",
"selected_question_count",
"requested_model",
"operation_id",
)
def _temp_path(
path: Path,
) -> Path:
return path.with_name(
f".{path.name}."
f"{os.getpid()}.tmp"
)
def save_json_results(
path: Path,
payload: dict[str, Any],
) -> None:
path.parent.mkdir(
parents=True,
exist_ok=True,
)
temp_path = _temp_path(
path
)
try:
with temp_path.open(
"w",
encoding="utf-8",
) as file:
json.dump(
payload,
file,
ensure_ascii=False,
indent=2,
)
file.write("\n")
file.flush()
os.fsync(
file.fileno()
)
os.replace(
temp_path,
path,
)
finally:
if temp_path.exists():
temp_path.unlink()
def save_csv_results(
path: Path,
results: list[dict[str, Any]],
) -> None:
path.parent.mkdir(
parents=True,
exist_ok=True,
)
temp_path = _temp_path(
path
)
fieldnames = [
"id",
"split",
"category",
"difficulty",
"question",
"requested_model",
"response_model",
"tool_called",
"tool_call_count",
"answer_contains_score",
"source_url_score",
"should_answer_score",
"overall_score",
"strict_pass",
"expected_answer_contains",
"expected_source_urls",
"rag_source_urls",
"answer",
"first_model_latency_seconds",
"tool_latency_seconds",
"final_model_latency_seconds",
"total_latency_seconds",
"prompt_tokens",
"completion_tokens",
"total_tokens",
"error",
]
try:
with temp_path.open(
"w",
encoding="utf-8",
newline="",
) as file:
writer = csv.DictWriter(
file,
fieldnames=fieldnames,
)
writer.writeheader()
for result in results:
usage = (
result.get("usage")
or {}
)
writer.writerow(
{
"id": result.get(
"id"
),
"split": result.get(
"split"
),
"category": result.get(
"category"
),
"difficulty": result.get(
"difficulty"
),
"question": result.get(
"question"
),
"requested_model": (
result.get(
"requested_model"
)
),
"response_model": (
result.get(
"response_model"
)
),
"tool_called": (
result.get(
"tool_called"
)
),
"tool_call_count": (
result.get(
"tool_call_count"
)
),
"answer_contains_score": (
result.get(
"answer_contains_score"
)
),
"source_url_score": (
result.get(
"source_url_score"
)
),
"should_answer_score": (
result.get(
"should_answer_score"
)
),
"overall_score": (
result.get(
"overall_score"
)
),
"strict_pass": (
result.get(
"strict_pass"
)
),
"expected_answer_contains": (
json.dumps(
result.get(
"expected_answer_contains",
[],
),
ensure_ascii=False,
)
),
"expected_source_urls": (
json.dumps(
result.get(
"expected_source_urls",
[],
),
ensure_ascii=False,
)
),
"rag_source_urls": (
json.dumps(
result.get(
"rag_source_urls",
[],
),
ensure_ascii=False,
)
),
"answer": result.get(
"answer",
"",
),
"first_model_latency_seconds": (
result.get(
"first_model_latency_seconds"
)
),
"tool_latency_seconds": (
result.get(
"tool_latency_seconds"
)
),
"final_model_latency_seconds": (
result.get(
"final_model_latency_seconds"
)
),
"total_latency_seconds": (
result.get(
"total_latency_seconds"
)
),
"prompt_tokens": usage.get(
"prompt_tokens",
0,
),
"completion_tokens": usage.get(
"completion_tokens",
0,
),
"total_tokens": usage.get(
"total_tokens",
0,
),
"error": result.get(
"error",
"",
),
}
)
file.flush()
os.fsync(
file.fileno()
)
os.replace(
temp_path,
path,
)
finally:
if temp_path.exists():
temp_path.unlink()
def build_error_result(
base_result: dict[str, Any],
exc: Exception,
) -> dict[str, Any]:
return {
**base_result,
"answer": "",
"tool_called": False,
"tool_call_count": 0,
"tool_calls": [],
"rag_source_urls": [],
"first_model_latency_seconds": 0.0,
"tool_latency_seconds": 0.0,
"final_model_latency_seconds": 0.0,
"total_latency_seconds": 0.0,
"usage": {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0,
},
"response_model": None,
"answer_matches": [],
"answer_contains_score": 0.0,
"source_matches": [],
"source_url_score": 0.0,
"should_answer_ok": False,
"should_answer_score": 0.0,
"returned_no_answer": False,
"tool_score": 0.0,
"overall_score": 0.0,
"strict_pass": False,
"error": (
f"{type(exc).__name__}: "
f"{exc}"
),
}
def build_base_result(
*,
question: dict[str, Any],
question_id: str,
question_text: str,
selected_override_ids: list[str],
requested_model: str,
) -> dict[str, Any]:
return {
"id": question_id,
"split": question.get(
"split"
),
"category": question.get(
"category",
"unknown",
),
"difficulty": question.get(
"difficulty",
"unknown",
),
"question": question_text,
"override_applied": (
question_id
in selected_override_ids
),
"requested_model": (
requested_model
),
"expected_documents": (
question.get(
"expected_documents",
[],
)
),
"expected_source_urls": (
question.get(
"expected_source_urls",
[],
)
),
"expected_answer_contains": (
question.get(
"expected_answer_contains",
[],
)
),
"should_answer": question.get(
"should_answer",
True,
),
"note": question.get(
"note"
),
}
def build_run_configuration(
*,
args: argparse.Namespace,
questions: list[dict[str, Any]],
selected_override_ids: list[str],
operation_id: str,
) -> dict[str, Any]:
return {
"schema_version": (
PARTIAL_SCHEMA_VERSION
),
"questions_file": str(
args.questions.resolve()
),
"overrides_file": str(
args.overrides.resolve()
),
"questions_fingerprint": (
questions_fingerprint(
questions
)
),
"selected_question_ids": [
str(item["id"])
for item in questions
],
"applied_override_ids": list(
selected_override_ids
),
"split": args.split,
"selected_question_count": len(
questions
),
"requested_model": (
args.model
),
"operation_id": (
operation_id
),
"openwebui_url": (
OPENWEBUI_URL
),
"rag_url": (
LOCAL_RAG_URL
),
"timeout": args.timeout,
"max_attempts": (
args.max_attempts
),
"backoff_base": (
args.backoff_base
),
"backoff_max": (
args.backoff_max
),
"delay": args.delay,
}
def load_partial_payload(
path: Path,
) -> dict[str, Any]:
if not path.exists():
raise FileNotFoundError(
"Partial výsledok "
f"neexistuje: {path}"
)
with path.open(
"r",
encoding="utf-8",
) as file:
payload = json.load(
file
)
if not isinstance(
payload,
dict,
):
raise ValueError(
"Partial výsledok musí "
"obsahovať JSON objekt."
)
configuration = payload.get(
"configuration"
)
if not isinstance(
configuration,
dict,
):
raise ValueError(
"Partial výsledok nemá "
"kompatibilnú configuration "
"sekciu. Starý partial spusti "
"znova s --overwrite."
)
results = payload.get(
"results"
)
if not isinstance(
results,
list,
):
raise ValueError(
"Partial výsledok nemá "
"platné results pole."
)
return payload
def validate_resume_compatibility(
partial_configuration: dict[
str,
Any,
],
current_configuration: dict[
str,
Any,
],
) -> None:
mismatches = [
key
for key in (
RESUME_COMPATIBILITY_KEYS
)
if (
partial_configuration.get(
key
)
!= current_configuration.get(
key
)
)
]
if mismatches:
raise ValueError(
"Partial výsledok nie je "
"kompatibilný s aktuálnym runom. "
"Nezhodné polia: "
+ ", ".join(
mismatches
)
)
def extract_resume_state(
partial_payload: dict[str, Any],
*,
selected_question_ids: set[str],
) -> tuple[
dict[str, dict[str, Any]],
list[str],
]:
successful: dict[
str,
dict[str, Any],
] = {}
failed_ids: list[str] = []
seen_ids: set[str] = set()
for index, raw_result in enumerate(
partial_payload["results"],
start=1,
):
if not isinstance(
raw_result,
dict,
):
raise ValueError(
"Partial "
f"results[{index}] "
"musí byť JSON objekt."
)
question_id = raw_result.get(
"id"
)
if (
not isinstance(
question_id,
str,
)
or not question_id
):
raise ValueError(
"Partial "
f"results[{index}] "
"nemá platné id."
)
if question_id in seen_ids:
raise ValueError(
"Partial obsahuje "
"duplicitné result ID: "
f"{question_id}"
)
seen_ids.add(
question_id
)
if (
question_id
not in selected_question_ids
):
raise ValueError(
"Partial obsahuje result "
"mimo aktuálneho výberu: "
f"{question_id}"
)
if raw_result.get(
"error"
):
failed_ids.append(
question_id
)
continue
successful[
question_id
] = raw_result
return (
successful,
failed_ids,
)
def ordered_results(
questions: list[dict[str, Any]],
results_by_id: dict[
str,
dict[str, Any],
],
) -> list[dict[str, Any]]:
return [
results_by_id[
str(question["id"])
]
for question in questions
if str(question["id"])
in results_by_id
]
def determine_run_status(
results: list[dict[str, Any]],
*,
expected_total: int,
) -> str:
if len(results) < expected_total:
return "partial"
if any(
result.get("error")
for result in results
):
return (
"complete_with_errors"
)
return "complete"
def build_partial_payload(
*,
status: str,
configuration: dict[str, Any],
results: list[dict[str, Any]],
expected_total: int,
) -> dict[str, Any]:
successful_count = sum(
1
for result in results
if not result.get(
"error"
)
)
return {
"generated_at": (
datetime.now(
timezone.utc
).isoformat()
),
"status": status,
"configuration": (
configuration
),
"attempted_so_far": len(
results
),
"successful_so_far": (
successful_count
),
"remaining_for_clean_completion": (
expected_total
- successful_count
),
"expected_total": (
expected_total
),
"summary": (
summarize_results(
results
)
),
"results": results,
}
def prepare_output_state(
*,
json_path: Path,
csv_path: Path,
partial_path: Path,
resume: bool,
overwrite: bool,
) -> None:
if resume:
if not partial_path.exists():
raise FileNotFoundError(
"--resume bol zadaný, "
"ale partial neexistuje: "
f"{partial_path}"
)
return
if overwrite:
for path in (
partial_path,
json_path,
csv_path,
):
if path.exists():
path.unlink()
return
if partial_path.exists():
raise FileExistsError(
"Existuje partial výsledok: "
f"{partial_path}. "
"Použi --resume alebo "
"--overwrite."
)
existing_final = [
path
for path in (
json_path,
csv_path,
)
if path.exists()
]
if existing_final:
raise FileExistsError(
"Výsledný súbor už existuje: "
+ ", ".join(
str(path)
for path in existing_final
)
+ ". Použi --overwrite."
)
def print_summary(
summary: dict[str, Any],
) -> None:
print()
print(
"RAG answer evaluation"
)
print(
"=" * 78
)
print(
f"Total: "
f"{summary['total']}"
)
print(
f"Completed: "
f"{summary['completed']}"
)
print(
f"Errors: "
f"{summary['errors']}"
)
print(
"Tool call rate: "
f"{summary['tool_call_rate']:.3f}"
)
print(
"Answer contains: "
f"{summary['answer_contains_score']:.3f}"
)
print(
"Source URL: "
f"{summary['source_url_score']:.3f}"
)
print(
"Should answer: "
f"{summary['should_answer_score']:.3f}"
)
print(
"Overall: "
f"{summary['overall_score']:.3f}"
)
print(
"Strict pass: "
f"{summary['strict_pass_count']}"
f"/{summary['total']} "
f"({summary['strict_pass_rate']:.3f})"
)
print(
"Mean latency: "
f"{summary['mean_latency_seconds']:.3f} s"
)
print(
"Total tokens: "
f"{summary['total_tokens']}"
)
print(
"=" * 78
)
print()

580
test/test_evaluate_rag_answers.py Normal file
View File

@ -0,0 +1,580 @@
from __future__ import annotations
import json
from pathlib import Path
from types import SimpleNamespace
from typing import Any
import pytest
import evaluation.evaluate_rag_answers as evaluator
def sample_question(
question_id: str = "q0001",
*,
question: str = (
"Aká je téma práce?"
),
) -> dict[str, Any]:
return {
"id": question_id,
"split": "dev",
"category": "specific_fact",
"difficulty": "medium",
"question": question,
"expected_documents": [
"pages/test/README.md"
],
"expected_source_urls": [
"https://example.test/source"
],
"expected_answer_contains": [
"test"
],
"should_answer": True,
"note": None,
}
def sample_configuration() -> dict[
str,
Any,
]:
return {
"schema_version": (
evaluator.PARTIAL_SCHEMA_VERSION
),
"questions_file": (
"/tmp/questions.json"
),
"overrides_file": (
"/tmp/overrides.json"
),
"questions_fingerprint": (
"abc123"
),
"selected_question_ids": [
"q0001",
"q0002",
],
"applied_override_ids": [],
"split": "dev",
"selected_question_count": 2,
"requested_model": (
"model120-fast"
),
"operation_id": (
"retrieve_zpwiki_context"
),
"openwebui_url": (
"https://ui.example/api"
),
"rag_url": (
"http://localhost:8000/rag"
),
"timeout": 180,
"max_attempts": 4,
"backoff_base": 1.0,
"backoff_max": 8.0,
"delay": 0.5,
}
def test_save_json_results_is_atomic_and_valid(
tmp_path: Path,
) -> None:
path = (
tmp_path
/ "result.partial.json"
)
payload = {
"status": "partial",
"results": [
{
"id": "q0001"
}
],
}
evaluator.save_json_results(
path,
payload,
)
loaded = json.loads(
path.read_text(
encoding="utf-8"
)
)
assert (
loaded
== payload
)
assert (
path.read_bytes()
.endswith(
b"\n"
)
)
assert (
list(
tmp_path.glob(
".*.tmp"
)
)
== []
)
def test_questions_fingerprint_is_stable_and_detects_change() -> None:
first = [
sample_question()
]
same = [
sample_question()
]
changed = [
sample_question(
question=(
"Iná otázka"
)
)
]
assert (
evaluator.questions_fingerprint(
first
)
== evaluator.questions_fingerprint(
same
)
)
assert (
evaluator.questions_fingerprint(
first
)
!= evaluator.questions_fingerprint(
changed
)
)
def test_validate_answer_questions_accepts_valid_question() -> None:
evaluator.validate_answer_questions(
[
sample_question()
]
)
def test_validate_answer_questions_rejects_invalid_types() -> None:
question = (
sample_question()
)
question[
"should_answer"
] = "yes"
with pytest.raises(
ValueError,
match="should_answer",
):
evaluator.validate_answer_questions(
[
question
]
)
question = (
sample_question()
)
question[
"expected_source_urls"
] = (
"https://example.test/source"
)
with pytest.raises(
ValueError,
match="expected_source_urls",
):
evaluator.validate_answer_questions(
[
question
]
)
def test_validate_resume_compatibility_accepts_matching_configuration() -> None:
configuration = (
sample_configuration()
)
evaluator.validate_resume_compatibility(
configuration,
dict(
configuration
),
)
def test_validate_resume_compatibility_rejects_model_mismatch() -> None:
partial_configuration = (
sample_configuration()
)
current_configuration = (
sample_configuration()
)
current_configuration[
"requested_model"
] = "model2"
with pytest.raises(
ValueError,
match="requested_model",
):
evaluator.validate_resume_compatibility(
partial_configuration,
current_configuration,
)
def test_extract_resume_state_skips_success_and_retries_errors() -> None:
partial_payload = {
"results": [
{
"id": "q0001",
"answer": "ok",
"error": None,
},
{
"id": "q0002",
"answer": "",
"error": "timeout",
},
]
}
(
successful,
failed_ids,
) = evaluator.extract_resume_state(
partial_payload,
selected_question_ids={
"q0001",
"q0002",
},
)
assert (
set(
successful
)
== {
"q0001"
}
)
assert (
failed_ids
== [
"q0002"
]
)
def test_extract_resume_state_rejects_duplicate_ids() -> None:
partial_payload = {
"results": [
{
"id": "q0001",
"error": None,
},
{
"id": "q0001",
"error": None,
},
]
}
with pytest.raises(
ValueError,
match="duplicitné",
):
evaluator.extract_resume_state(
partial_payload,
selected_question_ids={
"q0001"
},
)
def test_determine_run_status_distinguishes_states() -> None:
assert (
evaluator.determine_run_status(
[],
expected_total=2,
)
== "partial"
)
assert (
evaluator.determine_run_status(
[
{
"id": "q0001",
"error": None,
}
],
expected_total=2,
)
== "partial"
)
assert (
evaluator.determine_run_status(
[
{
"id": "q0001",
"error": None,
},
{
"id": "q0002",
"error": "timeout",
},
],
expected_total=2,
)
== "complete_with_errors"
)
assert (
evaluator.determine_run_status(
[
{
"id": "q0001",
"error": None,
},
{
"id": "q0002",
"error": None,
},
],
expected_total=2,
)
== "complete"
)
def test_prepare_output_state_requires_resume_or_overwrite_for_partial(
tmp_path: Path,
) -> None:
json_path = (
tmp_path
/ "results.json"
)
csv_path = (
tmp_path
/ "results.csv"
)
partial_path = (
tmp_path
/ "results.partial.json"
)
partial_path.write_text(
"{}\n",
encoding="utf-8",
)
with pytest.raises(
FileExistsError,
match="--resume",
):
evaluator.prepare_output_state(
json_path=(
json_path
),
csv_path=(
csv_path
),
partial_path=(
partial_path
),
resume=False,
overwrite=False,
)
evaluator.prepare_output_state(
json_path=(
json_path
),
csv_path=(
csv_path
),
partial_path=(
partial_path
),
resume=False,
overwrite=True,
)
assert (
not partial_path.exists()
)
def test_prepare_output_state_requires_existing_partial_for_resume(
tmp_path: Path,
) -> None:
with pytest.raises(
FileNotFoundError,
match="partial neexistuje",
):
evaluator.prepare_output_state(
json_path=(
tmp_path
/ "results.json"
),
csv_path=(
tmp_path
/ "results.csv"
),
partial_path=(
tmp_path
/ "results.partial.json"
),
resume=True,
overwrite=False,
)
def test_build_run_configuration_contains_fingerprint(
tmp_path: Path,
) -> None:
questions_path = (
tmp_path
/ "questions.json"
)
overrides_path = (
tmp_path
/ "overrides.json"
)
questions_path.write_text(
"[]\n",
encoding="utf-8",
)
overrides_path.write_text(
"{}\n",
encoding="utf-8",
)
args = SimpleNamespace(
questions=(
questions_path
),
overrides=(
overrides_path
),
split="dev",
model="model120-fast",
timeout=180,
max_attempts=4,
backoff_base=1.0,
backoff_max=8.0,
delay=0.5,
)
questions = [
sample_question()
]
configuration = (
evaluator.build_run_configuration(
args=args,
questions=(
questions
),
selected_override_ids=[],
operation_id=(
"retrieve_zpwiki_context"
),
)
)
assert (
configuration[
"schema_version"
]
== evaluator.PARTIAL_SCHEMA_VERSION
)
assert (
configuration[
"selected_question_ids"
]
== [
"q0001"
]
)
assert (
configuration[
"questions_fingerprint"
]
== evaluator.questions_fingerprint(
questions
)
)
def test_apply_question_overrides_rejects_empty_question(
tmp_path: Path,
) -> None:
questions = [
sample_question()
]
overrides_path = (
tmp_path
/ "overrides.json"
)
overrides_path.write_text(
json.dumps(
{
"q0001": {
"question": " "
}
},
ensure_ascii=False,
)
+ "\n",
encoding="utf-8",
)
with pytest.raises(
ValueError,
match="override question",
):
evaluator.apply_question_overrides(
questions,
overrides_path,
)