"""Intent-routing eval runner (E3). Feeds each golden case in `intent_dataset.json` to the live 6-intent router (`OrchestratorAgent.classify`), then scores correctness + records latency and token usage. Prints a per-case detail table and an aggregate summary, and writes a timestamped JSON report under `results/` (never overwritten — one file per run, so runs can be diffed over time). Run before every deploy that touches the router prompt or its few-shots. Invoke as a module (`-m`) so the repo root is on `sys.path` and `src` imports resolve — running the file path directly (`python eval/intent/run_eval.py`) puts only `eval/intent/` on the path and fails: uv run python -m eval.intent.run_eval uv run python -m eval.intent.run_eval --limit 6 # quick smoke test Tokens come straight from the model response (LangChain `usage_metadata` via a callback) — no Langfuse needed. The router is called unmodified: it already accepts a `callbacks=` list and forwards it into the chain config. """ from __future__ import annotations import argparse import asyncio import json import statistics import time from dataclasses import asdict, dataclass from datetime import datetime from pathlib import Path from typing import Any from langchain_core.callbacks import BaseCallbackHandler from langchain_core.outputs import LLMResult from src.agents.orchestration import OrchestratorAgent _HERE = Path(__file__).resolve().parent DATASET = _HERE / "intent_dataset.json" RESULTS_DIR = _HERE / "results" INTENTS = [ "chat", "help", "check", "unstructured_flow", "structured_flow", "out_of_scope", ] # Short labels so the EXPECT->GOT column stays narrow in the detail table. _ABBR = { "chat": "chat", "help": "help", "check": "check", "unstructured_flow": "unstruct", "structured_flow": "structF", "out_of_scope": "oos", "blocked": "blocked", } def _is_content_filter_error(err: Exception) -> bool: """True when an exception is Azure's content-filter / jailbreak rejection. Mirrors `chat_handler._is_content_filter_error` (string-match, not an import of the concrete openai error type, so it survives SDK/version changes). We keep a local copy rather than importing from `src.agents.chat_handler` to avoid pulling the whole handler's import graph into the eval runner. """ s = str(err).lower() return ( "content_filter" in s or "responsibleai" in s or "jailbreak" in s or "content management policy" in s ) class _UsageCollector(BaseCallbackHandler): """Sums token usage across the LLM calls made during one classify(). Reads `usage_metadata` off each returned message (the canonical LangChain field), falling back to `llm_output['token_usage']` for providers that only populate the legacy field. """ def __init__(self) -> None: self.input_tokens = 0 self.output_tokens = 0 self.total_tokens = 0 def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None: before = self.total_tokens for generation_list in response.generations: for generation in generation_list: message = getattr(generation, "message", None) usage = getattr(message, "usage_metadata", None) if message else None if usage: self.input_tokens += usage.get("input_tokens", 0) self.output_tokens += usage.get("output_tokens", 0) self.total_tokens += usage.get("total_tokens", 0) if self.total_tokens == before and response.llm_output: usage = response.llm_output.get("token_usage") or {} self.input_tokens += usage.get("prompt_tokens", 0) self.output_tokens += usage.get("completion_tokens", 0) self.total_tokens += usage.get("total_tokens", 0) @property def tokens(self) -> dict[str, int]: return { "input": self.input_tokens, "output": self.output_tokens, "total": self.total_tokens, } @dataclass class CaseResult: id: str lang: str message: str expected: str got: str correct: bool latency_ms: int tokens: dict[str, int] def load_cases(path: Path) -> list[dict[str, Any]]: """Read the `cases` array, skipping the leading `_*` doc keys and `schema`.""" data = json.loads(path.read_text(encoding="utf-8")) return list(data["cases"]) @dataclass class _LangfuseCtx: """Optional Langfuse sink — one session groups all cases of a run.""" session_id: str client: Any def _new_langfuse_handler(lf_ctx: _LangfuseCtx, case: dict[str, Any]) -> Any: """Per-case LangChain callback so each trace carries the case's labels.""" from langfuse.callback import CallbackHandler from src.config.settings import settings return CallbackHandler( public_key=settings.LANGFUSE_PUBLIC_KEY, secret_key=settings.LANGFUSE_SECRET_KEY, host=settings.LANGFUSE_HOST, session_id=lf_ctx.session_id, trace_name=f"intent_eval/{case['id']}", metadata={ "case_id": case["id"], "expected": case["expected_intent"], "lang": case["lang"], }, tags=["intent-eval", case["expected_intent"], case["lang"]], ) def _score_langfuse(lf_ctx: _LangfuseCtx, handler: Any, result: CaseResult) -> None: """Attach a 1/0 correctness score to the case's trace. Best-effort.""" try: lf_ctx.client.score( trace_id=handler.get_trace_id(), name="intent_correct", value=1 if result.correct else 0, comment=f"{result.expected} -> {result.got}", ) except Exception: # noqa: BLE001, S110 — scoring must never break the run pass async def run_case( agent: OrchestratorAgent, case: dict[str, Any], lf_ctx: _LangfuseCtx | None = None, ) -> CaseResult: """Classify one message; never throws. A raised exception is recorded as `ERROR:` and scored wrong — EXCEPT Azure's content-filter / jailbreak rejection on an `out_of_scope` case. That 400 is the *correct* end-to-end outcome: the real app catches it and returns a clean refusal (see `chat_handler._is_content_filter_error`), so the platform guardrail firing IS the desired `out_of_scope` behaviour. We record it as `blocked` and score it correct, keeping `out_of_scope` accuracy honest instead of penalising the router for inputs the guardrail caught before the model saw them. A content-filter block on any *other* expected intent is still a mismatch (unexpected block). """ collector = _UsageCollector() callbacks: list[Any] = [collector] lf_handler = _new_langfuse_handler(lf_ctx, case) if lf_ctx else None if lf_handler is not None: callbacks.append(lf_handler) expected = case["expected_intent"] start = time.perf_counter() got: str correct: bool try: decision = await agent.classify(case["message"], callbacks=callbacks) got = decision.intent correct = got == expected except Exception as exc: # noqa: BLE001 — one bad case shouldn't kill the run if _is_content_filter_error(exc): got = "blocked" correct = expected == "out_of_scope" else: got = f"ERROR:{type(exc).__name__}" correct = False latency_ms = round((time.perf_counter() - start) * 1000) result = CaseResult( id=case["id"], lang=case["lang"], message=case["message"], expected=expected, got=got, correct=correct, latency_ms=latency_ms, tokens=collector.tokens, ) if lf_ctx is not None and lf_handler is not None: _score_langfuse(lf_ctx, lf_handler, result) return result def _group_accuracy(results: list[CaseResult], key: str) -> dict[str, dict[str, Any]]: out: dict[str, dict[str, Any]] = {} keys = INTENTS if key == "expected" else sorted({getattr(r, key) for r in results}) for k in keys: sub = [r for r in results if getattr(r, key) == k] if not sub: continue passed = sum(r.correct for r in sub) out[k] = { "n": len(sub), "passed": passed, "accuracy": round(passed / len(sub), 3), } return out def summarize(results: list[CaseResult]) -> dict[str, Any]: n = len(results) passed = sum(r.correct for r in results) latencies = [r.latency_ms for r in results] tok_in = sum(r.tokens["input"] for r in results) tok_out = sum(r.tokens["output"] for r in results) tok_total = sum(r.tokens["total"] for r in results) return { "total": n, "passed": passed, "accuracy": round(passed / n, 3) if n else 0.0, "runtime_avg_ms": round(statistics.mean(latencies)) if latencies else 0, "runtime_total_s": round(sum(latencies) / 1000, 1), "tokens": { "input": tok_in, "output": tok_out, "total": tok_total, "avg_total_per_case": round(tok_total / n) if n else 0, }, "by_intent": _group_accuracy(results, "expected"), "by_lang": _group_accuracy(results, "lang"), } def _truncate(text: str, width: int) -> str: text = text.replace("\n", " ") return text if len(text) <= width else text[: width - 3] + "..." def format_table(results: list[CaseResult]) -> str: header = ( f"{'ID':<15} {'L':<3} {'QUESTION':<40} " f"{'EXPECT->GOT':<20} {'OK':<3} {'MS':>5} {'TOK':>6}" ) rule = "-" * len(header) lines = [rule, header, rule] for r in results: exp_got = f"{_ABBR.get(r.expected, r.expected)}->{_ABBR.get(r.got, r.got)}" ok = "ok" if r.correct else "X" lines.append( f"{r.id:<15} {r.lang:<3} {_truncate(r.message, 40):<40} " f"{_truncate(exp_got, 20):<20} {ok:<3} {r.latency_ms:>5} {r.tokens['total']:>6}" ) lines.append(rule) return "\n".join(lines) def format_summary(summary: dict[str, Any], results: list[CaseResult]) -> str: lines = ["SUMMARY"] lines.append( f" Overall {summary['passed']}/{summary['total']} correct" f" ({summary['accuracy'] * 100:.1f}%)" ) lines.append( f" Runtime avg {summary['runtime_avg_ms']} ms" f" | total {summary['runtime_total_s']} s" ) tok = summary["tokens"] lines.append( f" Tokens avg {tok['avg_total_per_case']}" f" | total {tok['total']} (in {tok['input']} / out {tok['output']})" ) lines.append("") lines.append(" By intent") for intent, m in summary["by_intent"].items(): lines.append( f" {intent:<18} {m['passed']}/{m['n']} {m['accuracy'] * 100:.0f}%" ) lines.append(" By language") for lang, m in summary["by_lang"].items(): lines.append( f" {lang:<18} {m['passed']}/{m['n']} {m['accuracy'] * 100:.0f}%" ) failures = [r for r in results if not r.correct] lines.append("") lines.append(f" FAILURES ({len(failures)})") for r in failures: lines.append(f" {r.id:<14} [{r.lang}] {r.expected:<12} -> {r.got}") return "\n".join(lines) def build_report( results: list[CaseResult], summary: dict[str, Any], meta: dict[str, Any] ) -> dict[str, Any]: run = {**meta, **{k: summary[k] for k in ( "total", "passed", "accuracy", "runtime_avg_ms", "runtime_total_s", "tokens" )}} return { "run": run, "by_intent": summary["by_intent"], "by_lang": summary["by_lang"], "cases": [asdict(r) for r in results], } def _model_name() -> str: try: from src.config.settings import settings return str(settings.azureai_deployment_name_4o) except Exception: # noqa: BLE001 — meta only; .env may be absent return "gpt-4o" async def main() -> None: parser = argparse.ArgumentParser(description="Intent-routing eval (E3)") parser.add_argument("--dataset", type=Path, default=DATASET) parser.add_argument("--limit", type=int, default=0, help="run first N cases only") parser.add_argument("--prompt-version", default="intent_router.md") parser.add_argument("--no-table", action="store_true", help="skip the detail table") parser.add_argument( "--langfuse", action="store_true", help="also send each case as a Langfuse trace + correctness score", ) args = parser.parse_args() cases = load_cases(args.dataset) if args.limit: cases = cases[: args.limit] started = datetime.now() print(f"Intent Routing Eval -- {started:%Y-%m-%d %H:%M:%S}") print(f"dataset: {args.dataset.name} ({len(cases)}) model: {_model_name()} " f"prompt: {args.prompt_version}") lf_ctx: _LangfuseCtx | None = None if args.langfuse: try: from src.observability.langfuse.langfuse import get_langfuse lf_ctx = _LangfuseCtx( session_id=f"intent_eval_{started:%Y%m%d_%H%M%S}", client=get_langfuse(), # type: ignore[no-untyped-call] ) print(f"langfuse: enabled (session {lf_ctx.session_id})") except Exception as exc: # noqa: BLE001 — Langfuse is optional print(f"langfuse: disabled ({type(exc).__name__}: {exc})") agent = OrchestratorAgent() results: list[CaseResult] = [] for case in cases: results.append(await run_case(agent, case, lf_ctx)) if lf_ctx is not None: try: lf_ctx.client.flush() except Exception: # noqa: BLE001, S110 — flush failure shouldn't fail the run pass summary = summarize(results) if not args.no_table: print(format_table(results)) print(format_summary(summary, results)) meta = { "timestamp": started.isoformat(timespec="seconds"), "dataset": args.dataset.name, "model": _model_name(), "prompt_version": args.prompt_version, "langfuse_session": lf_ctx.session_id if lf_ctx else None, } report = build_report(results, summary, meta) RESULTS_DIR.mkdir(parents=True, exist_ok=True) out_path = RESULTS_DIR / f"eval_result_{started:%Y-%m-%d_%H%M%S}.json" out_path.write_text( json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8" ) print(f"\n-> saved: {out_path.relative_to(_HERE.parent.parent)}") if __name__ == "__main__": asyncio.run(main())