Route Between a Small and a Large Model with Jev

Tallyboard's support assistant answers billing questions from policy.md: eleven numbered rules about prices, seats, refunds and failed payments. Two models on OpenRouter can do the job:

The large model costs fourteen times more per answer. Most requests are lookups ("How much is Team per seat?") that both models get right. The small model's misses are the requests that combine rules, such as a nonprofit discount applied on top of annual pricing, or a refund counted in used months.

Both models reason before they answer. llm.chat(..., reasoning_effort="low") keeps the small model's reasoning short, which is how you would run a cheap tier in production.

Do this

Open router.py.

1. Write extract(text). The model ends with ANSWER: <value>. Return the text after the last ANSWER: (strip spaces and * around it). The small model sometimes drops that line on one-word replies, so fall back to the last non-empty line. Return None for empty or missing text.

2. Write answer(model, request, chat=llm.chat). Make one chat call with a system message (SYSTEM, the instructions plus the policy) and a user message (the request). Use reasoning_effort="low" for SMALL and None for LARGE. Return {"model", "answer", "text", "cost", "ms"}.

3. Run. Five requests go to both models. The lookups come back right from both. Watch #44 (a nonprofit's saving from annual billing) and #56 (seats removed and added on the same day): the small model usually misses at least one of them, for about a fifteenth of the price.

Starter file: jev.py

"""A small client for Jev, TypeSafe's decision model, through the lab's model proxy.

You send a state and some typed questions; Jev sends back one typed answer per
question with probabilities attached. It never writes text.

    jev.post({"model": jev.MODEL, "state": {...}, "questions": {...}})

Identical requests are answered from .cache/jev/, so running a step twice, or
the check after a run, does not ask the model again. Delete .cache/ to start
fresh.
"""
import hashlib
import json
import os
import time
import urllib.error
import urllib.request

# The pod's proxy holds the API key; this lab never sees it.
BASE = os.environ.get("OPENAI_BASE_URL", "http://nim-proxy.labs.svc:8080/v1").rstrip("/")
URL = BASE + "/decisions"
MODEL = "~typesafe/jev-latest"
CACHE_DIR = os.path.join(".cache", "jev")


class JevError(Exception):
    """Jev rejected the request (a malformed question, usually) or could not be reached."""


def _cache_path(payload):
    key = hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()[:24]
    return os.path.join(CACHE_DIR, key + ".json")


def post(payload, use_cache=True):
    """Send one decision request and return the parsed response:
    {"model": ..., "answers": {name: {...}}, "usage": {...}}."""
    path = _cache_path(payload)
    if use_cache and os.path.exists(path):
        with open(path, encoding="utf-8") as f:
            return json.load(f)

    data = json.dumps(payload).encode()
    for attempt in range(6):
        req = urllib.request.Request(URL, data=data, headers={"Content-Type": "application/json"})
        try:
            with urllib.request.urlopen(req, timeout=30) as resp:
                body = json.loads(resp.read())
            break
        except urllib.error.HTTPError as e:
            detail = e.read().decode(errors="replace")
            if e.code in (429, 500, 502, 503, 529) and attempt < 5:
                # 429 is the proxy's per-minute budget: wait for the next window.
                time.sleep((5 if e.code == 429 else 2) * (attempt + 1))
                continue
            raise JevError(f"Jev answered HTTP {e.code}: {_readable(detail)}") from None
        except (urllib.error.URLError, TimeoutError) as e:
            if attempt < 5:
                time.sleep(2 * (attempt + 1))
                continue
            raise JevError(f"Could not reach the model proxy at {URL}: {e}") from None

    if use_cache:
        os.makedirs(CACHE_DIR, exist_ok=True)
        tmp = path + ".tmp"
        with open(tmp, "w", encoding="utf-8") as f:
            json.dump(body, f)
        os.replace(tmp, path)
    return body


def _readable(detail):
    """Validation errors arrive as a JSON list inside a JSON string; show the useful part."""
    try:
        msg = json.loads(detail)["error"]["message"]
        issues = json.loads(msg)
        return "; ".join(f"{'.'.join(str(p) for p in i.get('path', []))}: {i.get('message')}" for i in issues)
    except Exception:
        return detail[:300]