diff --git a/README.md b/README.md index 7bc478f..d6c536d 100644 --- a/README.md +++ b/README.md @@ -163,6 +163,32 @@ dependent entry. Responses carry `X-Cache: miss | exact | semantic`. | Semantic | Disabled by default; requires `public` privacy, deterministic generation, and an extraction, classification, or summarization task. | | Router decision | Reuses stable task classification; cleared when the policy version changes. | +## Evaluation + +Every optimization is graded against a committed dataset in +[`benchmarks/datasets`](benchmarks/datasets), never against sampled production traffic. +`execute()` runs each case through a caller-supplied transport and times it; `build_report()` +turns the outcomes into one `EvaluationReport` that always pairs quality with latency and +GPU cost, never one alone. + +```bash +python -c " +from llm_router.evaluation import build_report, execute, load_dataset +cases = load_dataset('benchmarks/datasets/extraction-v1.jsonl') +outcomes = execute(cases, lambda case: (case.expected, True, 0.05)) +print(build_report(outcomes, model_id='small-specialist', model_revision='rev-1').summary()) +" +``` + +- Constrained tasks (extraction, classification) are scored by exact match, generative tasks + by token overlap, and structured tasks score zero when the output is not valid JSON. +- `compare()` rejects a variant that buys latency or throughput with a quality or + structured-validity regression, however small; `render_comparison()` prints the verdict + with every regression reason. +- [`benchmarks/workloads`](benchmarks/workloads) holds reproducible k6 steady and burst load + definitions. The burst scenario asserts bounded queue behavior — an explicit `429`/`503` + rejection — rather than unbounded tail latency. + ## Runtime settings All settings use the `ROUTER_` prefix. diff --git a/benchmarks/datasets/extraction-v1.jsonl b/benchmarks/datasets/extraction-v1.jsonl new file mode 100644 index 0000000..265d6de --- /dev/null +++ b/benchmarks/datasets/extraction-v1.jsonl @@ -0,0 +1,5 @@ +{"id": "extract-001", "task": "extraction", "prompt": "Extract the invoice number and total from: Invoice INV-4417, total 182.50 USD. Reply as JSON.", "expected": "{\"invoice_number\": \"INV-4417\", \"total\": \"182.50 USD\"}", "structured": true} +{"id": "extract-002", "task": "extraction", "prompt": "Extract the claim id from: Claim CLM-9921 was filed on 2026-04-02. Reply as JSON.", "expected": "{\"claim_id\": \"CLM-9921\"}", "structured": true} +{"id": "classify-001", "task": "classification", "prompt": "Classify this ticket as billing, technical, or other: my card was charged twice.", "expected": "billing"} +{"id": "classify-002", "task": "classification", "prompt": "Classify this ticket as billing, technical, or other: the dashboard returns a 500 error.", "expected": "technical"} +{"id": "summarize-001", "task": "summarization", "prompt": "Summarize: revenue grew in every region while support costs fell for the third quarter running.", "expected": "revenue grew in every region and support costs fell"} diff --git a/benchmarks/workloads/burst.js b/benchmarks/workloads/burst.js new file mode 100644 index 0000000..7d0067c --- /dev/null +++ b/benchmarks/workloads/burst.js @@ -0,0 +1,49 @@ +// Burst load definition (section 16): verifies bounded queue behaviour and +// predictable rejection rather than unbounded tail latency. +// k6 run -e BASE_URL=http://127.0.0.1:8000 -e API_KEY=dev-key benchmarks/workloads/burst.js +import http from "k6/http"; +import { check } from "k6"; + +export const options = { + scenarios: { + burst: { + executor: "ramping-arrival-rate", + startRate: 10, + timeUnit: "1s", + preAllocatedVUs: 100, + maxVUs: 400, + stages: [ + { target: 10, duration: "1m" }, + { target: 200, duration: "30s" }, + { target: 200, duration: "1m" }, + { target: 10, duration: "1m" }, + ], + }, + }, + thresholds: { + // Under saturation the platform must reject predictably, not queue without bound. + "http_req_duration": ["p(99)<10000"], + "checks": ["rate>0.99"], + }, +}; + +const BASE_URL = __ENV.BASE_URL || "http://127.0.0.1:8000"; +const API_KEY = __ENV.API_KEY || "dev-key"; + +export default function () { + const response = http.post( + `${BASE_URL}/v1/chat/completions`, + JSON.stringify({ + model: "auto", + messages: [{ role: "user", content: "Classify this burst probe ticket." }], + max_tokens: 64, + routing: { task: "classification", privacy: "private" }, + }), + { headers: { "Content-Type": "application/json", Authorization: `Bearer ${API_KEY}` } }, + ); + + check(response, { + "no unhandled failure": (r) => [200, 429, 503].includes(r.status), + "rejection is explicit": (r) => r.status === 200 || r.json("error.type") !== undefined, + }); +} diff --git a/benchmarks/workloads/steady.js b/benchmarks/workloads/steady.js new file mode 100644 index 0000000..54b4545 --- /dev/null +++ b/benchmarks/workloads/steady.js @@ -0,0 +1,55 @@ +// Sustained load definition (section 16). Reproducible from this committed file: +// k6 run -e BASE_URL=http://127.0.0.1:8000 -e API_KEY=dev-key benchmarks/workloads/steady.js +import http from "k6/http"; +import { check } from "k6"; + +export const options = { + scenarios: { + steady: { + executor: "constant-arrival-rate", + rate: 20, + timeUnit: "1s", + duration: "5m", + preAllocatedVUs: 40, + maxVUs: 120, + }, + }, + thresholds: { + // Report quality and latency together; a passing run is not a quality claim. + "http_req_duration{expected_response:true}": ["p(95)<2000", "p(99)<5000"], + "http_req_failed": ["rate<0.01"], + }, +}; + +const BASE_URL = __ENV.BASE_URL || "http://127.0.0.1:8000"; +const API_KEY = __ENV.API_KEY || "dev-key"; + +// Prompt-length distribution: short classification, medium extraction, long RAG. +const PROMPTS = [ + { task: "classification", text: "Classify this ticket: my card was charged twice." }, + { task: "extraction", text: "Extract the invoice number from: Invoice INV-4417, total 182.50 USD." }, + { + task: "rag", + text: "According to the documents provided, summarize the retention policy. ".repeat(24), + }, +]; + +export default function () { + const prompt = PROMPTS[Math.floor(Math.random() * PROMPTS.length)]; + const response = http.post( + `${BASE_URL}/v1/chat/completions`, + JSON.stringify({ + model: "auto", + messages: [{ role: "user", content: prompt.text }], + max_tokens: 128, + routing: { task: prompt.task, privacy: "private" }, + }), + { headers: { "Content-Type": "application/json", Authorization: `Bearer ${API_KEY}` } }, + ); + + check(response, { + "served or rejected predictably": (r) => r.status === 200 || r.status === 503 || r.status === 429, + "overload carries retry guidance": (r) => r.status !== 503 || r.headers["Retry-After"] !== undefined, + "route is attributed": (r) => r.status !== 200 || r.headers["X-Route-Model"] !== undefined, + }); +} diff --git a/src/llm_router/evaluation.py b/src/llm_router/evaluation.py new file mode 100644 index 0000000..f608ea7 --- /dev/null +++ b/src/llm_router/evaluation.py @@ -0,0 +1,267 @@ +"""Quality, latency, and cost measurement from section 16. + +Quality and latency are always reported together, and every comparison states +its regressions explicitly so an optimization cannot be accepted on latency +alone. Nothing here samples production traffic; runs are driven from committed +dataset and workload definitions. +""" + +import json +import math +import time +from collections.abc import Callable, Iterable, Sequence +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from llm_router.models import TaskClass + + +class EvaluationError(RuntimeError): + """Raised when a dataset or comparison is unusable.""" + + +@dataclass(frozen=True) +class EvaluationCase: + """One graded example from a committed benchmark dataset.""" + + id: str + task: TaskClass + prompt: str + expected: str + structured: bool = False + + +@dataclass(frozen=True) +class CaseOutcome: + case: EvaluationCase + output: str + latency_ms: float + gpu_seconds: float = 0.0 + succeeded: bool = True + + +@dataclass(frozen=True) +class EvaluationReport: + """Quality and latency reported together, never separately.""" + + model_id: str + model_revision: str + adapter_id: str | None + quantization: str + cases: int + successes: int + quality_score: float + structured_validity: float + latency_p50_ms: float + latency_p95_ms: float + latency_p99_ms: float + throughput_rps: float + gpu_seconds_per_successful_request: float + metadata: dict[str, Any] = field(default_factory=dict) + + def summary(self) -> str: + adapter = self.adapter_id or "none" + return ( + f"{self.model_id}@{self.model_revision} adapter={adapter} " + f"quantization={self.quantization} quality={self.quality_score:.3f} " + f"structured={self.structured_validity:.3f} p95={self.latency_p95_ms:.1f}ms " + f"gpu_s/req={self.gpu_seconds_per_successful_request:.4f}" + ) + + +def percentile(values: Sequence[float], fraction: float) -> float: + """Nearest-rank percentile; deterministic and stable for small samples.""" + + if not values: + raise EvaluationError("cannot take a percentile of an empty sample") + ordered = sorted(values) + rank = max(1, math.ceil(fraction * len(ordered))) + return ordered[rank - 1] + + +def structured_output_valid(output: str) -> bool: + """Structured tasks must return a JSON object, not prose that looks like one.""" + + try: + return isinstance(json.loads(output), dict) + except ValueError: + return False + + +def score_case(outcome: CaseOutcome) -> float: + """Exact match for constrained tasks, token overlap for generative ones.""" + + if not outcome.succeeded: + return 0.0 + case = outcome.case + if case.structured and not structured_output_valid(outcome.output): + return 0.0 + if case.task in {TaskClass.CLASSIFICATION, TaskClass.EXTRACTION}: + return 1.0 if outcome.output.strip() == case.expected.strip() else 0.0 + expected = set(case.expected.lower().split()) + produced = set(outcome.output.lower().split()) + if not expected: + return 0.0 + return len(expected & produced) / len(expected | produced) + + +def load_dataset(path: str | Path) -> tuple[EvaluationCase, ...]: + """Read a JSON Lines dataset of graded cases.""" + + cases: list[EvaluationCase] = [] + for number, line in enumerate(Path(path).read_text(encoding="utf-8").splitlines(), start=1): + if not line.strip(): + continue + try: + document = json.loads(line) + cases.append( + EvaluationCase( + id=str(document["id"]), + task=TaskClass(document["task"]), + prompt=str(document["prompt"]), + expected=str(document["expected"]), + structured=bool(document.get("structured", False)), + ) + ) + except (ValueError, KeyError) as error: + raise EvaluationError(f"{path}:{number} is not a usable case: {error}") from error + if not cases: + raise EvaluationError(f"{path} contains no cases") + return tuple(cases) + + +def build_report( + outcomes: Iterable[CaseOutcome], + *, + model_id: str, + model_revision: str, + adapter_id: str | None = None, + quantization: str = "none", + wall_clock_seconds: float | None = None, + metadata: dict[str, Any] | None = None, +) -> EvaluationReport: + collected = list(outcomes) + if not collected: + raise EvaluationError("cannot report on an empty run") + + latencies = [outcome.latency_ms for outcome in collected] + successes = [outcome for outcome in collected if outcome.succeeded] + structured = [outcome for outcome in collected if outcome.case.structured] + elapsed = wall_clock_seconds if wall_clock_seconds else sum(latencies) / 1000 or 1e-9 + + return EvaluationReport( + model_id=model_id, + model_revision=model_revision, + adapter_id=adapter_id, + quantization=quantization, + cases=len(collected), + successes=len(successes), + quality_score=sum(score_case(outcome) for outcome in collected) / len(collected), + structured_validity=( + sum(structured_output_valid(outcome.output) for outcome in structured) / len(structured) + if structured + else 1.0 + ), + latency_p50_ms=percentile(latencies, 0.50), + latency_p95_ms=percentile(latencies, 0.95), + latency_p99_ms=percentile(latencies, 0.99), + throughput_rps=len(collected) / elapsed, + gpu_seconds_per_successful_request=( + sum(outcome.gpu_seconds for outcome in collected) / len(successes) if successes else 0.0 + ), + metadata=metadata or {}, + ) + + +def execute( + cases: Iterable[EvaluationCase], + invoke: Callable[[EvaluationCase], tuple[str, bool, float]], +) -> tuple[CaseOutcome, ...]: + """Run every case through a caller-supplied transport and time each one. + + The transport returns the produced text, whether the request succeeded, and + the GPU seconds it consumed; latency is measured here so every harness run + reports it the same way. + """ + + outcomes: list[CaseOutcome] = [] + for case in cases: + started = time.perf_counter() + output, succeeded, gpu_seconds = invoke(case) + outcomes.append( + CaseOutcome( + case=case, + output=output, + latency_ms=(time.perf_counter() - started) * 1000, + gpu_seconds=gpu_seconds, + succeeded=succeeded, + ) + ) + return tuple(outcomes) + + +@dataclass(frozen=True) +class Comparison: + """Variant versus baseline, with the quality cost of any speedup stated.""" + + baseline: EvaluationReport + variant: EvaluationReport + quality_delta: float + latency_p95_delta_ms: float + cost_delta_gpu_seconds: float + regressions: tuple[str, ...] + + @property + def accepted(self) -> bool: + return not self.regressions + + +def compare( + baseline: EvaluationReport, + variant: EvaluationReport, + *, + quality_tolerance: float = 0.01, + structured_tolerance: float = 0.0, +) -> Comparison: + """Reject a variant that buys latency with quality, however small the loss.""" + + quality_delta = variant.quality_score - baseline.quality_score + regressions: list[str] = [] + if quality_delta < -quality_tolerance: + regressions.append( + f"quality fell by {abs(quality_delta):.3f}, beyond the " + f"{quality_tolerance:.3f} tolerance" + ) + structured_delta = variant.structured_validity - baseline.structured_validity + if structured_delta < -structured_tolerance: + regressions.append(f"structured-output validity fell by {abs(structured_delta):.3f}") + if variant.successes < baseline.successes: + regressions.append( + f"successful requests fell from {baseline.successes} to {variant.successes}" + ) + + return Comparison( + baseline=baseline, + variant=variant, + quality_delta=quality_delta, + latency_p95_delta_ms=variant.latency_p95_ms - baseline.latency_p95_ms, + cost_delta_gpu_seconds=( + variant.gpu_seconds_per_successful_request - baseline.gpu_seconds_per_successful_request + ), + regressions=tuple(regressions), + ) + + +def render_comparison(comparison: Comparison) -> str: + verdict = "accepted" if comparison.accepted else "rejected" + lines = [ + f"baseline: {comparison.baseline.summary()}", + f"variant: {comparison.variant.summary()}", + f"quality delta: {comparison.quality_delta:+.3f}", + f"p95 latency delta: {comparison.latency_p95_delta_ms:+.1f} ms", + f"gpu seconds per successful request delta: {comparison.cost_delta_gpu_seconds:+.4f}", + f"verdict: {verdict}", + ] + lines.extend(f"regression: {reason}" for reason in comparison.regressions) + return "\n".join(lines) diff --git a/tests/unit/test_evaluation.py b/tests/unit/test_evaluation.py new file mode 100644 index 0000000..fa6fb4c --- /dev/null +++ b/tests/unit/test_evaluation.py @@ -0,0 +1,251 @@ +from pathlib import Path + +import pytest + +from llm_router.evaluation import ( + CaseOutcome, + EvaluationCase, + EvaluationError, + build_report, + compare, + execute, + load_dataset, + percentile, + render_comparison, + score_case, + structured_output_valid, +) +from llm_router.models import TaskClass + +DATASET = Path("benchmarks/datasets/extraction-v1.jsonl") + + +def case(**overrides: object) -> EvaluationCase: + values: dict[str, object] = { + "id": "case-1", + "task": TaskClass.CLASSIFICATION, + "prompt": "classify this", + "expected": "billing", + "structured": False, + } + values.update(overrides) + return EvaluationCase(**values) # type: ignore[arg-type] + + +def outcome(**overrides: object) -> CaseOutcome: + values: dict[str, object] = { + "case": case(), + "output": "billing", + "latency_ms": 120.0, + "gpu_seconds": 0.1, + "succeeded": True, + } + values.update(overrides) + return CaseOutcome(**values) # type: ignore[arg-type] + + +def report(**overrides: object) -> object: + outcomes = [outcome(latency_ms=float(index)) for index in range(1, 11)] + defaults: dict[str, object] = { + "model_id": "small-specialist", + "model_revision": "rev-1", + "quantization": "none", + } + defaults.update(overrides) + return build_report(outcomes, **defaults) # type: ignore[arg-type] + + +def test_committed_dataset_loads_with_task_and_structure_flags() -> None: + cases = load_dataset(DATASET) + + assert len(cases) == 5 + assert {item.task for item in cases} == { + TaskClass.EXTRACTION, + TaskClass.CLASSIFICATION, + TaskClass.SUMMARIZATION, + } + assert sum(item.structured for item in cases) == 2 + + +def test_dataset_loader_reports_unusable_lines(tmp_path: Path) -> None: + path = tmp_path / "cases.jsonl" + path.write_text('{"id": "a"}\n', encoding="utf-8") + + with pytest.raises(EvaluationError, match="not a usable case"): + load_dataset(path) + + +def test_dataset_loader_rejects_an_empty_file(tmp_path: Path) -> None: + path = tmp_path / "empty.jsonl" + path.write_text("\n\n", encoding="utf-8") + + with pytest.raises(EvaluationError, match="no cases"): + load_dataset(path) + + +def test_percentiles_use_nearest_rank_and_reject_empty_samples() -> None: + values = [10.0, 20.0, 30.0, 40.0, 50.0] + + assert percentile(values, 0.5) == 30.0 + assert percentile(values, 0.95) == 50.0 + with pytest.raises(EvaluationError, match="empty sample"): + percentile([], 0.5) + + +def test_structured_validity_requires_a_json_object() -> None: + assert structured_output_valid('{"a": 1}') is True + assert structured_output_valid("[1, 2]") is False + assert structured_output_valid("looks like {json}") is False + + +def test_constrained_tasks_are_scored_by_exact_match() -> None: + assert score_case(outcome()) == 1.0 + assert score_case(outcome(output="technical")) == 0.0 + assert score_case(outcome(succeeded=False)) == 0.0 + + +def test_structured_case_scores_zero_when_the_output_is_not_json() -> None: + structured = case(task=TaskClass.EXTRACTION, expected='{"id": "1"}', structured=True) + + assert score_case(CaseOutcome(case=structured, output='{"id": "1"}', latency_ms=1.0)) == 1.0 + assert score_case(CaseOutcome(case=structured, output="id is 1", latency_ms=1.0)) == 0.0 + + +def test_generative_tasks_are_scored_by_token_overlap() -> None: + generative = case(task=TaskClass.SUMMARIZATION, expected="revenue grew everywhere") + + partial = score_case(CaseOutcome(case=generative, output="revenue grew", latency_ms=1.0)) + empty = score_case( + CaseOutcome( + case=case(task=TaskClass.SUMMARIZATION, expected=""), output="x", latency_ms=1.0 + ) + ) + + assert 0.0 < partial < 1.0 + assert empty == 0.0 + + +def test_report_pairs_quality_with_latency_and_cost() -> None: + built = report() + + assert built.cases == 10 + assert built.successes == 10 + assert built.quality_score == 1.0 + assert built.latency_p50_ms == 5.0 + assert built.latency_p95_ms == 10.0 + assert built.gpu_seconds_per_successful_request == pytest.approx(0.1) + assert "quality=1.000" in built.summary() + assert "p95=10.0ms" in built.summary() + + +def test_report_rejects_an_empty_run() -> None: + with pytest.raises(EvaluationError, match="empty run"): + build_report([], model_id="m", model_revision="r") + + +def test_report_uses_wall_clock_for_throughput_when_supplied() -> None: + built = build_report([outcome()], model_id="m", model_revision="r", wall_clock_seconds=2.0) + + assert built.throughput_rps == pytest.approx(0.5) + + +def test_execute_times_each_case_and_reports_the_transport_result() -> None: + cases = load_dataset(DATASET) + + def invoke(item: EvaluationCase) -> tuple[str, bool, float]: + return item.expected, True, 0.2 + + outcomes = execute(cases, invoke) + + assert len(outcomes) == len(cases) + assert [outcome.case.id for outcome in outcomes] == [item.id for item in cases] + assert all( + outcome.output == item.expected for outcome, item in zip(outcomes, cases, strict=True) + ) + assert all(outcome.succeeded for outcome in outcomes) + assert all(outcome.gpu_seconds == 0.2 for outcome in outcomes) + assert all(outcome.latency_ms >= 0.0 for outcome in outcomes) + + +def test_execute_carries_a_failed_transport_call_into_the_outcome() -> None: + failing = case(id="broken") + + outcomes = execute([failing], lambda item: ("", False, 0.0)) + + assert outcomes[0].succeeded is False + assert outcomes[0].output == "" + + +def test_execute_feeds_build_report_directly() -> None: + cases = load_dataset(DATASET) + + outcomes = execute(cases, lambda item: (item.expected, True, 0.1)) + built = build_report(outcomes, model_id="small-specialist", model_revision="rev-1") + + assert built.cases == len(cases) + assert built.quality_score == 1.0 + + +def test_quantized_variant_is_rejected_when_quality_falls() -> None: + baseline = report() + degraded_outcomes = [outcome(output="technical") for _ in range(10)] + variant = build_report( + degraded_outcomes, model_id="small-specialist", model_revision="rev-1", quantization="awq" + ) + + comparison = compare(baseline, variant) # type: ignore[arg-type] + + assert comparison.accepted is False + assert any("quality fell" in reason for reason in comparison.regressions) + assert "verdict: rejected" in render_comparison(comparison) + + +def test_variant_within_tolerance_is_accepted_and_reports_its_deltas() -> None: + baseline = report() + faster = build_report( + [outcome(latency_ms=1.0, gpu_seconds=0.05) for _ in range(10)], + model_id="small-specialist", + model_revision="rev-1", + quantization="awq", + ) + + comparison = compare(baseline, faster) # type: ignore[arg-type] + + assert comparison.accepted is True + assert comparison.latency_p95_delta_ms < 0 + assert comparison.cost_delta_gpu_seconds < 0 + assert "verdict: accepted" in render_comparison(comparison) + + +def test_structured_validity_regression_is_reported_separately() -> None: + structured = case(task=TaskClass.EXTRACTION, expected='{"id": "1"}', structured=True) + baseline = build_report( + [CaseOutcome(case=structured, output='{"id": "1"}', latency_ms=5.0)], + model_id="m", + model_revision="r", + ) + variant = build_report( + [CaseOutcome(case=structured, output="id is 1", latency_ms=1.0)], + model_id="m", + model_revision="r", + quantization="gptq", + ) + + comparison = compare(variant, baseline) + reverse = compare(baseline, variant) + + assert comparison.accepted is True + assert any("structured-output validity" in reason for reason in reverse.regressions) + + +def test_fewer_successful_requests_is_always_a_regression() -> None: + baseline = report() + variant = build_report( + [outcome(succeeded=index > 0) for index in range(10)], + model_id="small-specialist", + model_revision="rev-1", + ) + + comparison = compare(baseline, variant) # type: ignore[arg-type] + + assert any("successful requests fell" in reason for reason in comparison.regressions)