b84b4839dc
Post a task to POST /route and print the full decision tree (classification, candidates, selected model, est cost, rejections) without dispatching a provider call. Supports --category/--tier/--context overrides and --json. Route-only and no-spend.
234 lines
7.5 KiB
Python
234 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
|
"""router_cli.py -- one-shot routing-decision printer.
|
|
|
|
POSTs a task to the running router's ``POST /route`` endpoint and prints the
|
|
full decision tree: classification, candidates considered, the selected model
|
|
(plus its estimated cost and proficiency), runners-up, and -- when nothing was
|
|
selected -- the rejection reason.
|
|
|
|
This is a **route-only, no-spend probe**. It never dispatches to a provider and
|
|
never calls a model. It is not a daemon; it runs once in the foreground and
|
|
exits.
|
|
|
|
Usage:
|
|
python router_cli.py "<task>" [--category CAT] [--tier N] [--context N] [--json]
|
|
|
|
The router base URL comes from the ROUTER_URL env var, defaulting to
|
|
``http://127.0.0.1:8080``. ``/route`` needs no auth (loopback-only bind).
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
import requests
|
|
|
|
DEFAULT_BASE_URL = "http://127.0.0.1:8080"
|
|
ROUTE_PATH = "/route"
|
|
|
|
|
|
# --- pure rendering helpers -----------------------------------------------
|
|
|
|
|
|
def _fmt_cost(cost):
|
|
"""A compact human-readable US$ for a cost figure, or a placeholder."""
|
|
if cost is None:
|
|
return "n/a"
|
|
if cost < 0.001:
|
|
return f"US${cost:.3e}"
|
|
return f"US${cost:.6f}"
|
|
|
|
|
|
def _fmt_proficiency(prof):
|
|
if prof is None:
|
|
return "n/a"
|
|
return f"{prof:.3f}"
|
|
|
|
|
|
def render_decision(data: dict) -> str:
|
|
"""Render a RouteResponse-shaped dict into a human decision tree."""
|
|
lines = []
|
|
cls = data.get("classification") or {}
|
|
lines.append("classification:")
|
|
lines.append(f" category : {cls.get('task_category', 'n/a')}")
|
|
lines.append(f" tier : {cls.get('task_tier', 'n/a')}")
|
|
lines.append(
|
|
f" required_context: {cls.get('required_context_tokens', 'n/a')}"
|
|
)
|
|
lines.append(f" confidence : {cls.get('confidence', 'n/a')}")
|
|
lines.append(f" source : {cls.get('source', 'n/a')}")
|
|
lines.append(f"latency_tolerance : {data.get('latency_tolerance', 'n/a')}")
|
|
lines.append(f"candidates_considered: {data.get('candidates_considered', 'n/a')}")
|
|
|
|
selected = data.get("selected")
|
|
if selected:
|
|
lines.append("selected:")
|
|
lines.append(f" model : {selected.get('model_id', 'n/a')}")
|
|
lines.append(f" provider : {selected.get('provider', 'n/a')}")
|
|
lines.append(f" tier : {selected.get('tier', 'n/a')}")
|
|
lines.append(f" est cost : {_fmt_cost(selected.get('cost'))}")
|
|
lines.append(
|
|
f" proficiency: {_fmt_proficiency(selected.get('proficiency_score'))}"
|
|
)
|
|
else:
|
|
lines.append("selected: (none)")
|
|
|
|
runners_up = data.get("runners_up") or []
|
|
if runners_up:
|
|
lines.append("runners-up:")
|
|
for ru in runners_up[:3]:
|
|
lines.append(
|
|
f" {ru.get('model_id', 'n/a')} "
|
|
f"(est cost {_fmt_cost(ru.get('cost'))}, "
|
|
f"proficiency {_fmt_proficiency(ru.get('proficiency_score'))})"
|
|
)
|
|
else:
|
|
lines.append("runners-up: (none)")
|
|
|
|
if not selected:
|
|
reason = data.get("detail") or data.get("rejected_reason")
|
|
lines.append("rejected_reason: " + (reason or "no model selected"))
|
|
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def _validate_overrides(tier, context):
|
|
"""Return (tier, context) validated, or raise ValueError with a message."""
|
|
if tier is not None:
|
|
if isinstance(tier, bool):
|
|
raise ValueError("--tier must be an integer between 1 and 3")
|
|
try:
|
|
tier = int(tier)
|
|
except (TypeError, ValueError):
|
|
raise ValueError("--tier must be an integer between 1 and 3")
|
|
if not (1 <= tier <= 3):
|
|
raise ValueError("--tier must be between 1 and 3")
|
|
if context is not None:
|
|
if isinstance(context, bool):
|
|
raise ValueError("--context must be a non-negative integer")
|
|
try:
|
|
context = int(context)
|
|
except (TypeError, ValueError):
|
|
raise ValueError("--context must be a non-negative integer")
|
|
if context < 0:
|
|
raise ValueError("--context must be a non-negative integer")
|
|
return tier, context
|
|
|
|
|
|
# --- the callable decision logic ------------------------------------------
|
|
|
|
|
|
def run(
|
|
task,
|
|
category=None,
|
|
tier=None,
|
|
context=None,
|
|
as_json=False,
|
|
post=None,
|
|
base_url=None,
|
|
):
|
|
"""Route ``task`` and print the decision tree.
|
|
|
|
Returns a process exit code (0 on success). ``post`` is an injectable
|
|
callback ``post(url, json=..., timeout=...) -> response`` replacing
|
|
``requests.post`` for testing; ``base_url`` defaults to the ROUTER_URL env
|
|
var or the loopback default. Route-only: never dispatches to a provider.
|
|
"""
|
|
try:
|
|
tier, context = _validate_overrides(tier, context)
|
|
except ValueError as e:
|
|
print(f"router-cli: {e}", file=sys.stderr)
|
|
return 2
|
|
|
|
if base_url is None:
|
|
base_url = os.environ.get("ROUTER_URL", DEFAULT_BASE_URL)
|
|
url = base_url.rstrip("/") + ROUTE_PATH
|
|
|
|
body = {"task": task}
|
|
if category is not None:
|
|
body["task_category"] = category
|
|
if tier is not None:
|
|
body["task_tier"] = tier
|
|
if context is not None:
|
|
body["required_context_tokens"] = context
|
|
|
|
if post is None:
|
|
post = lambda u, json=None, **kw: requests.post(u, json=json, timeout=30, **kw)
|
|
|
|
try:
|
|
resp = post(url, json=body)
|
|
except requests.exceptions.RequestException as e:
|
|
# requests' own ConnectionError/Timeout/HTTPError family.
|
|
print(f"router-cli: request to {url} failed: {e}", file=sys.stderr)
|
|
return 1
|
|
except ConnectionError as e:
|
|
# The injectable callback (and tests) may raise the builtin instead.
|
|
print(
|
|
f"router-cli: could not reach the router at {url} "
|
|
f"(connection error: {e}). Is the dispatcher running?",
|
|
file=sys.stderr,
|
|
)
|
|
return 1
|
|
|
|
if resp.status_code != 200:
|
|
reason = None
|
|
try:
|
|
reason = resp.json().get("detail")
|
|
except (ValueError, TypeError, AttributeError):
|
|
if getattr(resp, "text", None):
|
|
reason = resp.text
|
|
print(
|
|
f"router-cli: router returned {resp.status_code}: "
|
|
f"{reason or 'no model satisfies the routing filters'}",
|
|
file=sys.stderr,
|
|
)
|
|
return 1
|
|
|
|
data = resp.json()
|
|
if as_json:
|
|
print(json.dumps(data, indent=2))
|
|
else:
|
|
sys.stdout.write(render_decision(data))
|
|
return 0
|
|
|
|
|
|
# --- CLI entry ------------------------------------------------------------
|
|
|
|
def _build_parser():
|
|
p = argparse.ArgumentParser(
|
|
prog="router_cli.py",
|
|
description=(
|
|
"Route a task through the running router and print the decision "
|
|
"tree. Route-only: never dispatches to a model or spends quota."
|
|
),
|
|
)
|
|
p.add_argument("task", help="The task text to route.")
|
|
p.add_argument("--category", help="Override task_category (skips classifier).")
|
|
p.add_argument("--tier", type=int, help="Override task_tier (1-3).")
|
|
p.add_argument(
|
|
"--context", type=int, help="Override required_context_tokens."
|
|
)
|
|
p.add_argument(
|
|
"--json",
|
|
action="store_true",
|
|
help="Emit the raw RouteResponse JSON instead of the decision tree.",
|
|
)
|
|
return p
|
|
|
|
|
|
def main(argv=None):
|
|
parser = _build_parser()
|
|
args = parser.parse_args(argv)
|
|
return run(
|
|
args.task,
|
|
category=args.category,
|
|
tier=args.tier,
|
|
context=args.context,
|
|
as_json=args.json,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|