Praxis CI
be3df525d8
server/scenarios/schema.py defines the typed model: Scenario (id, path, market, language, title, difficulty, failure_mode, persona, setup, success_criteria, common_mistakes, branches[], debrief) + Branch (id, trigger.learner_signals, outcome, failure_mode, debrief_focus) + ScenarioDebrief (model=deepseek-v4-flash:cloud, mode=no_think, D-020). failure_mode field present per D-009. server/scenarios/loader.py loads YAML → Pydantic, validates at load time, raises typed ValidationError on bad input. scenarios/customer_service_refund_ca_v01.yaml — the v0.1 Canada Customer Service scenario (D-010): 'Angry customer requesting refund on a damaged product', one branch point (accept_resolution vs escalate), failure_mode=escalates_unresolved, success criteria, common mistakes, debrief config. Matches the RESEARCH.md example. 5 unit tests pass (valid parse, invalid raises typed error, branch outcome Literal, branch_by_id, real YAML load). load() returns a valid Scenario with both branches. ---ci--- phase: 1 milestone: v0.1 plan: 03 task: 03-01,03-02 status: execute persona: data-engineer requirements: covered: [REQ-SCEN-01, REQ-SCEN-FMT-01] ---/ci---
58 lines
1.9 KiB
Python
58 lines
1.9 KiB
Python
"""Scenario loader — YAML → Pydantic Scenario (D-018).
|
|
|
|
Loads a scenario by id from the scenarios/ directory, validates it against the
|
|
Pydantic schema, and returns a typed Scenario object. Used by the pipeline
|
|
(TASK-03-07) and the e2e smoke test.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
from server.scenarios.schema import Scenario, ValidationError
|
|
|
|
_DEFAULT_SCENARIOS_DIR = Path(__file__).resolve().parent.parent.parent / "scenarios"
|
|
|
|
|
|
def load(scenario_id: str, scenarios_dir: Path | None = None) -> Scenario:
|
|
"""Load and validate a scenario by id.
|
|
|
|
Args:
|
|
scenario_id: e.g. 'customer_service_refund_ca_v01' (the YAML filename stem).
|
|
scenarios_dir: override the scenarios directory (default: repo /scenarios).
|
|
|
|
Returns:
|
|
A validated Scenario object.
|
|
|
|
Raises:
|
|
FileNotFoundError: if the YAML file doesn't exist.
|
|
ValidationError: if the YAML fails schema validation (typed Pydantic error).
|
|
"""
|
|
base = scenarios_dir or _DEFAULT_SCENARIOS_DIR
|
|
path = base / f"{scenario_id}.yaml"
|
|
if not path.exists():
|
|
# Try the id-with-cs-prefix alias (RESEARCH example used 'cs_refund_ca_v01').
|
|
path = base / f"{scenario_id.replace('cs_', 'customer_service_')}.yaml"
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"Scenario YAML not found: {scenario_id} in {base}")
|
|
|
|
with path.open("r", encoding="utf-8") as f:
|
|
raw = yaml.safe_load(f)
|
|
|
|
return Scenario.model_validate(raw)
|
|
|
|
|
|
def load_all(scenarios_dir: Path | None = None) -> list[Scenario]:
|
|
"""Load all scenarios in the directory (for the future scenario library)."""
|
|
base = scenarios_dir or _DEFAULT_SCENARIOS_DIR
|
|
out: list[Scenario] = []
|
|
for p in sorted(base.glob("*.yaml")):
|
|
with p.open("r", encoding="utf-8") as f:
|
|
raw = yaml.safe_load(f)
|
|
out.append(Scenario.model_validate(raw))
|
|
return out
|
|
|
|
|
|
__all__ = ["load", "load_all", "ValidationError"] |