diff --git a/EXTENDING.md b/EXTENDING.md index 2dbd38b..375b19d 100644 --- a/EXTENDING.md +++ b/EXTENDING.md @@ -78,6 +78,16 @@ MCP URL, a plain streamable-HTTP URL) or a skill is a registry write: portal UI, API call, or a new entry in `seed_data.py`. No mechanism changes, nothing to rebuild. +### Add an application-effect dashboard scenario — data only + +Create an Evaluation dataset with a custom `scenario` name and `scoring` +configuration. Use `json_exact` with an `output_field` (including a dotted +field such as `decision.intent`) for deterministic classification, or +`llm_judge` with a business rubric for answer quality. Once the dataset runs, +Observability discovers the scenario and charts its actual results. See +[docs/observability.md](docs/observability.md#application-effect-dashboard) +for an API example and the code touch points for adding a new scoring method. + ### Add an integration *kind* — a known multi-file change The registry classifies each MCP entry by `kind` (`agentcore-runtime` | diff --git a/backend/app/api/evals.py b/backend/app/api/evals.py index b94d835..8c2ed01 100644 --- a/backend/app/api/evals.py +++ b/backend/app/api/evals.py @@ -1,6 +1,6 @@ """Evaluation endpoints.""" -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends, HTTPException, Query from app.dependencies import get_current_user, require_admin from app.models.schemas import EvalDatasetCreateRequest, EvalRunRequest @@ -22,6 +22,9 @@ def create_dataset(req: EvalDatasetCreateRequest, user: str = Depends(get_curren user=user, name=req.name, description=req.description, + scenario=req.scenario, + scoring=req.scoring.model_dump() if req.scoring else None, + synthetic=req.synthetic, cases=[c.model_dump() for c in req.cases], ) except ValueError as e: @@ -30,6 +33,16 @@ def create_dataset(req: EvalDatasetCreateRequest, user: str = Depends(get_curren return ds +@router.get("/datasets/{dataset_id}/runs") +def list_dataset_runs( + dataset_id: str, + limit: int = Query(20, ge=1, le=50), + user: str = Depends(get_current_user), +): + """Run summaries of one dataset, newest first (for version comparison).""" + return eval_service.list_dataset_runs(dataset_id, limit) + + @router.delete("/datasets/{dataset_id}") def delete_dataset(dataset_id: str, user: str = Depends(get_current_user)): if not eval_service.delete_dataset(dataset_id): @@ -39,8 +52,11 @@ def delete_dataset(dataset_id: str, user: str = Depends(get_current_user)): @router.get("/runs") -def list_runs(user: str = Depends(get_current_user)): - return eval_service.list_runs() +def list_runs( + limit: int = Query(20, ge=1, le=50), + user: str = Depends(get_current_user), +): + return eval_service.list_runs(limit) @router.get("/runs/{run_id}") @@ -57,6 +73,6 @@ async def start_run(req: EvalRunRequest, user: str = Depends(get_current_user)): try: run = eval_service.start_run(user=user, dataset_id=req.dataset_id, target=req.target) except KeyError: - raise HTTPException(status_code=404, detail="Dataset not found") + raise HTTPException(status_code=404, detail="Dataset or agent not found") audit_service.record(user, "eval.run.start", f"run:{run['id']}", f"dataset {run['dataset_name']} → {req.target}") return run diff --git a/backend/app/models/schemas.py b/backend/app/models/schemas.py index 5737c80..9573b5b 100644 --- a/backend/app/models/schemas.py +++ b/backend/app/models/schemas.py @@ -1,5 +1,7 @@ """API request/response models.""" +from typing import Literal + from pydantic import BaseModel, Field @@ -183,9 +185,22 @@ class EvalCase(BaseModel): expected: str = Field(default="", max_length=1000) +class EvalScoringConfig(BaseModel): + method: Literal["json_exact", "llm_judge"] = "llm_judge" + output_field: str = Field( + default="category", + max_length=80, + pattern=r"^[A-Za-z_][A-Za-z0-9_]*(\.[A-Za-z_][A-Za-z0-9_]*){0,3}$", + ) + rubric: str = Field(default="", max_length=1000) + + class EvalDatasetCreateRequest(BaseModel): name: str = Field(min_length=1, max_length=120) description: str = Field(default="", max_length=400) + scenario: str = Field(default="general", max_length=48, pattern=r"^[a-z][a-z0-9_-]*$") + scoring: EvalScoringConfig | None = None + synthetic: bool = False cases: list[EvalCase] = Field(min_length=1, max_length=20) diff --git a/backend/app/services/eval_service.py b/backend/app/services/eval_service.py index 0affd75..b7214d0 100644 --- a/backend/app/services/eval_service.py +++ b/backend/app/services/eval_service.py @@ -1,11 +1,33 @@ -"""Evaluation: fixed task suites scored by an LLM judge. +"""Evaluation: fixed task suites with dataset-selected scoring. Datasets (``PK=EVAL``) hold up to :data:`MAX_CASES` cases inline — each a prompt plus free-text expectation. A *run* executes every case against a -chosen target (kernel or published agent) through the invocation pipeline, -then asks the same headless kernel to act as a strict judge, scoring each -answer against the expectation. Runs (``PK=EVALRUN``) update progressively so -the portal can poll while a run executes in the background. +chosen target (kernel or published agent) through the invocation pipeline. +JSON field datasets use deterministic exact-match scoring; other datasets ask +the same headless kernel to act as a strict judge. + +Storage keeps every item far below DynamoDB's 400 KB limit: + +* ``PK=EVALRUN, SK=#`` — the run summary (status, counts, + agent version). Small, so listing recent runs stays one cheap Query page. +* ``PK=EVALRUN#`` — the run's own partition: ``SK=META`` points back + at the summary's sort key (lookup by id without a scan) and + ``SK=CASE#`` holds one case's full evidence (prompt, expectation, + answer, verdict). Cases are written once each, never rewritten. +* ``PK=AGENTPROMPT, SK=#V`` — the system prompt of a + published agent version, written once and shared by every run of it. +* ``PK=EVALDS#, SK=#`` — one pointer per run, + so a dataset's full run history is one Query however many other runs exist. + +Each case also records the infrastructure view of its own calls (ok, +duration, turns, cost, runtime session id for the agent call and the judge +call), and the summary aggregates them when the run ends. Outcome and +execution health of the same calls therefore sit on the same record, with no +join against the invocation ledger; the session id links a case to its +kernel spans. + +Runs written before this layout carry ``results`` and ``system_prompt`` on +the summary item; :meth:`EvalService.get_run` still reads them. Judging with the platform's own kernel keeps the sample dependency-free; the judge system prompt pins the output to a JSON verdict for parsing. @@ -15,6 +37,8 @@ import json import logging import re +import statistics +import time import uuid from datetime import datetime, timezone from decimal import Decimal @@ -27,7 +51,65 @@ PK_DS = "EVAL" PK_RUN = "EVALRUN" +PK_PROMPT = "AGENTPROMPT" MAX_CASES = 20 +MAX_ANSWER_CHARS = 8000 + + +def _run_pk(run_id: str) -> str: + return f"{PK_RUN}#{run_id}" + + +def _dataset_runs_pk(dataset_id: str) -> str: + return f"EVALDS#{dataset_id}" + + +def _plain(value): + """DynamoDB numbers come back as Decimal; return JSON-friendly numbers.""" + if isinstance(value, Decimal): + return int(value) if value == value.to_integral_value() else float(value) + if isinstance(value, dict): + return {k: _plain(v) for k, v in value.items()} + if isinstance(value, list): + return [_plain(v) for v in value] + return value + + +def _call_metrics(res: dict, started: float) -> dict: + """The execution facts of one invocation, as the ledger would record them.""" + usage = res.get("usage") or {} + cost = usage.get("total_cost_usd") + turns = usage.get("num_turns") + return { + "ok": bool(res.get("ok")), + "duration_ms": int(usage.get("duration_ms") or (time.monotonic() - started) * 1000), + "num_turns": int(turns) if turns is not None else None, + "cost_usd": Decimal(str(cost)) if cost is not None else None, + "runtime_session_id": str(res.get("runtime_session_id") or ""), + } + + +def _calls_summary(results: list[dict]) -> dict: + """Aggregate the per-case call metrics onto the run summary.""" + agent = [r["agent_call"] for r in results if r.get("agent_call")] + judge = [r["judge_call"] for r in results if r.get("judge_call")] + + def cost(calls: list[dict]) -> Decimal: + return sum((c["cost_usd"] for c in calls if c.get("cost_usd") is not None), Decimal(0)) + + return { + "agent_calls": len(agent), + "agent_ok": sum(1 for c in agent if c["ok"]), + "agent_duration_p50_ms": int(statistics.median(c["duration_ms"] for c in agent)) if agent else None, + "agent_cost_usd": cost(agent), + "judge_calls": len(judge), + "judge_ok": sum(1 for c in judge if c["ok"]), + "judge_cost_usd": cost(judge), + } + + +def _prompt_sk(agent_id: str, version: int) -> str: + return f"{agent_id}#V{int(version):06d}" JUDGE_SYSTEM = ( "You are a strict evaluation judge. You receive a task prompt, the expected " @@ -47,8 +129,10 @@ def _parse_verdict(text: str) -> dict: if match: try: v = json.loads(match.group(0)) + if not isinstance(v, dict) or not isinstance(v.get("pass"), bool): + raise TypeError("judge pass must be a boolean") return { - "pass": bool(v.get("pass")), + "pass": v["pass"], "score": max(0, min(10, int(v.get("score", 0)))), "reason": str(v.get("reason", ""))[:300], } @@ -57,6 +141,53 @@ def _parse_verdict(text: str) -> dict: return {"pass": False, "score": 0, "reason": f"unparseable judge output: {text[:120]}"} +def _default_scoring(scenario: str) -> dict: + # Existing datasets predate scoring configuration. + return { + "method": "json_exact" if scenario == "classification" else "llm_judge", + "output_field": "category", + "rubric": "", + } + + +def _parse_json_field(text: str, field: str) -> str: + """Read one bounded dotted field from a JSON object returned by the agent.""" + cleaned = text.strip() + if cleaned.startswith("```"): + cleaned = re.sub(r"^```(?:json)?\s*|\s*```$", "", cleaned, flags=re.IGNORECASE).strip() + try: + value = json.loads(cleaned) + except json.JSONDecodeError: + return "" + for part in field.split("."): + if not isinstance(value, dict): + return "" + value = value.get(part) + return value.strip()[:120] if isinstance(value, str) else "" + + +def _parse_category(text: str) -> str: + """Compatibility helper for the original classifier contract.""" + return _parse_json_field(text, "category") + + +def _validated_scoring(scenario: str, scoring: dict | None) -> dict: + config = {**_default_scoring(scenario), **(scoring or {})} + if config["method"] not in ("json_exact", "llm_judge"): + raise ValueError("unknown scoring method") + if not isinstance(config["output_field"], str) or not re.fullmatch( + r"[A-Za-z_][A-Za-z0-9_]*(\.[A-Za-z_][A-Za-z0-9_]*){0,3}", config["output_field"] + ): + raise ValueError("invalid scoring output_field") + if not isinstance(config["rubric"], str) or len(config["rubric"]) > 1000: + raise ValueError("invalid scoring rubric") + return { + "method": config["method"], + "output_field": config["output_field"], + "rubric": config["rubric"], + } + + class EvalService: def __init__(self) -> None: dynamodb = boto3.resource("dynamodb", region_name=settings.aws_region) @@ -74,6 +205,9 @@ def _ds_public(item: dict) -> dict: "id": item["SK"].partition("#")[2], "name": item.get("name", ""), "description": item.get("description", ""), + "scenario": item.get("scenario", "general"), + "scoring": item.get("scoring") or _default_scoring(item.get("scenario", "general")), + "synthetic": bool(item.get("synthetic", False)), "cases": item.get("cases", []), "created_by": item.get("created_by", ""), "created_at": item.get("created_at", ""), @@ -90,7 +224,14 @@ def list_datasets(self) -> list[dict]: reverse=True, ) - def create_dataset(self, *, user: str, name: str, description: str, cases: list[dict]) -> dict: + def create_dataset( + self, *, user: str, name: str, description: str, + cases: list[dict], scenario: str = "general", synthetic: bool = False, + scoring: dict | None = None, + ) -> dict: + if not re.fullmatch(r"[a-z][a-z0-9_-]{0,47}", scenario): + raise ValueError("invalid evaluation scenario") + scoring = _validated_scoring(scenario, scoring) cleaned = [ {"prompt": str(c.get("prompt", ""))[:2000], "expected": str(c.get("expected", ""))[:1000]} for c in cases[:MAX_CASES] @@ -103,6 +244,9 @@ def create_dataset(self, *, user: str, name: str, description: str, cases: list[ "SK": f"DS#{uuid.uuid4().hex[:12]}", "name": name[:120] or "dataset", "description": description[:400], + "scenario": scenario, + "scoring": scoring, + "synthetic": synthetic, "cases": cleaned, "created_by": user, "created_at": _now(), @@ -120,39 +264,133 @@ def delete_dataset(self, dataset_id: str) -> bool: # ---------------------------------------------------------------- runs @staticmethod - def _run_public(item: dict) -> dict: + def _run_public(item: dict, results: list[dict] | None = None, system_prompt: str | None = None) -> dict: + """Summary fields always; ``results``/``system_prompt`` only when given. + + Legacy items (pre per-case storage) carry both inline; they are used + when the caller does not pass the new-layout values. + """ + inline = item.get("results", []) return { "id": item.get("run_id", ""), "dataset_id": item.get("dataset_id", ""), "dataset_name": item.get("dataset_name", ""), + "scenario": item.get("scenario", "general"), + "scoring": item.get("scoring") or _default_scoring(item.get("scenario", "general")), + "synthetic": bool(item.get("synthetic", False)), "target": item.get("target", ""), + "agent_version": int(item["agent_version"]) if item.get("agent_version") is not None else None, + "system_prompt": system_prompt if system_prompt is not None else item.get("system_prompt", ""), "status": item.get("status", ""), "started_by": item.get("started_by", ""), "started_at": item.get("started_at", ""), "finished_at": item.get("finished_at", ""), - "results": item.get("results", []), + "results": results if results is not None else inline, + "evaluated": int(item["evaluated"]) if "evaluated" in item else len(inline), "passed": int(item.get("passed", 0)), "total": int(item.get("total", 0)), "avg_score": float(item["avg_score"]) if "avg_score" in item else None, + "calls": _plain(item["calls"]) if item.get("calls") else None, "error": item.get("error", ""), } def list_runs(self, limit: int = 20) -> list[dict]: + """Recent run summaries, newest first. Case evidence and the system + prompt are not included; fetch one run with :meth:`get_run`.""" resp = self.table.query( KeyConditionExpression="PK = :pk", ExpressionAttributeValues={":pk": PK_RUN}, ScanIndexForward=False, Limit=min(limit, 50), ) - return [self._run_public(i) for i in resp.get("Items", [])] + return [self._run_public(i, results=[], system_prompt="") for i in resp.get("Items", [])] def get_run(self, run_id: str) -> dict | None: - # SK is ts#id; scan the recent window for the id (sample-scale volumes) - for run in self.list_runs(50): - if run["id"] == run_id: - return run + """One run with every case's evidence and the agent's system prompt.""" + meta = self.table.get_item(Key={"PK": _run_pk(run_id), "SK": "META"}).get("Item") + if not meta: + return self._get_legacy_run(run_id) + item = self.table.get_item(Key={"PK": PK_RUN, "SK": meta["run_sk"]}).get("Item") + if not item: + return None + return self._run_public( + item, results=self._case_rows(run_id), system_prompt=self._system_prompt_for(item), + ) + + def list_dataset_runs(self, dataset_id: str, limit: int = 20) -> list[dict]: + """Run summaries of one dataset, newest first, independent of how many + runs other datasets have had since.""" + resp = self.table.query( + KeyConditionExpression="PK = :pk", + ExpressionAttributeValues={":pk": _dataset_runs_pk(dataset_id)}, + ScanIndexForward=False, + Limit=min(limit, 50), + ) + out = [] + for pointer in resp.get("Items", []): + item = self.table.get_item(Key={"PK": PK_RUN, "SK": pointer["SK"]}).get("Item") + if item: + out.append(self._run_public(item, results=[], system_prompt="")) + return out + + def _get_legacy_run(self, run_id: str) -> dict | None: + # Runs written before per-case storage have no META pointer; their + # results sit on the summary item. Look in the recent window only. + resp = self.table.query( + KeyConditionExpression="PK = :pk", + ExpressionAttributeValues={":pk": PK_RUN}, + ScanIndexForward=False, + Limit=50, + ) + for item in resp.get("Items", []): + if item.get("run_id") == run_id: + return self._run_public(item) return None + def _case_rows(self, run_id: str) -> list[dict]: + rows: list[dict] = [] + kwargs = { + "KeyConditionExpression": "PK = :pk AND begins_with(SK, :p)", + "ExpressionAttributeValues": {":pk": _run_pk(run_id), ":p": "CASE#"}, + } + while True: + resp = self.table.query(**kwargs) + for item in resp.get("Items", []): + rows.append(_plain({k: v for k, v in item.items() if k not in ("PK", "SK")})) + if "LastEvaluatedKey" not in resp: + return sorted(rows, key=lambda r: r["case"]) + kwargs["ExclusiveStartKey"] = resp["LastEvaluatedKey"] + + def _system_prompt_for(self, item: dict) -> str: + target = item.get("target", "") + version = item.get("agent_version") + if not target.startswith("agent:") or version is None: + return item.get("system_prompt", "") + snap = self.table.get_item( + Key={"PK": PK_PROMPT, "SK": _prompt_sk(target.partition(":")[2], version)} + ).get("Item") + return (snap or {}).get("system_prompt", item.get("system_prompt", "")) + + def _snapshot_prompt(self, agent_id: str, version: int, system_prompt: str) -> None: + """Store a version's system prompt once; later runs of it reuse it.""" + from botocore.exceptions import ClientError + + try: + self.table.put_item( + Item={ + "PK": PK_PROMPT, + "SK": _prompt_sk(agent_id, version), + "agent_id": agent_id, + "version": int(version), + "system_prompt": system_prompt, + "captured_at": _now(), + }, + ConditionExpression="attribute_not_exists(SK)", + ) + except ClientError as e: + if e.response.get("Error", {}).get("Code") != "ConditionalCheckFailedException": + raise + def _update_run(self, sk: str, **fields) -> None: # Alias every attribute name: "status" and "error" are DynamoDB # reserved words and would fail a bare UpdateExpression. @@ -176,36 +414,66 @@ def start_run(self, *, user: str, dataset_id: str, target: str) -> dict: raise KeyError("dataset not found") run_id = uuid.uuid4().hex[:12] sk = f"{_now()}#{run_id}" + agent_version = None + system_prompt = "" + if target.startswith("agent:"): + from app.services.agent_service import agent_service + + agent = agent_service.get_agent(target.partition(":")[2]) + if not agent: + raise KeyError("agent not found") + agent_version = agent["version"] + system_prompt = agent["system_prompt"] + self._snapshot_prompt(target.partition(":")[2], agent_version, system_prompt) item = { "PK": PK_RUN, "SK": sk, "run_id": run_id, "dataset_id": dataset_id, "dataset_name": ds.get("name", ""), + "scenario": ds.get("scenario", "general"), + "scoring": ds.get("scoring") or _default_scoring(ds.get("scenario", "general")), + "synthetic": bool(ds.get("synthetic", False)), "target": target, + "agent_version": agent_version, "status": "running", "started_by": user, "started_at": _now(), - "results": [], + "evaluated": 0, "passed": 0, "total": len(ds.get("cases", [])), } self.table.put_item(Item=item) + self.table.put_item(Item={"PK": _run_pk(run_id), "SK": "META", "run_sk": sk}) + self.table.put_item(Item={"PK": _dataset_runs_pk(dataset_id), "SK": sk, "run_id": run_id}) # fire-and-forget; progress lands in DDB. Requires a running event # loop, so the API route calling this must be ``async def`` (sync # routes run in a threadpool thread with no loop). - task = asyncio.get_running_loop().create_task(self._execute(sk, user, ds, target)) + task = asyncio.get_running_loop().create_task( + self._execute(sk, user, ds, target, agent_version=agent_version) + ) self._tasks.add(task) task.add_done_callback(self._tasks.discard) - return self._run_public(item) + return self._run_public(item, results=[], system_prompt=system_prompt) - async def _execute(self, sk: str, user: str, ds: dict, target: str) -> None: + async def _execute( + self, sk: str, user: str, ds: dict, target: str, agent_version: int | None = None, + ) -> None: from app.services.invocation_service import invoke # avoid import cycle run_id = sk.partition("#")[2] + scenario = ds.get("scenario", "general") + scoring = ds.get("scoring") or _default_scoring(scenario) results: list[dict] = [] try: for idx, case in enumerate(ds.get("cases", [])): + if agent_version is not None: + from app.services.agent_service import agent_service + + current = agent_service.get_agent(target.partition(":")[2]) + if not current or current["version"] != agent_version: + raise RuntimeError("agent version changed during evaluation") + started = time.monotonic() answer_res = await asyncio.to_thread( invoke, user=user, @@ -214,38 +482,79 @@ async def _execute(self, sk: str, user: str, ds: dict, target: str) -> None: prompt=case["prompt"], ref=f"eval:{run_id}", ) + agent_call = _call_metrics(answer_res, started) + judge_call = None + if agent_version is not None: + current = agent_service.get_agent(target.partition(":")[2]) + if not current or current["version"] != agent_version: + raise RuntimeError("agent version changed during evaluation") answer = (answer_res.get("result") or "").strip() - judge_prompt = ( - f"Task prompt:\n{case['prompt']}\n\n" - f"Expected outcome:\n{case['expected'] or '(none given — judge general correctness)'}\n\n" - f"Candidate answer:\n{answer or '(empty answer)'}" - ) - judge_res = await asyncio.to_thread( - invoke, - user=user, - source="eval", - target="agent-sdk", - prompt=judge_prompt, - system=JUDGE_SYSTEM, - max_turns=1, - ref=f"eval-judge:{run_id}", - ) - verdict = _parse_verdict(judge_res.get("result") or "") - results.append( - { - "case": idx, - "prompt": case["prompt"][:200], - "expected": case["expected"][:200], - "answer": answer[:500], - "pass": verdict["pass"], - "score": verdict["score"], - "reason": verdict["reason"], + if scoring["method"] == "json_exact": + predicted = ( + _parse_json_field(answer, scoring["output_field"]) + if answer_res.get("ok") else "" + ) + expected = case["expected"].strip() + passed = bool(predicted) and predicted == expected + verdict = { + "pass": passed, + "score": 10 if passed else 0, + "reason": ( + f"{scoring['output_field']} matches expected value" if passed else + "agent invocation failed" if not answer_res.get("ok") else + f"invalid or missing {scoring['output_field']} in JSON output" if not predicted else + f"expected {expected}, got {predicted}" + ), } - ) + elif not answer_res.get("ok"): + predicted = "" + verdict = {"pass": False, "score": 0, "reason": "agent invocation failed"} + else: + predicted = "" + judge_prompt = ( + f"Task prompt:\n{case['prompt']}\n\n" + f"Expected outcome:\n{case['expected'] or '(none given — judge general correctness)'}\n\n" + f"Evaluation rubric:\n{scoring.get('rubric') or '(use expected outcome)'}\n\n" + f"Candidate answer:\n{answer or '(empty answer)'}" + ) + started = time.monotonic() + judge_res = await asyncio.to_thread( + invoke, + user=user, + source="eval", + target="agent-sdk", + prompt=judge_prompt, + system=JUDGE_SYSTEM, + max_turns=1, + ref=f"eval-judge:{run_id}", + ) + judge_call = _call_metrics(judge_res, started) + verdict = ( + _parse_verdict(judge_res.get("result") or "") + if judge_res.get("ok") + else {"pass": False, "score": 0, "reason": "judge invocation failed"} + ) + row = { + "case": idx, + "prompt": case["prompt"], + "expected": case["expected"], + "answer": answer[:MAX_ANSWER_CHARS], + "pass": verdict["pass"], + "score": verdict["score"], + "reason": verdict["reason"], + "agent_call": agent_call, + } + if judge_call: + row["judge_call"] = judge_call + if scoring["method"] == "json_exact": + row["predicted_value"] = predicted + row["expected_value"] = case["expected"].strip() + results.append(row) + self.table.put_item(Item={"PK": _run_pk(run_id), "SK": f"CASE#{idx:03d}", **row}) self._update_run( sk, - results=results, + evaluated=len(results), passed=sum(1 for r in results if r["pass"]), ) scores = [r["score"] for r in results] @@ -254,10 +563,13 @@ async def _execute(self, sk: str, user: str, ds: dict, target: str) -> None: status="completed", finished_at=_now(), avg_score=(sum(scores) / len(scores)) if scores else 0.0, + calls=_calls_summary(results), ) except Exception as e: logger.exception("eval run failed: %s", sk) - self._update_run(sk, status="failed", finished_at=_now(), error=str(e)[:300]) + self._update_run( + sk, status="failed", finished_at=_now(), error=str(e)[:300], calls=_calls_summary(results), + ) eval_service = EvalService() diff --git a/backend/tests/test_eval_scenarios.py b/backend/tests/test_eval_scenarios.py new file mode 100644 index 0000000..2b9051e --- /dev/null +++ b/backend/tests/test_eval_scenarios.py @@ -0,0 +1,334 @@ +"""Focused tests for application scoring and persisted case evidence.""" + +import asyncio +import importlib +import json +import sys +import types +import unittest +from pathlib import Path +from unittest.mock import patch + +import boto3 +from botocore.exceptions import ClientError + +# Three bytes in UTF-8, like most CJK text: stresses item sizes the same way. +MULTIBYTE = "€" # U+20AC EURO SIGN + +RUN_SK = "ts#run1" +RUN_ID = "run1" + + +class FakeTable: + """Just enough of a DynamoDB Table: items keyed by (PK, SK), the Query + shapes eval_service uses, conditional puts, and SET updates.""" + + def __init__(self): + self.items: dict[tuple[str, str], dict] = {} + self.item = None # last put, for dataset assertions + + def put_item(self, **kwargs): + item = kwargs["Item"] + key = (item["PK"], item["SK"]) + if kwargs.get("ConditionExpression") == "attribute_not_exists(SK)" and key in self.items: + raise ClientError({"Error": {"Code": "ConditionalCheckFailedException"}}, "PutItem") + self.items[key] = dict(item) + self.item = item + + def get_item(self, **kwargs): + key = (kwargs["Key"]["PK"], kwargs["Key"]["SK"]) + return {"Item": dict(self.items[key])} if key in self.items else {} + + def update_item(self, **kwargs): + key = (kwargs["Key"]["PK"], kwargs["Key"]["SK"]) + names = kwargs["ExpressionAttributeNames"] + values = kwargs["ExpressionAttributeValues"] + item = self.items.setdefault(key, {"PK": key[0], "SK": key[1]}) + for index in range(len(names)): + item[names[f"#f{index}"]] = values[f":v{index}"] + + def query(self, **kwargs): + values = kwargs["ExpressionAttributeValues"] + prefix = values.get(":p", "") + rows = [dict(v) for (pk, sk), v in self.items.items() if pk == values[":pk"] and sk.startswith(prefix)] + rows.sort(key=lambda r: r["SK"], reverse=not kwargs.get("ScanIndexForward", True)) + return {"Items": rows[: kwargs["Limit"]] if "Limit" in kwargs else rows} + + # helpers for assertions + def summary(self): + return self.items[("EVALRUN", RUN_SK)] + + def cases(self): + return [v for (pk, sk), v in sorted(self.items.items()) if pk == f"EVALRUN#{RUN_ID}" and sk.startswith("CASE#")] + + +def load_eval_module(): + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + config = types.ModuleType("app.config") + config.settings = types.SimpleNamespace(aws_region="us-east-1", dynamo_table="test") + resource = types.SimpleNamespace(Table=lambda name: FakeTable()) + with patch.dict(sys.modules, {"app.config": config}), patch.object(boto3, "resource", return_value=resource): + return importlib.import_module("app.services.eval_service") + + +class EvalScenarioTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.module = load_eval_module() + + @classmethod + def tearDownClass(cls): + sys.modules.pop("app.services.eval_service", None) + package = sys.modules.get("app.services") + if package is not None: + package.__dict__.pop("eval_service", None) + + def setUp(self): + self.service = self.module.EvalService.__new__(self.module.EvalService) + self.service.table = FakeTable() + self.service._tasks = set() + + def execute(self, dataset, invoke, *, agent_version=None, agent_service=None): + invocation_module = types.ModuleType("app.services.invocation_service") + invocation_module.invoke = invoke + async def call_direct(function, **kwargs): + return function(**kwargs) + + modules = {"app.services.invocation_service": invocation_module} + if agent_service is not None: + agent_module = types.ModuleType("app.services.agent_service") + agent_module.agent_service = agent_service + modules["app.services.agent_service"] = agent_module + with patch.dict(sys.modules, modules), patch.object( + self.module.asyncio, "to_thread", call_direct + ): + asyncio.run(self.service._execute( + RUN_SK, "tester", dataset, "agent:demo", agent_version=agent_version, + )) + table = self.service.table + return {**table.summary(), "results": table.cases()} + + def test_category_contract(self): + parse = self.module._parse_category + self.assertEqual(parse('{"category":"billing"}'), "billing") + self.assertEqual(parse('```json\n{"category":"technical"}\n```'), "technical") + self.assertEqual(parse('{"category":42}'), "") + self.assertEqual(parse("technical"), "") + self.assertEqual( + self.module._parse_json_field('{"decision":{"intent":"refund"}}', "decision.intent"), + "refund", + ) + + def test_judge_string_false_is_not_a_pass(self): + verdict = self.module._parse_verdict('{"pass":"false","score":8,"reason":"wrong type"}') + self.assertFalse(verdict["pass"]) + self.assertEqual(verdict["score"], 0) + + def test_classification_uses_exact_match_without_judge(self): + answers = { + "a": '{"category":"billing"}', + "b": '{"category":"billing"}', + "c": "not JSON", + } + calls = [] + + def invoke(**kwargs): + calls.append(kwargs) + self.assertEqual(kwargs["target"], "agent:demo") + return {"ok": True, "result": answers[kwargs["prompt"]]} + + fields = self.execute({ + "scenario": "classification", + "cases": [ + {"prompt": "a", "expected": "billing"}, + {"prompt": "b", "expected": "technical"}, + {"prompt": "c", "expected": "account"}, + ], + }, invoke) + self.assertEqual(len(calls), 3) + self.assertEqual(fields["status"], "completed") + self.assertEqual(fields["passed"], 1) + self.assertEqual([r["predicted_value"] for r in fields["results"]], ["billing", "billing", ""]) + self.assertEqual([r["score"] for r in fields["results"]], [10, 0, 0]) + + def test_support_uses_real_judge_verdict(self): + calls = [] + + def invoke(**kwargs): + calls.append(kwargs) + if kwargs["target"] == "agent:demo": + return {"ok": True, "result": "Please contact human support."} + return {"ok": True, "result": '{"pass":true,"score":8,"reason":"Escalated appropriately"}'} + + fields = self.execute({ + "scenario": "support", + "scoring": {"method": "llm_judge", "output_field": "category", "rubric": "Escalate unsupported requests."}, + "cases": [{"prompt": "Can I get a refund after 15 days?", "expected": "Escalate to human support"}], + }, invoke) + self.assertEqual(len(calls), 2) + self.assertIn("Escalate unsupported requests.", calls[1]["prompt"]) + self.assertEqual(fields["passed"], 1) + self.assertEqual(fields["results"][0]["answer"], "Please contact human support.") + self.assertEqual(fields["results"][0]["reason"], "Escalated appropriately") + + def test_custom_scenario_reads_configured_json_field(self): + fields = self.execute({ + "scenario": "claims-routing", + "scoring": {"method": "json_exact", "output_field": "decision.intent", "rubric": ""}, + "cases": [{"prompt": "route this", "expected": "refund"}], + }, lambda **kwargs: {"ok": True, "result": '{"decision":{"intent":"refund"}}'}) + self.assertEqual(fields["passed"], 1) + self.assertEqual(fields["results"][0]["predicted_value"], "refund") + + def test_user_defined_scenario_is_saved_with_scoring_contract(self): + dataset = self.service.create_dataset( + user="tester", + name="claims-routing", + description="Claim intent", + scenario="claims-routing", + scoring={"method": "json_exact", "output_field": "decision.intent", "rubric": ""}, + cases=[{"prompt": "Refund this", "expected": "refund"}], + ) + self.assertEqual(dataset["scoring"]["output_field"], "decision.intent") + self.assertEqual(self.service.table.item["scenario"], "claims-routing") + with self.assertRaises(ValueError): + self.service.create_dataset( + user="tester", name="bad", description="", scenario="claims-routing", + scoring={"method": "json_exact", "output_field": "decision[0]", "rubric": ""}, + cases=[{"prompt": "Refund this", "expected": "refund"}], + ) + + def test_version_change_fails_run(self): + versions = iter((1, 2)) + agent = types.SimpleNamespace(get_agent=lambda agent_id: {"version": next(versions)}) + with patch.object(self.module.logger, "exception"): + fields = self.execute( + {"scenario": "classification", "cases": [{"prompt": "a", "expected": "billing"}]}, + lambda **kwargs: {"ok": True, "result": '{"category":"billing"}'}, + agent_version=1, + agent_service=agent, + ) + self.assertEqual(fields["status"], "failed") + self.assertIn("version changed", fields["error"]) + + +class EvalStorageTests(unittest.TestCase): + """Run evidence is stored per case so no item nears the 400 KB limit.""" + + @classmethod + def setUpClass(cls): + cls.module = load_eval_module() + + def setUp(self): + self.service = self.module.EvalService.__new__(self.module.EvalService) + self.service.table = FakeTable() + self.service._tasks = set() + self.agent = {"id": "ag1", "version": 3, "system_prompt": "You classify. " * 1500} + self.agents = types.SimpleNamespace(get_agent=lambda agent_id: dict(self.agent)) + self.service.table.put_item(Item={ + "PK": "EVAL", "SK": "DS#ds1", "name": "multibyte", "scenario": "support", + "scoring": {"method": "llm_judge", "output_field": "category", "rubric": ""}, + "cases": [{"prompt": MULTIBYTE * 2000, "expected": MULTIBYTE * 900} for _ in range(20)], + }) + + def run_dataset(self): + invocation_module = types.ModuleType("app.services.invocation_service") + + def invoke(**kwargs): + if kwargs["target"] == "agent-sdk": + return {"ok": True, "result": '{"pass":true,"score":9,"reason":"ok"}', + "usage": {"duration_ms": 1500, "num_turns": 1, "total_cost_usd": 0.002}, + "runtime_session_id": "judge-sid"} + return {"ok": True, "result": MULTIBYTE * 16000, + "usage": {"duration_ms": 9000, "num_turns": 2, "total_cost_usd": 0.0125}, + "runtime_session_id": "agent-sid"} + + invocation_module.invoke = invoke + agent_module = types.ModuleType("app.services.agent_service") + agent_module.agent_service = self.agents + + async def call_direct(function, **kwargs): + return function(**kwargs) + + async def go(): + run = self.service.start_run(user="tester", dataset_id="ds1", target="agent:ag1") + await asyncio.gather(*self.service._tasks) + return run + + with patch.dict(sys.modules, { + "app.services.invocation_service": invocation_module, + "app.services.agent_service": agent_module, + }), patch.object(self.module.asyncio, "to_thread", call_direct): + return asyncio.run(go()) + + def test_worst_case_multibyte_run_keeps_every_item_small(self): + run = self.run_dataset() + self.assertEqual(run["status"], "running") + sizes = {key: len(json.dumps(item, ensure_ascii=False, default=str).encode()) for key, item in self.service.table.items.items()} + # The dataset item holds the inputs (unchanged layout, ~180 KB at most); + # every run-side item stays small no matter how long the answers are. + self.assertLess(max(sizes.values()), 400_000) + self.assertLess(max(v for k, v in sizes.items() if k[0] != "EVAL"), 40_000) + summary = next(v for (pk, _), v in self.service.table.items.items() if pk == "EVALRUN") + self.assertNotIn("results", summary) + self.assertNotIn("system_prompt", summary) + self.assertEqual((summary["status"], summary["evaluated"], summary["passed"]), ("completed", 20, 20)) + + def test_get_run_by_id_returns_full_evidence_and_prompt(self): + run = self.run_dataset() + full = self.service.get_run(run["id"]) + self.assertEqual(len(full["results"]), 20) + self.assertEqual(full["results"][0]["answer"], (MULTIBYTE * 16000)[:8000]) + self.assertEqual(full["system_prompt"], self.agent["system_prompt"]) + listed = self.service.list_runs(50) + self.assertEqual((listed[0]["results"], listed[0]["system_prompt"], listed[0]["evaluated"]), ([], "", 20)) + + def test_prompt_snapshot_is_stored_once_per_version(self): + self.run_dataset() + self.run_dataset() + prompts = [k for k in self.service.table.items if k[0] == "AGENTPROMPT"] + self.assertEqual(prompts, [("AGENTPROMPT", "ag1#V000003")]) + + def test_each_case_records_its_calls_and_the_run_aggregates_them(self): + run = self.run_dataset() + full = self.service.get_run(run["id"]) + case = full["results"][0] + self.assertEqual(case["agent_call"], { + "ok": True, "duration_ms": 9000, "num_turns": 2, "cost_usd": 0.0125, "runtime_session_id": "agent-sid", + }) + self.assertEqual(case["judge_call"]["runtime_session_id"], "judge-sid") + calls = full["calls"] + self.assertEqual((calls["agent_calls"], calls["agent_ok"], calls["judge_calls"]), (20, 20, 20)) + self.assertEqual(calls["agent_duration_p50_ms"], 9000) + self.assertAlmostEqual(calls["agent_cost_usd"], 0.25) + self.assertAlmostEqual(calls["judge_cost_usd"], 0.04) + json.dumps(full) # API-serialisable without a Decimal encoder + + def test_dataset_history_is_independent_of_other_runs(self): + first = self.run_dataset() + self.service.table.put_item(Item={ + "PK": "EVAL", "SK": "DS#other", "name": "other", "scenario": "support", + "scoring": {"method": "llm_judge", "output_field": "category", "rubric": ""}, + "cases": [{"prompt": "x", "expected": "y"}], + }) + for i in range(60): # more than the 50-run window of list_runs + self.service.table.put_item(Item={"PK": "EVALRUN", "SK": f"9999-{i:03d}#o{i}", "run_id": f"o{i}", + "dataset_id": "other", "status": "completed"}) + self.assertNotIn(first["id"], [r["id"] for r in self.service.list_runs(50)]) + second = self.run_dataset() + history = self.service.list_dataset_runs("ds1") + self.assertEqual([r["id"] for r in history], [second["id"], first["id"]]) + + def test_legacy_inline_run_is_still_readable(self): + self.service.table.put_item(Item={ + "PK": "EVALRUN", "SK": "2026-01-01#old1", "run_id": "old1", "status": "completed", + "system_prompt": "legacy prompt", "passed": 1, "total": 1, + "results": [{"case": 0, "prompt": "p", "expected": "e", "answer": "a", "pass": True, "score": 10, "reason": "r"}], + }) + run = self.service.get_run("old1") + self.assertEqual((run["evaluated"], run["system_prompt"], len(run["results"])), (1, "legacy prompt", 1)) + self.assertIsNone(self.service.get_run("missing")) + + +if __name__ == "__main__": + unittest.main() diff --git a/docs/observability.md b/docs/observability.md index 09defa5..4f0ce18 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -153,3 +153,162 @@ correlated by `spanId`; the kernel's stdout lines in the same group also carry - Business counters that only the workflow engine knows (funnel counts per phase) are still application-side; publish them as EMF metrics in the same `bedrock-agentcore` namespace to put them on the same dashboard. + +## Application-effect dashboard + +The portal's **Observability** page reads evaluation runs and displays +application outcomes separately from execution health. It discovers scenario +tabs from the datasets that have actually run; no scenario name is built into +the dashboard. Each dataset selects one of two scoring methods: + +- **`json_exact`** reads a configured string field from the agent's JSON + response (for example `category` or `decision.intent`) and compares it + exactly with the case's expected value. The dashboard counts expected and + actual values, exact-match rate, invalid outputs and failures. It makes no + judge-model call. Scores are only 0 or 10, so no mean score is shown. +- **`llm_judge`** scores each answer against the case's expectation and an + optional dataset rubric. The dashboard shows pass rate, mean score, + low-score cases, full answers and reasons. Judge scores are screening + signals; calibrate them against a small human-reviewed set before treating + them as a quality target. + +`json_exact` is a strict contract. Dataset authors should know the rules: + +- The whole answer must parse as one JSON object. A surrounding Markdown + code fence is stripped; any other prose before or after the object makes + the answer an invalid output. +- The field must be a string. Its value is trimmed and compared + **case-sensitively** with the trimmed expected value, so `Billing` fails + against `billing`. Write the expected values in the exact form the agent + is told to return. +- A dotted `output_field` walks nested objects, up to four levels. + +The judge runs on the platform's own headless kernel with the same default +model family as most agents under test. That keeps the sample free of extra +dependencies, but the judge is not independent of what it grades and can +share its blind spots. In the difficult-customer runs it passed an answer +that invented a response deadline. Treat a pass as "nothing obviously wrong" +and keep reviewing the stored answers. + +Users can create a custom scenario and choose its scoring contract in the +**Evaluation** page. The equivalent API request is: + +```json +{ + "name": "claims-routing-v1", + "scenario": "claims-routing", + "scoring": {"method": "json_exact", "output_field": "decision.intent", "rubric": ""}, + "cases": [{"prompt": "I need a refund for order A", "expected": "refund"}] +} +``` + +The agent must then return a JSON object with `decision.intent`. A new +scenario appears in Observability when its first evaluation run starts. +For an LLM-judged scenario, use `method = llm_judge` and set `rubric` to the +business-specific judging criteria. + +When `scoring` is omitted, the scenario name picks a default: `classification` +becomes `json_exact` on `category`, and every other scenario becomes +`llm_judge` with no rubric. The default keeps datasets created before +scoring configuration working. Pass `scoring` explicitly for new datasets so +the contract is visible in the request. + +This is the no-code extension path. For a new scoring algorithm, extend +`EvalScoringConfig` in `backend/app/models/schemas.py`, the scoring dispatch +in `backend/app/services/eval_service.py`, and the method selector in +`frontend/src/pages/EvalPage.tsx`. The dashboard draws a value distribution +for exact JSON scoring and a score breakdown for judge scoring. A custom +algorithm may also need a corresponding dashboard component. The platform +does not execute arbitrary user-uploaded evaluator code. + +Each run records the published agent's version at start, and the system +prompt of that version is stored once and shared by every run of it. Each +case's prompt, expectation, full answer and verdict is stored as its own item, +so a run of 20 long multilingual cases stays well inside DynamoDB's item size +limit. A version change while a run is executing fails the run so it cannot be +presented as a single-version comparison. + +Each case also records the execution facts of its own calls, the same ones +the invocation ledger keeps: ok, duration, turns, cost and runtime session id, +for the agent call and for the judge call. The run summary aggregates them +when it ends. The page shows that infrastructure view next to the pass rate +for the same cases, for example "20/20 agent calls returned ok, 6 of those +answers broke the expectation", with no join against the ledger. The session +id on a case is also the `session.id` on its kernel spans when trace delivery +is enabled. A dataset's run history is read through its own index, so the +previous run is found however many runs of other datasets came after it. + +The dashboard compares completed runs of the **same dataset**. Every pass rate +is shown with its sample size, because one case moves the rate of an 8-case +dataset by over 12 points. A drop is flagged only when it amounts to at least +two cases. This is an offline benchmark: +the general invocation ledger still stores only a 200-character user-prompt +preview and execution metrics. It has no automatic business label or +production-conversation quality score. + +To generate results without customer data, run the synthetic benchmark against +a **deployed** portal: + +```bash +PORTAL_URL=https:// \ +PORTAL_TOKEN= \ +python3 scripts/run_observability_demo.py +``` + +For local open-auth development, omit `PORTAL_TOKEN`. The script publishes a +classifier and a support agent, creates two reusable synthetic datasets, +starts real AgentCore invocations, waits for scored results, and reports how +many of the case calls it finds in the invocation ledger. Re-running bumps the +agents' versions and adds comparable runs to the same datasets. It does not +fabricate predictions, answers, scores, token usage, latency, or cost. Model +calls incur the platform's normal charges and quota. + +Agent and dataset names start with `--prefix`, `obs-demo-` by default. An +administrator can republish an agent another user published, so the script +checks first: if an agent with one of its names exists and was not created +by these scripts, it stops instead of overwriting it. Choose another prefix +in that case. + +The ledger endpoint returns the latest 200 rows across the platform. On a +deployment with other traffic, part of a run's calls can be older than that. +The script then reports the count it found as "in window" rather than +failing. + +### Difficult-customer benchmarks + +`scripts/run_difficult_customer_evals.py` builds on the two demo agents (run +the demo script first, with the same `--prefix`) and adds three more +scenarios: `oncall` (severity triage, `json_exact` on +`severity`), `rnd` (an engineering assistant) and `sales-success`. It runs +96 adversarial cases in two rounds: + +- **hard**: prompt injection, forged system messages, output-format + overrides, social engineering, requests beyond policy, and several + languages. +- **extreme**: contradictory stakeholders, profanity, personal attacks, + blame-shifting, and claimed promises that nobody can verify. The judge + rubric also fails an answer that mirrors abuse, admits liability it + cannot verify, over-apologises into a commitment, or lectures the user. + +```bash +PORTAL_URL=https:// PORTAL_TOKEN= \ +python3 scripts/run_difficult_customer_evals.py --round all +``` + +The script ends by printing, for each run, the infrastructure view of its +calls next to its pass rate. Both come from the run itself, so they always +cover the same cases. The usual result is that every call succeeds at the +infrastructure layer, with no errors and normal latency and cost. The evaluation layer still flags answers +that broke a business rule, for example an upsell to a customer claiming +compensation, a response deadline the policy never states, or a severity +talked down by an angry message. That gap is what the page is for. Judge +verdicts are screening signals. The page keeps every full answer so a +reviewer can confirm or overturn each one. + +Agents and datasets are reused, so a re-run is comparable with earlier +runs of the same version. `--republish` publishes the three scenario agents +as a new version, which lets you show the same-dataset comparison. + +The page stays empty until runs actually return results. A workspace with no +portal URL or AWS credentials can build and test the feature but cannot make +claims about CloudWatch trace delivery or measured agent quality. diff --git a/frontend/src/components/observability/ApplicationOutcomes.tsx b/frontend/src/components/observability/ApplicationOutcomes.tsx new file mode 100644 index 0000000..6ce2071 --- /dev/null +++ b/frontend/src/components/observability/ApplicationOutcomes.tsx @@ -0,0 +1,309 @@ +import { useEffect, useMemo, useState } from 'react' +import { Link } from 'react-router' +import { AlertTriangle, CheckCircle2, XCircle } from 'lucide-react' +import { api, type EvalRun } from '@/services/api' +import { fmtTs } from '@/services/format' + +/** A pass-rate drop is flagged only when it amounts to at least this many cases, + * so one flipped case in a 6-case dataset does not read as a regression. */ +const MIN_DROP_CASES = 2 +const INVALID = '(invalid output)' + +function Tile({ label, value, hint }: { label: string; value: string; hint?: string }) { + return ( +
+

