diff --git a/engine/.gitignore b/engine/.gitignore index 0602cd44db..d7e7ae99f0 100644 --- a/engine/.gitignore +++ b/engine/.gitignore @@ -46,3 +46,7 @@ logs/ # OS .DS_Store Thumbs.db + +# Eval run outputs - measurements, not source. +evals/*/results/ +evals/*/results-*/ diff --git a/engine/evals/planner/README.md b/engine/evals/planner/README.md new file mode 100644 index 0000000000..d487c5c279 --- /dev/null +++ b/engine/evals/planner/README.md @@ -0,0 +1,45 @@ +# Planner eval + +Measures the second decision in an edit request: once the orchestrator has routed to +`pdf_edit`, which of the 73 operations gets chosen. + +It exists to quantify one specific thing. The production planner prompt is roughly 6,800 +tokens; a default Ollama accepts about 2,050 and silently drops the head. This eval +measures what that costs, rather than assuming it costs something. + +```bash +uv run --group engine python evals/planner/runner.py +``` + +## The three strategies + +| Strategy | Prompt shape | +| --- | --- | +| `prod` | Production order - request first, then the full catalogue. Overruns the context. | +| `request_last` | Identical content, request moved to the tail, which survives truncation. | +| `shortlist` | Top-N operations by embedding similarity (`nomic-embed-text`), request last. Fits comfortably. | + +`--shortlist N` sets the candidate count; the runner prints recall@N first, so a low score +there is visible before the accuracy numbers are read. + +## Why position is reported + +`dataset.py` records each expected operation's index in `OPERATIONS`, and the menu is +rendered in that order. Conversions occupy 0-25 and lose their descriptions first; +security operations at 63-72 always survive. Accuracy is therefore broken out by catalogue +band, and `mean_picked_index` shows whether a strategy's answers are being dragged toward +the end of the list. + +The response schema enumerates all 73 endpoints regardless of what the prompt contains - +exactly as production does, since `ToolEndpoint` is a `StrEnum`. A truncated prompt does +not stop the model naming an operation; it only stops it reading what that operation does. + +## Caveats + +- Cases are single-operation. Production can chain, and chained plans are not measured here. +- `also_ok` marks operations that genuinely satisfy a request too (CSV for a spreadsheet ask). +- Latency is measured under `--concurrency`, so it is throughput rather than user-facing. +- **`prompt_tokens_reported` is not a truncation measure.** Ollama counts only + newly-evaluated tokens, so cases sharing a long prefix report far fewer than they sent. + Measure the ceiling with a single controlled request instead: send one oversized prompt + and read `usage.prompt_tokens` off it alone. diff --git a/engine/evals/planner/dataset.py b/engine/evals/planner/dataset.py new file mode 100644 index 0000000000..5e64c5da21 --- /dev/null +++ b/engine/evals/planner/dataset.py @@ -0,0 +1,106 @@ +"""Labelled tool-selection cases for the pdf_edit planner. + +Each case is a request that one operation answers. ``expected`` is that operation's +``ToolEndpoint`` name; ``also_ok`` holds operations that genuinely satisfy the request too. + +Cases carry the expected operation's index in ``OPERATIONS`` so accuracy can be reported by +catalogue position. Ollama drops the head of an over-long prompt, and the menu is rendered +in ``OPERATIONS`` order, so an early operation loses its description while a late one keeps +it. Position is therefore a variable under test, not a curiosity. +""" + +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class PlannerCase: + id: str + message: str + expected: str + also_ok: frozenset[str] = frozenset() + + +def _p(id: str, message: str, expected: str, also_ok: frozenset[str] = frozenset()) -> PlannerCase: + return PlannerCase(id, message, expected, also_ok) + + +# Conversions - indexes 0-25, the front of the catalogue and the first thing truncation eats. +_CONVERSIONS = [ + _p("to-word", "Turn this into an editable Word document", "PDF_TO_WORD"), + _p("to-excel", "I need this as a spreadsheet I can sort", "PDF_TO_XLSX", frozenset({"PDF_TO_CSV"})), + _p("to-images", "Save every page as a separate PNG", "PDF_TO_IMG"), + _p("to-text", "Give me the plain text out of this", "PDF_TO_TEXT"), + _p("to-pdfa", "Make this archival standard compliant for our records", "PDF_TO_PDFA"), + _p("to-html", "Convert this into a web page", "PDF_TO_HTML"), + _p("to-markdown", "I want this as markdown for our docs repo", "PDF_TO_MARKDOWN"), + _p("to-epub", "Turn this into an ebook I can read on my Kindle", "PDF_TO_EPUB"), + _p("to-slides", "Make this into a slide deck", "PDF_TO_PRESENTATION"), + _p("from-images", "Combine these photos into one PDF", "IMG_TO_PDF"), + _p("from-url", "Save this web address as a PDF", "URL_TO_PDF"), + _p("to-accessible", "Make this compliant for screen reader users", "PDF_TO_UA"), +] + +# Page-level structure - indexes 26-42, the middle of the catalogue. +_STRUCTURE = [ + _p("merge", "Join these two files into one", "MERGE_PDFS"), + _p("split-pages", "Break this into one file per page", "SPLIT_PAGES"), + _p("split-size", "Split this so no file is bigger than 5MB", "SPLIT_BY_SIZE_OR_COUNT"), + _p("split-chapters", "Separate this book out by chapter", "SPLIT_PDF_BY_CHAPTERS"), + _p("rotate", "Turn every page ninety degrees clockwise", "ROTATE_PDF"), + _p("remove-pages", "Get rid of pages 4 through 9", "REMOVE_PAGES"), + _p("rearrange", "Put the pages in reverse order", "REARRANGE_PAGES"), + _p("crop", "Trim the white margins off every page", "CROP"), + _p("nup", "Print this four pages to a sheet", "MULTI_PAGE_LAYOUT"), + _p("booklet", "Lay this out as a folded booklet for printing", "BOOKLET_IMPOSITION"), + _p("edit-text", "Change the word Draft to Final throughout", "EDIT_TEXT"), + _p("scale", "Resize every page to A4", "SCALE_PAGES"), +] + +# Content operations - indexes 43-62. +_CONTENT = [ + _p("compress", "Shrink this down, it is too big to email", "COMPRESS_PDF"), + _p("ocr", "Make this scan searchable", "OCR_PDF"), + _p("page-numbers", "Number the pages at the bottom", "ADD_PAGE_NUMBERS"), + _p("stamp", "Put our logo in the top corner of each page", "ADD_STAMP"), + _p("flatten", "Make the form fields non-editable", "FLATTEN"), + _p("remove-blanks", "Strip out the empty pages", "REMOVE_BLANKS"), + _p("extract-images", "Pull out all the pictures in this", "EXTRACT_IMAGES"), + _p("attachments", "Get the files attached to this PDF", "EXTRACT_ATTACHMENTS"), + _p("metadata", "Change the author name in the document properties", "UPDATE_METADATA"), + _p("repair", "This file is corrupted, can you fix it", "REPAIR"), + _p("auto-rotate", "Some pages are sideways, straighten them out", "AUTO_ROTATE_PDF"), + _p("scanner", "Make this look like it came off a scanner", "SCANNER_EFFECT"), +] + +# Security and reporting - indexes 63-72, the tail that always survives truncation. +_SECURITY = [ + _p("password", "Lock this with the password hunter2", "ADD_PASSWORD"), + _p("remove-password", "Take the password off this file", "REMOVE_PASSWORD"), + _p("watermark", "Put DRAFT across every page", "ADD_WATERMARK"), + _p("redact", "Black out every mention of the client name", "REDACT_EXECUTE"), + _p("sanitize", "Strip any embedded scripts out of this", "SANITIZE_PDF"), + _p("unsign", "Remove the digital signature from this", "REMOVE_CERT_SIGN"), + _p("timestamp", "Add a trusted timestamp to this document", "TIMESTAMP_PDF"), + _p("a11y-report", "Check this against accessibility rules and report back", "ACCESSIBILITY_REPORT"), +] + +CASES: list[PlannerCase] = [*_CONVERSIONS, *_STRUCTURE, *_CONTENT, *_SECURITY] + + +# Catalogue bands, by position in OPERATIONS. Named rather than computed so a reordering of +# the enum shows up as a failing assertion instead of silently changing what is measured. +BANDS: dict[str, range] = { + "front (0-25, conversions)": range(0, 26), + "middle (26-42, structure)": range(26, 43), + "back (43-62, content)": range(43, 63), + "tail (63-72, security)": range(63, 73), +} + + +def band_for(index: int) -> str: + for name, span in BANDS.items(): + if index in span: + return name + raise ValueError(f"index {index} is outside every band") diff --git a/engine/evals/planner/recall.py b/engine/evals/planner/recall.py new file mode 100644 index 0000000000..e1cee510bb --- /dev/null +++ b/engine/evals/planner/recall.py @@ -0,0 +1,128 @@ +"""Sweep operation-retrieval settings for the planner shortlist. + +Recall is the planner's ceiling: the model was correct on every case whose answer reached +the candidate list, so an operation that retrieval misses is an operation the planner +cannot pick. Recall needs embeddings only - no generation - so a full sweep costs seconds +and can be run before spending anything on the model. + + uv run --group engine python evals/planner/recall.py +""" + +from __future__ import annotations + +import asyncio +import json +import sys +from pathlib import Path + +import httpx + +_EVALS_ROOT = Path(__file__).resolve().parents[1] +if str(_EVALS_ROOT) not in sys.path: + sys.path.insert(0, str(_EVALS_ROOT)) +_SRC = _EVALS_ROOT.parent / "src" +if str(_SRC) not in sys.path: + sys.path.insert(0, str(_SRC)) + +from planner.dataset import CASES # noqa: E402 +from planner.runner import ( # noqa: E402 + ALL_OPS, + EMBED_MODEL, + EMBED_URL, + INDEX_OF, + bm25_scores, + cosine, + rank_fusion, + retrieval_text, + tokenize, +) + +from stirling.models import OPERATIONS, ToolEndpoint # noqa: E402 + + +def op_text(op: ToolEndpoint, *, variant: str) -> str: + """How an operation is described to the retriever.""" + schema = OPERATIONS[op].model_json_schema() + description = (schema.get("description") or "").strip() + name = op.name.replace("_", " ").lower() + if variant == "name": + return name + if variant == "name_desc": + return f"{name}. {description}" + return retrieval_text(op) + + +async def embed_all(client: httpx.AsyncClient, texts: list[str]) -> list[list[float]]: + response = await client.post(EMBED_URL, json={"model": EMBED_MODEL, "input": texts}, timeout=600.0) + return response.json()["embeddings"] + + +async def main() -> None: + variants = ["name", "name_desc", "name_desc_params"] + cutoffs = [5, 10, 12, 15, 20, 30] + results: dict[str, dict[str, dict[int, float]]] = {} + best_rankings: dict[str, dict[str, list[list[str]]]] = {} + + async with httpx.AsyncClient() as http: + case_vectors = await embed_all(http, [c.message for c in CASES]) + + for variant in variants: + texts = [op_text(op, variant=variant) for op in ALL_OPS] + op_vectors = await embed_all(http, texts) + corpus = [tokenize(t) for t in texts] + results[variant] = {"vector": {}, "bm25": {}, "hybrid": {}} + + per_case: dict[str, list[list[str]]] = {"vector": [], "bm25": [], "hybrid": []} + for case, vector in zip(CASES, case_vectors, strict=True): + vec_rank = sorted(ALL_OPS, key=lambda op: -cosine(vector, op_vectors[INDEX_OF[op.name]])) + vec_names = [op.name for op in vec_rank] + lex = bm25_scores(case.message, corpus) + lex_names = [op.name for op in sorted(ALL_OPS, key=lambda op: -lex[INDEX_OF[op.name]])] + per_case["vector"].append(vec_names) + per_case["bm25"].append(lex_names) + per_case["hybrid"].append(rank_fusion(vec_names, lex_names)) + + best_rankings[variant] = per_case + for method, rankings in per_case.items(): + for cutoff in cutoffs: + hits = sum( + 1 for case, ranking in zip(CASES, rankings, strict=True) if case.expected in ranking[:cutoff] + ) + results[variant][method][cutoff] = round(hits / len(CASES), 4) + + header = " ".join(f"@{c:<5}" for c in cutoffs) + print(f"{'embedding text':<20} {'method':<8} {header}") + for variant, methods in results.items(): + for method, by_cutoff in methods.items(): + row = " ".join(f"{by_cutoff[c]:<6.1%}" for c in cutoffs) + print(f"{variant:<20} {method:<8} {row}") + + best = max( + ((v, m, c, s) for v, ms in results.items() for m, cs in ms.items() for c, s in cs.items()), + key=lambda row: (row[3], -row[2]), + ) + print(f"\nBest: {best[0]} + {best[1]} @{best[2]} = {best[3]:.1%}") + + # What the winner still cannot reach, so the next change has somewhere to aim. + best_variant, best_method, best_cutoff, _ = best + misses = [ + {"case": case.id, "expected": case.expected, "rank": ranking.index(case.expected) + 1} + for case, ranking in zip(CASES, best_rankings[best_variant][best_method], strict=True) + if case.expected not in ranking[:best_cutoff] + ] + if misses: + print(f"\nStill missed at @{best_cutoff} ({len(misses)} of {len(CASES)}) - true rank in brackets:") + for miss in sorted(misses, key=lambda m: m["rank"]): + print(f" {miss['case']:<16} {miss['expected']:<24} [{miss['rank']}]") + + out = Path("evals/planner/results/recall.json") + out.parent.mkdir(parents=True, exist_ok=True) + out.write_text( + json.dumps({"cutoffs": cutoffs, "results": results, "misses_at_best": misses}, indent=2), + encoding="utf-8", + ) + print(f"\nWrote {out}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/engine/evals/planner/runner.py b/engine/evals/planner/runner.py new file mode 100644 index 0000000000..adfe779354 --- /dev/null +++ b/engine/evals/planner/runner.py @@ -0,0 +1,338 @@ +"""Measure pdf_edit tool selection, and how much of it the context ceiling costs. + + uv run --group engine python evals/planner/runner.py + +Strategies share one decision - pick the operation that answers the request - and differ +only in how the 73-operation catalogue is presented. The production prompt overruns what +Ollama will accept, so this quantifies the damage rather than assuming it. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import math +import re +import sys +import time +from collections import Counter, defaultdict +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any + +import httpx + +_EVALS_ROOT = Path(__file__).resolve().parents[1] +if str(_EVALS_ROOT) not in sys.path: + sys.path.insert(0, str(_EVALS_ROOT)) +_SRC = _EVALS_ROOT.parent / "src" +if str(_SRC) not in sys.path: + sys.path.insert(0, str(_SRC)) + +from planner.dataset import CASES, PlannerCase, band_for # noqa: E402 +from routing.client import OllamaRouter # noqa: E402 + +from stirling.agents.pdf_edit import PdfEditAgent # noqa: E402 +from stirling.models import OPERATIONS, ToolEndpoint # noqa: E402 + +ALL_OPS: list[ToolEndpoint] = list(OPERATIONS.keys()) +INDEX_OF: dict[str, int] = {op.name: i for i, op in enumerate(ALL_OPS)} +BY_VALUE: dict[str, ToolEndpoint] = {op.value: op for op in ALL_OPS} + +EMBED_URL = "http://localhost:11434/api/embed" +EMBED_MODEL = "nomic-embed-text" + +_SYSTEM = ( + "Plan PDF edit requests. Choose the single operation that best answers the request. " + "Each operation is listed with its description. Treat that list as authoritative: an " + "operation can only do what its description allows." +) + + +@dataclass +class PlannerObservation: + strategy: str + case_id: str + expected: str + predicted: str + correct: bool + expected_index: int + predicted_index: int + band: str + prompt_tokens_sent: int + # Ollama reports only newly-evaluated tokens, so a shared prefix across cases makes + # this read low. Use the controlled probes, not this, to measure truncation. + prompt_tokens_reported: int + latency_s: float + + +_WORD = re.compile(r"[a-z0-9]+") +# Splits SCREAMING_SNAKE endpoint names into their parts so PDF_TO_WORD matches "word". +_STOPWORDS = frozenset({"pdf", "the", "this", "a", "an", "of", "to", "and", "for", "it", "into", "out"}) + + +def tokenize(text: str) -> list[str]: + return [w for w in _WORD.findall(text.lower().replace("_", " ")) if w not in _STOPWORDS] + + +def bm25_scores(query: str, corpus: list[list[str]]) -> list[float]: + """Plain BM25 over the operation texts, as the lexical half of hybrid retrieval.""" + k1, b = 1.5, 0.75 + lengths = [len(doc) for doc in corpus] + avg_len = sum(lengths) / len(lengths) + doc_freq: Counter[str] = Counter() + for doc in corpus: + doc_freq.update(set(doc)) + n = len(corpus) + scores = [0.0] * n + for term in tokenize(query): + df = doc_freq.get(term, 0) + if not df: + continue + idf = math.log(1 + (n - df + 0.5) / (df + 0.5)) + for i, doc in enumerate(corpus): + tf = doc.count(term) + if tf: + scores[i] += idf * (tf * (k1 + 1)) / (tf + k1 * (1 - b + b * lengths[i] / avg_len)) + return scores + + +def rank_fusion(*rankings: list[str], k: int = 60) -> list[str]: + """Reciprocal rank fusion - combines rankings without needing comparable scores.""" + fused: dict[str, float] = {} + for ranking in rankings: + for position, name in enumerate(ranking): + fused[name] = fused.get(name, 0.0) + 1.0 / (k + position + 1) + return sorted(fused, key=lambda name: -fused[name]) + + +def op_line(op: ToolEndpoint) -> str: + schema = OPERATIONS[op].model_json_schema() + description = (schema.get("description") or "").strip() + return f"- {op.name} ({op.value}): {description}" if description else f"- {op.name} ({op.value})" + + +def retrieval_text(op: ToolEndpoint) -> str: + """What the retriever indexes - richer than what the prompt shows. + + Parameter descriptions carry the words users actually type ("watermark opacity", + "OCR language"), and including them lifted recall@12 from 77.3% to 88.6%. They stay out + of the prompt itself, which only needs enough to tell the candidates apart. + """ + model = OPERATIONS[op] + description = (model.model_json_schema().get("description") or "").strip() + # Field metadata, not the JSON schema: enum-typed fields render as a bare "$ref" and + # lose their description there. + params = [(f.description or "").strip() for f in model.model_fields.values() if (f.description or "").strip()] + return f"{op.name.replace('_', ' ').lower()}. {description} {' '.join(params)}".strip() + + +def endpoint_schema(ops: list[ToolEndpoint]) -> dict[str, Any]: + """Mirrors production: ToolEndpoint is a StrEnum, so the schema carries the path values. + + The names stay selectable even when the prompt describing them is truncated away, which + is exactly the production failure - the model can still name an operation it can no + longer read about. + """ + return { + "type": "json_schema", + "json_schema": { + "name": "plan", + "strict": True, + "schema": { + "type": "object", + "properties": {"operation": {"type": "string", "enum": [op.value for op in ops]}}, + "required": ["operation"], + "additionalProperties": False, + }, + }, + } + + +async def embed(client: httpx.AsyncClient, texts: list[str]) -> list[list[float]]: + response = await client.post(EMBED_URL, json={"model": EMBED_MODEL, "input": texts}, timeout=600.0) + return response.json()["embeddings"] + + +def cosine(a: list[float], b: list[float]) -> float: + dot = sum(x * y for x, y in zip(a, b, strict=True)) + na = sum(x * x for x in a) ** 0.5 + nb = sum(y * y for y in b) ** 0.5 + return dot / (na * nb) if na and nb else 0.0 + + +def prod_prompt(case: PlannerCase) -> str: + """The production shape: request first, then the full catalogue.""" + menu = PdfEditAgent._get_supported_operations_prompt(ALL_OPS) + return ( + f"Conversation history:\nNone\n" + f"User request: {case.message}\n" + f"Files: document.pdf\n" + f"Supported operations:\n{menu}\n" + f"Extracted page text:\nNone" + ) + + +def request_last_prompt(case: PlannerCase) -> str: + """Same content, request moved to the tail - the part that survives truncation.""" + menu = PdfEditAgent._get_supported_operations_prompt(ALL_OPS) + return ( + f"Supported operations:\n{menu}\n" + f"Conversation history:\nNone\n" + f"Files: document.pdf\n" + f"Extracted page text:\nNone\n" + f"User request: {case.message}" + ) + + +def shortlist_prompt(case: PlannerCase, ops: list[ToolEndpoint], *, rich: bool = False) -> str: + """Candidates only. ``rich`` spends the space a short list frees on describing them. + + The plain line carries the operation's own description; ``rich`` adds the parameter text + on top. With only 12-20 candidates there is room for both and the prompt still sits far + inside the context ceiling. + """ + render = (lambda op: f"- {op.name} ({op.value}): {retrieval_text(op)}") if rich else op_line + menu = "\n".join(render(op) for op in ops) + return ( + f"Candidate operations:\n{menu}\nConversation history:\nNone\nFiles: document.pdf\nUser request: {case.message}" + ) + + +async def main() -> None: + parser = argparse.ArgumentParser(description="pdf_edit tool-selection eval.") + parser.add_argument("--out", default="evals/planner/results") + parser.add_argument("--model", default="qwen3:8b") + parser.add_argument("--concurrency", type=int, default=3) + parser.add_argument("--shortlist", type=int, default=12) + parser.add_argument("--strategies", default="prod,request_last,shortlist") + parser.add_argument("--retrieval", default="hybrid", choices=["vector", "hybrid"]) + args = parser.parse_args() + + import tiktoken + + encoding = tiktoken.get_encoding("cl100k_base") + semaphore = asyncio.Semaphore(args.concurrency) + observations: list[PlannerObservation] = [] + + async with httpx.AsyncClient() as http: + router = OllamaRouter(http, model=args.model) + + print("Embedding the operation catalogue...", flush=True) + texts = [retrieval_text(op) for op in ALL_OPS] + op_vectors = await embed(http, texts) + corpus = [tokenize(t) for t in texts] + case_vectors = await embed(http, [c.message for c in CASES]) + shortlists: dict[str, list[ToolEndpoint]] = {} + by_name = {op.name: op for op in ALL_OPS} + for case, vector in zip(CASES, case_vectors, strict=True): + vector_rank = [ + op.name for op in sorted(ALL_OPS, key=lambda op: -cosine(vector, op_vectors[INDEX_OF[op.name]])) + ] + if args.retrieval == "vector": + ranked_names = vector_rank + else: + lexical = bm25_scores(case.message, corpus) + lexical_rank = [op.name for op in sorted(ALL_OPS, key=lambda op: -lexical[INDEX_OF[op.name]])] + ranked_names = rank_fusion(vector_rank, lexical_rank) + shortlists[case.id] = [by_name[n] for n in ranked_names[: args.shortlist]] + recall = sum(1 for c in CASES if c.expected in {op.name for op in shortlists[c.id]}) / len(CASES) + print(f"Shortlist recall@{args.shortlist} ({args.retrieval}): {recall:.1%}", flush=True) + + async def run(strategy: str, case: PlannerCase) -> PlannerObservation: + if strategy == "prod": + prompt, ops = prod_prompt(case), ALL_OPS + elif strategy == "request_last": + prompt, ops = request_last_prompt(case), ALL_OPS + else: + ops = shortlists[case.id] + prompt = shortlist_prompt(case, ops, rich=strategy == "shortlist_rich") + sent = len(encoding.encode(_SYSTEM)) + len(encoding.encode(prompt)) + async with semaphore: + result = await router.call( + _SYSTEM, + prompt, + response_format=endpoint_schema(ops), + max_tokens=4096, + temperature=0.0, + thinking=True, + ) + predicted = "__failed__" + if not result.error and result.finish_reason != "length": + try: + value = json.loads(result.content).get("operation") + predicted = BY_VALUE[value].name if value in BY_VALUE else "__offmenu__" + except ValueError: + predicted = "__unparsable__" + correct = predicted == case.expected or predicted in case.also_ok + return PlannerObservation( + strategy=strategy, + case_id=case.id, + expected=case.expected, + predicted=predicted, + correct=correct, + expected_index=INDEX_OF[case.expected], + predicted_index=INDEX_OF.get(predicted, -1), + band=band_for(INDEX_OF[case.expected]), + prompt_tokens_sent=sent, + prompt_tokens_reported=result.input_tokens, + latency_s=round(result.latency_s, 3), + ) + + for strategy in [s.strip() for s in args.strategies.split(",")]: + started = time.monotonic() + rows = await asyncio.gather(*[run(strategy, case) for case in CASES]) + observations.extend(rows) + print( + f"{strategy:14s} acc={sum(r.correct for r in rows) / len(rows):6.1%} " + f"sent={rows[0].prompt_tokens_sent:5d} " + f"wall={time.monotonic() - started:6.1f}s", + flush=True, + ) + + summary: dict[str, Any] = {} + grouped: dict[str, list[PlannerObservation]] = defaultdict(list) + for obs in observations: + grouped[obs.strategy].append(obs) + for name, rows in grouped.items(): + bands = {} + for band in {r.band for r in rows}: + band_rows = [r for r in rows if r.band == band] + bands[band] = { + "n": len(band_rows), + "accuracy": round(sum(r.correct for r in band_rows) / len(band_rows), 4), + } + picked = [r.predicted_index for r in rows if r.predicted_index >= 0] + summary[name] = { + "accuracy": round(sum(r.correct for r in rows) / len(rows), 4), + "prompt_tokens_sent": rows[0].prompt_tokens_sent, + "prompt_tokens_reported_avg": round(sum(r.prompt_tokens_reported for r in rows) / len(rows), 1), + "avg_latency_s": round(sum(r.latency_s for r in rows) / len(rows), 2), + # Where in the catalogue its answers come from: truncation should drag this late. + "mean_picked_index": round(sum(picked) / len(picked), 1) if picked else None, + "by_band": dict(sorted(bands.items())), + "wrong": [ + {"case": r.case_id, "expected": r.expected, "picked": r.predicted} for r in rows if not r.correct + ][:15], + "top_wrong_picks": Counter(r.predicted for r in rows if not r.correct).most_common(5), + } + + out_dir = Path(args.out) + out_dir.mkdir(parents=True, exist_ok=True) + (out_dir / "summary.json").write_text( + json.dumps({"model": args.model, "case_count": len(CASES), "summary": summary}, indent=2), + encoding="utf-8", + ) + with (out_dir / "observations.jsonl").open("w", encoding="utf-8") as handle: + for obs in observations: + handle.write(json.dumps(asdict(obs)) + "\n") + expected_mean = sum(INDEX_OF[c.expected] for c in CASES) / len(CASES) + print(f"\nMean catalogue index of the chosen operation (expected mean {expected_mean:.1f}):") + for name, stats in summary.items(): + print(f" {name:14s} {stats['mean_picked_index']}") + print(f"\nWrote {out_dir / 'summary.json'}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/engine/evals/routing/README.md b/engine/evals/routing/README.md new file mode 100644 index 0000000000..0925a212df --- /dev/null +++ b/engine/evals/routing/README.md @@ -0,0 +1,74 @@ +# Routing eval + +Measures the orchestrator's top-level capability decision - the one that sends a turn to +`pdf_edit`, `pdf_question`, `pdf_review`, `pdf_create`, `user_spec`, or `unsupported`. + +It exists because the reported failure ("asking a question and it runs a tool instead") is +a routing failure, and prompt changes aimed at it were otherwise unmeasurable. + +## Running it + +Needs a reachable Ollama with the model pulled. Nothing else - no engine server, no database. + +```bash +uv run --group engine python evals/routing/runner.py +``` + +Useful flags: + +| Flag | Purpose | +| --- | --- | +| `--strategies thinking_off,fewshot_thinking_off` | Run a subset. | +| `--repeats 5` | Run each case N times; feeds the `unstable_cases` count. | +| `--concurrency 3` | Parallel in-flight requests. | +| `--model qwen2.5:7b` | Compare models on the same cases. | +| `--limit 8` | Smoke test. | + +Then turn the results into the cost table: + +```bash +uv run --group engine python evals/routing/cost.py +``` + +Outputs land in `evals/routing/results/`: `summary.json` (per-strategy scores), +`observations.jsonl` (one row per case per strategy, for re-scoring without re-spending), +and `cost.json`. + +## What is scored + +`dataset.py` holds 82 labelled turns in four bands: + +- **clear** - unambiguous either way. +- **boundary** - question and edit phrased almost identically ("Are there any blank pages?" + vs "Remove the blank pages"). This is the reported failure mode. +- **contextual** - only resolvable from conversation history ("yes do that", "now rotate it"). +- **adversarial** - a tool's name appears in a turn that is not asking for that tool + ("Which section talks about the merger?"). + +Beyond plain accuracy the runner reports: + +- **destructive misroutes** - a read-only turn sent to a mutating capability. Weighted + separately because that direction changes the user's file; the reverse only answers. +- **hard failures** - truncated, unparsable, or off-menu model output. +- **unstable cases** - with `--repeats > 1`, cases where the same input routed differently + across runs. + +`tolerated` on a case marks a genuinely defensible second answer, scored as a near-miss +rather than a failure, so the boundary cases do not punish reasonable disagreement. + +## Adding a strategy + +Implement the `Strategy` protocol in `strategies.py` (a `name`, a `description`, and an +async `route`) and add it to `build_strategies()`. The prompts come from +`stirling.agents.orchestrator`, so the baseline tracks production rather than a copy of it. + +## Caveats + +- Latency is measured under `--concurrency`, so it reflects throughput, not the latency a + single user sees. Re-run with `--concurrency 1` for user-facing latency. +- `reasoning_effort: "none"` is the only thinking switch Ollama honours on the + OpenAI-compatible path for qwen3; `chat_template_kwargs={"enable_thinking": false}` and a + `/no_think` suffix are both ignored. +- The eval sends prompts directly rather than through pydantic-ai, so it does not exercise + pydantic-ai's output-validation retries. A `__truncated__` result here is what production + spends a retry on. diff --git a/engine/evals/routing/__init__.py b/engine/evals/routing/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/engine/evals/routing/client.py b/engine/evals/routing/client.py new file mode 100644 index 0000000000..1f8b27a7d9 --- /dev/null +++ b/engine/evals/routing/client.py @@ -0,0 +1,146 @@ +"""Thin Ollama client for the routing eval. + +The eval talks to ``/v1/chat/completions`` directly rather than through pydantic-ai so it +can drive knobs pydantic-ai does not surface (``reasoning_effort``) and can measure the +reasoning tokens Ollama reports outside ``completion_tokens``. The prompts themselves are +imported from the production orchestrator, so what is measured is the real prompt. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass, field +from typing import Any + +import httpx + +DEFAULT_BASE_URL = "http://localhost:11434/v1/chat/completions" +DEFAULT_MODEL = "qwen3:8b" + + +@dataclass +class CallResult: + """One model round trip.""" + + content: str + latency_s: float + input_tokens: int + output_tokens: int + # Ollama returns qwen3's chain of thought in a separate `reasoning` field and (except + # when the cap truncates) leaves it out of completion_tokens, so it is counted here. + thinking_chars: int + finish_reason: str + error: str | None = None + + +@dataclass +class Budget: + """Running totals for one strategy run over one case.""" + + calls: int = 0 + input_tokens: int = 0 + output_tokens: int = 0 + thinking_chars: int = 0 + latency_s: float = 0.0 + errors: list[str] = field(default_factory=list) + + def add(self, result: CallResult) -> None: + self.calls += 1 + self.input_tokens += result.input_tokens + self.output_tokens += result.output_tokens + self.thinking_chars += result.thinking_chars + self.latency_s += result.latency_s + if result.error: + self.errors.append(result.error) + + +def enum_schema(name: str, values: list[str], *, with_message: bool = False) -> dict[str, Any]: + """A json-schema response format holding a single enum choice. + + Mirrors what pydantic-ai's NativeOutput sends for the router's ``_RouteDecision``. + """ + properties: dict[str, Any] = {"capability": {"type": "string", "enum": values}} + required = ["capability"] + if with_message: + properties["message"] = {"type": "string"} + required.append("message") + return { + "type": "json_schema", + "json_schema": { + "name": name, + "strict": True, + "schema": { + "type": "object", + "properties": properties, + "required": required, + "additionalProperties": False, + }, + }, + } + + +class OllamaRouter: + def __init__( + self, + client: httpx.AsyncClient, + *, + model: str = DEFAULT_MODEL, + base_url: str = DEFAULT_BASE_URL, + ) -> None: + self._client = client + self._model = model + self._base_url = base_url + + async def call( + self, + system_prompt: str, + user_prompt: str, + *, + response_format: dict[str, Any] | None, + max_tokens: int, + temperature: float | None, + thinking: bool, + ) -> CallResult: + body: dict[str, Any] = { + "model": self._model, + "messages": [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": user_prompt}, + ], + "max_tokens": max_tokens, + } + if response_format is not None: + body["response_format"] = response_format + if temperature is not None: + body["temperature"] = temperature + if not thinking: + # The only knob Ollama honours on the OpenAI-compatible path for qwen3. + # chat_template_kwargs{enable_thinking} and a /no_think suffix are both ignored. + body["reasoning_effort"] = "none" + + started = time.monotonic() + try: + response = await self._client.post(self._base_url, json=body, timeout=600.0) + except httpx.HTTPError as exc: + return CallResult("", time.monotonic() - started, 0, 0, 0, "transport", f"{type(exc).__name__}: {exc}") + elapsed = time.monotonic() - started + + try: + payload = response.json() + except ValueError: + return CallResult("", elapsed, 0, 0, 0, "bad-json", f"HTTP {response.status_code}: non-JSON body") + if "choices" not in payload: + return CallResult("", elapsed, 0, 0, 0, "no-choices", f"HTTP {response.status_code}: {payload}"[:400]) + + choice = payload["choices"][0] + message = choice.get("message") or {} + usage = payload.get("usage") or {} + reasoning = message.get("reasoning") or message.get("reasoning_content") or "" + return CallResult( + content=message.get("content") or "", + latency_s=elapsed, + input_tokens=int(usage.get("prompt_tokens") or 0), + output_tokens=int(usage.get("completion_tokens") or 0), + thinking_chars=len(reasoning), + finish_reason=str(choice.get("finish_reason") or ""), + ) diff --git a/engine/evals/routing/cost.py b/engine/evals/routing/cost.py new file mode 100644 index 0000000000..3116264a0e --- /dev/null +++ b/engine/evals/routing/cost.py @@ -0,0 +1,184 @@ +"""Cost model for the routing decision. + +Two questions the eval alone does not answer: + +1. What does each routing strategy cost per conversation, in tokens, in local GPU + seconds, and in dollars if the same orchestrator runs on the hosted default + (engine/.env ships ``anthropic:claude-haiku-4-5``)? +2. How big is that cost next to the rest of the request it is deciding for? + +Run after runner.py: + + uv run --group engine python evals/routing/cost.py +""" + +from __future__ import annotations + +import json +import sys +from dataclasses import dataclass +from pathlib import Path + +_EVALS_ROOT = Path(__file__).resolve().parents[1] +if str(_EVALS_ROOT) not in sys.path: + sys.path.insert(0, str(_EVALS_ROOT)) +_SRC = _EVALS_ROOT.parent / "src" +if str(_SRC) not in sys.path: + sys.path.insert(0, str(_SRC)) + + +# Claude Haiku 4.5 list price, the hosted default in engine/.env. USD per million tokens. +HAIKU_INPUT_PER_MTOK = 1.00 +HAIKU_OUTPUT_PER_MTOK = 5.00 + + +@dataclass(frozen=True) +class StageProfile: + """One model call in a request, sized from the real rendered prompts.""" + + name: str + input_tokens: int + output_tokens: int + note: str + + +def measure_pipeline_stages() -> list[StageProfile]: + """Token profile of a single-operation pdf_edit request, measured not guessed.""" + import tiktoken + + from stirling.agents.orchestrator import _ROUTER_SYSTEM_PROMPT + from stirling.agents.pdf_edit import PdfEditAgent + from stirling.models import OPERATIONS + + encoding = tiktoken.get_encoding("cl100k_base") + operations = list(OPERATIONS.keys()) + + def count(text: str) -> int: + return len(encoding.encode(text)) + + menu = PdfEditAgent._get_supported_operations_prompt(operations) + flat_names = PdfEditAgent._get_operations_prompt(operations) + + # A representative turn: short message, one file, no history. + turn_overhead = count("Conversation history:\nNone\nUser message: Rotate all pages 90 degrees\nFiles: a.pdf\n") + + router_in = count(_ROUTER_SYSTEM_PROMPT) + turn_overhead + count("Available artifacts:\n- none") + # The planner's system prompt carries the flat name+path list; the user prompt carries + # the full menu with every parameter description. + planner_in = count(flat_names) + count(menu) + turn_overhead + 200 + + # One parameter call per planned operation. It re-sends the same turn plus the + # selected operation's schema, and pydantic-ai passes the model the full schema. + param_schema_tokens = max(count(json.dumps(OPERATIONS[op].model_json_schema())) for op in operations) + median_schema = sorted(count(json.dumps(OPERATIONS[op].model_json_schema())) for op in operations)[ + len(operations) // 2 + ] + + return [ + StageProfile("router", router_in, 8, "6-way enum decision"), + StageProfile("edit planner", planner_in, 60, f"73-operation menu ({count(menu)} tok) + flat list"), + StageProfile("parameter call", median_schema + turn_overhead + 100, 60, "one per planned operation"), + StageProfile( + "parameter call (worst schema)", + param_schema_tokens + turn_overhead + 100, + 60, + "largest operation schema", + ), + ] + + +def dollars(input_tokens: float, output_tokens: float) -> float: + return input_tokens / 1e6 * HAIKU_INPUT_PER_MTOK + output_tokens / 1e6 * HAIKU_OUTPUT_PER_MTOK + + +def build_cost_table(summary_path: Path) -> dict: + payload = json.loads(summary_path.read_text(encoding="utf-8")) + summary = payload["summary"] + + rows = [] + for name, stats in summary.items(): + per_conv_in = stats["avg_input_tokens"] + per_conv_out = stats["avg_output_tokens"] + accuracy = stats["accuracy"] + usd_1k = dollars(per_conv_in * 1000, per_conv_out * 1000) + # A wrong route is wasted spend plus a wasted downstream pipeline, so the honest + # unit is cost per correctly-routed conversation. + usd_1k_correct = usd_1k / accuracy if accuracy else float("inf") + rows.append( + { + "strategy": name, + "description": payload["strategy_descriptions"].get(name, ""), + "accuracy": accuracy, + "accuracy_with_tolerated": stats["accuracy_with_tolerated"], + "destructive_misroutes": stats["destructive_misroutes"], + "hard_failures": stats["hard_failures"], + "avg_calls": stats["avg_calls"], + "avg_input_tokens": per_conv_in, + "avg_output_tokens": per_conv_out, + "avg_thinking_chars": stats["avg_thinking_chars"], + "avg_latency_s": stats["avg_latency_s"], + "p90_latency_s": stats["p90_latency_s"], + "usd_per_1k_conversations": round(usd_1k, 4), + "usd_per_1k_correct": round(usd_1k_correct, 4), + "by_band": stats["by_band"], + "top_confusions": stats["top_confusions"], + } + ) + rows.sort(key=lambda r: -r["accuracy"]) + return {"meta": payload, "rows": rows} + + +def main() -> None: + results = Path("evals/routing/results/summary.json") + stages = measure_pipeline_stages() + + print("=== Measured pipeline token profile (single-operation pdf_edit request) ===") + total_in = 0 + for stage in stages: + if "worst" in stage.name: + continue + total_in += stage.input_tokens + print(f" {stage.name:24s} in={stage.input_tokens:6d} out={stage.output_tokens:4d} {stage.note}") + print(f" {'TOTAL (3 calls)':24s} in={total_in:6d}") + router_share = stages[0].input_tokens / total_in + print(f" router share of input tokens: {router_share:.1%}") + + if not results.exists(): + print("\nNo eval results yet - run runner.py first.") + return + + table = build_cost_table(results) + print("\n=== Per-strategy cost (measured) ===") + print( + f"{'strategy':20s} {'acc':>6s} {'dstr':>5s} {'calls':>6s} {'in':>7s} " + f"{'out':>6s} {'lat_s':>7s} {'$/1k':>8s} {'$/1k ok':>8s}" + ) + for row in table["rows"]: + print( + f"{row['strategy']:20s} {row['accuracy']:6.1%} {row['destructive_misroutes']:5d} " + f"{row['avg_calls']:6.2f} {row['avg_input_tokens']:7.0f} {row['avg_output_tokens']:6.0f} " + f"{row['avg_latency_s']:7.2f} {row['usd_per_1k_conversations']:8.3f} {row['usd_per_1k_correct']:8.3f}" + ) + + out = Path("evals/routing/results/cost.json") + out.write_text( + json.dumps( + { + "pricing": { + "model": "claude-haiku-4-5", + "input_per_mtok": HAIKU_INPUT_PER_MTOK, + "output_per_mtok": HAIKU_OUTPUT_PER_MTOK, + }, + "pipeline_stages": [vars(s) for s in stages], + "strategies": table["rows"], + "meta": {k: v for k, v in table["meta"].items() if k != "summary"}, + }, + indent=2, + ), + encoding="utf-8", + ) + print(f"\nWrote {out}") + + +if __name__ == "__main__": + main() diff --git a/engine/evals/routing/dataset.py b/engine/evals/routing/dataset.py new file mode 100644 index 0000000000..e1d7c40e44 --- /dev/null +++ b/engine/evals/routing/dataset.py @@ -0,0 +1,319 @@ +"""Labelled routing cases for the orchestrator's top-level capability decision. + +Each case is one user turn as the orchestrator sees it. ``expected`` is the correct +capability; ``tolerated`` lists routes that are defensible for genuinely ambiguous +turns and are scored as near-misses rather than failures. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import StrEnum + +# Mirrors orchestrator._RouteCapability. Kept as plain strings so the eval can score a +# model that returns something off-menu without blowing up. +CAPABILITIES = ("pdf_edit", "pdf_question", "user_spec", "pdf_review", "pdf_create", "unsupported") + + +class Band(StrEnum): + """Difficulty band, so accuracy is reported per band rather than as one blurred number.""" + + CLEAR = "clear" + BOUNDARY = "boundary" + CONTEXTUAL = "contextual" + ADVERSARIAL = "adversarial" + + +@dataclass(frozen=True) +class RoutingCase: + id: str + message: str + expected: str + band: Band + files: tuple[str, ...] = ("document.pdf",) + history: tuple[tuple[str, str], ...] = () + tolerated: frozenset[str] = frozenset() + note: str = "" + + +def _c( + id: str, + message: str, + expected: str, + band: Band, + *, + files: tuple[str, ...] = ("document.pdf",), + history: tuple[tuple[str, str], ...] = (), + tolerated: frozenset[str] = frozenset(), + note: str = "", +) -> RoutingCase: + return RoutingCase(id, message, expected, band, files, history, tolerated, note) + + +# --- Clear questions: reading the document, nothing mutated ------------------------- +_CLEAR_QUESTIONS = [ + _c("q-what-about", "What is this document about?", "pdf_question", Band.CLEAR), + _c("q-who-signed", "Who signed this contract?", "pdf_question", Band.CLEAR), + _c("q-total", "What is the total amount on this invoice?", "pdf_question", Band.CLEAR), + _c("q-summarise", "Summarise this for me", "pdf_question", Band.CLEAR), + _c("q-deadline", "When is the payment deadline?", "pdf_question", Band.CLEAR), + _c("q-parties", "List the parties named in this agreement", "pdf_question", Band.CLEAR), + _c("q-explain", "Explain clause 7 in plain English", "pdf_question", Band.CLEAR), + _c("q-howmany-pages", "How many pages does this have?", "pdf_question", Band.CLEAR), + _c("q-mention", "Does this mention anything about termination?", "pdf_question", Band.CLEAR), + _c("q-language", "What language is this written in?", "pdf_question", Band.CLEAR), +] + +# --- Clear edits: the user wants a changed file back -------------------------------- +_CLEAR_EDITS = [ + _c("e-rotate", "Rotate all pages 90 degrees", "pdf_edit", Band.CLEAR), + _c("e-to-word", "Convert this to Word", "pdf_edit", Band.CLEAR), + _c("e-compress", "Compress this file, it is too big to email", "pdf_edit", Band.CLEAR), + _c("e-split", "Split this into one file per page", "pdf_edit", Band.CLEAR), + _c("e-merge", "Merge these two files together", "pdf_edit", Band.CLEAR, files=("a.pdf", "b.pdf")), + _c("e-watermark", "Add a DRAFT watermark to every page", "pdf_edit", Band.CLEAR), + _c("e-password", "Password protect this with hunter2", "pdf_edit", Band.CLEAR), + _c("e-delete-p3", "Delete page 3", "pdf_edit", Band.CLEAR), + _c("e-ocr", "Run OCR on this so I can search it", "pdf_edit", Band.CLEAR), + _c("e-to-images", "Export every page as a PNG", "pdf_edit", Band.CLEAR), +] + +# --- The reported failure mode: question vs edit on near-identical surface forms ----- +_BOUNDARY = [ + _c( + "b-blank-ask", + "Are there any blank pages in this?", + "pdf_question", + Band.BOUNDARY, + note="Asks whether, does not ask to remove.", + ), + _c("b-blank-do", "Remove the blank pages", "pdf_edit", Band.BOUNDARY), + _c("b-signed-ask", "Is this document signed?", "pdf_question", Band.BOUNDARY), + _c("b-signed-do", "Sign this document for me", "pdf_edit", Band.BOUNDARY), + _c("b-pw-ask", "Is this file password protected?", "pdf_question", Band.BOUNDARY), + _c("b-pw-do", "Remove the password from this file", "pdf_edit", Band.BOUNDARY), + _c("b-pii-ask", "Does this contain any personal data?", "pdf_question", Band.BOUNDARY), + _c("b-pii-do", "Redact all the personal data", "pdf_edit", Band.BOUNDARY), + _c( + "b-can-you-compress", + "Can you compress this?", + "pdf_edit", + Band.BOUNDARY, + note="Polite imperative, not a capability question.", + ), + _c( + "b-how-much-smaller", + "How much smaller could this file get?", + "pdf_question", + Band.BOUNDARY, + tolerated=frozenset({"unsupported"}), + ), + _c("b-which-pages", "Which pages mention the budget?", "pdf_question", Band.BOUNDARY), + _c("b-extract-pages", "Extract the pages that mention the budget", "pdf_edit", Band.BOUNDARY), + _c("b-count-forms", "How many form fields are in this?", "pdf_question", Band.BOUNDARY), + _c("b-flatten-forms", "Flatten the form fields", "pdf_edit", Band.BOUNDARY), + _c("b-orientation-ask", "Are any pages upside down?", "pdf_question", Band.BOUNDARY), + _c("b-orientation-do", "Fix the pages that are upside down", "pdf_edit", Band.BOUNDARY), + _c( + "b-should-i-split", + "Should I split this into separate documents?", + "pdf_question", + Band.BOUNDARY, + tolerated=frozenset({"unsupported"}), + note="Advice, not an instruction.", + ), + _c("b-what-format", "What format is this file in?", "pdf_question", Band.BOUNDARY), +] + +# --- Create a brand-new document, usually with no input file ------------------------ +_CREATE = [ + _c("c-invoice", "Write me an invoice for 500 pounds of consulting work", "pdf_create", Band.CLEAR, files=()), + _c("c-letter", "Draft a resignation letter", "pdf_create", Band.CLEAR, files=()), + _c("c-report", "Create a one page status report about our Q3 launch", "pdf_create", Band.CLEAR, files=()), + _c("c-contract", "Make me a simple freelance contract template", "pdf_create", Band.CLEAR, files=()), + _c("c-nda", "Generate an NDA between Acme Ltd and Beta Inc", "pdf_create", Band.CLEAR, files=()), + _c( + "c-invoice-withfile", + "Create an invoice like this one but for 900 pounds", + "pdf_create", + Band.BOUNDARY, + tolerated=frozenset({"pdf_edit", "pdf_question"}), + note="File present but the output is a new doc.", + ), + _c("c-cover", "Write a cover letter for a product manager role", "pdf_create", Band.CLEAR, files=()), + _c("c-agenda", "Put together a meeting agenda for Thursday", "pdf_create", Band.CLEAR, files=()), +] + +# --- Review: return the PDF with comments/annotations attached ---------------------- +_REVIEW = [ + _c("r-review", "Review this and leave comments", "pdf_review", Band.CLEAR), + _c("r-annotate", "Annotate anything unclear with sticky notes", "pdf_review", Band.CLEAR), + _c("r-feedback", "Give me feedback on this draft as comments on the document", "pdf_review", Band.CLEAR), + _c("r-flag", "Flag any risky clauses directly in the PDF", "pdf_review", Band.CLEAR), + _c( + "r-markup", + "Mark up the sections that need work", + "pdf_review", + Band.BOUNDARY, + tolerated=frozenset({"pdf_edit"}), + ), + _c( + "r-whats-wrong", + "What is wrong with this contract?", + "pdf_question", + Band.BOUNDARY, + tolerated=frozenset({"pdf_review"}), + note="Wants an answer, not an annotated file.", + ), + _c("r-proofread", "Proofread this and put your corrections in the margin", "pdf_review", Band.BOUNDARY), +] + +# --- Agent spec authoring ----------------------------------------------------------- +_SPEC = [ + _c("s-agent", "Create an agent that watermarks every file I upload", "user_spec", Band.CLEAR, files=()), + _c("s-automation", "Set up an automation to compress incoming invoices", "user_spec", Band.CLEAR, files=()), + _c("s-define", "Define a new agent for redacting client names", "user_spec", Band.CLEAR, files=()), + _c( + "s-workflow", + "I want a workflow that OCRs and then splits by chapter", + "user_spec", + Band.BOUNDARY, + tolerated=frozenset({"pdf_edit"}), + ), + _c("s-edit-spec", "Change my watermark agent to use red text", "user_spec", Band.CLEAR, files=()), + _c("s-once", "Watermark this one file", "pdf_edit", Band.BOUNDARY, note="One-off action, not a reusable spec."), +] + +# --- About the assistant, or out of scope ------------------------------------------- +_UNSUPPORTED = [ + _c("u-model", "What model are you running on?", "unsupported", Band.CLEAR, files=()), + _c("u-who", "Who made you?", "unsupported", Band.CLEAR, files=()), + _c("u-capabilities", "What can you do?", "unsupported", Band.CLEAR, files=()), + _c("u-weather", "What is the weather in London tomorrow?", "unsupported", Band.CLEAR, files=()), + _c("u-code", "Write me a Python script to sort a list", "unsupported", Band.CLEAR, files=()), + _c("u-hello", "hey", "unsupported", Band.CLEAR, files=()), + _c("u-thanks", "thanks, that worked", "unsupported", Band.CLEAR, tolerated=frozenset({"pdf_question"})), +] + +# --- Turns that only make sense against the conversation history --------------------- +_CONTEXTUAL = [ + _c( + "x-do-it", + "yes do that", + "pdf_edit", + Band.CONTEXTUAL, + history=( + ("user", "Can you compress this file?"), + ("assistant", "I can compress it to about 40% of the current size. Shall I go ahead?"), + ), + ), + _c( + "x-and-page2", + "and what about page 2?", + "pdf_question", + Band.CONTEXTUAL, + history=( + ("user", "What does page 1 say?"), + ("assistant", "Page 1 is the cover sheet for the Acme service agreement."), + ), + ), + _c( + "x-now-rotate", + "now rotate it", + "pdf_edit", + Band.CONTEXTUAL, + history=(("user", "Split this into single pages"), ("assistant", "Done, I split it into 12 files.")), + ), + _c( + "x-why", + "why did you pick that one?", + "unsupported", + Band.CONTEXTUAL, + tolerated=frozenset({"pdf_question"}), + history=(("user", "Compress this"), ("assistant", "I used the lossless compression profile.")), + ), + _c( + "x-same-again", + "do the same to this one", + "pdf_edit", + Band.CONTEXTUAL, + files=("second.pdf",), + history=(("user", "Add a watermark saying CONFIDENTIAL"), ("assistant", "Watermark added.")), + ), + _c( + "x-more-detail", + "can you give me more detail on that?", + "pdf_question", + Band.CONTEXTUAL, + history=(("user", "What are the payment terms?"), ("assistant", "Net 30 from invoice date.")), + ), +] + +# --- Phrasings built to trip a keyword-ish router ----------------------------------- +_ADVERSARIAL = [ + _c( + "a-word-question", + "Does this document explain how to convert a PDF to Word?", + "pdf_question", + Band.ADVERSARIAL, + note="Contains 'convert to Word' but is a content question.", + ), + _c( + "a-merge-mention", + "Which section talks about the merger?", + "pdf_question", + Band.ADVERSARIAL, + note="'merger' is not the merge tool.", + ), + _c("a-split-mention", "What does it say about splitting the estate?", "pdf_question", Band.ADVERSARIAL), + _c("a-rotate-mention", "Is there anything in here about staff rotation?", "pdf_question", Band.ADVERSARIAL), + _c( + "a-sign-mention", "What are the signature requirements described in clause 4?", "pdf_question", Band.ADVERSARIAL + ), + _c("a-compress-mention", "Does the report discuss compression algorithms?", "pdf_question", Band.ADVERSARIAL), + _c( + "a-polite-delete", + "Could you please delete the last page?", + "pdf_edit", + Band.ADVERSARIAL, + note="Question mark, but an instruction.", + ), + _c("a-polite-landscape", "I would really appreciate it if this were in landscape", "pdf_edit", Band.ADVERSARIAL), + _c( + "a-declarative-q", + "I need to know who approved this", + "pdf_question", + Band.ADVERSARIAL, + note="Declarative, but a question.", + ), + _c("a-redact-mention", "Explain the redaction policy described on page 6", "pdf_question", Band.ADVERSARIAL), +] + +CASES: list[RoutingCase] = [ + *_CLEAR_QUESTIONS, + *_CLEAR_EDITS, + *_BOUNDARY, + *_CREATE, + *_REVIEW, + *_SPEC, + *_UNSUPPORTED, + *_CONTEXTUAL, + *_ADVERSARIAL, +] + + +def by_band() -> dict[Band, list[RoutingCase]]: + grouped: dict[Band, list[RoutingCase]] = {band: [] for band in Band} + for case in CASES: + grouped[case.band].append(case) + return grouped + + +# Misrouting a read-only turn into a mutating capability changes the user's file; the +# reverse merely answers. Scored separately so a strategy cannot win on raw accuracy +# while getting the dangerous direction wrong. +MUTATING = frozenset({"pdf_edit", "pdf_create", "pdf_review"}) + + +def is_destructive_miss(case: RoutingCase, predicted: str) -> bool: + return case.expected not in MUTATING and predicted in MUTATING and predicted not in case.tolerated diff --git a/engine/evals/routing/runner.py b/engine/evals/routing/runner.py new file mode 100644 index 0000000000..b3d3f36650 --- /dev/null +++ b/engine/evals/routing/runner.py @@ -0,0 +1,200 @@ +"""Run the routing eval and write results to JSON. + + uv run --group engine python evals/routing/runner.py --out evals/routing/results + +Every strategy sees every case. Results are written per call so a run can be re-scored or +re-charted without paying for the model calls again. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import sys +import time +from collections import Counter, defaultdict +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any + +import httpx + +_EVALS_ROOT = Path(__file__).resolve().parents[1] +if str(_EVALS_ROOT) not in sys.path: + sys.path.insert(0, str(_EVALS_ROOT)) +_SRC = _EVALS_ROOT.parent / "src" +if str(_SRC) not in sys.path: + sys.path.insert(0, str(_SRC)) + +from routing.client import DEFAULT_BASE_URL, DEFAULT_MODEL, OllamaRouter # noqa: E402 +from routing.dataset import CASES, Band, RoutingCase, is_destructive_miss # noqa: E402 +from routing.strategies import Strategy, build_strategies # noqa: E402 + + +@dataclass +class Observation: + strategy: str + case_id: str + band: str + expected: str + predicted: str + correct: bool + tolerated: bool + destructive: bool + failed: bool + calls: int + input_tokens: int + output_tokens: int + thinking_chars: int + latency_s: float + repeat: int + errors: list[str] + + +def score(case: RoutingCase, predicted: str) -> tuple[bool, bool, bool, bool]: + """(correct, tolerated, destructive_miss, hard_failure).""" + failed = predicted.startswith("__") + correct = predicted == case.expected + tolerated = (not correct) and predicted in case.tolerated + destructive = (not failed) and is_destructive_miss(case, predicted) + return correct, tolerated, destructive, failed + + +async def run_one( + router: OllamaRouter, + strategy: Strategy, + case: RoutingCase, + repeat: int, + semaphore: asyncio.Semaphore, +) -> Observation: + async with semaphore: + result = await strategy.route(router, case) + correct, tolerated, destructive, failed = score(case, result.predicted) + return Observation( + strategy=strategy.name, + case_id=case.id, + band=str(case.band), + expected=case.expected, + predicted=result.predicted, + correct=correct, + tolerated=tolerated, + destructive=destructive, + failed=failed, + calls=result.budget.calls, + input_tokens=result.budget.input_tokens, + output_tokens=result.budget.output_tokens, + thinking_chars=result.budget.thinking_chars, + latency_s=round(result.budget.latency_s, 3), + repeat=repeat, + errors=result.budget.errors, + ) + + +def summarise(observations: list[Observation]) -> dict[str, Any]: + by_strategy: dict[str, list[Observation]] = defaultdict(list) + for obs in observations: + by_strategy[obs.strategy].append(obs) + + summary: dict[str, Any] = {} + for name, rows in by_strategy.items(): + total = len(rows) + bands: dict[str, dict[str, float]] = {} + for band in Band: + band_rows = [r for r in rows if r.band == str(band)] + if band_rows: + bands[str(band)] = { + "n": len(band_rows), + "accuracy": round(sum(r.correct for r in band_rows) / len(band_rows), 4), + } + confusion = Counter((r.expected, r.predicted) for r in rows if not r.correct) + # Stability: how often repeats of the same case disagree with each other. + grouped: dict[str, set[str]] = defaultdict(set) + for row in rows: + grouped[row.case_id].add(row.predicted) + unstable = sum(1 for preds in grouped.values() if len(preds) > 1) + + summary[name] = { + "n": total, + "accuracy": round(sum(r.correct for r in rows) / total, 4), + "accuracy_with_tolerated": round(sum(r.correct or r.tolerated for r in rows) / total, 4), + "destructive_misroutes": sum(r.destructive for r in rows), + "hard_failures": sum(r.failed for r in rows), + "unstable_cases": unstable, + "cases_seen": len(grouped), + "avg_calls": round(sum(r.calls for r in rows) / total, 3), + "avg_input_tokens": round(sum(r.input_tokens for r in rows) / total, 1), + "avg_output_tokens": round(sum(r.output_tokens for r in rows) / total, 1), + "avg_thinking_chars": round(sum(r.thinking_chars for r in rows) / total, 1), + "avg_latency_s": round(sum(r.latency_s for r in rows) / total, 2), + "p90_latency_s": round(sorted(r.latency_s for r in rows)[int(0.9 * (total - 1))], 2), + "total_latency_s": round(sum(r.latency_s for r in rows), 1), + "by_band": bands, + "top_confusions": [{"expected": e, "predicted": p, "count": c} for (e, p), c in confusion.most_common(8)], + } + return summary + + +async def main() -> None: + parser = argparse.ArgumentParser(description="Routing eval for the orchestrator's capability decision.") + parser.add_argument("--out", default="evals/routing/results", help="Output directory.") + parser.add_argument("--model", default=DEFAULT_MODEL) + parser.add_argument("--base-url", default=DEFAULT_BASE_URL) + parser.add_argument("--repeats", type=int, default=1, help="Runs per case, for stability measurement.") + parser.add_argument("--concurrency", type=int, default=3) + parser.add_argument("--strategies", default="", help="Comma-separated subset of strategy names.") + parser.add_argument("--limit", type=int, default=0, help="Cap the number of cases (smoke tests).") + args = parser.parse_args() + + strategies = build_strategies() + if args.strategies: + wanted = {s.strip() for s in args.strategies.split(",")} + strategies = [s for s in strategies if s.name in wanted] + missing = wanted - {s.name for s in strategies} + if missing: + parser.error(f"Unknown strategies: {sorted(missing)}") + cases = CASES[: args.limit] if args.limit else CASES + + out_dir = Path(args.out) + out_dir.mkdir(parents=True, exist_ok=True) + + semaphore = asyncio.Semaphore(args.concurrency) + started = time.monotonic() + observations: list[Observation] = [] + + async with httpx.AsyncClient() as http: + router = OllamaRouter(http, model=args.model, base_url=args.base_url) + for strategy in strategies: + strategy_started = time.monotonic() + tasks = [ + run_one(router, strategy, case, repeat, semaphore) for repeat in range(args.repeats) for case in cases + ] + results = await asyncio.gather(*tasks) + observations.extend(results) + accuracy = sum(r.correct for r in results) / len(results) + print( + f"{strategy.name:20s} acc={accuracy:6.1%} " + f"destructive={sum(r.destructive for r in results):3d} " + f"fail={sum(r.failed for r in results):3d} " + f"wall={time.monotonic() - strategy_started:6.1f}s", + flush=True, + ) + + payload = { + "model": args.model, + "repeats": args.repeats, + "concurrency": args.concurrency, + "case_count": len(cases), + "wall_clock_s": round(time.monotonic() - started, 1), + "strategy_descriptions": {s.name: s.description for s in strategies}, + "summary": summarise(observations), + } + (out_dir / "summary.json").write_text(json.dumps(payload, indent=2), encoding="utf-8") + with (out_dir / "observations.jsonl").open("w", encoding="utf-8") as handle: + for obs in observations: + handle.write(json.dumps(asdict(obs)) + "\n") + print(f"\nWrote {out_dir / 'summary.json'} and observations.jsonl ({len(observations)} rows)") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/engine/evals/routing/strategies.py b/engine/evals/routing/strategies.py new file mode 100644 index 0000000000..075eda3273 --- /dev/null +++ b/engine/evals/routing/strategies.py @@ -0,0 +1,329 @@ +"""Routing strategies under test. + +``prod_baseline`` reproduces what ships today. Every other strategy is a candidate fix, +implemented so the eval measures the actual change rather than a description of it. +""" + +from __future__ import annotations + +import json +from collections import Counter +from dataclasses import dataclass +from typing import Protocol + +from routing.client import Budget, CallResult, OllamaRouter, enum_schema +from routing.dataset import CAPABILITIES, RoutingCase +from stirling.agents.orchestrator import _ROUTER_SYSTEM_PROMPT +from stirling.contracts import AiFile, ConversationMessage, format_conversation_history, format_file_names + +# Production ceiling for the router's tier (engine/.env STIRLING_FAST_MODEL_MAX_TOKENS). +PROD_MAX_TOKENS = 2048 +# Headroom for the strategies that deliberately let qwen3 think: measured chains run +# 1,400-2,200 characters, and the cap covers reasoning plus content together. +THINK_MAX_TOKENS = 4096 + + +@dataclass +class StrategyResult: + predicted: str + budget: Budget + detail: str = "" + + +class Strategy(Protocol): + name: str + description: str + + async def route(self, router: OllamaRouter, case: RoutingCase) -> StrategyResult: ... + + +def build_user_prompt(case: RoutingCase) -> str: + """The orchestrator's own prompt shape (OrchestratorAgent._build_prompt).""" + history = format_conversation_history([ConversationMessage(role=r, content=c) for r, c in case.history]) + files = format_file_names([AiFile(id=f"id-{name}", name=name) for name in case.files]) + return ( + f"Conversation history:\n{history}\nUser message: {case.message}\nFiles: {files}\nAvailable artifacts:\n- none" + ) + + +def _parse_capability(result: CallResult, allowed: list[str]) -> str: + """Pull the capability out of a constrained-decode response, or report why not.""" + if result.error: + return "__error__" + if result.finish_reason == "length": + # The cap fired mid-generation: pydantic-ai sees invalid output and burns a retry. + return "__truncated__" + try: + parsed = json.loads(result.content) + except ValueError: + return "__unparsable__" + value = parsed.get("capability") + return value if value in allowed else "__offmenu__" + + +# Worded differently from every eval case, so the measured gain is generalisation +# rather than memorisation of the test set. +_FEWSHOT = """ +Decide by what the user wants BACK, not by which PDF words appear in the message. +- They want an ANSWER, or information read out of the document -> pdf_question. +- They want a CHANGED FILE handed back -> pdf_edit. +- A document merely being mentioned is not a request to act on it. + +Examples: + "How long is the notice period?" -> pdf_question + "Shorten the notice period to 30 days" -> pdf_edit + "Is there a table of contents?" -> pdf_question + "Add a table of contents" -> pdf_edit + "Does it say anything about encryption?" -> pdf_question + "Encrypt this file" -> pdf_edit + "What sections cover the merger?" -> pdf_question + "Combine these into one file" -> pdf_edit + "Tell me if the numbers add up" -> pdf_question + "Put a note on every paragraph that needs work" -> pdf_review + "Build me a purchase order from scratch" -> pdf_create + "Make a rule that stamps every upload" -> user_spec + "Which version of the app is this?" -> unsupported +""".strip() + + +class _EnumRouter: + """Single-call enum routing. The knobs are what distinguishes the strategies.""" + + def __init__( + self, + name: str, + description: str, + *, + thinking: bool, + temperature: float | None, + max_tokens: int, + fewshot: bool = False, + narrow_by_files: bool = False, + ) -> None: + self.name = name + self.description = description + self._thinking = thinking + self._temperature = temperature + self._max_tokens = max_tokens + self._fewshot = fewshot + self._narrow = narrow_by_files + + def _allowed(self, case: RoutingCase) -> list[str]: + if self._narrow and not case.files: + # Nothing is attached, so the three capabilities that operate on an input file + # are impossible. Dropping them from the enum makes them undecodable rather + # than merely discouraged. + return ["user_spec", "pdf_create", "unsupported"] + return list(CAPABILITIES) + + def _system_prompt(self) -> str: + return f"{_ROUTER_SYSTEM_PROMPT}\n\n{_FEWSHOT}" if self._fewshot else _ROUTER_SYSTEM_PROMPT + + async def route(self, router: OllamaRouter, case: RoutingCase) -> StrategyResult: + allowed = self._allowed(case) + budget = Budget() + result = await router.call( + self._system_prompt(), + build_user_prompt(case), + response_format=enum_schema("route", allowed, with_message=False), + max_tokens=self._max_tokens, + temperature=self._temperature, + thinking=self._thinking, + ) + budget.add(result) + return StrategyResult(_parse_capability(result, allowed), budget) + + +_STAGE1_SYSTEM = ( + "Decide one thing only: after this turn, does the user expect a FILE back that they did " + "not have before (a changed document, a new document, or the document with annotations " + "added), or do they expect an ANSWER in the chat?\n" + 'Answer "file" or "answer".\n' + "A document being mentioned or described is not a request to produce a file. " + "Questions about what a document contains, means, or whether something is present are " + '"answer". Instructions to change, convert, protect, split, combine, annotate, or ' + 'author a document are "file".' +) + +_STAGE2_FILE_SYSTEM = ( + "The user wants a file back. Choose which one:\n" + "- pdf_edit: change or convert the attached document(s).\n" + "- pdf_create: author a brand new document from scratch.\n" + "- pdf_review: hand the document back with review comments or sticky notes on it.\n" + "- user_spec: define a reusable agent or automation rule, rather than acting once now." +) + +_STAGE2_ANSWER_SYSTEM = ( + "The user wants an answer in chat. Choose which one:\n" + "- pdf_question: the answer comes from reading the attached document(s).\n" + "- unsupported: the question is about the assistant itself, or is unrelated to the " + "attached documents; put a short helpful reply in 'message'." +) + + +class BinaryChain: + """Two narrow decisions instead of one six-way one. + + Costs a second call, but each decision has a much smaller space, and the first one is + exactly the distinction that is being got wrong today. + """ + + name = "binary_chain" + description = "Stage 1: file-or-answer. Stage 2: pick within that half." + + def __init__(self, *, thinking: bool = False, fewshot: bool = True) -> None: + self._thinking = thinking + self._fewshot = fewshot + + async def route(self, router: OllamaRouter, case: RoutingCase) -> StrategyResult: + budget = Budget() + prompt = build_user_prompt(case) + stage1_system = f"{_STAGE1_SYSTEM}\n\n{_FEWSHOT}" if self._fewshot else _STAGE1_SYSTEM + stage1 = await router.call( + stage1_system, + prompt, + response_format={ + "type": "json_schema", + "json_schema": { + "name": "intent", + "strict": True, + "schema": { + "type": "object", + "properties": {"expects": {"type": "string", "enum": ["file", "answer"]}}, + "required": ["expects"], + "additionalProperties": False, + }, + }, + }, + max_tokens=THINK_MAX_TOKENS if self._thinking else PROD_MAX_TOKENS, + temperature=0.0, + thinking=self._thinking, + ) + budget.add(stage1) + if stage1.error or stage1.finish_reason == "length": + return StrategyResult("__error__" if stage1.error else "__truncated__", budget, "stage1") + try: + expects = json.loads(stage1.content).get("expects") + except ValueError: + return StrategyResult("__unparsable__", budget, "stage1") + + if expects == "file": + allowed = ["pdf_edit", "pdf_create", "pdf_review", "user_spec"] + if not case.files: + allowed = ["pdf_create", "user_spec"] + system = _STAGE2_FILE_SYSTEM + else: + allowed = ["pdf_question", "unsupported"] + if not case.files: + allowed = ["unsupported"] + system = _STAGE2_ANSWER_SYSTEM + + if len(allowed) == 1: + return StrategyResult(allowed[0], budget, f"stage1={expects}, stage2 skipped") + + stage2 = await router.call( + system, + prompt, + response_format=enum_schema("route", allowed), + max_tokens=THINK_MAX_TOKENS if self._thinking else PROD_MAX_TOKENS, + temperature=0.0, + thinking=self._thinking, + ) + budget.add(stage2) + return StrategyResult(_parse_capability(stage2, allowed), budget, f"stage1={expects}") + + +class SelfConsistency: + """Sample the same decision N times and take the majority.""" + + def __init__(self, n: int = 3, *, temperature: float = 0.6, thinking: bool = False, fewshot: bool = True) -> None: + self.name = f"vote{n}" + self.description = f"{n} samples at temperature {temperature}, majority wins." + self._n = n + self._temperature = temperature + self._inner = _EnumRouter( + f"vote{n}-inner", + "", + thinking=thinking, + temperature=temperature, + max_tokens=THINK_MAX_TOKENS if thinking else PROD_MAX_TOKENS, + fewshot=fewshot, + ) + + async def route(self, router: OllamaRouter, case: RoutingCase) -> StrategyResult: + budget = Budget() + votes: list[str] = [] + for _ in range(self._n): + single = await self._inner.route(router, case) + budget.calls += single.budget.calls + budget.input_tokens += single.budget.input_tokens + budget.output_tokens += single.budget.output_tokens + budget.thinking_chars += single.budget.thinking_chars + budget.latency_s += single.budget.latency_s + budget.errors.extend(single.budget.errors) + votes.append(single.predicted) + valid = [v for v in votes if v in CAPABILITIES] + if not valid: + return StrategyResult(votes[0] if votes else "__error__", budget, f"votes={votes}") + winner, _ = Counter(valid).most_common(1)[0] + return StrategyResult(winner, budget, f"votes={votes}") + + +def build_strategies() -> list[Strategy]: + return [ + _EnumRouter( + "prod_baseline", + "Exactly what ships today: 6-way enum, no temperature set, 2048 cap, thinking left on.", + thinking=True, + temperature=None, + max_tokens=PROD_MAX_TOKENS, + ), + _EnumRouter( + "temp0", + "Production prompt, temperature pinned to 0.", + thinking=True, + temperature=0.0, + max_tokens=PROD_MAX_TOKENS, + ), + _EnumRouter( + "thinking_off", + "Temperature 0 and reasoning_effort=none.", + thinking=False, + temperature=0.0, + max_tokens=PROD_MAX_TOKENS, + ), + _EnumRouter( + "think_budget", + "Thinking on, temperature 0, cap raised to 4096 so the chain cannot be truncated.", + thinking=True, + temperature=0.0, + max_tokens=THINK_MAX_TOKENS, + ), + _EnumRouter( + "fewshot_thinking_off", + "Few-shot boundary examples, thinking off.", + thinking=False, + temperature=0.0, + max_tokens=PROD_MAX_TOKENS, + fewshot=True, + ), + _EnumRouter( + "fewshot_think", + "Few-shot boundary examples, thinking on with a 4096 cap.", + thinking=True, + temperature=0.0, + max_tokens=THINK_MAX_TOKENS, + fewshot=True, + ), + _EnumRouter( + "fewshot_narrowed", + "Few-shot, thinking off, plus enum narrowed by whether a file is attached.", + thinking=False, + temperature=0.0, + max_tokens=PROD_MAX_TOKENS, + fewshot=True, + narrow_by_files=True, + ), + BinaryChain(thinking=False, fewshot=True), + SelfConsistency(3, temperature=0.6, thinking=False, fewshot=True), + ]