diff --git a/src/strands_evals/evaluation_data_store.py b/src/strands_evals/evaluation_data_store.py index 05f6a4c2..8bb7e233 100644 --- a/src/strands_evals/evaluation_data_store.py +++ b/src/strands_evals/evaluation_data_store.py @@ -9,24 +9,46 @@ class EvaluationDataStore(Protocol): Implementations can use any storage backend (local files, S3, databases, etc.) as long as they implement the load and save methods. + + A case may be evaluated more than once — for statistical confidence you run the same case + N times and keep every result. `run_index` selects which run to read or write; it defaults + to `0`, so single-run callers can ignore it entirely and see the same behavior as before. + `completed_run_count` reports how many runs a case already has, which lets a caller resume an + interrupted N-run sweep by running only the runs that are missing. """ - def load(self, case_name: str) -> EvaluationData | None: - """Load cached evaluation data by case name. + def load(self, case_name: str, run_index: int = 0) -> EvaluationData | None: + """Load cached evaluation data by case name and run index. Args: case_name: The name of the case to load results for. + run_index: Which run of the case to load. Defaults to `0` (the first/only run). Returns: The cached EvaluationData if found, None otherwise. """ ... - def save(self, case_name: str, result: EvaluationData) -> None: - """Save evaluation data for a case. + def save(self, case_name: str, result: EvaluationData, run_index: int = 0) -> None: + """Save evaluation data for one run of a case. Args: case_name: The name of the case to save results for. result: The EvaluationData to save. + run_index: Which run of the case this result is. Defaults to `0`. + """ + ... + + def completed_run_count(self, case_name: str) -> int: + """Return how many consecutive runs of a case are already stored. + + Counts stored runs starting at `run_index=0` and stopping at the first gap, so the + result is the next `run_index` a caller should write when topping up to N runs. + + Args: + case_name: The name of the case to count runs for. + + Returns: + The number of consecutive stored runs, or `0` if none exist. """ ... diff --git a/src/strands_evals/local_file_task_result_store.py b/src/strands_evals/local_file_task_result_store.py index 3080e04e..1f3c8614 100644 --- a/src/strands_evals/local_file_task_result_store.py +++ b/src/strands_evals/local_file_task_result_store.py @@ -1,39 +1,142 @@ +import json from pathlib import Path from .types.evaluation import EvaluationData +# A case's runs live in a per-case subdirectory as run_0.json, run_1.json, .... A pre-existing +# flat `{case_name}.json` from the single-run layout is treated as that case's run 0, so old +# stores keep loading without a migration step. +_RUN_PREFIX = "run_" +_MANIFEST_NAME = "_manifest.json" + + +class ConfigDriftError(RuntimeError): + """Raised when a store's recorded config no longer matches the current experiment. + + Mixing results produced under different evaluators, cases, or model config into one report + silently compares apples to oranges. When the store already holds results written under a + different `config_hash`, resuming raises this instead of appending, unless the caller passes + `overwrite=True` to discard the stale results and start fresh. + """ + class LocalFileTaskResultStore: """Task result store backed by local JSON files. - Saves one JSON file per case in the specified directory, - using the case name as the filename. + Each case gets its own subdirectory holding one JSON file per run (`run_0.json`, + `run_1.json`, ...), so a case can be evaluated N times and every result is kept. A legacy + flat `{case_name}.json` file from the original single-run layout is read as that case's + run 0, so existing stores keep working unchanged. + + An optional `config_hash` guards against config drift: pass the hash of the current + experiment's evaluators/cases/model config and the store records it in a `_manifest.json`. + A later run whose hash differs raises `ConfigDriftError` rather than mixing incompatible + results, unless the store is opened with `overwrite=True`. """ - def __init__(self, directory: str | Path): + def __init__(self, directory: str | Path, config_hash: str | None = None, overwrite: bool = False): + """Open (or create) a store rooted at `directory`. + + Args: + directory: Directory that holds the per-case result subdirectories. + config_hash: Optional hash of the current experiment config. When given, it is + checked against the stored manifest (drift raises unless `overwrite=True`) and + then recorded. + overwrite: When True, discard any existing results and manifest before use. Use this + to start fresh after an intentional config change. + + Raises: + ConfigDriftError: If `config_hash` is given, differs from the stored manifest, and + `overwrite` is False. + """ self._directory = Path(directory) self._directory.mkdir(parents=True, exist_ok=True) - def load(self, case_name: str) -> EvaluationData | None: - """Load a cached task result from a JSON file. + if overwrite: + self._clear() + + if config_hash is not None: + self._guard_config(config_hash, overwrite=overwrite) + + def _clear(self) -> None: + """Remove all stored case results and the manifest, leaving an empty store directory.""" + for child in self._directory.iterdir(): + if child.is_dir(): + for f in child.glob(f"{_RUN_PREFIX}*.json"): + f.unlink() + # Drop the now-empty case directory; ignore if other files remain. + if not any(child.iterdir()): + child.rmdir() + elif child.suffix == ".json": + child.unlink() + + def _manifest_path(self) -> Path: + return self._directory / _MANIFEST_NAME + + def _guard_config(self, config_hash: str, overwrite: bool) -> None: + """Check the stored config hash against `config_hash`, then record the current one.""" + manifest_path = self._manifest_path() + if manifest_path.exists() and not overwrite: + stored = json.loads(manifest_path.read_text()).get("config_hash") + if stored is not None and stored != config_hash: + raise ConfigDriftError( + "Experiment configuration changed since results were last written " + f"(stored config_hash={stored}, current={config_hash}). " + "Open the store with overwrite=True to discard cached results and start fresh." + ) + + manifest_path.write_text(json.dumps({"config_hash": config_hash}, indent=2)) + + def _case_dir(self, case_name: str) -> Path: + return self._directory / case_name + + def _run_path(self, case_name: str, run_index: int) -> Path: + return self._case_dir(case_name) / f"{_RUN_PREFIX}{run_index}.json" + + def _legacy_path(self, case_name: str) -> Path: + return self._directory / f"{case_name}.json" + + def load(self, case_name: str, run_index: int = 0) -> EvaluationData | None: + """Load a cached task result for one run of a case. Args: case_name: The name of the case to load results for. + run_index: Which run to load. Defaults to `0`. Returns: - The cached EvaluationData if the file exists, None otherwise. + The cached EvaluationData if the run exists, None otherwise. Run 0 also reads a + legacy flat `{case_name}.json` file when no per-run file is present. """ - path = self._directory / f"{case_name}.json" - if not path.exists(): - return None - return EvaluationData.model_validate_json(path.read_text()) + path = self._run_path(case_name, run_index) + if path.exists(): + return EvaluationData.model_validate_json(path.read_text()) + + if run_index == 0: + legacy = self._legacy_path(case_name) + if legacy.exists(): + return EvaluationData.model_validate_json(legacy.read_text()) + + return None - def save(self, case_name: str, result: EvaluationData) -> None: - """Save a task result to a JSON file. + def save(self, case_name: str, result: EvaluationData, run_index: int = 0) -> None: + """Save a task result for one run of a case. Args: case_name: The name of the case to save results for. result: The EvaluationData to save. + run_index: Which run this result is. Defaults to `0`. """ - path = self._directory / f"{case_name}.json" + path = self._run_path(case_name, run_index) + path.parent.mkdir(parents=True, exist_ok=True) path.write_text(result.model_dump_json(indent=2)) + + def completed_run_count(self, case_name: str) -> int: + """Return the number of consecutive stored runs for a case. + + Counts from `run_index=0` upward and stops at the first missing run, so the return value + is the next index to write. A legacy flat file counts as run 0. + """ + count = 0 + while self.load(case_name, count) is not None: + count += 1 + return count diff --git a/tests/strands_evals/cli/test_run.py b/tests/strands_evals/cli/test_run.py index 9378f772..672312ad 100644 --- a/tests/strands_evals/cli/test_run.py +++ b/tests/strands_evals/cli/test_run.py @@ -313,8 +313,8 @@ def test_run_data_store_creates_directory(experiment_file: Path, tmp_path: Path) ) assert exit_code == 0 assert store_dir.is_dir() - # One file per case (case 'c1' in the fixture). - assert (store_dir / "c1.json").exists() + # One directory per case and one file per run (case 'c1' in the fixture). + assert (store_dir / "c1" / "run_0.json").exists() def test_run_exit_zero_overrides_fail_on(experiment_file: Path, tmp_path: Path): diff --git a/tests/strands_evals/test_experiment.py b/tests/strands_evals/test_experiment.py index e1b0645b..9495612e 100644 --- a/tests/strands_evals/test_experiment.py +++ b/tests/strands_evals/test_experiment.py @@ -1889,13 +1889,19 @@ class DictEvaluationDataStore: """Simple in-memory store for testing.""" def __init__(self): - self._data: dict[str, EvaluationData] = {} + self._data: dict[tuple[str, int], EvaluationData] = {} - def load(self, case_name: str) -> EvaluationData | None: - return self._data.get(case_name) + def load(self, case_name: str, run_index: int = 0) -> EvaluationData | None: + return self._data.get((case_name, run_index)) - def save(self, case_name: str, result: EvaluationData) -> None: - self._data[case_name] = result + def save(self, case_name: str, result: EvaluationData, run_index: int = 0) -> None: + self._data[(case_name, run_index)] = result + + def completed_run_count(self, case_name: str) -> int: + count = 0 + while (case_name, count) in self._data: + count += 1 + return count class TestEvaluationDataStore: diff --git a/tests/strands_evals/test_local_file_task_result_store.py b/tests/strands_evals/test_local_file_task_result_store.py index ab0fe547..c1e914ee 100644 --- a/tests/strands_evals/test_local_file_task_result_store.py +++ b/tests/strands_evals/test_local_file_task_result_store.py @@ -2,7 +2,7 @@ import pytest -from strands_evals.local_file_task_result_store import LocalFileTaskResultStore +from strands_evals.local_file_task_result_store import ConfigDriftError, LocalFileTaskResultStore from strands_evals.types import EvaluationData @@ -44,10 +44,95 @@ def test_creates_directory(self, tmp_path): LocalFileTaskResultStore(directory=new_dir) assert new_dir.exists() - def test_save_writes_json_file(self, store, evaluation_data, tmp_path): + def test_save_writes_run_file(self, store, evaluation_data, tmp_path): store.save("math_case", evaluation_data) - file_path = tmp_path / "results" / "math_case.json" + file_path = tmp_path / "results" / "math_case" / "run_0.json" assert file_path.exists() data = json.loads(file_path.read_text()) assert data["input"] == "What is 2+2?" assert data["actual_output"] == "4" + + +class TestPerRunKeying: + """A case can be evaluated N times; each run is stored and read independently.""" + + def _data(self, name, output): + return EvaluationData(input="q", actual_output=output, name=name, expected_output="q") + + def test_runs_are_stored_and_loaded_independently(self, store): + store.save("c", self._data("c", "run-a"), run_index=0) + store.save("c", self._data("c", "run-b"), run_index=1) + + assert store.load("c", 0).actual_output == "run-a" + assert store.load("c", 1).actual_output == "run-b" + assert store.load("c", 2) is None + + def test_completed_run_count_counts_consecutive_runs(self, store): + assert store.completed_run_count("c") == 0 + store.save("c", self._data("c", "0"), run_index=0) + store.save("c", self._data("c", "1"), run_index=1) + assert store.completed_run_count("c") == 2 + + def test_completed_run_count_stops_at_first_gap(self, store): + store.save("c", self._data("c", "0"), run_index=0) + store.save("c", self._data("c", "2"), run_index=2) # gap at 1 + assert store.completed_run_count("c") == 1 + + def test_run_files_live_in_per_case_subdir(self, store, tmp_path): + store.save("c", self._data("c", "x"), run_index=3) + assert (tmp_path / "results" / "c" / "run_3.json").exists() + + def test_legacy_flat_file_is_read_as_run_zero(self, tmp_path): + # A store written by the old single-run layout: flat {case_name}.json. + results_dir = tmp_path / "results" + results_dir.mkdir() + legacy = self._data("legacy", "old") + (results_dir / "legacy.json").write_text(legacy.model_dump_json()) + + store = LocalFileTaskResultStore(directory=results_dir) + assert store.load("legacy", 0).actual_output == "old" + assert store.completed_run_count("legacy") == 1 + + +class TestConfigDrift: + """An optional config hash prevents mixing results from incompatible experiment configs.""" + + def _data(self, name): + return EvaluationData(input="q", actual_output="a", name=name, expected_output="q") + + def test_first_open_records_hash(self, tmp_path): + d = tmp_path / "results" + LocalFileTaskResultStore(directory=d, config_hash="hash-1") + manifest = json.loads((d / "_manifest.json").read_text()) + assert manifest["config_hash"] == "hash-1" + + def test_reopen_with_same_hash_is_allowed(self, tmp_path): + d = tmp_path / "results" + s1 = LocalFileTaskResultStore(directory=d, config_hash="hash-1") + s1.save("c", self._data("c")) + s2 = LocalFileTaskResultStore(directory=d, config_hash="hash-1") + assert s2.load("c") is not None + + def test_reopen_with_different_hash_raises(self, tmp_path): + d = tmp_path / "results" + LocalFileTaskResultStore(directory=d, config_hash="hash-1") + with pytest.raises(ConfigDriftError, match="configuration changed"): + LocalFileTaskResultStore(directory=d, config_hash="hash-2") + + def test_overwrite_discards_results_and_updates_hash(self, tmp_path): + d = tmp_path / "results" + s1 = LocalFileTaskResultStore(directory=d, config_hash="hash-1") + s1.save("c", self._data("c")) + assert s1.load("c") is not None + + s2 = LocalFileTaskResultStore(directory=d, config_hash="hash-2", overwrite=True) + assert s2.load("c") is None + manifest = json.loads((d / "_manifest.json").read_text()) + assert manifest["config_hash"] == "hash-2" + + def test_no_hash_skips_the_guard(self, tmp_path): + d = tmp_path / "results" + LocalFileTaskResultStore(directory=d, config_hash="hash-1") + # Opening without a hash must not raise even though a manifest exists. + store = LocalFileTaskResultStore(directory=d) + assert store is not None