151 lines
5.2 KiB
Python
151 lines
5.2 KiB
Python
"""Select replayable demo scenarios from the shared scenario catalog."""
|
|
|
|
import re
|
|
import unicodedata
|
|
from dataclasses import dataclass
|
|
from difflib import SequenceMatcher
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
_REPOSITORY_ROOT = Path(__file__).parents[4]
|
|
SCENARIO_PATH = _REPOSITORY_ROOT / "eval" / "scenarios.yaml"
|
|
FIXTURE_ROOT = _REPOSITORY_ROOT / "eval" / "fixtures"
|
|
_WORD_PATTERN = re.compile(r"[a-z0-9]+")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Scenario:
|
|
"""One shared suggestion and its optional parent scenario."""
|
|
|
|
key: str
|
|
chip: str
|
|
query: str
|
|
after: str | None = None
|
|
|
|
@property
|
|
def is_refinement(self) -> bool:
|
|
"""Return whether this scenario refines an earlier result."""
|
|
return self.after is not None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ScenarioMatch:
|
|
"""Describe which recorded scenario will answer a demo request."""
|
|
|
|
key: str
|
|
is_exact: bool
|
|
scenario_query: str
|
|
chip: str
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def load_scenarios() -> tuple[Scenario, ...]:
|
|
"""Load and validate the shared scenario catalog once."""
|
|
payload = yaml.safe_load(SCENARIO_PATH.read_text(encoding="ascii"))
|
|
if not isinstance(payload, dict) or payload.get("version") != 1:
|
|
raise ValueError("Scenario catalog must have version 1")
|
|
entries = payload.get("scenarios")
|
|
if not isinstance(entries, list) or not entries:
|
|
raise ValueError("Scenario catalog must contain scenarios")
|
|
scenarios = tuple(_parse_scenario(entry) for entry in entries)
|
|
keys = {scenario.key for scenario in scenarios}
|
|
if len(keys) != len(scenarios):
|
|
raise ValueError("Scenario keys must be unique")
|
|
if any(scenario.after not in keys for scenario in scenarios if scenario.after is not None):
|
|
raise ValueError("Scenario catalog contains an unknown parent")
|
|
return scenarios
|
|
|
|
|
|
def select_scenario(query: str) -> ScenarioMatch:
|
|
"""Select an exact scenario or the nearest available recorded fixture."""
|
|
scenarios = _replayable_scenarios()
|
|
normalized_query = normalize_text(query)
|
|
for scenario in scenarios:
|
|
if normalized_query in {normalize_text(scenario.query), normalize_text(scenario.chip)}:
|
|
return _to_match(scenario, is_exact=True)
|
|
|
|
selected = max(
|
|
scenarios,
|
|
key=lambda scenario: max(
|
|
SequenceMatcher(None, normalized_query, normalize_text(scenario.query)).ratio(),
|
|
SequenceMatcher(None, normalized_query, normalize_text(scenario.chip)).ratio(),
|
|
),
|
|
)
|
|
return _to_match(selected, is_exact=False)
|
|
|
|
|
|
def select_replay_scenario(
|
|
query: str,
|
|
has_previous_recommendations: bool,
|
|
) -> ScenarioMatch:
|
|
"""Map a request with prior results to its recorded refinement when available."""
|
|
selected = select_scenario(query)
|
|
scenarios = _replayable_scenarios()
|
|
selected_scenario = next(scenario for scenario in scenarios if scenario.key == selected.key)
|
|
if selected_scenario.is_refinement or not has_previous_recommendations:
|
|
return selected
|
|
refinement = next(
|
|
(scenario for scenario in scenarios if scenario.after == selected_scenario.key),
|
|
None,
|
|
)
|
|
return _to_match(refinement, is_exact=False) if refinement is not None else selected
|
|
|
|
|
|
def suggestion_items() -> list[dict[str, str]]:
|
|
"""Return chip and query pairs for non-refinement scenarios."""
|
|
return [
|
|
{"chip": scenario.chip, "query": scenario.query}
|
|
for scenario in load_scenarios()
|
|
if not scenario.is_refinement
|
|
]
|
|
|
|
|
|
def normalize_text(value: str) -> str:
|
|
"""Normalize text for exact and similarity matching."""
|
|
decomposed = unicodedata.normalize("NFKD", value.casefold())
|
|
ascii_text = decomposed.encode("ascii", errors="ignore").decode("ascii")
|
|
return " ".join(_WORD_PATTERN.findall(ascii_text))
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def _replayable_scenarios() -> tuple[Scenario, ...]:
|
|
fixture_keys = {path.name for path in FIXTURE_ROOT.iterdir() if path.is_dir()}
|
|
scenarios = tuple(scenario for scenario in load_scenarios() if scenario.key in fixture_keys)
|
|
if not scenarios:
|
|
raise ValueError("Demo mode requires at least one scenario fixture")
|
|
return scenarios
|
|
|
|
|
|
def _parse_scenario(payload: object) -> Scenario:
|
|
if not isinstance(payload, dict):
|
|
raise ValueError("Each scenario must be a mapping")
|
|
required = {"key", "chip", "query", "facets"}
|
|
if not required <= set(payload):
|
|
raise ValueError("Scenario is missing a required field")
|
|
after = payload.get("after")
|
|
if after is not None and not isinstance(after, str):
|
|
raise ValueError("Scenario parent must be a key")
|
|
return Scenario(
|
|
key=_required_string(payload, "key"),
|
|
chip=_required_string(payload, "chip"),
|
|
query=_required_string(payload, "query"),
|
|
after=after,
|
|
)
|
|
|
|
|
|
def _required_string(payload: dict[object, object], key: str) -> str:
|
|
value = payload.get(key)
|
|
if not isinstance(value, str) or not value:
|
|
raise ValueError(f"Scenario {key} must be a non-empty string")
|
|
return value
|
|
|
|
|
|
def _to_match(scenario: Scenario, is_exact: bool) -> ScenarioMatch:
|
|
return ScenarioMatch(
|
|
key=scenario.key,
|
|
is_exact=is_exact,
|
|
scenario_query=scenario.query,
|
|
chip=scenario.chip,
|
|
)
|