diff --git a/.gitignore b/.gitignore index 2284503..594b4e0 100644 --- a/.gitignore +++ b/.gitignore @@ -6,3 +6,4 @@ __pycache__/ dist/ build/ .env +benchmarks/results/**/decisions.jsonl diff --git a/README.md b/README.md index 5d3cba2..9f735fe 100644 --- a/README.md +++ b/README.md @@ -35,6 +35,23 @@ chat completion to make the routing decision. - **Auditable decisions.** Inspect the complete state, question, and opaque choice mapping before sending anything. +## Small result + +On a 191-query complete-case slice of +[LLMRouterBench](https://github.com/ynulihao/LLMRouterBench), adding K=4 +retrieved observations per candidate improved ReflexRoute's mean quality from +`0.5497` to `0.6832` (+24.3% relative). Routing remained inexpensive: + +| Mode | Mean quality | Router latency P50 | Router cost / 1,000 decisions | +| --- | ---: | ---: | ---: | +| Zero-shot | 0.5497 | 722 ms | $0.108 | +| Few-shot, K=4 | 0.6832 | 1,333 ms | $0.403 | + +This is a small exploratory measurement of ReflexRoute's two modes, not a +comparison against other routers. Candidate outputs were cached, so the run +only paid `$0.09757` for 382 Jev decisions. See the +[setup, cost definition, caveats, and reproduction command](benchmarks/README.md). + ## Install ReflexRoute requires Python 3.10 or newer. @@ -59,6 +76,19 @@ Set your OpenRouter API key: export OPENROUTER_API_KEY="your-key" ``` +## 60-second demo + +The demo routes one query twice: first from built-in priors, then after adding +eight illustrative user observations. + +```bash +python examples/adaptation_demo.py +``` + +It prints the selected model, distribution, confidence, routing latency, and +the Jev charge for each decision. It makes two Decisions API calls but does not +invoke either selected model. + ## Quick start ### Zero-shot routing diff --git a/benchmarks/README.md b/benchmarks/README.md new file mode 100644 index 0000000..4be2228 --- /dev/null +++ b/benchmarks/README.md @@ -0,0 +1,61 @@ +# Small benchmark + +This is a deliberately small, low-cost measurement of ReflexRoute itself. It +shows the cold-start and few-shot modes without claiming a comparison against +other routing systems. + +## Result + +| Mode | Mean quality | Router latency P50 / P95 | Router cost / 1,000 decisions | Selected-model cost / 1,000 queries | +| --- | ---: | ---: | ---: | ---: | +| Zero-shot | 0.5497 | 722 / 1,376 ms | $0.108 | $11.786 | +| Few-shot, K=4 | 0.6832 | 1,333 / 2,033 ms | $0.403 | $18.111 | + +Adding four retrieved observations per candidate increased mean quality by +0.1335, or 24.3% relative to the zero-shot result. The complete run made 382 +Jev decisions and incurred $0.09757 in actual routing charges. + +These are exploratory measurements, not broad quality or latency guarantees. + +## Setup + +- Source: [LLMRouterBench](https://github.com/ynulihao/LLMRouterBench) cached + model outcomes. +- Evaluation: 191 validation queries with complete quality and cost records for + all four candidates. +- Candidates: Llama 3.1 8B Instruct, DeepSeek R1 0528, Gemini 2.5 Flash, and + Gemini 2.5 Pro. +- Few-shot context: 256 training queries, producing 1,024 observations; local + TF-IDF retrieves K=4 observations independently for each candidate. +- Quality: mean of the benchmark's per-query score, normalized to [0, 1]. +- Cost: the router column is the amount billed for Jev decisions; selected-model + cost comes from the benchmark's cached executions. No candidate model was + called again for this run. +- Latency: wall-clock routing latency only. It excludes candidate-model + generation latency and was measured from one network location on 2026-09-20. + +The machine-readable aggregate is in +[`results/mini_v1/summary.json`](results/mini_v1/summary.json). Per-query +checkpoints are intentionally gitignored. + +## Reproduce + +Install the benchmark dependency: + +```bash +python -m pip install -e '.[benchmark]' +``` + +Prepare `train.pkl` and `val.pkl` dense tables under `data/llmrouterbench`, or +pass another directory. Each row needs `sample_id`, `eval_name`, `query_text`, +and the performance/cost columns used in the script. + +```bash +export OPENROUTER_API_KEY="your-key" +python benchmarks/mini_llmrouterbench.py \ + --data-dir data/llmrouterbench \ + --output-dir benchmarks/results/mini_v1 +``` + +Completed decisions are appended to `decisions.jsonl`, so rerunning the same +command resumes without paying for successful calls again. diff --git a/benchmarks/mini_llmrouterbench.py b/benchmarks/mini_llmrouterbench.py new file mode 100644 index 0000000..dbef242 --- /dev/null +++ b/benchmarks/mini_llmrouterbench.py @@ -0,0 +1,395 @@ +#!/usr/bin/env python3 +"""Run a small, resumable ReflexRoute benchmark on LLMRouterBench. + +Only ReflexRoute's Jev decisions make network requests. Model quality and model +execution cost are read from LLMRouterBench's precomputed outcomes, which keeps +this benchmark inexpensive and reproducible. +""" + +from __future__ import annotations + +import argparse +from concurrent.futures import ThreadPoolExecutor, as_completed +from dataclasses import dataclass +import json +import math +from pathlib import Path +import random +import statistics +import sys +import threading +import time +from typing import Any, Iterable, Mapping, Sequence + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +if str(REPOSITORY_ROOT) not in sys.path: + sys.path.insert(0, str(REPOSITORY_ROOT)) + +try: + import pandas as pd +except ImportError as exc: # pragma: no cover - exercised by benchmark users + raise SystemExit( + "This benchmark needs pandas. Install it with `python -m pip install pandas`." + ) from exc + +from reflexroute import HistoryRecord, Router + + +SEED = 20260920 +DEFAULT_DATA_DIR = Path("data/llmrouterbench") +MODEL_SPECS = ( + (6, "meta-llama/llama-3.1-8b-instruct"), + (18, "deepseek/deepseek-r1-0528"), + (21, "google/gemini-2.5-flash"), + (22, "google/gemini-2.5-pro"), +) +MODE_TO_METHOD = { + "zero-shot": "reflexroute_zero_shot", + "few-shot": "reflexroute_few_shot_k4", +} + + +@dataclass(frozen=True) +class Example: + sample_id: str + eval_name: str + query: str + performance: dict[str, float] + cost: dict[str, float] + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--data-dir", type=Path, default=DEFAULT_DATA_DIR) + parser.add_argument( + "--output-dir", type=Path, default=Path("benchmarks/results/mini_v1") + ) + parser.add_argument("--context-size", type=int, default=256) + parser.add_argument("--history-top-k", type=int, default=4) + parser.add_argument("--workers", type=int, default=12) + parser.add_argument("--retries", type=int, default=3) + parser.add_argument("--limit", type=int) + parser.add_argument( + "--modes", + nargs="+", + choices=tuple(MODE_TO_METHOD), + default=list(MODE_TO_METHOD), + ) + return parser.parse_args() + + +def finite(value: Any) -> bool: + try: + return math.isfinite(float(value)) + except (TypeError, ValueError): + return False + + +def complete_examples(frame: Any) -> list[Example]: + required = [ + f"model_{index}_{metric}" + for index, _ in MODEL_SPECS + for metric in ("performance", "cost") + ] + mask = frame[required].notna().all(axis=1) + examples: list[Example] = [] + for _, row in frame.loc[mask].iterrows(): + examples.append( + Example( + sample_id=str(row["sample_id"]), + eval_name=str(row["eval_name"]), + query=str(row["query_text"]), + performance={ + model: float(row[f"model_{index}_performance"]) + for index, model in MODEL_SPECS + }, + cost={ + model: float(row[f"model_{index}_cost"]) + for index, model in MODEL_SPECS + }, + ) + ) + return examples + + +def sample_context(examples: Sequence[Example], size: int) -> list[Example]: + if size <= 0: + raise ValueError("context-size must be positive") + if size >= len(examples): + return list(examples) + + # Preserve the source benchmark's task mixture while keeping selection stable. + rng = random.Random(SEED) + groups: dict[str, list[Example]] = {} + for example in examples: + groups.setdefault(example.eval_name, []).append(example) + for group in groups.values(): + rng.shuffle(group) + + exact = {name: size * len(group) / len(examples) for name, group in groups.items()} + quota = {name: min(len(groups[name]), math.floor(value)) for name, value in exact.items()} + remaining = size - sum(quota.values()) + order = sorted(groups, key=lambda name: (-(exact[name] - quota[name]), name)) + for name in order: + if remaining == 0: + break + if quota[name] < len(groups[name]): + quota[name] += 1 + remaining -= 1 + return [item for name in sorted(groups) for item in groups[name][: quota[name]]] + + +def make_history(context: Iterable[Example]) -> list[HistoryRecord]: + return [ + HistoryRecord( + query=example.query, + model=model, + performance=example.performance[model], + cost=example.cost[model], + ) + for example in context + for _, model in MODEL_SPECS + ] + + +def percentile(values: Sequence[float], proportion: float) -> float: + if not values: + return 0.0 + ordered = sorted(values) + position = (len(ordered) - 1) * proportion + lower = math.floor(position) + upper = math.ceil(position) + if lower == upper: + return ordered[lower] + weight = position - lower + return ordered[lower] * (1 - weight) + ordered[upper] * weight + + +def usage_cost(usage: Mapping[str, Any]) -> float | None: + """Read a billed dollar amount without guessing from token counts.""" + + for key in ("cost", "total_cost"): + value = usage.get(key) + if finite(value): + return float(value) + return None + + +def append_jsonl(path: Path, value: Mapping[str, Any], lock: threading.Lock) -> None: + encoded = json.dumps(value, ensure_ascii=False, sort_keys=True) + with lock: + with path.open("a", encoding="utf-8") as handle: + handle.write(encoded + "\n") + handle.flush() + + +def load_jsonl(path: Path) -> list[dict[str, Any]]: + if not path.exists(): + return [] + rows: list[dict[str, Any]] = [] + with path.open(encoding="utf-8") as handle: + for line_number, line in enumerate(handle, start=1): + if not line.strip(): + continue + try: + value = json.loads(line) + except json.JSONDecodeError as exc: + raise RuntimeError(f"Invalid checkpoint line {line_number}: {exc}") from exc + if isinstance(value, dict): + rows.append(value) + return rows + + +def route_once( + router: Router, + mode: str, + example: Example, + retries: int, +) -> dict[str, Any]: + last_error: Exception | None = None + for attempt in range(1, retries + 1): + started = time.perf_counter() + try: + result = router.route(example.query) + elapsed = time.perf_counter() - started + return { + "mode": mode, + "method": MODE_TO_METHOD[mode], + "sample_id": example.sample_id, + "eval_name": example.eval_name, + "selected_model": result.model, + "probabilities": result.probabilities, + "confidence": result.confidence, + "usage": result.usage, + "router_cost": usage_cost(result.usage), + "router_latency_seconds": elapsed, + "realized_quality": example.performance[result.model], + "selected_model_cost": example.cost[result.model], + "attempts": attempt, + } + except Exception as exc: # API errors are retried and checkpointed by caller + last_error = exc + if attempt < retries: + time.sleep(min(2 ** (attempt - 1), 4)) + assert last_error is not None + raise last_error + + +def summarize(rows: Sequence[Mapping[str, Any]]) -> dict[str, Any]: + methods = sorted({str(row["method"]) for row in rows}) + result: dict[str, Any] = {} + for method in methods: + subset = [row for row in rows if row["method"] == method] + latencies = [float(row["router_latency_seconds"]) for row in subset] + known_router_costs = [ + float(row["router_cost"]) + for row in subset + if finite(row.get("router_cost")) + ] + router_cost_total = sum(known_router_costs) + selected_cost_total = sum(float(row["selected_model_cost"]) for row in subset) + count = len(subset) + selections: dict[str, int] = {} + for row in subset: + model = str(row["selected_model"]) + selections[model] = selections.get(model, 0) + 1 + result[method] = { + "n": count, + "quality_mean": statistics.fmean(float(row["realized_quality"]) for row in subset), + "router_latency_p50_ms": percentile(latencies, 0.50) * 1000, + "router_latency_p95_ms": percentile(latencies, 0.95) * 1000, + "router_cost_total_usd": router_cost_total, + "router_cost_per_1000_usd": router_cost_total / count * 1000, + "router_cost_coverage": len(known_router_costs) / count, + "selected_model_cost_mean_usd": selected_cost_total / count, + "selected_model_cost_per_1000_usd": selected_cost_total / count * 1000, + "total_cost_per_1000_usd": (router_cost_total + selected_cost_total) / count * 1000, + "selections": dict(sorted(selections.items())), + } + return result + + +def print_table(summary: Mapping[str, Mapping[str, Any]]) -> None: + print("\nmethod n quality p50 ms p95 ms route $/1k model $/1k total $/1k") + print("-" * 104) + for method, metrics in summary.items(): + print( + f"{method:28} {metrics['n']:3d} {metrics['quality_mean']:.4f}" + f" {metrics['router_latency_p50_ms']:7.1f}" + f" {metrics['router_latency_p95_ms']:7.1f}" + f" {metrics['router_cost_per_1000_usd']:11.4f}" + f" {metrics['selected_model_cost_per_1000_usd']:11.4f}" + f" {metrics['total_cost_per_1000_usd']:11.4f}" + ) + + +def main() -> None: + args = parse_args() + if args.workers <= 0 or args.retries <= 0 or args.history_top_k <= 0: + raise SystemExit("workers, retries, and history-top-k must be positive") + + train_path = args.data_dir / "train.pkl" + val_path = args.data_dir / "val.pkl" + if not train_path.exists() or not val_path.exists(): + raise SystemExit(f"LLMRouterBench train.pkl/val.pkl not found under {args.data_dir}") + + args.output_dir.mkdir(parents=True, exist_ok=True) + checkpoint_path = args.output_dir / "decisions.jsonl" + summary_path = args.output_dir / "summary.json" + + context_pool = complete_examples(pd.read_pickle(train_path)) + targets = complete_examples(pd.read_pickle(val_path)) + context = sample_context(context_pool, args.context_size) + if args.limit is not None: + targets = targets[: max(0, args.limit)] + if not targets: + raise SystemExit("No complete validation examples selected") + + history = make_history(context) + checkpoint_rows = load_jsonl(checkpoint_path) + completed = { + (str(row.get("mode")), str(row.get("sample_id"))) + for row in checkpoint_rows + if row.get("method") in MODE_TO_METHOD.values() + } + target_ids = {example.sample_id for example in targets} + router_rows = [ + row + for row in checkpoint_rows + if str(row.get("sample_id")) in target_ids and row.get("mode") in args.modes + ] + + routers: dict[str, Router] = {} + model_names = [model for _, model in MODEL_SPECS] + for mode in args.modes: + router = Router(models=model_names, history_top_k=args.history_top_k, timeout=30.0) + if mode == "few-shot": + router.set_history(history) + routers[mode] = router + + jobs = [ + (mode, example) + for mode in args.modes + for example in targets + if (mode, example.sample_id) not in completed + ] + print( + f"context={len(context)} rows ({len(history)} observations), " + f"targets={len(targets)}, cached={len(router_rows)}, pending={len(jobs)}" + ) + lock = threading.Lock() + errors: list[tuple[str, str, str]] = [] + if jobs: + with ThreadPoolExecutor(max_workers=args.workers) as executor: + future_to_job = { + executor.submit(route_once, routers[mode], mode, example, args.retries): ( + mode, + example, + ) + for mode, example in jobs + } + finished = 0 + for future in as_completed(future_to_job): + mode, example = future_to_job[future] + try: + row = future.result() + except Exception as exc: + errors.append((mode, example.sample_id, str(exc))) + else: + append_jsonl(checkpoint_path, row, lock) + router_rows.append(row) + finished += 1 + if finished % 10 == 0 or finished == len(jobs): + print(f"completed {finished}/{len(jobs)} new decisions; errors={len(errors)}", flush=True) + + summary = summarize(router_rows) + payload = { + "benchmark": "LLMRouterBench mini", + "run_date": "2026-09-20", + "seed": SEED, + "context_size": len(context), + "history_records": len(history), + "history_top_k": args.history_top_k, + "target_count": len(targets), + "models": [ + {"dataset_index": index, "openrouter_id": model} + for index, model in MODEL_SPECS + ], + "methods": summary, + "errors": [ + {"mode": mode, "sample_id": sample_id, "error": error} + for mode, sample_id, error in errors + ], + } + summary_path.write_text( + json.dumps(payload, indent=2, ensure_ascii=False, sort_keys=True) + "\n", + encoding="utf-8", + ) + print_table(summary) + print(f"\nWrote {summary_path} and {checkpoint_path}") + if errors: + raise SystemExit(f"{len(errors)} routing decisions failed; rerun to resume them") + + +if __name__ == "__main__": + main() diff --git a/benchmarks/results/mini_v1/summary.json b/benchmarks/results/mini_v1/summary.json new file mode 100644 index 0000000..1dcb2b9 --- /dev/null +++ b/benchmarks/results/mini_v1/summary.json @@ -0,0 +1,66 @@ +{ + "benchmark": "LLMRouterBench mini", + "context_size": 256, + "errors": [], + "history_records": 1024, + "history_top_k": 4, + "methods": { + "reflexroute_few_shot_k4": { + "n": 191, + "quality_mean": 0.6832460732984293, + "router_cost_coverage": 1.0, + "router_cost_per_1000_usd": 0.4032351832460733, + "router_cost_total_usd": 0.07701792, + "router_latency_p50_ms": 1333.2483787089586, + "router_latency_p95_ms": 2033.195577096194, + "selected_model_cost_mean_usd": 0.018111016335078535, + "selected_model_cost_per_1000_usd": 18.111016335078535, + "selections": { + "deepseek/deepseek-r1-0528": 61, + "google/gemini-2.5-flash": 98, + "google/gemini-2.5-pro": 27, + "meta-llama/llama-3.1-8b-instruct": 5 + }, + "total_cost_per_1000_usd": 18.514251518324606 + }, + "reflexroute_zero_shot": { + "n": 191, + "quality_mean": 0.5497382198952879, + "router_cost_coverage": 1.0, + "router_cost_per_1000_usd": 0.10759806282722509, + "router_cost_total_usd": 0.020551229999999993, + "router_latency_p50_ms": 721.7784440144897, + "router_latency_p95_ms": 1376.4288143720478, + "selected_model_cost_mean_usd": 0.011785609816753922, + "selected_model_cost_per_1000_usd": 11.785609816753922, + "selections": { + "deepseek/deepseek-r1-0528": 80, + "google/gemini-2.5-flash": 58, + "google/gemini-2.5-pro": 2, + "meta-llama/llama-3.1-8b-instruct": 51 + }, + "total_cost_per_1000_usd": 11.893207879581148 + } + }, + "models": [ + { + "dataset_index": 6, + "openrouter_id": "meta-llama/llama-3.1-8b-instruct" + }, + { + "dataset_index": 18, + "openrouter_id": "deepseek/deepseek-r1-0528" + }, + { + "dataset_index": 21, + "openrouter_id": "google/gemini-2.5-flash" + }, + { + "dataset_index": 22, + "openrouter_id": "google/gemini-2.5-pro" + } + ], + "run_date": "2026-09-20", + "seed": 20260920, + "target_count": 191 +} diff --git a/examples/adaptation_demo.py b/examples/adaptation_demo.py new file mode 100644 index 0000000..baa2f81 --- /dev/null +++ b/examples/adaptation_demo.py @@ -0,0 +1,117 @@ +"""Show the same routing decision before and after adding a few observations. + +Run with: + OPENROUTER_API_KEY=... python examples/adaptation_demo.py + +This demo makes two Jev Decisions API calls. It only selects a model; it does +not invoke the selected model. +""" + +from __future__ import annotations + +from pathlib import Path +import sys +import time + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +if str(REPOSITORY_ROOT) not in sys.path: + sys.path.insert(0, str(REPOSITORY_ROOT)) + +from reflexroute import HistoryRecord, Router, RoutingResult + + +MODELS = [ + "meta-llama/llama-3.1-8b-instruct", + "deepseek/deepseek-r1-0528", + "google/gemini-2.5-flash", + "google/gemini-2.5-pro", +] +QUERY = "Implement a Python async rate limiter and explain its failure modes." + +# Illustrative observations from an application's own evaluation loop. +HISTORY = [ + HistoryRecord( + query="Debug a concurrent Python worker pool with intermittent deadlocks.", + model="meta-llama/llama-3.1-8b-instruct", + performance=0.42, + cost=0.0001, + ), + HistoryRecord( + query="Debug a concurrent Python worker pool with intermittent deadlocks.", + model="deepseek/deepseek-r1-0528", + performance=0.61, + cost=0.017, + ), + HistoryRecord( + query="Debug a concurrent Python worker pool with intermittent deadlocks.", + model="google/gemini-2.5-flash", + performance=0.97, + cost=0.006, + ), + HistoryRecord( + query="Debug a concurrent Python worker pool with intermittent deadlocks.", + model="google/gemini-2.5-pro", + performance=0.73, + cost=0.074, + ), + HistoryRecord( + query="Summarize a customer support email in three bullets.", + model="meta-llama/llama-3.1-8b-instruct", + performance=0.76, + cost=0.0001, + ), + HistoryRecord( + query="Summarize a customer support email in three bullets.", + model="deepseek/deepseek-r1-0528", + performance=0.85, + cost=0.012, + ), + HistoryRecord( + query="Summarize a customer support email in three bullets.", + model="google/gemini-2.5-flash", + performance=0.95, + cost=0.004, + ), + HistoryRecord( + query="Summarize a customer support email in three bullets.", + model="google/gemini-2.5-pro", + performance=0.94, + cost=0.052, + ), +] + + +def route_and_time(router: Router) -> tuple[RoutingResult, float]: + started = time.perf_counter() + result = router.route(QUERY) + return result, time.perf_counter() - started + + +def show(label: str, result: RoutingResult, elapsed: float) -> None: + billed = result.usage.get("cost") + billed_text = f"${float(billed):.6f}" if isinstance(billed, (int, float)) else "n/a" + print(f"\n{label}") + print(f" selected {result.model}") + print(f" confidence {result.confidence:.2f}") + print(f" latency {elapsed * 1000:.0f} ms") + print(f" route cost {billed_text}") + print(" probabilities") + for model, probability in sorted( + result.probabilities.items(), key=lambda item: item[1], reverse=True + ): + print(f" {probability:5.1%} {model}") + + +def main() -> None: + router = Router(models=MODELS, history_top_k=2) + + zero_shot, zero_shot_time = route_and_time(router) + show("Zero-shot · built-in priors", zero_shot, zero_shot_time) + + router.set_history(HISTORY) + few_shot, few_shot_time = route_and_time(router) + show("Few-shot · priors + 8 user observations", few_shot, few_shot_time) + + +if __name__ == "__main__": + main() diff --git a/pyproject.toml b/pyproject.toml index b751372..ffca8f2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -30,6 +30,7 @@ Changelog = "https://github.com/AIGNLAI/ReflexRoute/blob/main/CHANGELOG.md" [project.optional-dependencies] dev = ["pytest>=7.0"] +benchmark = ["pandas>=2.0"] [project.scripts] reflexroute = "reflexroute.cli:main"