diff --git a/docs/model-routing.md b/docs/model-routing.md new file mode 100644 index 0000000..b60ac85 --- /dev/null +++ b/docs/model-routing.md @@ -0,0 +1,228 @@ +# RFC: `llm_route` — suite-wide local-first model routing with auditable escalation + +Status: **RFC / reference implementation.** Ships as an **optional** module +`maine_forms_engine.llm_route` (stdlib-only, zero new dependencies). Nothing in +`maine_forms_engine.fill` imports it — the engine's deterministic, no-model-at-fill-time +contract is untouched. This document is the design rationale; the code is +[`src/maine_forms_engine/llm_route.py`](../src/maine_forms_engine/llm_route.py) with +tests in [`tests/test_llm_route.py`](../tests/test_llm_route.py). + +> **May graduate to its own repo.** The module has no dependency on the engine and could +> live as a tiny standalone `llm-route` package (git-tag-pinned like the engine), imported +> by every repo in the suite. It is proposed here first because the engine is the hub the +> form repos already depend on, which makes it the lowest-friction place to trial the API. +> Maintainers should say which they prefer. + +--- + +## 1. Problem + +The 16-repo suite has grown **nine different base-URL env conventions** and **20+ +hardcoded model strings**, and every LLM-touching tool hand-rolls its own OpenAI-compatible +client (≥6 copies). There is **no shared abstraction for choosing a model per task or for +escalating when a cheap/local model is likely wrong** — each repo reinvents retry-on-empty, +enum-reject, and consensus logic in isolation. A model choice is never recorded as an +auditable decision. + +### Current-state inventory: base-URL env conventions + +| # | Convention | Where it lives today | Notes | +|---|------------|----------------------|-------| +| 1 | `MCF_LLM_ENDPOINTS` | maine-corporation-forms | list form | +| 2 | `AUDIT_QWEN_BASE_URL` | corp audit tools | model-named env | +| 3 | `ROUTER_BASE_URL` / `ROUTER_MODEL` / `ROUTER_API_KEY` | maine-forms-router | closest to canonical; already has `OPENAI_*` fallback | +| 4 | `HALLUCHECK_BASE_URL` | LLM_Hallucination_Checker | | +| 5 | `INSPECTOR_BASE_URL` | hallucheck inspector | second env in the same repo | +| 6 | `VLM_API_BASE` | (a **hardcoded constant**, not env) | must be un-hardcoded | +| 7 | `LOCAL_VL_ENDPOINTS` | vision tooling | list form | +| 8 | `OPENAI_BASE_URL` / `OPENAI_MODEL` / `OPENAI_API_KEY` | several | de-facto lowest common denominator | +| 9 | `ANTHROPIC_BASE_URL` | hallucheck anthropic branch | provider-specific | + +Plus probate `config.py` hardcodes a local fleet layout as Python constants in a public +repo, and 20+ model-name string literals are scattered across drafting/verify call sites. + +### The canonical trio (this RFC) + +One convention, provider-agnostic, self-hostable, **no hardcoded hosts**: + +``` +LLM_ROUTE__BASE_URL # OpenAI-compatible /v1 base, e.g. http://localhost:11434/v1 +LLM_ROUTE__MODEL # model name as the endpoint expects it +LLM_ROUTE__API_KEY # optional; "none" if the endpoint ignores it +``` + +Tiers default to `LOCAL` and `FRONTIER` and are **extensible** (insert e.g. a `MID` rung by +passing `Router(ladder=[...])`). If no tier-specific env is set, each tier falls back to the +ubiquitous `OPENAI_BASE_URL` / `OPENAI_MODEL` / `OPENAI_API_KEY`, so a single-endpoint user +needs **no new env at all**. There are no hardcoded hosts, no provider SDKs, no private +infrastructure references anywhere in the module. + +--- + +## 2. Design + +Four small pieces (see the module docstring for the full contract): + +* **`TaskClass`** — `EXTRACT`, `CLASSIFY`, `DRAFT`, `VERIFY`. Sets the *starting* rung. + EXTRACT/CLASSIFY are local-first; VERIFY starts at `FRONTIER` (high-liability path). +* **`Signals`** — per-call evidence: `input_chars` / `approx_tokens`, `schema_strict`, + `prior_failures`, `ambiguity` (0–1), `needs_vision`. All default to "no pressure", so + `Signals()` gives plain local-first behavior. +* **`ModelTier`** — one rung: `name`, `base_url` (from env), `model`, `api_key`, + `max_context_hint`, `cost_hint`, `supports_vision`. Empty base/model ⇒ *unconfigured*. +* **`Router`** — `choose(task, signals) -> tier` (pure policy) and + `complete(task, messages, signals, validate=, on_decision=) -> Completion` (walks the + ladder). The returned `Completion` carries `tier_used`, `attempts`, `escalated`, and a + per-hop `route_log`. + +### Policy (`choose`) + +* **LOCAL-first** for EXTRACT / CLASSIFY / structured work. +* **Escalate one rung** per active signal: `prior_failures > 0`, `ambiguity > 0.7`, + `approx_tokens > tier.max_context_hint`, or `needs_vision` without local vision. +* **DRAFT** starts LOCAL but jumps a rung when `schema_strict`. +* **VERIFY** starts at `FRONTIER` by default — this mirrors the suite's existing + Qwen-draft → Opus-adjudicate split, where the high-liability judgment is not trusted to + the cheap rung. + +### Execution (`complete`) + +Walks from the policy's entry rung upward. Escalation is triggered by: + +1. an **unconfigured** rung (skipped, with a recorded reason — never silently dropped), +2. a **transport error**, or +3. a caller-supplied **`validate(text) -> bool`** returning falsy. + +`validate` generalizes the suite's existing retry-on-empty and enum-validation-reject +patterns (see forms-router `_llm` — empty JSON array retries, invalid form ids are +filtered). A **budget guard** caps escalations (`max_escalations`) and optionally total +cost (`cost_ceiling`). + +**Never silent.** Every hop appends a dict to `route_log` (and fires `on_decision`). +If **all** tiers are unconfigured, `complete` raises `NoTierConfigured` — the caller owns +its deterministic fallback (forms-router drops to lexical; hallucheck stays fail-closed). +The router never fabricates success. + +--- + +## 3. Escalation-signal catalog + +Each existing repo already computes one of these signals ad hoc; the router makes them a +uniform escalation vocabulary. + +| Signal | `Signals` field | Existing analog in the suite | +|--------|-----------------|------------------------------| +| Schema / enum validation failure | `validate` callback | forms-router `_llm` enum-filter + retry-on-empty; corp/probate `route_form.py` | +| Consensus disagreement | `prior_failures` / re-call | hallucheck `inspect_consensus(..., samples=3)` fail-biased vote | +| Lexical-margin ambiguity | `ambiguity` (0–1) | forms-router lexical scoring (tie-break / low-margin) | +| Context overflow | `approx_tokens` vs `max_context_hint` | forms-router's 591-form catalog prompt is already long-context | +| Vision requirement | `needs_vision` + `supports_vision` | the VLM / `LOCAL_VL_ENDPOINTS` tooling | +| Prior failure (any) | `prior_failures` | generic retry loops across corp/court tools | + +A caller wires consensus disagreement in by counting minority votes and passing them as +`prior_failures` (or by returning falsy from `validate` when agreement is below threshold), +so hallucheck's sampling behavior becomes an escalation trigger without special-casing. + +--- + +## 4. Ledger auditability wiring + +`complete(..., on_decision=cb)` calls `cb(hop_dict)` for **every** routing decision. The +hop dict carries `tier`, `configured`, `called`, `ok`, `validated`, `reason`, and `cost` — +exactly what an audit backing needs. It is designed to plug straight into +`legal_logic_layer.Ledger.record(source="llm", ...)`: + +* `schema.py` already lists **`llm_inference`** in `BACKING_TYPES`, and +* `APPLIES_KINDS` already contains an **unused `"route"` kind**. + +So a model choice becomes a first-class, reviewable ledger decision: + +```python +# In a consumer that also uses legal-logic-layer (llm_route itself imports neither): +def to_ledger(hop, ledger): + ledger.record( + f"routed to {hop['tier']}: {hop['reason']}", + source="llm", # forces needs_review=True in the ledger + backing=[{"type": "llm_inference"}], + # applies_to kind "route" — already a valid APPLIES_KIND, currently unused + ) + +router.complete(task, messages, signals, on_decision=lambda h: to_ledger(h, ledger)) +``` + +`llm_route` **does not import** `legal_logic_layer` — the callback is the only seam, +keeping the module zero-dependency and the wiring the consumer's choice. + +--- + +## 5. Migration plan + +Order and effort (S/M/L) follow the ecosystem audit's ranking. Each step swaps a repo's +hand-rolled client + bespoke env for `Router`, with no behavior change beyond gaining the +canonical env trio and the escalation ladder. + +| Order | Repo | Effort | What changes | +|-------|------|--------|--------------| +| 1 | maine-forms-router | **S** | Replace `_llm_config` / `_llm_call` with `Router.complete`; catalog prompt = long-context escalation; retry-on-empty + enum-reject become `validate`. One call site. | +| 2 | maine-government-feeds | **S** | Also moves off raw `httpx` onto the stdlib client; unifies `classify_items.py` env. | +| 3 | LLM_Hallucination_Checker | **M** | Reference **VERIFY** route; keep fail-closed; `inspect_consensus` disagreement → escalation signal. | +| 4 | corp audit tools | **M** | Un-name `AUDIT_QWEN_BASE_URL` / `VLM_API_BASE`; consolidate `route_form.py`. | +| 5 | maine-probate-forms | **L** | Un-hardcode `config.py` fleet constants; many call sites. | +| 6 | maine-court-forms | **L** | Most call sites in the suite; migrate last once the API is proven. | + +Each migration keeps the repo's existing deterministic fallback (lexical routing, +fail-closed verify) — `NoTierConfigured` hands control back rather than guessing. + +--- + +## 6. Configuration example + +Point tiers at any OpenAI-compatible endpoint. A common self-hosted setup: a +Gemma-3-27B-class local model via Ollama as `LOCAL`, and any frontier model as `FRONTIER`. + +```bash +# LOCAL tier: a Gemma-3-27B-class model served by Ollama's OpenAI-compatible API +export LLM_ROUTE_LOCAL_BASE_URL="http://localhost:11434/v1" +export LLM_ROUTE_LOCAL_MODEL="gemma3:27b" +export LLM_ROUTE_LOCAL_API_KEY="ollama" # Ollama ignores it; any string works + +# FRONTIER tier: any OpenAI-compatible frontier endpoint (cloud or self-hosted gateway) +export LLM_ROUTE_FRONTIER_BASE_URL="https://your-openai-compatible-endpoint/v1" +export LLM_ROUTE_FRONTIER_MODEL="your-frontier-model" +export LLM_ROUTE_FRONTIER_API_KEY="sk-..." +``` + +The same works with vLLM, llama.cpp's server, LiteLLM, or any cloud that speaks +OpenAI chat-completions. Set only `OPENAI_BASE_URL` / `OPENAI_MODEL` and every tier falls +back to it (single-endpoint mode). Examples are `localhost` / placeholder only — no host +is hardcoded in the module. + +```python +from maine_forms_engine.llm_route import Router, TaskClass, Signals, default_ladder + +router = Router(ladder=default_ladder()) # LOCAL then FRONTIER, from env +completion = router.complete( + TaskClass.CLASSIFY, + [{"role": "user", "content": "Classify this matter: ..."}], + Signals(input_chars=2400, schema_strict=True), + validate=lambda text: text.strip() in {"probate", "family", "corporate"}, +) +print(completion.tier_used, completion.escalated) +for hop in completion.route_log: + print(hop["tier"], hop["reason"]) +``` + +--- + +## 7. What this deliberately does NOT do + +* **No provider SDKs.** stdlib `urllib` only; OpenAI-compatible chat-completions only. +* **No change to deterministic fill.** Nothing in `maine_forms_engine.fill` imports this; + the engine still consults **no model at fill time**. +* **No hardcoded hosts or private-infrastructure references.** All endpoints come from env. +* **No hidden fallback.** All tiers unconfigured ⇒ raise; the caller owns its deterministic + fallback. The router never fakes a completion. +* **No streaming, no function-calling, no token accounting** in the reference cut — those + are additive and out of scope for the RFC. + +*Part of the 2026-07-06 suite-wide audit.* diff --git a/src/maine_forms_engine/llm_route.py b/src/maine_forms_engine/llm_route.py new file mode 100644 index 0000000..fca1f1c --- /dev/null +++ b/src/maine_forms_engine/llm_route.py @@ -0,0 +1,458 @@ +"""llm_route — optional, stdlib-only, local-first model routing with auditable escalation. + +RFC module (see ``docs/model-routing.md``). This is an **optional** add-on: nothing in +``maine_forms_engine.fill`` imports it, and it adds **zero** runtime dependencies. The +engine's fill path stays deterministic — no model is consulted at fill time. This module +is for the *callers* around the engine (the router, the drafting/verify tools) that today +each hand-roll their own OpenAI-compatible client with their own env convention. + +Design goals +------------ +* **One env convention** for the whole suite. Tiers are configured by a canonical trio:: + + LLM_ROUTE__BASE_URL e.g. LLM_ROUTE_LOCAL_BASE_URL=http://localhost:11434/v1 + LLM_ROUTE__MODEL e.g. LLM_ROUTE_LOCAL_MODEL=gemma3:27b + LLM_ROUTE__API_KEY e.g. LLM_ROUTE_LOCAL_API_KEY=ollama (optional) + + with a fallback to the ubiquitous ``OPENAI_BASE_URL`` / ``OPENAI_MODEL`` / + ``OPENAI_API_KEY`` so a single-endpoint user needs no new env at all. + +* **Local-first, escalate on evidence.** Cheap/local tiers answer structured extraction + and classification; the router climbs one rung when a signal says the local rung is + likely to be wrong (prior failures, high ambiguity, context overflow, vision need) or + when a caller-supplied ``validate`` rejects the output. + +* **Never silent.** Every :class:`Completion` carries which tier answered, how many + attempts it took, whether it escalated, and a per-hop ``route_log``. An unconfigured + tier is *skipped with a recorded reason*, not hidden. If **all** tiers are unconfigured + the router raises :class:`NoTierConfigured` — the caller owns its deterministic + fallback; this module never fakes success. + +STDLIB ONLY: ``urllib``, ``json``, ``dataclasses``, ``enum``, ``os``. This mirrors the +sibling ``maine-forms-router`` client so the module can graduate to its own repo unchanged. +""" + +from __future__ import annotations + +import json +import os +import urllib.error +import urllib.request +from dataclasses import dataclass, field +from enum import Enum +from typing import Any, Callable, Optional + +__all__ = [ + "TaskClass", + "Signals", + "ModelTier", + "Completion", + "Router", + "NoTierConfigured", + "load_tier", + "default_ladder", +] + +# The canonical tier order, cheapest/most-local first. Extra tiers can be inserted +# between these (e.g. a mid-size regional model) by naming them in Router(ladder=[...]). +DEFAULT_TIER_ORDER = ("LOCAL", "FRONTIER") + + +class TaskClass(Enum): + """What the model is being asked to do — drives the *starting* rung. + + EXTRACT / CLASSIFY are structured, low-liability, local-first tasks. DRAFT produces + prose a human reviews. VERIFY is the high-liability adjudication path and starts at + FRONTIER by default (mirrors the suite's Qwen-draft -> Opus-adjudicate pattern). + """ + + EXTRACT = "extract" + CLASSIFY = "classify" + DRAFT = "draft" + VERIFY = "verify" + + +@dataclass +class Signals: + """Per-call evidence the policy uses to decide the starting rung and escalations. + + All fields default to the "no escalation pressure" value so a caller can pass an + empty ``Signals()`` and get plain local-first behavior. + """ + + input_chars: int = 0 + #: Rough token estimate. If not given, derived from input_chars (~4 chars/token), + #: the same cheap heuristic the router uses for its long catalog prompt. + approx_tokens: Optional[int] = None + #: Output must satisfy a strict schema/enum (raises the bar for the local rung). + schema_strict: bool = False + #: How many times a prior attempt (this call or upstream) already failed. + prior_failures: int = 0 + #: Lexical/consensus ambiguity in [0, 1]; > 0.7 escalates. Mirrors forms-router's + #: lexical-margin and hallucheck's consensus-disagreement signals. + ambiguity: float = 0.0 + #: The task needs image/vision input. + needs_vision: bool = False + + def tokens(self) -> int: + if self.approx_tokens is not None: + return self.approx_tokens + # ~4 chars/token is the standard rough English heuristic. + return (self.input_chars + 3) // 4 + + +@dataclass +class ModelTier: + """One rung of the ladder. Configured from env by :func:`load_tier`. + + ``base_url``/``model`` empty => the tier is *unconfigured* and will be skipped + (with a recorded reason) rather than called. + """ + + name: str + base_url: str = "" + model: str = "" + api_key: str = "" + #: Advisory local context budget in tokens; a prompt above this escalates off this + #: rung. Not a hard API limit — just the policy's "this rung will truncate" hint. + max_context_hint: int = 8192 + #: Relative cost per call (USD-ish, advisory). Used only for the optional budget + #: ceiling; local tiers are ~0. + cost_hint: float = 0.0 + #: Whether this rung can accept vision input. + supports_vision: bool = False + + @property + def configured(self) -> bool: + return bool(self.base_url and self.model) + + +@dataclass +class Completion: + """The result of :meth:`Router.complete`. Never silent about how it got here.""" + + text: str + tier_used: Optional[str] + attempts: int + escalated: bool + #: One dict per hop: tier, configured, called, ok, reason, validated, cost. + route_log: list[dict] = field(default_factory=list) + + def to_dict(self) -> dict: + return { + "text": self.text, + "tier_used": self.tier_used, + "attempts": self.attempts, + "escalated": self.escalated, + "route_log": self.route_log, + } + + +class NoTierConfigured(RuntimeError): + """Raised by :meth:`Router.complete` when no tier in the ladder is configured. + + The caller decides its deterministic fallback (e.g. forms-router drops to lexical). + The router never fabricates a successful completion. + """ + + +# --------------------------------------------------------------- env config + + +def _env(tier: str, suffix: str) -> str: + return os.environ.get(f"LLM_ROUTE_{tier.upper()}_{suffix}", "") + + +def load_tier(name: str, **overrides: Any) -> ModelTier: + """Build a :class:`ModelTier` from the canonical env trio. + + ``LLM_ROUTE__BASE_URL`` / ``_MODEL`` / ``_API_KEY``. Falls back to the + ubiquitous ``OPENAI_BASE_URL`` / ``OPENAI_MODEL`` / ``OPENAI_API_KEY`` so a + single-endpoint deployment needs no tier-specific env. ``overrides`` (e.g. + ``max_context_hint``, ``cost_hint``, ``supports_vision``) win over env. + """ + + base = _env(name, "BASE_URL") or os.environ.get("OPENAI_BASE_URL", "") + model = _env(name, "MODEL") or os.environ.get("OPENAI_MODEL", "") + key = _env(name, "API_KEY") or os.environ.get("OPENAI_API_KEY", "") + tier = ModelTier( + name=name.upper(), + base_url=base.rstrip("/"), + model=model, + api_key=key or "none", + ) + for k, v in overrides.items(): + setattr(tier, k, v) + return tier + + +def default_ladder(order: tuple[str, ...] = DEFAULT_TIER_ORDER, + **per_tier: dict) -> list[ModelTier]: + """Build the standard ladder from env. ``per_tier`` maps tier name -> overrides, + e.g. ``default_ladder(LOCAL={"max_context_hint": 32768, "supports_vision": True})``. + """ + + ladder = [] + for name in order: + ladder.append(load_tier(name, **per_tier.get(name, {}))) + # A FRONTIER rung is assumed to be big-context + vision-capable unless told otherwise; + # this only affects the *hint*-based escalation triggers, not any API behavior. + for t in ladder: + if t.name == "FRONTIER" and "FRONTIER" not in per_tier: + t.max_context_hint = max(t.max_context_hint, 128_000) + t.supports_vision = True + return ladder + + +# --------------------------------------------------------------- transport + + +def _chat_completion(tier: ModelTier, messages: list[dict], *, + temperature: float = 0.0, max_tokens: int = 1024, + timeout: float = 60.0) -> str: + """One OpenAI-compatible /chat/completions POST via urllib. Mirrors + maine-forms-router._llm_call so the wire contract is identical across the suite.""" + + body = json.dumps({ + "model": tier.model, + "messages": messages, + "temperature": temperature, + "max_tokens": max_tokens, + }).encode() + req = urllib.request.Request( + f"{tier.base_url}/chat/completions", + data=body, + headers={ + "Content-Type": "application/json", + "Authorization": f"Bearer {tier.api_key}", + }, + ) + with urllib.request.urlopen(req, timeout=timeout) as resp: + data = json.loads(resp.read()) + return data["choices"][0]["message"]["content"] + + +# --------------------------------------------------------------- the router + + +@dataclass +class Router: + """Local-first router over a tier ladder with auditable, budgeted escalation.""" + + ladder: list[ModelTier] = field(default_factory=default_ladder) + #: Cap on how many *additional* rungs above the start we will try. + max_escalations: int = 2 + #: Optional cumulative cost ceiling (sum of cost_hint of called tiers). None = off. + cost_ceiling: Optional[float] = None + #: Injectable transport (for tests). Signature: (tier, messages) -> str. + transport: Callable[..., str] = _chat_completion + + # --- policy --------------------------------------------------------- + + def _start_index(self, task: TaskClass, signals: Signals) -> int: + """Which rung to *start* at, before escalation. VERIFY starts one rung up.""" + idx = 0 + if task == TaskClass.VERIFY: + idx = min(1, len(self.ladder) - 1) # frontier-by-default + elif task == TaskClass.DRAFT: + # Drafting is prose a human reviews; local is fine to start, but a strict + # schema pushes it up a rung immediately. + if signals.schema_strict: + idx = min(idx + 1, len(self.ladder) - 1) + return idx + + def _escalation_pressure(self, task: TaskClass, signals: Signals, + tier: ModelTier) -> list[str]: + """Reasons (if any) the *current* rung is a poor fit -> escalate one rung. + + Returned as a list of human-readable reasons for the route_log; empty => stay. + """ + reasons = [] + if signals.prior_failures > 0: + reasons.append(f"prior_failures={signals.prior_failures}") + if signals.ambiguity > 0.7: + reasons.append(f"ambiguity={signals.ambiguity:.2f}>0.7") + if signals.tokens() > tier.max_context_hint: + reasons.append( + f"approx_tokens={signals.tokens()}>context_hint={tier.max_context_hint}") + if signals.needs_vision and not tier.supports_vision: + reasons.append("needs_vision without local vision") + return reasons + + def choose(self, task: TaskClass, signals: Optional[Signals] = None) -> ModelTier: + """Pick the tier the policy would *start* at, after applying escalation pressure + from the signals. Pure and side-effect-free; :meth:`complete` uses it as the + entry rung then walks up on validation failure. Returns the chosen configured or + unconfigured tier (caller can inspect ``.configured``).""" + signals = signals or Signals() + idx = self._start_index(task, signals) + # Apply signal-driven pressure once from the start rung (bounded by ladder top). + while idx < len(self.ladder) - 1 and self._escalation_pressure( + task, signals, self.ladder[idx]): + idx += 1 + return self.ladder[idx] + + # --- execution ------------------------------------------------------ + + def complete(self, task: TaskClass, messages: list[dict], + signals: Optional[Signals] = None, *, + validate: Optional[Callable[[str], bool]] = None, + on_decision: Optional[Callable[[dict], None]] = None, + **transport_kw: Any) -> Completion: + """Walk the ladder from the policy's entry rung, returning the first output that + (optionally) passes ``validate``. + + Escalation happens when: + * a rung is unconfigured (skipped, recorded), or + * the transport errors, or + * ``validate(text)`` returns falsy (generalizes the suite's retry-on-empty and + enum-validation-reject patterns). + + ``on_decision`` receives each hop's dict as it happens — this is the seam for + ``legal_logic_layer.Ledger.record(source="llm", ...)`` (no import here; keep + zero-dep). Budget guard: at most ``max_escalations`` rungs above the start, and + an optional cumulative ``cost_ceiling``. + + Raises :class:`NoTierConfigured` if no configured tier was reachable at all. + """ + signals = signals or Signals() + start = self._start_index(task, signals) + route_log: list[dict] = [] + attempts = 0 + spent = 0.0 + escalations_used = 0 + any_configured_called = False + + idx = start + # Fold the initial signal-driven pressure into the starting rung too, so choose() + # and complete() agree on where we begin. + while idx < len(self.ladder) - 1 and self._escalation_pressure( + task, signals, self.ladder[idx]): + idx += 1 + start = idx + + while idx < len(self.ladder): + tier = self.ladder[idx] + hop: dict = { + "task": task.value, + "tier": tier.name, + "index": idx, + "configured": tier.configured, + "called": False, + "ok": False, + "validated": None, + "reason": "", + "cost": 0.0, + } + + if not tier.configured: + hop["reason"] = "tier unconfigured (no base_url/model in env)" + route_log.append(hop) + if on_decision: + on_decision(hop) + idx += 1 + continue + + if self.cost_ceiling is not None and spent + tier.cost_hint > self.cost_ceiling: + hop["reason"] = (f"cost ceiling {self.cost_ceiling} would be exceeded " + f"(spent={spent}, tier={tier.cost_hint})") + route_log.append(hop) + if on_decision: + on_decision(hop) + break + + attempts += 1 + any_configured_called = True + hop["called"] = True + try: + text = self.transport(tier, messages, **transport_kw) + except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, + OSError, KeyError, ValueError, json.JSONDecodeError) as exc: + spent += tier.cost_hint + hop["cost"] = tier.cost_hint + hop["reason"] = f"transport error: {type(exc).__name__}: {exc}" + route_log.append(hop) + if on_decision: + on_decision(hop) + idx, escalations_used = self._advance(idx, start, escalations_used, + route_log, on_decision) + if idx is None: + break + continue + + spent += tier.cost_hint + hop["cost"] = tier.cost_hint + ok = True + if validate is not None: + try: + ok = bool(validate(text)) + except Exception as exc: # a broken validator must not crash routing + ok = False + hop["reason"] = f"validate() raised {type(exc).__name__}: {exc}" + hop["validated"] = ok + + if ok: + hop["ok"] = True + if not hop["reason"]: + hop["reason"] = "accepted" + route_log.append(hop) + if on_decision: + on_decision(hop) + return Completion( + text=text, + tier_used=tier.name, + attempts=attempts, + escalated=idx > start, + route_log=route_log, + ) + + if not hop["reason"]: + hop["reason"] = "validate() rejected output" + route_log.append(hop) + if on_decision: + on_decision(hop) + idx, escalations_used = self._advance(idx, start, escalations_used, + route_log, on_decision) + if idx is None: + break + + if not any_configured_called: + raise NoTierConfigured( + "no configured tier in ladder " + f"({[t.name for t in self.ladder]}); set LLM_ROUTE__BASE_URL " + "and _MODEL (or OPENAI_BASE_URL/OPENAI_MODEL). route_log=" + f"{route_log}" + ) + + # Configured tiers were tried but none passed validation / budget ran out. + # Return the last text-bearing failure honestly (empty text, escalated=True). + return Completion( + text="", + tier_used=None, + attempts=attempts, + escalated=True, + route_log=route_log, + ) + + def _advance(self, idx: int, start: int, escalations_used: int, + route_log: list[dict], + on_decision: Optional[Callable[[dict], None]]): + """Move to the next rung if the escalation budget allows; else stop. + + Returns ``(new_idx_or_None, escalations_used)``. Records a budget-stop hop when + the ladder cannot climb further. + """ + if escalations_used >= self.max_escalations or idx + 1 >= len(self.ladder): + stop = { + "tier": None, + "called": False, + "ok": False, + "reason": (f"escalation budget exhausted " + f"(used={escalations_used}, max={self.max_escalations}, " + f"at rung {idx} of {len(self.ladder) - 1})"), + } + route_log.append(stop) + if on_decision: + on_decision(stop) + return None, escalations_used + return idx + 1, escalations_used + 1 diff --git a/tests/test_llm_route.py b/tests/test_llm_route.py new file mode 100644 index 0000000..f7b17a0 --- /dev/null +++ b/tests/test_llm_route.py @@ -0,0 +1,341 @@ +"""Tests for the optional llm_route module. No network: the transport is either +injected as a stub or urllib is monkeypatched. Covers the choose() policy matrix, +escalation on validate-fail, budget stop, the all-unconfigured raise, and route_log +completeness.""" +import json +import urllib.error + +import pytest + +from maine_forms_engine.llm_route import ( + Completion, + ModelTier, + NoTierConfigured, + Router, + Signals, + TaskClass, + default_ladder, + load_tier, +) + + +# --------------------------------------------------------------- fixtures + + +def _tier(name, *, configured=True, **kw): + if configured: + kw.setdefault("base_url", f"http://{name.lower()}.invalid/v1") + kw.setdefault("model", f"{name.lower()}-model") + kw.setdefault("api_key", "none") + return ModelTier(name=name, **kw) + + +def _echo_transport(reply): + """Return a transport that always returns ``reply`` and records calls.""" + calls = [] + + def transport(tier, messages, **kw): + calls.append(tier.name) + return reply + + transport.calls = calls + return transport + + +def _local_frontier(**router_kw): + ladder = [ + _tier("LOCAL", max_context_hint=8192, cost_hint=0.0, supports_vision=False), + _tier("FRONTIER", max_context_hint=128000, cost_hint=0.05, supports_vision=True), + ] + return ladder, Router(ladder=ladder, **router_kw) + + +MSGS = [{"role": "user", "content": "hi"}] + + +# --------------------------------------------------------------- env / config + + +def test_load_tier_canonical_env(monkeypatch): + monkeypatch.setenv("LLM_ROUTE_LOCAL_BASE_URL", "http://localhost:11434/v1/") + monkeypatch.setenv("LLM_ROUTE_LOCAL_MODEL", "gemma3:27b") + monkeypatch.setenv("LLM_ROUTE_LOCAL_API_KEY", "ollama") + t = load_tier("local") + assert t.name == "LOCAL" + assert t.base_url == "http://localhost:11434/v1" # trailing slash stripped + assert t.model == "gemma3:27b" + assert t.api_key == "ollama" + assert t.configured + + +def test_load_tier_openai_fallback(monkeypatch): + for k in list(__import__("os").environ): + if k.startswith("LLM_ROUTE_"): + monkeypatch.delenv(k, raising=False) + monkeypatch.setenv("OPENAI_BASE_URL", "http://localhost:8000/v1") + monkeypatch.setenv("OPENAI_MODEL", "qwen2.5") + monkeypatch.delenv("OPENAI_API_KEY", raising=False) + t = load_tier("LOCAL") + assert t.base_url == "http://localhost:8000/v1" + assert t.model == "qwen2.5" + assert t.api_key == "none" # empty -> "none" sentinel like the sibling router + assert t.configured + + +def test_load_tier_unconfigured(monkeypatch): + for k in list(__import__("os").environ): + if k.startswith(("LLM_ROUTE_", "OPENAI_")): + monkeypatch.delenv(k, raising=False) + t = load_tier("FRONTIER") + assert not t.configured + + +def test_default_ladder_frontier_defaults(monkeypatch): + for k in list(__import__("os").environ): + if k.startswith(("LLM_ROUTE_", "OPENAI_")): + monkeypatch.delenv(k, raising=False) + monkeypatch.setenv("LLM_ROUTE_FRONTIER_BASE_URL", "http://f.invalid/v1") + monkeypatch.setenv("LLM_ROUTE_FRONTIER_MODEL", "big") + ladder = default_ladder() + frontier = [t for t in ladder if t.name == "FRONTIER"][0] + assert frontier.max_context_hint >= 128000 + assert frontier.supports_vision is True + + +# --------------------------------------------------------------- choose() matrix + + +def test_choose_extract_classify_start_local(): + _, r = _local_frontier() + assert r.choose(TaskClass.EXTRACT, Signals()).name == "LOCAL" + assert r.choose(TaskClass.CLASSIFY, Signals()).name == "LOCAL" + + +def test_choose_verify_starts_frontier(): + _, r = _local_frontier() + assert r.choose(TaskClass.VERIFY, Signals()).name == "FRONTIER" + + +def test_choose_draft_schema_strict_escalates(): + _, r = _local_frontier() + assert r.choose(TaskClass.DRAFT, Signals()).name == "LOCAL" + assert r.choose(TaskClass.DRAFT, Signals(schema_strict=True)).name == "FRONTIER" + + +def test_choose_prior_failures_escalates(): + _, r = _local_frontier() + assert r.choose(TaskClass.EXTRACT, Signals(prior_failures=1)).name == "FRONTIER" + + +def test_choose_high_ambiguity_escalates(): + _, r = _local_frontier() + assert r.choose(TaskClass.CLASSIFY, Signals(ambiguity=0.9)).name == "FRONTIER" + assert r.choose(TaskClass.CLASSIFY, Signals(ambiguity=0.5)).name == "LOCAL" + + +def test_choose_context_overflow_escalates(): + _, r = _local_frontier() + big = Signals(approx_tokens=20000) # over LOCAL's 8192 hint + assert r.choose(TaskClass.EXTRACT, big).name == "FRONTIER" + + +def test_choose_vision_without_local_vision_escalates(): + _, r = _local_frontier() + assert r.choose(TaskClass.EXTRACT, Signals(needs_vision=True)).name == "FRONTIER" + + +def test_signals_token_estimate_from_chars(): + assert Signals(input_chars=400).tokens() == 100 # ~4 chars/token + assert Signals(approx_tokens=7).tokens() == 7 # explicit wins + + +# --------------------------------------------------------------- complete() + + +def test_complete_local_first_success(): + ladder, r = _local_frontier() + r.transport = _echo_transport("ok") + c = r.complete(TaskClass.EXTRACT, MSGS, Signals()) + assert isinstance(c, Completion) + assert c.tier_used == "LOCAL" + assert c.attempts == 1 + assert c.escalated is False + assert r.transport.calls == ["LOCAL"] + assert c.route_log[-1]["ok"] is True + + +def test_complete_escalates_on_validate_fail(): + ladder, r = _local_frontier() + # LOCAL returns "bad", FRONTIER returns "good"; validate accepts only "good". + def transport(tier, messages, **kw): + return "good" if tier.name == "FRONTIER" else "bad" + r.transport = transport + c = r.complete(TaskClass.EXTRACT, MSGS, Signals(), + validate=lambda t: t == "good") + assert c.tier_used == "FRONTIER" + assert c.attempts == 2 + assert c.escalated is True + # route_log: LOCAL rejected, FRONTIER accepted + assert c.route_log[0]["tier"] == "LOCAL" + assert c.route_log[0]["validated"] is False + assert c.route_log[-1]["tier"] == "FRONTIER" + assert c.route_log[-1]["ok"] is True + + +def test_complete_skips_unconfigured_tier(): + ladder = [ + _tier("LOCAL", configured=False), + _tier("FRONTIER"), + ] + r = Router(ladder=ladder) + r.transport = _echo_transport("answer") + c = r.complete(TaskClass.EXTRACT, MSGS, Signals()) + assert c.tier_used == "FRONTIER" + # LOCAL recorded as skipped with a reason, not silently dropped + local_hop = c.route_log[0] + assert local_hop["tier"] == "LOCAL" + assert local_hop["configured"] is False + assert local_hop["called"] is False + assert "unconfigured" in local_hop["reason"] + + +def test_complete_all_unconfigured_raises(): + ladder = [_tier("LOCAL", configured=False), _tier("FRONTIER", configured=False)] + r = Router(ladder=ladder) + with pytest.raises(NoTierConfigured): + r.complete(TaskClass.EXTRACT, MSGS, Signals()) + + +def test_complete_budget_stops_escalation(): + ladder = [_tier("LOCAL"), _tier("MID"), _tier("FRONTIER")] + r = Router(ladder=ladder, max_escalations=1) + r.transport = _echo_transport("nope") # always fails validate + c = r.complete(TaskClass.EXTRACT, MSGS, Signals(), + validate=lambda t: False) + # start=LOCAL, one escalation allowed -> tries LOCAL, MID, then budget stop + assert c.tier_used is None + assert c.attempts == 2 + assert c.escalated is True + assert any("budget exhausted" in h["reason"] for h in c.route_log) + + +def test_complete_cost_ceiling_blocks_expensive_tier(): + ladder = [ + _tier("LOCAL", cost_hint=0.0), + _tier("FRONTIER", cost_hint=1.0), + ] + r = Router(ladder=ladder, cost_ceiling=0.5) + r.transport = _echo_transport("x") + c = r.complete(TaskClass.EXTRACT, MSGS, Signals(), + validate=lambda t: False) # force escalation attempt + assert c.tier_used is None + # FRONTIER blocked by cost ceiling, recorded + assert any("cost ceiling" in h["reason"] for h in c.route_log) + + +def test_complete_transport_error_escalates(): + ladder, r = _local_frontier() + + def transport(tier, messages, **kw): + if tier.name == "LOCAL": + raise urllib.error.URLError("connection refused") + return "recovered" + r.transport = transport + c = r.complete(TaskClass.EXTRACT, MSGS, Signals()) + assert c.tier_used == "FRONTIER" + assert "transport error" in c.route_log[0]["reason"] + assert c.route_log[0]["called"] is True + + +def test_on_decision_callback_receives_every_hop(): + ladder, r = _local_frontier() + + def transport(tier, messages, **kw): + return "good" if tier.name == "FRONTIER" else "bad" + r.transport = transport + seen = [] + r.complete(TaskClass.EXTRACT, MSGS, Signals(), + validate=lambda t: t == "good", + on_decision=seen.append) + # LOCAL reject + FRONTIER accept => 2 decisions, mirroring route_log + assert [h["tier"] for h in seen] == ["LOCAL", "FRONTIER"] + assert seen[-1]["ok"] is True + + +def test_ledger_shaped_callback_wiring(): + """The on_decision dict carries exactly what a ledger llm_inference backing needs: + tier + escalation + reason. This test documents the seam (no legal_logic_layer import).""" + ladder, r = _local_frontier() + r.transport = _echo_transport("ok") + records = [] + + def fake_ledger_record(hop): + # what legal_logic_layer.Ledger.record(source="llm", backing=[{type:llm_inference}]) + # would consume: + records.append({ + "source": "llm", + "backing_type": "llm_inference", + "applies_kind": "route", + "tier": hop["tier"], + "reason": hop["reason"], + }) + r.complete(TaskClass.CLASSIFY, MSGS, Signals(), on_decision=fake_ledger_record) + assert records[0]["tier"] == "LOCAL" + assert records[0]["applies_kind"] == "route" + assert records[0]["backing_type"] == "llm_inference" + + +def test_route_log_completeness_fields(): + ladder, r = _local_frontier() + r.transport = _echo_transport("ok") + c = r.complete(TaskClass.EXTRACT, MSGS, Signals()) + hop = c.route_log[-1] + for key in ("task", "tier", "index", "configured", "called", "ok", + "validated", "reason", "cost"): + assert key in hop + + +def test_completion_to_dict_roundtrips(): + ladder, r = _local_frontier() + r.transport = _echo_transport("ok") + c = r.complete(TaskClass.EXTRACT, MSGS, Signals()) + d = c.to_dict() + assert set(d) == {"text", "tier_used", "attempts", "escalated", "route_log"} + # JSON-serializable end to end (matters for ledger emit) + assert json.loads(json.dumps(d))["tier_used"] == "LOCAL" + + +# --------------------------------------------------------------- transport wire shape + + +def test_real_transport_builds_openai_request(monkeypatch): + """Verify _chat_completion posts an OpenAI-compatible body via urllib (mocked).""" + from maine_forms_engine import llm_route + + captured = {} + + class FakeResp: + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def read(self): + return json.dumps( + {"choices": [{"message": {"content": "hello"}}]}).encode() + + def fake_urlopen(req, timeout=None): + captured["url"] = req.full_url + captured["body"] = json.loads(req.data) + captured["auth"] = req.headers.get("Authorization") + return FakeResp() + + monkeypatch.setattr(llm_route.urllib.request, "urlopen", fake_urlopen) + tier = _tier("LOCAL") + out = llm_route._chat_completion(tier, MSGS) + assert out == "hello" + assert captured["url"].endswith("/chat/completions") + assert captured["body"]["model"] == tier.model + assert captured["body"]["messages"] == MSGS + assert captured["auth"] == "Bearer none"