API reference¶

def enforce_budgets(state: Any, questions: dict[str, dict[str, Any]], backend: Backend) -> None:
    """Fail fast with ``422`` when the request exceeds the backend's token budgets.

    Compares ``backend.count(state, questions)`` against the backend's
    ``max_total_tokens`` / ``max_state_question_tokens`` (Jev: 64k total,
    32k state+longest-question). Backends that predate the budget
    interface skip enforcement. A broken counter fails open (warn + skip)
    so a counting bug can never turn a small request into a ``500``.
    """
    count = getattr(backend, "count", None)
    max_total = getattr(backend, "max_total_tokens", None)
    max_longest = getattr(backend, "max_state_question_tokens", None)
    if not callable(count) or max_total is None or max_longest is None:
        return
    try:
        total, state_longest = (int(n) for n in count(state, questions))
    except Exception:
        logger.warning("budget count failed; skipping enforcement", exc_info=True)
        return
    if state_longest > max_longest:
        raise compat.CompatError(
            f"context budget exceeded: state plus longest question is ~{state_longest} "
            f"tokens (limit {max_longest})",
            field="questions",
        )
    if total > max_total:
        raise compat.CompatError(
            f"context budget exceeded: state plus all questions is ~{total} tokens "
            f"(limit {max_total})",
            field="questions",
        )

def evaluate(request: SystemOneRequest, backend: Backend, settings: Settings) -> dict[str, Any]:
    """Run one SystemOne request and return a Jev-shaped response dict."""
    serving_model = compat.resolve_model(
        request.model, backend.serving_model, settings.extra_models_set
    )
    questions = request.model_dump(mode="json", exclude_none=False)["questions"]
    enforce_budgets(request.state, questions, backend)
    try:
        raw = backend.predict(request.state, questions)
    except ValueError as exc:
        mapped = _as_budget_error(exc)
        if mapped is exc:
            raise
        raise mapped from exc
    return compat.shape_response(
        questions,
        raw["answers"],
        raw.get("usage", {}),
        serving_model,
    )

def list_models(settings: Settings, backend: Backend) -> dict[str, Any]:
    """Return the models listing: the serving checkpoint plus Jev aliases."""
    from .schemas import ModelsResponse

    cards = [
        {
            "name": backend.serving_model,
            "description": "Laya decision model served with a Jev-compatible API.",
            "release_date": "unknown",
        }
    ]
    for alias in sorted(compat.JEV_KNOWN_MODELS | settings.extra_models_set):
        if alias == backend.serving_model:
            continue
        cards.append(
            {
                "name": alias,
                "description": f"Alias resolving to {backend.serving_model}.",
                "release_date": "unknown",
            }
        )
    return ModelsResponse.model_validate({"models": cards}).model_dump(mode="json")

def resolve_model(
    requested: str, serving_model: str, extra_models: frozenset[str] = frozenset()
) -> str:
    """Validate the requested ``model`` and return the id to report.

    Known Jev names (plus operator-configured extras) resolve to
    ``serving_model`` — the checkpoint actually answering, e.g.
    ``"laya-english"``. Unknown names raise :class:`CompatError`.
    """
    if requested in JEV_KNOWN_MODELS or requested in extra_models or requested == serving_model:
        return serving_model
    raise CompatError(
        f"unknown model {requested!r}; expected a Jev model name or the serving model {serving_model!r}",
        field="model",
    )

def shape_choice(question_id: str, raw: dict[str, Any], options: list[str]) -> ChoiceAnswer:
    """Shape a raw ``choice`` answer; drops Laya-only ``action``."""
    choice = raw.get("choice")
    if choice not in options:
        raise CompatError(
            f"choice answer {question_id!r} selected {choice!r}, which is not one of {options!r}",
            field=f"answers.{question_id}.choice",
        )
    probs = raw.get("probabilities", {})
    if set(probs) != set(options):
        raise CompatError(
            f"choice answer {question_id!r} probabilities cover {sorted(probs)!r}, expected {options!r}",
            field=f"answers.{question_id}.probabilities",
        )
    return ChoiceAnswer(
        type="choice",
        choice=choice,
        probabilities={
            opt: _clamp01(probs[opt], f"answers.{question_id}.probabilities.{opt}")
            for opt in options
        },
        confidence=_clamp01(raw.get("confidence"), f"answers.{question_id}.confidence"),
    )

def shape_noul(question_id: str, raw: dict[str, Any]) -> NoulAnswer:
    """Shape a raw ``noul`` answer; drops Laya-only ``confidence``/``action``."""
    if "noul" not in raw:
        raise CompatError(
            f"noul answer {question_id!r} is missing 'noul'", field=f"answers.{question_id}"
        )
    return NoulAnswer(type="noul", noul=_clamp01(raw["noul"], f"answers.{question_id}.noul"))

