From 8852e363932a8f67f17367bf48cd8deaab0d0a1a Mon Sep 17 00:00:00 2001 From: Robin Date: Wed, 26 Aug 2026 22:08:20 -0700 Subject: [PATCH] update --- EVALUATIONS.md | 95 +++++++++++++-- agentx/evaluations/client.py | 19 ++- agentx/evaluations/datasets.py | 8 ++ agentx/evaluations/models.py | 3 + agentx/evaluations/runner.py | 179 ++++++++++++++++++++++------ agentx/monitor/client.py | 65 ++++++++-- agentx/monitor/judge_scorers.py | 28 ++++- agentx/monitor/online_evaluators.py | 10 +- agentx/monitor/review_queue.py | 93 +++++++++++++++ agentx/testing.py | 4 +- tests/test_integrations.py | 14 +++ tests/test_review_queue.py | 73 ++++++++++++ tests/test_runner_features.py | 146 +++++++++++++++++++++++ 13 files changed, 673 insertions(+), 64 deletions(-) create mode 100644 agentx/monitor/review_queue.py create mode 100644 tests/test_review_queue.py create mode 100644 tests/test_runner_features.py diff --git a/EVALUATIONS.md b/EVALUATIONS.md index aebd6d2..0e18e5b 100644 --- a/EVALUATIONS.md +++ b/EVALUATIONS.md @@ -197,6 +197,82 @@ client.evaluations.datasets.builder(name="Support Agent v2").add_case( --- +### Agent trajectory and retrieval-context checks (deterministic, no judge spend) + +Two per-case expectations are scored server-side against the case's **linked trace** (return +`{"output": ..., "trace_id": span.trace_id}` from your agent function, with the call wrapped in +`client.tracer.trace(..., sync=True)`): + +```python +client.evaluations.datasets.builder(name="Agent checks").add_case( + query="Refund order 4412 and email a confirmation.", + expected_results="Refund issued and confirmation sent.", + # The tool calls a correct run should make, matched against the trace's REAL calls. + # Modes (agentevals semantics): strict | unordered | superset | subset. + expected_tools=["lookup_order", "issue_refund", "send_email"], + trajectory_match_mode="strict", + # What a correct retriever should have fetched - compared to the actual retrieved context + # with token Jaccard. Catches retriever regressions even when the answer text is identical. + expected_retrieval_context=["Refunds: 30 days, no restocking fee."], +) +``` + +Each produces a scorer row on the result (`Trajectory match ()` pass/fail, and +`Context match (jaccard)` 0-1). Both are deterministic: no LLM judge call, no spend, and they +run on every result that carries the needed evidence. + +--- + +### Dataset splits (cheap PR runs vs. nightly full runs) + +Tag cases with named subsets and run just one subset: + +```python +builder.add_case(query="smoke case", splits=["smoke"]) +builder.add_case(query="full-only case") + +client.evaluations.run(dataset_id, subject, split="smoke").execute(my_fn).finalize() +``` + +Original case indexes are preserved, so per-case comparisons line up between a split run and a +full run. The connector-driven dashboard run accepts the same `split`. + +--- + +### Concurrency and output reuse + +```python +run.execute(my_fn, concurrency=4) # thread-pooled agent calls, ordered submission +run.execute(my_fn, reuse_outputs_from="run_abc123") # replay a previous run's outputs +``` + +`reuse_outputs_from` replays the recorded output for every case whose query text is unchanged +(errored rows and changed/new cases run normally) and the judge re-scores everything with THIS +run's grading config - which makes iterating on scorers essentially free. Reused results carry +`metadata.reusedFromRun`. + +Interrupted runs resume: `execute()` asks the engine which idempotency keys were already +accepted and skips those cases, so a crash or a failed batch (which now raises +`EvaluationSubmissionError` instead of finishing silently empty) never re-pays for finished +work - just call `execute()` again on the same context. + +--- + +### Human review queue (label-and-calibrate from code) + +```python +item = client.monitor.review_queue.queue(trace_id, note="spot-check this") +for item in client.monitor.review_queue.list(status="pending"): + client.monitor.review_queue.label(item.id, "bad", corrected_score=2, note="hallucinated") +``` + +Labels (with the judge's own score for the same trace) feed per-scorer calibration +(`client.monitor.judge_scorers.calibration(scorer_id)`) and become judge-tuning evidence. +Project-level numbers come from `client.monitor.calibration(window="7d")`, whose exact wire +keys are `comparedCount`, `agreementRate`, `falsePositiveRate`, `falseNegativeRate`. + +--- + ### LLM Judge Scorers - reusable grading configs By default, a dataset runs against the grading config it was created with (`number_of_requests`, `acceptance_criteria`, similarity metrics, etc. - see above). If you want to grade the **same dataset** against **different configs** (e.g. a strict config vs. a lenient one, or reuse one config across many datasets), create a standalone **LLM Judge Scorer** and pass its id to `.run()`. @@ -285,13 +361,13 @@ Calibration, tuning and the live-scoring history hang off the same scorer id - t ```python client.monitor.judge_scorers.calibration(scorer.id, window="7d") # verdicts vs. recorded ground truth proposal = client.monitor.judge_scorers.tune(scorer.id) # LLM call, slow -client.monitor.judge_scorers.validate_tuning(scorer.id, proposal) # re-judge with candidate criteria -client.monitor.judge_scorers.publish_tuning(scorer.id, proposal) # write it onto the rubric +verdict = client.monitor.judge_scorers.validate_tuning(scorer.id, proposal) # re-judge with candidate criteria +client.monitor.judge_scorers.publish_tuning(scorer.id, proposal, validation=verdict) # write it onto the rubric client.monitor.judge_scorers.ratings(scorer.id, window="7d") # -> list[OnlineEvaluatorRatingPoint] client.monitor.judge_scorers.events(scorer.id, window="7d") # -> list[OnlineEvaluatorEvent] ``` -Those six cover live-traffic scoring, so calling them on an offline-only scorer raises `AgentXJudgeScorersError` naming the fix (`update(scorer_id, online={"enabled": True})`). `publish_tuning` writes to the shared rubric, so it applies everywhere the scorer is used: online scoring, offline dataset runs and the playground alike. +Those six cover live-traffic scoring, so calling them on an offline-only scorer raises `AgentXJudgeScorersError` naming the fix (`update(scorer_id, online={"enabled": True})`). `publish_tuning` writes to the shared rubric, so it applies everywhere the scorer is used: online scoring, offline dataset runs and the playground alike. Publish is provenance-gated: the engine refuses an unvalidated publish, and a measured regression, unless you pass `force=True`; the validation verdict is stamped into the rubric's version history. #### Engine compatibility (self-host) @@ -854,12 +930,9 @@ Your callable can return any of: | `dict` with `"output"` key | Output text from `output`, rest stored as metadata | | `EvaluationResult` | Full control - pass rating, justification, trace, timings | -### Security and redaction - -The SDK automatically scrubs secrets from outputs and metadata before uploading: -- `sk-...` API keys -- Bearer tokens -- Authorization headers -- Password-like fields +### What gets uploaded -Raw agent outputs, prompts, and CoT reasoning are **never uploaded** - only the text response, metadata you explicitly include, and optional observable trace summaries. +The SDK uploads exactly what your callable returns: the output text, any metadata you include, +and (when tracing is enabled) the trace you instrumented. **The SDK does not scrub or redact +anything** - if your agent's output or metadata can contain secrets, redact them in your own +code before returning, or keep them out of the returned payload entirely. diff --git a/agentx/evaluations/client.py b/agentx/evaluations/client.py index 004a945..8dfd354 100644 --- a/agentx/evaluations/client.py +++ b/agentx/evaluations/client.py @@ -60,6 +60,13 @@ class AgentXValidationError(AgentXEvaluationsError): pass +class EvaluationSubmissionError(AgentXEvaluationsError): + """A result batch could not be submitted (after one retry). The run is left unfinalized; + re-running execute() on the same context resumes past already-submitted cases.""" + + pass + + def _resolve_scorer_id(scorer_id: Optional[str], evaluation_settings_id: Optional[str]) -> Optional[str]: """One grader, two spellings: ``scorer_id`` is the post-consolidation name for what the wire still calls ``evaluationSettingsId`` (the ids are identical by design). Both kwargs are @@ -297,10 +304,11 @@ def init_run( python_version: Optional[str] = None, scorer_id: Optional[str] = None, evaluation_settings_id: Optional[str] = None, + split: Optional[str] = None, ) -> EvaluationRun: """``scorer_id`` names the LLM Judge Scorer grading this run (its id doubles as the wire's ``evaluationSettingsId``). ``evaluation_settings_id`` is the pre-consolidation - alias and keeps working.""" + alias and keeps working. ``split`` records the named case subset this run covers.""" from agentx.version import VERSION grader_id = _resolve_scorer_id(scorer_id, evaluation_settings_id) @@ -318,6 +326,8 @@ def init_run( } if grader_id: payload["evaluationSettingsId"] = grader_id + if split: + payload["split"] = split data = self._request("POST", "/runs", json=self._with_workspace(payload)) return EvaluationRun(**data) @@ -436,6 +446,13 @@ def get_missing_results(self, run_id: str) -> List[Dict[str, Any]]: data = self._request("GET", f"/runs/{run_id}/missing-results") return data if isinstance(data, list) else data.get("missing", []) + def get_submitted_keys(self, run_id: str) -> List[str]: + """Idempotency keys this run has already accepted - what execute() uses to resume a + crashed or interrupted run without re-running (and re-paying for) finished cases.""" + data = self._request("GET", f"/runs/{run_id}/missing-results") + keys = data.get("submittedKeys", []) if isinstance(data, dict) else [] + return [k for k in keys if isinstance(k, str)] + # ------------------------------------------------------------------ # Self-host analysis fallback # diff --git a/agentx/evaluations/datasets.py b/agentx/evaluations/datasets.py index b1c205e..74bcc6c 100644 --- a/agentx/evaluations/datasets.py +++ b/agentx/evaluations/datasets.py @@ -105,9 +105,15 @@ def add_case( expected_tools: Optional[List[str]] = None, trajectory_match_mode: str = "strict", expected_retrieval_context: Optional[Union[str, List[str]]] = None, + splits: Optional[List[str]] = None, ) -> "DatasetBuilder": """Add a case. `judge_guideline` is optional grading guidance specific to this question. + `splits` tags this case with named subsets (e.g. ``["smoke"]``): a run started with + ``client.evaluations.run(dataset_id, subject, split="smoke")`` executes only the tagged + cases (original case indexes are preserved, so per-case comparisons still line up with + full runs). An untagged case belongs to no split and only runs in full runs. + `expected_tools` declares the tool calls a correct run of this case should make. When a result links its trace (return `{"output": ..., "trace_id": span.trace_id}` from the agent function), the engine matches the trace's actual tool-call sequence against it and @@ -149,6 +155,8 @@ def add_case( main["expectedTrajectory"] = {"tools": expected_tools, "mode": trajectory_match_mode} if expected_retrieval_context: main["expectedRetrievalContext"] = expected_retrieval_context + if splits: + main["splits"] = splits self._payload["questions"].append( { "main_question": main, diff --git a/agentx/evaluations/models.py b/agentx/evaluations/models.py index a98e8cf..f58af20 100644 --- a/agentx/evaluations/models.py +++ b/agentx/evaluations/models.py @@ -52,6 +52,9 @@ class TestCase(BaseModel): expected_delegations: Optional[List[str]] = Field(default=None, alias="expectedDelegations") judge_guideline: Optional[str] = Field(default=None, alias="judgeGuideline") smoke_test: Optional[SmokeTestSettings] = Field(default=None, alias="smokeTest") + # Named subsets this case belongs to (e.g. ["smoke"], ["full", "regression"]). + # ``run(dataset_id, split="smoke")`` runs only cases tagged with that split. + splits: Optional[List[str]] = None class Config: populate_by_name = True diff --git a/agentx/evaluations/runner.py b/agentx/evaluations/runner.py index 5e743ae..51fbc57 100644 --- a/agentx/evaluations/runner.py +++ b/agentx/evaluations/runner.py @@ -4,12 +4,12 @@ import os import time import uuid -from typing import Any, Callable, Dict, List, Optional, Set, Union +from typing import Any, Callable, Dict, Iterator, List, Optional, Set, Union from agentx.evaluations.adapters.raw import RawCallableAdapter from agentx.evaluations.adapters.precomputed import PrecomputedAdapter from agentx.evaluations.adapters.http_endpoint import HttpEndpointAdapter -from agentx.evaluations.client import EvaluationsClient +from agentx.evaluations.client import EvaluationsClient, EvaluationSubmissionError from agentx.evaluations.models import ( AnalysisStatus, Dataset, @@ -96,11 +96,14 @@ def __init__( run: EvaluationRun, subject: EvaluationSubject, evaluation_settings: Optional[EvaluationSettings] = None, + split: Optional[str] = None, ): self._client = client self._dataset = dataset self._run = run self._subject = subject + # Named case subset for this run - only cases tagged with it are built/executed. + self._split = split # When set, this run was started with an independently chosen grading # config (evaluation_settings_id) - its fields take precedence over the # dataset's own for anything execution-time reads (see _build_cases). @@ -118,25 +121,63 @@ def __init__( # Step 1: execute # ------------------------------------------------------------------ - def execute(self, adapter: AdapterLike) -> "EvaluationRunContext": + def execute( + self, + adapter: AdapterLike, + concurrency: int = 1, + reuse_outputs_from: Optional[str] = None, + ) -> "EvaluationRunContext": """Run all cases locally and submit batches to AgentX. The whole loop runs inside the eval-run scope (tracing/eval_scope.py): any trace the agent function creates is stamped source="eval-run" + monitor=False automatically, so eval traffic never skews production monitoring and no one has to remember a flag. + + ``concurrency`` > 1 runs the agent callable across a thread pool (results are still + submitted in case order, and the eval-run scope is propagated into the workers). + ``reuse_outputs_from`` replays a previous run's recorded outputs for cases whose query + text is unchanged instead of re-running (and re-paying for) the agent - the judge still + re-scores them, which makes iterating on scorers cheap. Changed or new cases run + normally. """ from agentx.tracing.eval_scope import enter_eval_run, exit_eval_run scope_token = enter_eval_run(self._run.run_id) try: - return self._execute_inner(adapter) + return self._execute_inner(adapter, concurrency=concurrency, reuse_outputs_from=reuse_outputs_from) finally: exit_eval_run(scope_token) - def _execute_inner(self, adapter: AdapterLike) -> "EvaluationRunContext": + def _fetch_reusable_outputs(self, run_id: str) -> Dict[tuple, str]: + """(query, run_number, is_smoke_variant) -> output text, from a previous run's rows. + Keyed on the query TEXT so a reworded case never silently reuses a stale answer.""" + try: + prior = self._client.get_run(run_id) + except Exception as exc: + logger.warning("reuse_outputs_from: could not load run %s (%s) - running everything", run_id, exc) + return {} + reusable: Dict[tuple, str] = {} + for row in prior.get("results", []) or []: + if row.get("error") or row.get("status") == "failed": + continue + output = (row.get("output") or {}).get("text") + query = (row.get("input") or {}).get("query") + if not output or not query: + continue + key = (query, row.get("runNumber") or 1, bool(row.get("isSmokeTestVariant"))) + reusable[key] = output + return reusable + + def _execute_inner( + self, + adapter: AdapterLike, + concurrency: int = 1, + reuse_outputs_from: Optional[str] = None, + ) -> "EvaluationRunContext": normalized = _wrap_adapter(adapter) - cases = _build_cases(self._dataset, self._run, self._evaluation_settings) + cases = _build_cases(self._dataset, self._run, self._evaluation_settings, split=self._split) max_batch = self._run.limits.max_batch_size + reusable = self._fetch_reusable_outputs(reuse_outputs_from) if reuse_outputs_from else {} # Banner sep = "─" * 60 @@ -144,7 +185,7 @@ def _execute_inner(self, adapter: AdapterLike) -> "EvaluationRunContext": framework = self._subject.framework or "custom" runtime = self._subject.runtime or "local" display = self._subject.display_name or "" - n_q = len(self._dataset.questions) + n_q = len({c.question_index for c in cases if not c.is_smoke_test_variant}) n_r = ( self._evaluation_settings.number_of_requests if self._evaluation_settings @@ -171,6 +212,45 @@ def _execute_inner(self, adapter: AdapterLike) -> "EvaluationRunContext": batch: List[EvaluationResult] = [] total = len(cases) + def produce(case: EvaluationCase) -> EvaluationResult: + # Cached replay: same query text at the same repetition reuses the recorded output + # (the server still re-scores it with THIS run's grading config). + cached = reusable.get((case.query, case.run_number, case.is_smoke_test_variant)) + if cached is not None: + return normalize_result( + case, {"output": cached, "metadata": {"reusedFromRun": reuse_outputs_from}} + ) + return normalized(case) + + if concurrency > 1: + import concurrent.futures + import contextvars + + def in_scope(case: EvaluationCase) -> EvaluationResult: + # ContextVars (the eval-run scope) do not cross thread boundaries on their own - + # each worker task runs inside a copy of the submitting thread's context so the + # agent's traces still get stamped source="eval-run". + return contextvars.copy_context().run(produce, case) + + pending = [ + case + for case in cases + if _idem_key(self._run.run_id, case.case_id, case.run_number) not in already_done + ] + executor = concurrent.futures.ThreadPoolExecutor(max_workers=concurrency) + # map() yields in submission order, so batching/submission below stays deterministic. + mapped = executor.map(in_scope, pending) + + def ordered() -> "Iterator[EvaluationResult]": + try: + yield from mapped + finally: + executor.shutdown(wait=True) + + results_iter = ordered() + else: + results_iter = None # sequential path below produces inline + for idx, case in enumerate(cases, start=1): idem_key = _idem_key(self._run.run_id, case.case_id, case.run_number) @@ -179,7 +259,7 @@ def _execute_inner(self, adapter: AdapterLike) -> "EvaluationRunContext": _print_progress(idx, total, case, skipped=True) continue - result = normalized(case) + result = next(results_iter) if results_iter is not None else produce(case) result.idempotency_key = idem_key # Tag the result with the case's model so the server can group it into # the Sovereignty & Portability matrix (the callable may also set it). @@ -207,31 +287,46 @@ def _flush_batch(self, batch: List[EvaluationResult]) -> None: batch_id = str(uuid.uuid4()) n = len(batch) with Spinner(f"Scoring - AI is rating {n} result{'s' if n != 1 else ''}"): - try: - resp = self._client.append_results(self._run.run_id, batch_id, batch) - if resp.live_statistics is not None: - self._live_stats = resp.live_statistics - _say( - f" {green('✓')} Scored {resp.accepted} result{'s' if resp.accepted != 1 else ''}" - ) - logger.info( - "Batch %s: accepted=%d duplicates=%d failed=%d", - batch_id[:8], - resp.accepted, - resp.duplicates, - resp.failed_validation, - ) - except Exception as exc: - _say(f" {red('✗')} Scoring failed: {dim(str(exc))}") - logger.error("Failed to submit batch %s: %s", batch_id[:8], exc) + last_exc: Optional[Exception] = None + for attempt in (1, 2): + try: + resp = self._client.append_results(self._run.run_id, batch_id, batch) + if resp.live_statistics is not None: + self._live_stats = resp.live_statistics + _say( + f" {green('✓')} Scored {resp.accepted} result{'s' if resp.accepted != 1 else ''}" + ) + logger.info( + "Batch %s: accepted=%d duplicates=%d failed=%d", + batch_id[:8], + resp.accepted, + resp.duplicates, + resp.failed_validation, + ) + return + except Exception as exc: + last_exc = exc + if attempt == 1: + logger.warning("Batch %s submission failed, retrying once: %s", batch_id[:8], exc) + # A batch that cannot be submitted must FAIL the run, not print a red line and carry on: + # execute() used to finish "successfully" having uploaded nothing. Failing fast also + # stops paying for agent calls whose results would hit the same broken engine; a + # re-execute() of this context resumes past everything already accepted (idempotency + # keys are deterministic and the engine returns the submitted set). + _say(f" {red('✗')} Scoring failed: {dim(str(last_exc))}") + logger.error("Failed to submit batch %s after retry: %s", batch_id[:8], last_exc) + raise EvaluationSubmissionError( + f"Failed to submit a batch of {n} result(s) to the engine after a retry: {last_exc}. " + "The run was left unfinalized; re-running execute() resumes past already-submitted cases." + ) from last_exc def _fetch_submitted_keys(self) -> Set[str]: + """Keys already accepted by this run - the engine's /missing-results route returns them + so a re-execute() after a crash skips (and never re-pays for) finished cases.""" try: - missing = self._client.get_missing_results(self._run.run_id) - # missing-results returns cases NOT yet submitted - we want the inverse - # but if the endpoint isn't live yet, just return empty set - return set() + return set(self._client.get_submitted_keys(self._run.run_id)) except Exception: + # Older engines without the route: no resume, identical to the historical behavior. return set() # ------------------------------------------------------------------ @@ -517,11 +612,16 @@ def run( subject: Union[Dict[str, Any], EvaluationSubject], scorer_id: Optional[str] = None, evaluation_settings_id: Optional[str] = None, + split: Optional[str] = None, ) -> EvaluationRunContext: """Start a run of ``dataset_id`` against ``subject``. Pass ``scorer_id`` (an LLM Judge Scorer's id, e.g. from ``client.monitor.judge_scorers``) to grade with a specific scorer instead of the dataset's default. ``evaluation_settings_id`` is the - pre-consolidation alias for the same id and keeps working.""" + pre-consolidation alias for the same id and keeps working. + + ``split`` runs only the cases tagged with that named subset (``add_case(..., + splits=["smoke"])``) - the cheap-PR-run vs nightly-full-run workflow. Original case + indexes are preserved so per-case comparisons line up with full runs.""" from agentx.evaluations.client import _resolve_scorer_id if isinstance(subject, dict): @@ -532,18 +632,24 @@ def run( evaluation_settings = ( self._client.get_evaluation_settings(grader_id) if grader_id else None ) - run = self._client.init_run(dataset_id, subject, scorer_id=grader_id) + run = self._client.init_run(dataset_id, subject, scorer_id=grader_id, split=split) + case_count = ( + sum(1 for q in dataset.questions if split in (q.main_question.splits or [])) + if split + else len(dataset.questions) + ) logger.info( - "Started evaluation run %s on dataset %s (%d case(s), %d repetition(s))", + "Started evaluation run %s on dataset %s (%d case(s)%s, %d repetition(s))", run.run_id, dataset_id, - len(dataset.questions), + case_count, + f' in split "{split}"' if split else "", evaluation_settings.number_of_requests if evaluation_settings else dataset.number_of_requests, ) return EvaluationRunContext( - self._client, dataset, run, subject, evaluation_settings=evaluation_settings + self._client, dataset, run, subject, evaluation_settings=evaluation_settings, split=split ) @@ -568,6 +674,7 @@ def _build_cases( dataset: Dataset, run: EvaluationRun, evaluation_settings: Optional[EvaluationSettings] = None, + split: Optional[str] = None, ) -> List[EvaluationCase]: cases: List[EvaluationCase] = [] # When an independent evaluation_settings was chosen (evaluation_settings_id @@ -596,6 +703,10 @@ def _build_cases( } for q_idx, question in enumerate(dataset.questions): mq = question.main_question + # Split filtering preserves q_idx: a "smoke" run's case 7 is the same case 7 a full run + # scores, so per-case comparisons line up across the two. + if split and split not in (mq.splits or []): + continue for run_num in range(1, n_runs + 1): for model in models: suffix = f"::{model}" if model else "" diff --git a/agentx/monitor/client.py b/agentx/monitor/client.py index 66a0ac2..7550e91 100644 --- a/agentx/monitor/client.py +++ b/agentx/monitor/client.py @@ -39,6 +39,35 @@ class AgentXValidationError(AgentXMonitorError): pass +class CalibrationSummary(dict): + """Judge Calibration numbers (dict subclass, so existing key access keeps working). + Properties mirror the wire's exact camelCase keys.""" + + @property + def compared_count(self) -> int: + return int(self.get("comparedCount") or 0) + + @property + def agreement_rate(self): + return self.get("agreementRate") + + @property + def false_positive_rate(self): + return self.get("falsePositiveRate") + + @property + def false_negative_rate(self): + return self.get("falseNegativeRate") + + @property + def reported_count(self) -> int: + return int(self.get("reportedCount") or 0) + + @property + def review_label_count(self) -> int: + return int(self.get("reviewLabelCount") or 0) + + class MonitorClient: """Low-level HTTP client for the Monitor API (``/monitor``). Accessed via ``client.monitor`` on the top-level :class:`agentx.AgentX` instance; most callers @@ -82,6 +111,11 @@ def __init__( self.patterns = MonitorPatternClient(self) self.signals = MonitorSignalClient(self) + from agentx.monitor.review_queue import ReviewQueueClient + + # The human-review queue (list / queue / label / dismiss) - what makes the + # label-and-calibrate loop scriptable instead of dashboard-only. + self.review_queue = ReviewQueueClient(self) from agentx.monitor.scorers import ScorersClient # Scorers-catalog administration as code: template enable/disable, code/external scorer # CRUD and dry runs - full parity with the dashboard's Scorers page (P1.3). @@ -230,15 +264,19 @@ def kpis(self, window: str = "7d") -> dict: plus deltas vs the prior window and the run-outcome breakdown.""" return self._request("GET", "/kpis", params={"window": window}) - def calibration(self, window: str = "7d") -> dict: + def calibration(self, window: str = "7d") -> "CalibrationSummary": """Project-level judge calibration over a window ("24h", "7d", or "30d"): how often AgentX's own verdicts agreed with real-world ground truth reported later (ops outcomes - via ``client.outcomes`` and end-user downvotes). Returns the dashboard's Judge - Calibration numbers: compared count, agreement, falsePositiveRate, falseNegativeRate. - Per-evaluator calibration lives on ``client.monitor.online_evaluators.calibration``.""" - return self._request( - "GET", "/agent-monitoring/calibration", - base=self._api_root(), params={"window": window}, + via ``client.outcomes``, end-user downvotes, and human review labels). Returns the + dashboard's Judge Calibration numbers with these exact keys: ``comparedCount``, + ``agreementRate``, ``falsePositiveRate``, ``falseNegativeRate`` (plus + ``reportedCount``/``reviewLabelCount``/``noVerdictCount``). Per-scorer calibration + lives on ``client.monitor.judge_scorers.calibration(scorer_id)``.""" + return CalibrationSummary( + self._request( + "GET", "/agent-monitoring/calibration", + base=self._api_root(), params={"window": window}, + ) ) # ------------------------------------------------------------------ @@ -352,10 +390,19 @@ def validate_online_evaluator_tuning( base=self._api_root(), json={**criteria, "window": window}, timeout=600, ) - def publish_online_evaluator_tuning(self, evaluator_id: str, criteria: dict) -> dict: + def publish_online_evaluator_tuning( + self, evaluator_id: str, criteria: dict, *, validation: Optional[dict] = None, force: bool = False + ) -> dict: + # The engine gates publish on validation provenance (and refuses a measured regression) + # unless forced - see judge_scorers.publish_tuning for the full story. + payload = dict(criteria) + if validation is not None: + payload["validation"] = validation + if force: + payload["force"] = True return self._request( "POST", f"/agent-monitoring/online-evaluators/{evaluator_id}/tune/publish", - base=self._api_root(), json=criteria, timeout=60, + base=self._api_root(), json=payload, timeout=60, ) def update_profile(self, agent_id: str, payload: dict) -> MonitorProfile: diff --git a/agentx/monitor/judge_scorers.py b/agentx/monitor/judge_scorers.py index d8653ed..00ed507 100644 --- a/agentx/monitor/judge_scorers.py +++ b/agentx/monitor/judge_scorers.py @@ -280,12 +280,30 @@ def validate_tuning(self, scorer_id: str, criteria: Dict[str, Any], window: str timeout=600, ) - def publish_tuning(self, scorer_id: str, criteria: Dict[str, Any]) -> dict: + def publish_tuning( + self, + scorer_id: str, + criteria: Dict[str, Any], + *, + validation: Optional[Dict[str, Any]] = None, + force: bool = False, + ) -> dict: """Write tuned criteria onto the scorer's rubric - it applies everywhere the scorer is - used: online scoring, offline dataset runs, and the playground.""" - return self._request( - "POST", f"/online-evaluators/{self._profile_id(scorer_id)}/tune/publish", json=dict(criteria) - ) + used: online scoring, offline dataset runs, and the playground. + + The engine gates publish on provenance: pass ``validation`` (the dict returned by + ``validate_tuning``, or at least its ``verdict``/``netAgreementGain``) so the version + history records what the change measurably did; a ``regressed`` verdict is refused. + ``force=True`` publishes without (or despite) validation - deliberate escape hatch.""" + payload = dict(criteria) + if validation is not None: + payload["validation"] = { + "verdict": validation.get("verdict"), + "netAgreementGain": validation.get("netAgreementGain"), + } + if force: + payload["force"] = True + return self._request("POST", f"/online-evaluators/{self._profile_id(scorer_id)}/tune/publish", json=payload) def ratings(self, scorer_id: str, window: str = "7d") -> "List[OnlineEvaluatorRatingPoint]": """Bucketed average-rating-over-time for this scorer's live checks - same typed points diff --git a/agentx/monitor/online_evaluators.py b/agentx/monitor/online_evaluators.py index 5f1fefc..1adb05f 100644 --- a/agentx/monitor/online_evaluators.py +++ b/agentx/monitor/online_evaluators.py @@ -157,9 +157,13 @@ def validate_tuning(self, evaluator_id: str, criteria: dict, window: str = "7d") is {acceptanceCriteria, rejectionCriteria, evaluationCriteria} from tune().""" return self._client.validate_online_evaluator_tuning(evaluator_id, criteria, window) - def publish_tuning(self, evaluator_id: str, criteria: dict) -> dict: - """Publish tuned criteria onto the evaluator's config (the human-approval step).""" - return self._client.publish_online_evaluator_tuning(evaluator_id, criteria) + def publish_tuning( + self, evaluator_id: str, criteria: dict, *, validation: Optional[dict] = None, force: bool = False + ) -> dict: + """Publish tuned criteria onto the evaluator's config (the human-approval step). + Pass ``validation`` (the ``validate_tuning`` result) - the engine refuses an unvalidated + publish, and a ``regressed`` verdict, unless ``force=True``.""" + return self._client.publish_online_evaluator_tuning(evaluator_id, criteria, validation=validation, force=force) def ratings(self, evaluator_id: str, window: str = "7d") -> List[OnlineEvaluatorRatingPoint]: """Bucketed average-rating-over-time for this evaluator. ``window`` is one of diff --git a/agentx/monitor/review_queue.py b/agentx/monitor/review_queue.py new file mode 100644 index 0000000..dbfea00 --- /dev/null +++ b/agentx/monitor/review_queue.py @@ -0,0 +1,93 @@ +from __future__ import annotations + +import logging +from typing import Any, Dict, List, Optional, TYPE_CHECKING + +if TYPE_CHECKING: + from agentx.monitor.client import MonitorClient + +logger = logging.getLogger(__name__) + + +class ReviewQueueItem(dict): + """Wire object for one human-review item (dict subclass so unknown fields round-trip).""" + + @property + def id(self) -> str: + return self["_id"] + + @property + def trace_id(self) -> Optional[str]: + return self.get("traceId") + + @property + def status(self) -> Optional[str]: + return self.get("status") + + @property + def label(self) -> Optional[str]: + return self.get("label") + + @property + def judge_score_at_queue(self) -> Optional[float]: + return self.get("judgeScoreAtQueue") + + +class ReviewQueueClient: + """Surfaced as ``client.monitor.review_queue``: the human-review queue behind the dashboard's + Review tab, scriptable - so the label-and-calibrate loop (sample traces, label them + good/bad, optionally re-score) can run end to end from code. Labels feed judge calibration + and become judge-tuning evidence. + + The engine refuses duplicates (409, a trace already pending) and a full queue (429, pending + cap reached); both surface as raised errors with the engine's reason. + """ + + def __init__(self, client: "MonitorClient"): + self._client = client + + def list(self, status: Optional[str] = None, source: Optional[str] = None, limit: int = 100) -> List[ReviewQueueItem]: + """Queue items, newest first. ``status``: "pending" | "labeled" | "skipped" | "all" + (server default: pending). ``source``: "manual" | "rule" | "all".""" + params: Dict[str, Any] = {"limit": limit} + if status is not None: + params["status"] = status + if source is not None: + params["source"] = source + data = self._client._request("GET", "/agent-monitoring/review-queue", base=self._client._api_root(), params=params) + return [ReviewQueueItem(item) for item in data.get("items", [])] + + def queue(self, trace_id: str, note: Optional[str] = None) -> ReviewQueueItem: + """Send a trace to human review (the SDK-side twin of the dashboard's "Send to review").""" + payload: Dict[str, Any] = {"traceId": trace_id, "source": "manual"} + if note: + payload["note"] = note + data = self._client._request("POST", "/agent-monitoring/review-queue", base=self._client._api_root(), json=payload) + return ReviewQueueItem(data.get("item", data)) + + def label( + self, + item_id: str, + label: str, + *, + corrected_score: Optional[float] = None, + note: Optional[str] = None, + ) -> ReviewQueueItem: + """Record the human verdict on a queued item. ``label`` is "good" or "bad"; + ``corrected_score`` (0-10) optionally re-scores the judge's own rating for the trace - + the pair that calibration consumes.""" + if label not in ("good", "bad"): + raise ValueError('label must be "good" or "bad"') + payload: Dict[str, Any] = {"label": label} + if corrected_score is not None: + payload["correctedScore"] = corrected_score + if note is not None: + payload["note"] = note + data = self._client._request( + "PATCH", f"/agent-monitoring/review-queue/{item_id}", base=self._client._api_root(), json=payload + ) + return ReviewQueueItem(data.get("item", data)) + + def dismiss(self, item_id: str) -> None: + """Remove an item from the queue without a verdict (does not feed calibration).""" + self._client._request("DELETE", f"/agent-monitoring/review-queue/{item_id}", base=self._client._api_root()) diff --git a/agentx/testing.py b/agentx/testing.py index 6ccec09..00a0642 100644 --- a/agentx/testing.py +++ b/agentx/testing.py @@ -46,7 +46,9 @@ def _format_failures(gate: Any) -> str: for check in checks: get = check.get if isinstance(check, dict) else lambda k, d=None: getattr(check, k, d) status = "PASS" if get("passed") else "FAIL" - lines.append(f" [{status}] {get('name', 'check')}: {get('detail', '')}") + # The engine names each check under the key "check" ("fail-under" / "no-regression"); + # "name" is kept as a fallback for any older payload shape. + lines.append(f" [{status}] {get('check') or get('name', 'check')}: {get('detail', '')}") average = getattr(gate, "average_rating", None) if average is not None: lines.append(f" average rating: {average}") diff --git a/tests/test_integrations.py b/tests/test_integrations.py index ae2cc26..909267e 100644 --- a/tests/test_integrations.py +++ b/tests/test_integrations.py @@ -741,6 +741,16 @@ def _reset_litellm_callback_state(litellm) -> None: litellm._async_input_callback = [] +def _wait_for_send(tracer, timeout: float = 5.0) -> None: + """LiteLLM dispatches success/failure callbacks on a background thread (or a + fire-and-forget task) even for sync completions, so asserting on tracer._send + immediately after the call races the dispatcher - reliably losing under a loaded + full-suite run. Poll until the mock is called instead of sleeping a fixed beat.""" + deadline = time.time() + timeout + while time.time() < deadline and not tracer._send.called: + time.sleep(0.02) + + def test_litellm_sync_completion_traces_call(): litellm = pytest.importorskip("litellm") _reset_litellm_callback_state(litellm) @@ -757,6 +767,7 @@ def test_litellm_sync_completion_traces_call(): litellm.callbacks = [] assert response.choices[0].message.content == "Hello there!" + _wait_for_send(tracer) tracer._send.assert_called_once() _, kwargs = tracer._send.call_args assert kwargs["output"] == "Hello there!" @@ -790,6 +801,7 @@ async def run(): litellm.callbacks = [] assert response.choices[0].message.content == "Hello async!" + _wait_for_send(tracer) tracer._send.assert_called_once() _, kwargs = tracer._send.call_args assert kwargs["output"] == "Hello async!" @@ -818,6 +830,7 @@ def test_litellm_streaming_traces_aggregated_response(): litellm.callbacks = [] assert len(chunks) > 1 + _wait_for_send(tracer) tracer._send.assert_called_once() _, kwargs = tracer._send.call_args assert kwargs["output"] == "Hello streamed!" @@ -839,6 +852,7 @@ def test_litellm_failure_records_error(): finally: litellm.callbacks = [] + _wait_for_send(tracer) tracer._send.assert_called_once() _, kwargs = tracer._send.call_args assert "boom" in kwargs["error"] diff --git a/tests/test_review_queue.py b/tests/test_review_queue.py new file mode 100644 index 0000000..cf4907c --- /dev/null +++ b/tests/test_review_queue.py @@ -0,0 +1,73 @@ +"""Unit tests for client.monitor.review_queue (list / queue / label / dismiss) - wire-level, +no engine required. The engine-side contract is pinned by its review-queue routes.""" + +from typing import Any, Dict, List + +import pytest + +from agentx.monitor.review_queue import ReviewQueueClient, ReviewQueueItem + + +class FakeMonitorClient: + def __init__(self, responses: List[Any]): + self.calls: List[Dict[str, Any]] = [] + self._responses = responses + + def _api_root(self) -> str: + return "http://engine:4700/api/v1" + + def _request(self, method: str, path: str, base: str = "", **kwargs: Any) -> Any: + self.calls.append({"method": method, "path": path, "base": base, **kwargs}) + return self._responses.pop(0) if self._responses else {} + + +def test_list_hits_the_queue_with_filters(): + fake = FakeMonitorClient([{"items": [{"_id": "r1", "traceId": "t1", "status": "pending"}], "pending": 1}]) + items = ReviewQueueClient(fake).list(status="pending", source="rule", limit=25) # type: ignore[arg-type] + call = fake.calls[0] + assert call["method"] == "GET" + assert call["path"] == "/agent-monitoring/review-queue" + assert call["base"] == "http://engine:4700/api/v1" + assert call["params"] == {"limit": 25, "status": "pending", "source": "rule"} + assert isinstance(items[0], ReviewQueueItem) + assert items[0].id == "r1" + assert items[0].trace_id == "t1" + + +def test_queue_sends_trace_and_note(): + fake = FakeMonitorClient([{"item": {"_id": "r2", "traceId": "t9"}}]) + item = ReviewQueueClient(fake).queue("t9", note="looks off") # type: ignore[arg-type] + call = fake.calls[0] + assert call["method"] == "POST" + assert call["json"] == {"traceId": "t9", "source": "manual", "note": "looks off"} + assert item.id == "r2" + + +def test_label_validates_and_sends_the_calibration_pair(): + fake = FakeMonitorClient([{"item": {"_id": "r3", "label": "bad", "judgeScoreAtQueue": 8.0}}]) + client = ReviewQueueClient(fake) # type: ignore[arg-type] + item = client.label("r3", "bad", corrected_score=2, note="hallucinated policy") + call = fake.calls[0] + assert call["method"] == "PATCH" + assert call["path"] == "/agent-monitoring/review-queue/r3" + assert call["json"] == {"label": "bad", "correctedScore": 2, "note": "hallucinated policy"} + assert item.label == "bad" + assert item.judge_score_at_queue == 8.0 + + with pytest.raises(ValueError): + client.label("r3", "meh") + + +def test_dismiss_deletes_the_item(): + fake = FakeMonitorClient([""]) + ReviewQueueClient(fake).dismiss("r4") # type: ignore[arg-type] + call = fake.calls[0] + assert call["method"] == "DELETE" + assert call["path"] == "/agent-monitoring/review-queue/r4" + + +def test_registered_on_the_monitor_client(): + from agentx.monitor.client import MonitorClient + + monitor = MonitorClient(api_key="k", base_url="http://engine:4700/api/v1/monitor") + assert isinstance(monitor.review_queue, ReviewQueueClient) diff --git a/tests/test_runner_features.py b/tests/test_runner_features.py new file mode 100644 index 0000000..8c8edbd --- /dev/null +++ b/tests/test_runner_features.py @@ -0,0 +1,146 @@ +"""Runner-level tests for dataset splits, concurrent execution, output reuse, and the +fail-fast batch submission - all against a fake EvaluationsClient, no engine required.""" + +import threading +import time +from typing import Any, Dict, List, Optional + +import pytest + +from agentx.evaluations.client import EvaluationSubmissionError +from agentx.evaluations.models import ( + BatchAppendResponse, + Dataset, + EvaluationRun, + EvaluationSubject, +) +from agentx.evaluations.runner import EvaluationRunContext, _build_cases + + +def make_dataset(**overrides: Any) -> Dataset: + payload: Dict[str, Any] = { + "_id": "ds-1", + "name": "split dataset", + "questions": [ + {"main_question": {"query": "q0", "splits": ["smoke"]}}, + {"main_question": {"query": "q1"}}, + {"main_question": {"query": "q2", "splits": ["smoke", "full"]}}, + ], + } + payload.update(overrides) + return Dataset(**payload) + + +def make_run() -> EvaluationRun: + return EvaluationRun(runId="run-1", datasetId="ds-1") + + +class FakeClient: + def __init__(self, prior_run: Optional[Dict[str, Any]] = None, fail_batches: int = 0): + self.batches: List[List[Any]] = [] + self._prior_run = prior_run + self._fail_remaining = fail_batches + + def get_submitted_keys(self, run_id: str) -> List[str]: + return [] + + def append_results(self, run_id: str, batch_id: str, results: List[Any]) -> BatchAppendResponse: + if self._fail_remaining > 0: + self._fail_remaining -= 1 + raise RuntimeError("engine down") + self.batches.append(list(results)) + return BatchAppendResponse( + runId=run_id, batchId=batch_id, accepted=len(results), duplicates=0, failedValidation=0 + ) + + def get_run(self, run_id: str) -> Dict[str, Any]: + assert self._prior_run is not None + return self._prior_run + + +def make_context(client: FakeClient, split: Optional[str] = None) -> EvaluationRunContext: + return EvaluationRunContext( + client, # type: ignore[arg-type] + make_dataset(), + make_run(), + EvaluationSubject(), + split=split, + ) + + +def test_build_cases_filters_by_split_and_keeps_indexes(): + cases = _build_cases(make_dataset(), make_run(), split="smoke") + assert [c.question_index for c in cases] == [0, 2] + assert [c.query for c in cases] == ["q0", "q2"] + + all_cases = _build_cases(make_dataset(), make_run()) + assert [c.question_index for c in all_cases] == [0, 1, 2] + + +def test_execute_runs_only_the_split(monkeypatch): + monkeypatch.setenv("AGENTX_EVAL_QUIET", "1") + client = FakeClient() + ctx = make_context(client, split="smoke") + seen: List[str] = [] + + def agent(case): + seen.append(case.query) + return f"answer to {case.query}" + + ctx.execute(agent) + assert seen == ["q0", "q2"] + submitted = [r for batch in client.batches for r in batch] + assert [r.question_index for r in submitted] == [0, 2] + + +def test_concurrent_execution_preserves_submission_order(monkeypatch): + monkeypatch.setenv("AGENTX_EVAL_QUIET", "1") + client = FakeClient() + ctx = make_context(client) + threads: List[str] = [] + + def agent(case): + threads.append(threading.current_thread().name) + # The FIRST case is the slowest - order must still hold. + time.sleep(0.2 if case.query == "q0" else 0.01) + return f"answer to {case.query}" + + ctx.execute(agent, concurrency=3) + submitted = [r for batch in client.batches for r in batch] + assert [r.question_index for r in submitted] == [0, 1, 2] + assert any(name != "MainThread" for name in threads) + + +def test_reuse_outputs_from_replays_matching_queries(monkeypatch): + monkeypatch.setenv("AGENTX_EVAL_QUIET", "1") + prior = { + "results": [ + {"input": {"query": "q0"}, "output": {"text": "cached answer 0"}, "runNumber": 1}, + # q1's prior row errored - must NOT be reused. + {"input": {"query": "q1"}, "output": {"text": "bad"}, "runNumber": 1, "status": "failed"}, + ] + } + client = FakeClient(prior_run=prior) + ctx = make_context(client) + ran: List[str] = [] + + def agent(case): + ran.append(case.query) + return f"fresh answer to {case.query}" + + ctx.execute(agent, reuse_outputs_from="run-0") + # q0 replayed from cache; q1 (failed before) and q2 (no cache) ran for real. + assert ran == ["q1", "q2"] + submitted = [r for batch in client.batches for r in batch] + assert submitted[0].output == {"text": "cached answer 0"} + assert submitted[0].metadata.get("reusedFromRun") == "run-0" + assert submitted[1].output == {"text": "fresh answer to q1"} + + +def test_flush_batch_failure_raises_after_one_retry(monkeypatch): + monkeypatch.setenv("AGENTX_EVAL_QUIET", "1") + client = FakeClient(fail_batches=2) # first attempt + its retry both fail + ctx = make_context(client) + + with pytest.raises(EvaluationSubmissionError): + ctx.execute(lambda case: "x")