{label}

+

{value}

+ {hint &&

{hint}

} +
+ ) +} + +function rate(passed: number, n: number): string { + return n ? `${Math.round((passed / n) * 100)}% (${passed}/${n})` : '—' +} + +function isStructured(run: EvalRun | undefined): boolean { + if (!run) return false + return (run.scoring?.method || (run.scenario === 'classification' ? 'json_exact' : 'llm_judge')) === 'json_exact' +} + +/** + * Application-layer outcomes read from evaluation runs: whether answers were + * right, not whether the call returned. Loads independently of the ledger so + * an evaluation API error never blanks the infrastructure metrics. + */ +export default function ApplicationOutcomes({ refreshKey }: { refreshKey: number }) { + const [runs, setRuns] = useState([]) + const [detail, setDetail] = useState(null) + const [datasetRuns, setDatasetRuns] = useState([]) + const [scenario, setScenario] = useState('') + const [runId, setRunId] = useState('') + const [error, setError] = useState('') + + const loadRuns = () => { + api.listEvalRuns(50).then((rs) => { setRuns(rs); setError('') }).catch((e) => setError(String(e))) + } + useEffect(loadRuns, [refreshKey]) + + const hasActiveRuns = runs.some((r) => r.status === 'running') + useEffect(() => { + if (!hasActiveRuns) return + const timer = window.setInterval(loadRuns, 20000) + return () => window.clearInterval(timer) + }, [hasActiveRuns]) + + const scenarios = [...new Set(runs.map((r) => r.scenario || 'general'))] + const activeScenario = scenarios.includes(scenario) ? scenario : scenarios[0] || '' + const choices = runs.filter((r) => (r.scenario || 'general') === activeScenario) + const summary = choices.find((r) => r.id === runId) || choices[0] + + // The list carries summaries only; fetch the selected run's evidence, and + // fetch it again whenever more of its cases have been scored. + useEffect(() => { + if (!summary) return setDetail(null) + api.getEvalRun(summary.id).then(setDetail).catch((e) => setError(String(e))) + }, [summary?.id, summary?.evaluated, summary?.status]) + + // A dataset's own history, so the previous run is found however many runs + // of other datasets came after it. + useEffect(() => { + if (!summary) return setDatasetRuns([]) + api.listDatasetRuns(summary.dataset_id).then(setDatasetRuns).catch(() => setDatasetRuns([])) + }, [summary?.dataset_id, summary?.status, refreshKey]) + + const run = detail && summary && detail.id === summary.id ? detail : undefined + const structured = isStructured(summary) + const outputField = summary?.scoring?.output_field || 'category' + const results = run?.results ?? [] + const evaluated = summary?.evaluated ?? 0 + const passed = summary?.passed ?? 0 + const failedCases = results.filter((r) => !r.pass) + + const history = [...new Map( + [...datasetRuns, ...choices.filter((r) => r.dataset_id === summary?.dataset_id)].map((r) => [r.id, r]), + ).values()].sort((a, b) => b.started_at.localeCompare(a.started_at)).slice(0, 8) + const calls = run?.calls ?? summary?.calls ?? null + const previous = history.find((r) => + summary && r.id !== summary.id && r.status === 'completed' && r.started_at < summary.started_at && r.evaluated > 0, + ) + const dropCases = summary?.status === 'completed' && previous && evaluated + ? Math.round((previous.passed / previous.evaluated - passed / evaluated) * evaluated) + : 0 + + const distribution = useMemo(() => { + if (!structured) return [] + const counts = new Map() + for (const row of results) { + const expected = row.expected_value || row.expected_label || row.expected.trim() + const predicted = row.predicted_value || row.predicted_label || INVALID + const e = counts.get(expected) || { expected: 0, predicted: 0, correct: 0 } + e.expected += 1 + if (row.pass) e.correct += 1 + counts.set(expected, e) + const p = counts.get(predicted) || { expected: 0, predicted: 0, correct: 0 } + p.predicted += 1 + counts.set(predicted, p) + } + return [...counts].map(([label, values]) => ({ label, ...values })) + .sort((a, b) => b.predicted - a.predicted || a.label.localeCompare(b.label)) + }, [results, structured]) + const invalid = results.filter((r) => !(r.predicted_value || r.predicted_label)).length + const lowScores = results.filter((r) => r.score <= 5).length + const meanScore = results.length ? (results.reduce((sum, r) => sum + r.score, 0) / results.length).toFixed(1) : '—' + const maxCount = Math.max(1, ...distribution.map((d) => Math.max(d.expected, d.predicted))) + const weakest = [...distribution].filter((d) => d.expected > 0) + .sort((a, b) => a.correct / a.expected - b.correct / b.expected)[0] + + return ( +
+
+
+

Application outcomes

+

+ Whether the answers were right, from evaluation runs. A successful invocation above only means the call completed. +

+
+ Open Evaluation +
+ + {error &&
{error}
} + + {!summary && !error && ( +
+ No evaluation runs yet. Create a dataset and start a run on the Evaluation page; results from real invocations appear here. +
+ )} + + {summary && ( + <> +
+
+ {scenarios.map((key) => ( + + ))} +
+ +
+ +
+ {summary.dataset_name} + {summary.synthetic && synthetic cases} + {structured ? `exact match on ${outputField}` : 'LLM judge'} + target {summary.target}{summary.agent_version != null ? ` · agent v${summary.agent_version}` : ''} + · {summary.status} · {fmtTs(summary.started_at)} +
+ {summary.status === 'failed' && ( +

Run failed: {summary.error || 'unknown error'}

+ )} + + {calls && calls.agent_calls > 0 && ( +
+ Same calls, infrastructure view:{' '} + {calls.agent_ok}/{calls.agent_calls} agent calls returned ok + {calls.agent_duration_p50_ms != null && ` · median ${(calls.agent_duration_p50_ms / 1000).toFixed(1)}s`} + {` · $${calls.agent_cost_usd.toFixed(4)}`} + {calls.judge_calls > 0 && ` (+ ${calls.judge_calls} judge calls, $${calls.judge_cost_usd.toFixed(4)})`}.{' '} + {calls.agent_ok > passed + ? {calls.agent_ok - passed} of the answers that returned ok broke the expectation. + : 'Every answer that returned ok also passed.'} +
+ )} + +
+ + + {structured + ? d.predicted > 0 && d.label !== INVALID).length)} hint="returned by the agent" /> + : } + +
+ + {history.length > 1 && ( +
+

Runs of this dataset

+

Same cases and expectations, so versions are comparable. Click a row to open it.

+ + + + + + {!structured && } + + + {history.map((entry) => ( + setRunId(entry.id)}> + + + + {!structured && } + + + ))} +
Started (UTC)Agent versionPassedMean scoreStatus
{fmtTs(entry.started_at)}{entry.agent_version == null ? '—' : `v${entry.agent_version}`}{rate(entry.passed, entry.evaluated)}{entry.avg_score == null ? '—' : `${entry.avg_score.toFixed(1)}/10`}{entry.status}
+
+ )} + + {structured && ( +
+

{outputField} distribution

+

Blue: returned by the agent. Grey: expected. Right: returned / expected · accuracy for that value.

+ {distribution.length === 0 &&

Waiting for the first result…

} +
+ {distribution.map((d) => ( +
+ {d.label} +
+
+
+
+ {d.predicted} / {d.expected} · {d.expected ? `${Math.round((d.correct / d.expected) * 100)}%` : '—'} +
+ ))} +
+
+ )} + +
+
+