def shape_response(
    request_questions: dict[str, dict[str, Any]],
    raw_answers: dict[str, dict[str, Any]],
    usage: dict[str, Any],
    serving_model: str,
) -> dict[str, Any]:
    """Shape a full backend result into a Jev response dict (validated by caller).

    ``request_questions`` are the *validated* question dicts (with ``type``
    and ``criteria``); they define the expected answer shapes. ``raw_answers``
    is the backend's per-question output. Unknown/extra backend keys
    (``action``, ``routing``) are ignored.
    """
    if set(raw_answers) != set(request_questions):
        raise CompatError(
            f"backend answered {sorted(raw_answers)!r}, expected {sorted(request_questions)!r}",
            field="answers",
        )
    answers: dict[str, Any] = {}
    for qid, qdef in request_questions.items():
        raw = raw_answers[qid]
        qtype = qdef["type"]
        if qtype == "noul":
            answers[qid] = shape_noul(qid, raw)
        elif qtype == "choice":
            criteria = qdef["criteria"]
            options = list(criteria) if isinstance(criteria, dict) else list(criteria)
            answers[qid] = shape_choice(qid, raw, options)
        elif qtype == "score":
            answers[qid] = shape_score(qid, raw, list(qdef["criteria"]))
        else:  # pragma: no cover - schema validation rejects this first
            raise CompatError(f"unknown question type {qtype!r}", field=f"questions.{qid}.type")
    response = SystemOneResponse(
        model=serving_model,
        answers=answers,
        usage=Usage(
            input_tokens=max(0, int(usage.get("input_tokens", 0))),
            output_tokens=max(0, int(usage.get("output_tokens", 0))),
        ),
    )
    return response.model_dump(mode="json", exclude_none=False)

def shape_score(
    question_id: str,
    raw: dict[str, Any],
    levels: list[StructuredText],
) -> ScoreAnswer:
    """Shape a raw ``score`` answer with stringified level keys; drops ``action``."""
    expected = [str(i) for i in range(len(levels))]
    probs = raw.get("probabilities", {})
    if set(map(str, probs)) != set(expected):
        raise CompatError(
            f"score answer {question_id!r} probabilities cover {sorted(map(str, probs))!r}, "
            f"expected levels {expected!r}",
            field=f"answers.{question_id}.probabilities",
        )
    try:
        score = float(raw["score"])
    except (KeyError, TypeError, ValueError) as exc:
        raise CompatError(
            f"score answer {question_id!r} is missing a numeric 'score'",
            field=f"answers.{question_id}.score",
        ) from exc
    if not 0.0 <= score <= float(len(levels) - 1):
        raise CompatError(
            f"score answer {question_id!r} has score {score!r} outside [0, {len(levels) - 1}]",
            field=f"answers.{question_id}.score",
        )
    return ScoreAnswer(
        type="score",
        score=score,
        legend={str(i): level for i, level in enumerate(levels)},
        probabilities={
            str(i): _clamp01(
                probs[i] if i in probs else probs[str(i)],
                f"answers.{question_id}.probabilities.{i}",
            )
            for i in range(len(levels))
        },
        confidence=_clamp01(raw.get("confidence"), f"answers.{question_id}.confidence"),
    )

def build_backend(settings: Settings) -> Backend:
    """Instantiate the backend selected by ``settings.backend``."""
    if settings.backend == "fake":
        return FakeBackend(serving_model=settings.serving_model)
    if settings.backend == "laya":
        return LayaBackend(settings)
    raise ValueError(f"unknown backend {settings.backend!r}; expected 'fake' or 'laya'")

def count_request(state: Any, questions: dict[str, dict[str, Any]]) -> tuple[int, int]:
    """Weight-free (total, state+longest-question) word count of a request."""
    state_n = len(render_text(state).split())
    per_question = [len(render_question_text(q).split()) for q in questions.values()]
    longest = max(per_question, default=0)
    return state_n + sum(per_question), state_n + longest

def is_overload(exc: BaseException) -> bool:
    """Return whether ``exc`` looks like transient overload (retryable)."""
    haystack = f"{type(exc).__name__} {exc}".lower()
    return any(marker in haystack for marker in _OVERLOAD_MARKERS)

def render_question_text(qdef: dict[str, Any]) -> str:
    """Rendered size basis of one question: instructions + criteria descriptions."""
    parts = [render_text(qdef.get("instructions", ""))]
    criteria = qdef.get("criteria")
    if isinstance(criteria, dict):
        parts.extend(render_text(v) for v in criteria.values() if v is not None)
    elif isinstance(criteria, list):
        parts.extend(render_text(level) for level in criteria)
    return " ".join(part for part in parts if part)

def render_text(value: Any) -> str:
    """Render one value as text, mirroring ``laya.common.render_criterion``."""
    if isinstance(value, str):
        return value
    return json.dumps(value, ensure_ascii=False, separators=(", ", ": "), default=str)

def main() -> None:
    """Serve the app with uvicorn (host/port via UVIcorn env or defaults)."""
    settings = Settings()
    app = create_app(settings)
    uvicorn.run(app, host="0.0.0.0", port=8000)