Findings

+ {previous && dropCases >= MIN_DROP_CASES && ( +

+ + Pass rate fell from {rate(previous.passed, previous.evaluated)} on the previous run of this dataset to {rate(passed, evaluated)}. +

+ )} + {structured && weakest && weakest.correct < weakest.expected && ( +

Weakest value: {weakest.label}, {weakest.correct}/{weakest.expected} correct.

+ )} + {failedCases.length > 0 ? ( +

+ + {failedCases.length} of {results.length} cases failed + {structured && invalid > 0 ? `; ${invalid} returned no valid ${outputField} field.` : '. Open them below for the answer and the reason.'} +

+ ) :

{results.length ? 'All scored cases passed.' : 'Waiting for results…'}

} +
+
+

What was evaluated

+

The agent's system prompt as published for this version. Each case's user prompt is below.

+ {!structured && summary.scoring?.rubric &&

Judge rubric: {summary.scoring.rubric}

} +
+ Show system prompt + 
+                  {run?.system_prompt || 'Not recorded (raw kernel target or an older run).'}
+
+
+
+
+ +
+

Cases and verdicts

+ {!run &&

Loading…

} + {run && results.length === 0 &&

Waiting for the first result…

} +
+ {[...results].sort((a, b) => Number(a.pass) - Number(b.pass) || a.case - b.case).map((row) => ( +
+ + {row.pass ? : } + {row.prompt} + {!structured && {row.score}/10} + +
+

Prompt: {row.prompt}

+

Expected: {row.expected}

+ {structured &&

{outputField}: {row.predicted_value || row.predicted_label || INVALID}

} +

Answer: {row.answer || '(empty answer)'}

+

Verdict: {row.reason}

+ {row.agent_call && ( +

+ Call: {row.agent_call.ok ? 'ok' : 'failed'} · {(row.agent_call.duration_ms / 1000).toFixed(1)}s + {row.agent_call.num_turns != null && ` · ${row.agent_call.num_turns} turns`} + {row.agent_call.cost_usd != null && ` · $${row.agent_call.cost_usd.toFixed(4)}`} + {row.agent_call.runtime_session_id && <> · session {row.agent_call.runtime_session_id}} +

+ )} +
+
+ ))} +
+
+ + )} +
+ ) +} diff --git a/frontend/src/pages/EvalPage.tsx b/frontend/src/pages/EvalPage.tsx index c7c22fb..7fddd1d 100644 --- a/frontend/src/pages/EvalPage.tsx +++ b/frontend/src/pages/EvalPage.tsx @@ -1,7 +1,7 @@ import { useEffect, useRef, useState } from 'react' import { FlaskConical, Loader2, Play, Plus, RefreshCw, Trash2 } from 'lucide-react' import { Modal, SectionTitle } from '@/components/common/ui' -import { api, type EvalDataset, type EvalRun, type PublishedAgent } from '@/services/api' +import { api, type EvalDataset, type EvalRun, type EvalScoring, type PublishedAgent } from '@/services/api' import { fmtTs } from '@/services/format' export default function EvalPage() { @@ -15,6 +15,11 @@ export default function EvalPage() { // create form const [name, setName] = useState('') + const [scenarioChoice, setScenarioChoice] = useState('general') + const [customScenario, setCustomScenario] = useState('') + const [scoringMethod, setScoringMethod] = useState('llm_judge') + const [outputField, setOutputField] = useState('category') + const [rubric, setRubric] = useState('') const [casesText, setCasesText] = useState('What is 2+2? => 4\nCapital of France? => Paris') const [creating, setCreating] = useState(false) const [createError, setCreateError] = useState('') @@ -22,11 +27,19 @@ export default function EvalPage() { // run form const [runTarget, setRunTarget] = useState('agent-sdk') const [startingId, setStartingId] = useState('') + const [details, setDetails] = useState>({}) + const detailsRef = useRef(details) + detailsRef.current = details const refresh = () => { api.listEvalDatasets().then(setDatasets).catch((e) => setError(String(e))) api.listEvalRuns().then((rs) => { setRuns(rs) + // refresh the evidence of a loaded run while it is still scoring + for (const r of rs) { + const shown = detailsRef.current[r.id] + if (shown && shown.evaluated !== r.evaluated) loadDetail(r.id) + } // keep polling while something is running if (rs.some((r) => r.status === 'running')) { if (pollRef.current == null) pollRef.current = window.setInterval(refresh, 5000) @@ -37,6 +50,15 @@ export default function EvalPage() { }).catch(() => {}) api.listAgents().then(setAgents).catch(() => {}) } + const loadDetail = (id: string) => { + api.getEvalRun(id).then((full) => setDetails((prev) => ({ ...prev, [id]: full }))).catch(() => {}) + } + const toggleRun = (id: string) => { + if (expandedRun === id) return setExpandedRun('') + setExpandedRun(id) + loadDetail(id) + } + useEffect(() => { refresh() return () => { @@ -57,9 +79,20 @@ export default function EvalPage() { return { prompt, expected } }) .filter((c) => c.prompt) - await api.createEvalDataset({ name, cases }) + const scenario = scenarioChoice === 'custom' ? customScenario.trim() : scenarioChoice + await api.createEvalDataset({ + name, + scenario, + scoring: { method: scoringMethod, output_field: outputField.trim() || 'category', rubric }, + cases, + }) setShowCreate(false) setName('') + setScenarioChoice('general') + setCustomScenario('') + setScoringMethod('llm_judge') + setOutputField('category') + setRubric('') refresh() } catch (e) { setCreateError(String(e)) @@ -85,7 +118,7 @@ export default function EvalPage() {
@@ -105,7 +138,7 @@ export default function EvalPage() {

{d.name}

-

{d.cases.length} cases

+

{d.cases.length} cases · {d.scenario || 'general'} · {d.scoring?.method || 'llm_judge'}{d.synthetic ? ' · synthetic' : ''}