diff --git a/packages/gooddata-eval/src/gooddata_eval/cli/agentic_runner.py b/packages/gooddata-eval/src/gooddata_eval/cli/agentic_runner.py index 0bd6f5cf8..af67ce4cc 100644 --- a/packages/gooddata-eval/src/gooddata_eval/cli/agentic_runner.py +++ b/packages/gooddata-eval/src/gooddata_eval/cli/agentic_runner.py @@ -11,6 +11,7 @@ from gooddata_eval.core.agentic._langfuse import make_langfuse_client from gooddata_eval.core.agentic._trace_linker import BackgroundTraceLinker, SubmitTraceLink, run_trace_link_inline from gooddata_eval.core.agentic.alert_skill import evaluate_agentic_alert_skill +from gooddata_eval.core.agentic.anomaly_detection import evaluate_agentic_anomaly_detection from gooddata_eval.core.agentic.conversation import ConversationFixture, evaluate_agentic_conversation from gooddata_eval.core.agentic.general_question import evaluate_agentic_general_question from gooddata_eval.core.agentic.guardrail import evaluate_agentic_guardrail @@ -44,6 +45,7 @@ class _LfKw(TypedDict, total=False): "agentic_guardrail", "agentic_conversation", "agentic_kda_skill", + "agentic_anomaly_detection", } ) @@ -228,6 +230,17 @@ def _dispatch_agentic( agent_id=agent_id, **lf_kw, ) + elif kind == "agentic_anomaly_detection": + return evaluate_agentic_anomaly_detection( + host=host, + token=token, + workspace_id=workspace_id, + question=item.question, + expected_output=eo if isinstance(eo, dict) else {}, + k=k, + agent_id=agent_id, + **lf_kw, + ) elif kind == "agentic_conversation": fixture_data = eo.get("fixture") or eo if isinstance(eo, dict) else {} return evaluate_agentic_conversation( diff --git a/packages/gooddata-eval/src/gooddata_eval/core/agentic/anomaly_detection.py b/packages/gooddata-eval/src/gooddata_eval/core/agentic/anomaly_detection.py new file mode 100644 index 000000000..eea66ffb3 --- /dev/null +++ b/packages/gooddata-eval/src/gooddata_eval/core/agentic/anomaly_detection.py @@ -0,0 +1,629 @@ +# (C) 2026 GoodData Corporation. All rights reserved. +"""Agentic anomaly-detection skill evaluation runner. + +The skill is a single tool over a chart the agent builds first: + + create_adhoc_visualization(...) a measure over a time dimension + execute_anomaly_detection(visualization_ref, max_points) + +Its parameters say nothing about *what* to look for -- there is no threshold, no +sensitivity, no expected anomaly -- so unlike forecasting and what-if, the tool call itself +carries almost no assertable intent. What is assertable is the chart the detection ran on: +the measure and the time granularity are the whole of the question "did it look at the +right series", and getting either wrong makes the result meaningless however well the +detection itself performed. + +Granularity is resolved with the tool's own token map and in its own order -- field tokens +first, then a relative date filter's granularity -- so the evaluation reads the chart the +way the service does. It differs only in which token wins inside a single reference, and +only where the service itself is not deterministic: see ``_granularity_from``. + +The count of flagged points is reported but never asserted: whether a real series contains +anomalies is a property of the data, not of the agent, and a fixture that demanded some +would start failing the day the warehouse refreshed. +""" + +from __future__ import annotations + +import logging +import os +import re +from dataclasses import dataclass, field +from typing import Any + +from gooddata_eval.core.agentic._trace_linker import ( + RunIdentity, + RunTraceContext, + SubmitTraceLink, + open_trace_window, + run_trace_link_inline, + submit_trace_scoring, + utc_now, +) +from gooddata_eval.core.chat.render import render_answer_text +from gooddata_eval.core.chat.sse_client import ChatClient +from gooddata_eval.core.config import ReasoningEffort +from gooddata_eval.core.models import ( + AgenticAssertionError, + AgenticEvalOutcome, + ChatResult, + ReasoningStepEvent, + ToolCallEvent, + build_latency_breakdown, + shift_and_index_events, +) + +_log = logging.getLogger(__name__) + +_DEFAULT_K = 1 +# Observed live, the agent completes this in one turn (search, build, detect). The budget +# is disambiguation headroom for questions that name an ambiguous measure. +_DEFAULT_MAX_ITERATIONS = 4 + +# Copied verbatim from gen-ai's own token map, so a chart this scores as monthly is the one +# the tool analysed monthly. Note there is deliberately no "date" key: `label/process_date` +# with no granularity suffix matches nothing, and the tool then refuses the call rather than +# guessing daily -- an evaluator that guessed instead would score a granularity the service +# never used. "week_us" is unreachable through tokenization (it splits into "week" and "us") +# but is kept so this stays a copy rather than an edit. +_TOKEN_TO_GRANULARITY = { + "hour": "HOUR", + "day": "DAY", + "week": "WEEK", + "week_us": "WEEK", + "month": "MONTH", + "quarter": "QUARTER", + "year": "YEAR", +} + + +def _build_clarification_prompt(agent_message: str, expected_output: dict) -> str: + """The simulated-user reply, mentioning only the hints the fixture actually supplies.""" + hints: list[str] = [] + metric = expected_output.get("metric") + if metric: + hints.append(f"the measure to analyse is {metric}") + granularity = expected_output.get("granularity") + if granularity: + hints.append(f"the time granularity is {granularity}") + period = expected_output.get("period") + if period: + hints.append(f"the period to cover is {period}") + reference = "; ".join(hints) + return ( + f"You are simulating a user in a conversation with a BI assistant that detects anomalies " + f"in a metric over time. The assistant asked: '{agent_message}'. " + + (f"For reference, {reference}. " if reference else "") + + "Reply briefly as the user, answering whichever of those the assistant actually asked about." + ) + + +def generate_simulated_anomaly_response(agent_message: str, expected_output: dict) -> str: + """Generate a user reply to keep the anomaly conversation going (gpt-4o-mini). + + Always OpenAI regardless of the workspace's own model: harness plumbing, not the system + under test. + """ + try: + from openai import OpenAI # noqa: PLC0415 + except ImportError as exc: + raise RuntimeError("openai package is required for generate_simulated_anomaly_response") from exc + + api_key = os.environ.get("OPENAI_API_KEY") + if not api_key: + raise OSError("OPENAI_API_KEY environment variable is not set") + + client = OpenAI(api_key=api_key) + response = client.chat.completions.create( + model="gpt-4o-mini", + messages=[{"role": "user", "content": _build_clarification_prompt(agent_message, expected_output)}], + max_tokens=150, + temperature=0, + timeout=30, + ) + return response.choices[0].message.content or "Please proceed with the most complete option." + + +def _extract_anomaly_calls(tool_call_events: list[ToolCallEvent]) -> tuple[dict | None, dict | None]: + """Return (visualization_args, execute_result) for the LAST create/execute pair. + + A new chart clears any earlier detection result: that result described the chart it + followed. Taking the last of each independently would score a fresh series against a + detection that never ran on it. + """ + viz_args: dict | None = None + execute_result: dict | None = None + for tc in tool_call_events: + if tc.function_name == "create_adhoc_visualization": + args = tc.parsed_arguments() or {} + viz = args.get("visualization") + viz_args = viz if isinstance(viz, dict) else args + execute_result = None + elif tc.function_name == "execute_anomaly_detection" and tc.result: + execute_result = tc.parsed_result() + return viz_args, execute_result + + +def _granularity_from(value: str) -> str | None: + """The granularity a field reference names, taking the LAST recognised token. + + gen-ai tokenizes the same string into a *set* and returns the first match it iterates, + which is stable for a reference naming one granularity -- the normal case, including + every date label in the eval workspace. It is not stable for a reference naming two, + e.g. a snake_case ``first_day_quarter.month`` whose tokens contain "day", "quarter" and + "month": there the service's answer depends on set iteration order. Reading the last + token instead takes the suffix, which is what a dotted label means, and is + deterministic either way -- a scorer must not be a coin flip even where the thing it + scores is one. + """ + parts = [part for part in re.split(r"[^a-zA-Z0-9]+", value.lower()) if part] + for part in reversed(parts): + if part in _TOKEN_TO_GRANULARITY: + return _TOKEN_TO_GRANULARITY[part] + return None + + +def _metric_uris(viz_args: dict | None) -> set[str]: + """The metric URIs the chart measures, resolved through its field aliases. + + Reads raw tool-call arguments rather than a parsed CreatedVisualization, so a field may + be a bare URI string as well as an object. + """ + query = (viz_args or {}).get("query") + fields = query.get("fields") if isinstance(query, dict) else None + if not isinstance(fields, dict): + return set() + uris: set[str] = set() + for alias in (viz_args or {}).get("metrics") or list(fields): + name = alias.get("field") if isinstance(alias, dict) else alias + field_def = fields.get(name) + if isinstance(field_def, dict) and field_def.get("using"): + uris.add(str(field_def["using"])) + elif isinstance(field_def, str): + uris.add(field_def) + return {u for u in uris if u.startswith(("metric/", "fact/"))} + + +def _inferred_granularity(viz_args: dict | None) -> str | None: + """The granularity the detection tool would read off this chart. + + Follows gen-ai's own resolution ORDER -- field ``using``/``title`` tokens first, then a + relative date filter's ``granularity`` -- so a chart scored here as monthly is the one + the service analysed monthly. Within a single reference it differs on purpose; see + ``_granularity_from``. + """ + query = (viz_args or {}).get("query") + if not isinstance(query, dict): + return None + + fields = query.get("fields") + if isinstance(fields, dict): + for field_def in fields.values(): + if isinstance(field_def, dict): + reference = f"{field_def.get('using') or ''} {field_def.get('title') or ''}" + elif isinstance(field_def, str): + reference = field_def + else: + continue + granularity = _granularity_from(reference) + if granularity is not None: + return granularity + + filter_by = query.get("filter_by") + if isinstance(filter_by, dict): + for filter_def in filter_by.values(): + if not isinstance(filter_def, dict) or filter_def.get("type") != "date_filter": + continue + granularity = _granularity_from(str(filter_def.get("granularity", ""))) + if granularity is not None: + return granularity + return None + + +def _point_count(execute_result: dict | None) -> int | None: + data = (execute_result or {}).get("data") + if not isinstance(data, dict): + return None + count = data.get("point_count") + return count if isinstance(count, int) else None + + +@dataclass +class AnomalyEvaluation: + """Scores for a single anomaly-detection run. + + ``triggered``/``executed``/``success``/``turn_completed`` are the shared process checks. + ``metric_correct`` and ``granularity_correct`` ask whether the detection ran on the + right series at all; each is True when the fixture did not pin it, and ``asserted`` + records which ones it did. + """ + + triggered: bool + executed: bool + success: bool + turn_completed: bool + metric_correct: bool + granularity_correct: bool + asserted: list[str] = field(default_factory=list) + disambiguated: bool = False + + @property + def strict_pass(self) -> bool: + return all( + [ + self.triggered, + self.executed, + self.success, + self.turn_completed, + self.metric_correct, + self.granularity_correct, + ] + ) + + +@dataclass +class AnomalyRunResult: + """Outcome of one run (one conversation, up to max_iterations messages).""" + + conversation_id: str + evaluation: AnomalyEvaluation + actual_visualization: dict | None + actual_execute_result: dict | None + turn_wall_clock_sec: float | None = None + reasoning_steps: list[str] = field(default_factory=list) + response_id: str | None = None + tool_call_events: list[ToolCallEvent] = field(default_factory=list) + reasoning_step_events: list[ReasoningStepEvent] = field(default_factory=list) + + +@dataclass +class AgenticAnomalySummary: + """Aggregated outcome of K runs for one anomaly-detection item.""" + + run_results: list[AnomalyRunResult] + pass_at_k: bool + pass_power_k: bool + best: AnomalyRunResult + + +def _evaluate_run( + viz_args: dict | None, + execute_result: dict | None, + expected_output: dict, + turn_completed: bool, + disambiguated: bool = False, +) -> AnomalyEvaluation: + triggered = execute_result is not None or viz_args is not None + executed = execute_result is not None + success = executed and execute_result.get("success") is True + asserted: list[str] = [] + + expected_metric = expected_output.get("metric") + if not expected_metric: + metric_correct = True + else: + asserted.append("metric") + wanted = {expected_metric} if isinstance(expected_metric, str) else set(expected_metric) + metric_correct = bool(_metric_uris(viz_args) & wanted) + + expected_granularity = expected_output.get("granularity") + if not expected_granularity: + granularity_correct = True + else: + asserted.append("granularity") + granularity_correct = _inferred_granularity(viz_args) == str(expected_granularity).upper() + + return AnomalyEvaluation( + triggered=triggered, + executed=executed, + success=success, + turn_completed=turn_completed, + metric_correct=metric_correct, + granularity_correct=granularity_correct, + asserted=asserted, + disambiguated=disambiguated, + ) + + +def run_agentic_anomaly_detection( + host: str, + token: str, + workspace_id: str, + question: str, + expected_output: dict, + k: int = _DEFAULT_K, + max_iterations: int = _DEFAULT_MAX_ITERATIONS, + initial_conversation_id: str | None = None, + reasoning_effort: ReasoningEffort | None = None, + agent_id: str | None = None, +) -> AgenticAnomalySummary: + """Run the anomaly-detection agentic evaluation K times and return a summary. + + A run ends when execute_anomaly_detection returns. Short of that it keeps sending + simulated replies up to ``max_iterations``, without classifying whether the agent's + text was a question: missing a genuine one hard-fails the run, while answering a final + answer costs one harmless extra turn. + """ + if k < 1: + raise ValueError(f"k must be >= 1, got {k}") + run_results: list[AnomalyRunResult] = [] + client = ChatClient( + host=host, token=token, workspace_id=workspace_id, reasoning_effort=reasoning_effort, agent_id=agent_id + ) + + def _run_once(conv_id: str) -> AnomalyRunResult: + viz_args: dict | None = None + execute_result: dict | None = None + turn_wall_clock_sec: float | None = None + turn_completed = False + disambiguated = False + current_question = question + reasoning_steps: list[str] = [] + response_id: str | None = None + all_tool_call_events: list[ToolCallEvent] = [] + all_reasoning_step_events: list[ReasoningStepEvent] = [] + turn_offset = 0.0 # each turn's call_ts/ts restarts near 0 -- shift by prior turns' wall time + tool_index_offset = 0 + reasoning_index_offset = 0 + + def _accumulate(result: ChatResult) -> None: + nonlocal turn_offset, tool_index_offset, reasoning_index_offset + turn_offset, tool_index_offset, reasoning_index_offset = shift_and_index_events( + result, + turn_offset=turn_offset, + tool_index_offset=tool_index_offset, + reasoning_index_offset=reasoning_index_offset, + ) + all_tool_call_events.extend(result.tool_call_events or []) + all_reasoning_step_events.extend(result.reasoning_step_events or []) + + for iteration in range(max_iterations): + try: + chat_result = client.send_message(conv_id, current_question) + except Exception as exc: # noqa: BLE001 -- end this run, not the whole item + _log.warning("Anomaly send_message failed for conversation %s: %s", conv_id, exc) + partial = getattr(exc, "partial_result", None) + if partial is not None: + reasoning_steps.extend(partial.reasoning_steps or []) + response_id = partial.response_id or response_id + _accumulate(partial) + viz_args, execute_result = _extract_anomaly_calls(all_tool_call_events) + turn_completed = False + break + reasoning_steps.extend(chat_result.reasoning_steps or []) + response_id = chat_result.response_id or response_id + _accumulate(chat_result) + # Over every turn so far, not just this one: the agent may build the chart on + # one turn and detect on the next, and reading a single turn would drop the + # series the detection actually ran on. + viz_args, execute_result = _extract_anomaly_calls(all_tool_call_events) + response_text = render_answer_text(chat_result) + turn_completed = chat_result.stream_ended and bool(response_text) + if execute_result is not None: + # The turn that ran the detection, not an earlier disambiguation turn. + turn_wall_clock_sec = chat_result.turn_wall_clock_sec + break + if not response_text: + break + if iteration >= max_iterations - 1: + break + try: + current_question = generate_simulated_anomaly_response(response_text, expected_output) + disambiguated = True + except Exception as exc: # noqa: BLE001 -- harness-side fault; end only this run + _log.warning("Simulated anomaly user reply failed for conversation %s: %s", conv_id, exc) + break + + return AnomalyRunResult( + conversation_id=conv_id, + evaluation=_evaluate_run(viz_args, execute_result, expected_output, turn_completed, disambiguated), + actual_visualization=viz_args, + actual_execute_result=execute_result, + turn_wall_clock_sec=turn_wall_clock_sec, + reasoning_steps=reasoning_steps, + response_id=response_id, + tool_call_events=all_tool_call_events, + reasoning_step_events=all_reasoning_step_events, + ) + + try: + conv_id_0 = initial_conversation_id if initial_conversation_id is not None else client.create_conversation() + try: + run_results.append(_run_once(conv_id_0)) + finally: + if initial_conversation_id is None: # only delete conversations we created + client.delete_conversation(conv_id_0) + + for _ in range(1, k): + conv_id = client.create_conversation() + try: + run_results.append(_run_once(conv_id)) + finally: + client.delete_conversation(conv_id) + finally: + client.close() + + pass_at_k = any(r.evaluation.strict_pass for r in run_results) + pass_power_k = all(r.evaluation.strict_pass for r in run_results) + best = max( + run_results, + key=lambda r: sum( + [ + r.evaluation.triggered, + r.evaluation.executed, + r.evaluation.success, + r.evaluation.turn_completed, + r.evaluation.metric_correct, + r.evaluation.granularity_correct, + ] + ), + ) + return AgenticAnomalySummary( + run_results=run_results, + pass_at_k=pass_at_k, + pass_power_k=pass_power_k, + best=best, + ) + + +class AnomalyDetectionAssertionError(AgenticAssertionError): + """Raised when an anomaly-detection evaluation fails.""" + + +def _detail(best: AnomalyRunResult) -> dict[str, Any]: + ev = best.evaluation + return { + "triggered": ev.triggered, + "executed": ev.executed, + "success": ev.success, + "turn_completed": ev.turn_completed, + "metric_correct": ev.metric_correct, + "granularity_correct": ev.granularity_correct, + # Which content checks the fixture pinned -- without it a run that verified nothing + # reads the same as one where everything matched. + "asserted": ev.asserted, + "disambiguated": ev.disambiguated, + "actual_metrics": sorted(_metric_uris(best.actual_visualization)), + "actual_granularity": _inferred_granularity(best.actual_visualization), + # Reported, never asserted: whether a real series contains anomalies is a property + # of the data, so a fixture demanding some would fail on the next warehouse refresh. + "anomaly_point_count": _point_count(best.actual_execute_result), + "actual_execute_result": best.actual_execute_result, + "latency_breakdown": build_latency_breakdown(best.tool_call_events, best.reasoning_step_events), + } + + +def evaluate_agentic_anomaly_detection( + host: str, + token: str, + workspace_id: str, + question: str, + expected_output: dict, + k: int = _DEFAULT_K, + max_iterations: int = _DEFAULT_MAX_ITERATIONS, + initial_conversation_id: str | None = None, + agent_id: str | None = None, + langfuse: object | None = None, + dataset_item_id: str = "", + dataset_name: str = "anomaly_detection", + run_timestamp: str | None = None, + model_version_override: str | None = None, + run_metadata_extra: dict | None = None, + reasoning_effort: ReasoningEffort | None = None, + submit_trace_link: SubmitTraceLink = run_trace_link_inline, +) -> AgenticEvalOutcome: + """Run anomaly-detection evaluation, log to Langfuse, and raise on failure.""" + langfuse, window_start = open_trace_window(langfuse) + summary = run_agentic_anomaly_detection( + host=host, + token=token, + workspace_id=workspace_id, + question=question, + expected_output=expected_output, + k=k, + max_iterations=max_iterations, + initial_conversation_id=initial_conversation_id, + reasoning_effort=reasoning_effort, + agent_id=agent_id, + ) + + if langfuse is not None and dataset_item_id: + # Pinned on the calling thread: a deferred poll must not widen its query window. + window_end = utc_now() + + def _write_scores(ctx: RunTraceContext) -> None: + for run_idx, run in enumerate(summary.run_results): + pt = ctx.trace(run.conversation_id) + ev = run.evaluation + strict_checks = { + "anomaly_triggered": ev.triggered, + "anomaly_executed": ev.executed, + "anomaly_success": ev.success, + "anomaly_turn_completed": ev.turn_completed, + } + # Only the content checks the fixture actually pinned. An unasserted check + # is True internally so it cannot fail a run, but publishing that as a + # BOOLEAN 1 would claim the evaluator verified something it never looked at. + strict_checks.update( + { + key: value + for name, key, value in ( + ("metric", "anomaly_metric_correct", ev.metric_correct), + ("granularity", "anomaly_granularity_correct", ev.granularity_correct), + ) + if name in ev.asserted + } + ) + with ctx.observe(pt, run_idx) as tid: + for score_name, value in strict_checks.items(): + ctx.score(tid, name=score_name, value=float(value), data_type="BOOLEAN") + ctx.quality( + tid, + strict_checks=strict_checks, + # pt.latency covers the whole conversation, which is the item's real + # elapsed cost when the agent needed clarification turns to get + # there; turn_wall_clock_sec (the goal turn alone) is the fallback. + # This is what 7 of the 8 existing kinds do -- kda_skill is the + # outlier and documents its own reason. Cost is not gated on + # ev.triggered: a run that answered without ever reaching the tool + # still spent tokens, and hiding that understates what the item cost. + latency_sec=pt.latency if pt else run.turn_wall_clock_sec, + cost_usd=pt.total_cost if pt else None, + ) + + # Before the pass@K raise: a failing item's scores are the ones worth having. + submit_trace_scoring( + submit_trace_link, + RunIdentity( + host, + token, + workspace_id, + dataset_name, + run_timestamp, + model_version_override, + run_metadata_extra, + reasoning_effort, + ), + langfuse=langfuse, + dataset_item_id=dataset_item_id, + conversation_ids=[r.conversation_id for r in summary.run_results], + window_start=window_start, + window_end=window_end, + suffix_runs=len(summary.run_results) > 1, + write_scores=_write_scores, + # The question this run answered, so a score is readable without resolving the + # conversation back to its item. + item_input=question, + ) + + best = summary.best + ev = best.evaluation + detail = _detail(best) + runs_passed = sum(1 for r in summary.run_results if r.evaluation.strict_pass) + + if not summary.pass_at_k: + message = ( + f"Anomaly detection assertion failed. strict_pass={ev.strict_pass} " + f"(triggered={ev.triggered}, executed={ev.executed}, success={ev.success}, " + f"turn_completed={ev.turn_completed}, metric_correct={ev.metric_correct}, " + f"granularity_correct={ev.granularity_correct}). " + f"Analysed {detail['actual_metrics']} at {detail['actual_granularity']}. " + f"Actual execute result: {best.actual_execute_result}." + ) + exc = AnomalyDetectionAssertionError(message) + exc.reasoning_steps = best.reasoning_steps + exc.conversation_id = best.conversation_id + exc.response_id = best.response_id + exc.detail = detail + exc.runs_passed = runs_passed + exc.runs_effective = len(summary.run_results) + raise exc + + return AgenticEvalOutcome( + runs_passed=runs_passed, + runs_effective=len(summary.run_results), + reasoning_steps=best.reasoning_steps, + conversation_id=best.conversation_id, + response_id=best.response_id, + detail=detail, + ) diff --git a/packages/gooddata-eval/tests/test_agentic_anomaly_detection.py b/packages/gooddata-eval/tests/test_agentic_anomaly_detection.py new file mode 100644 index 000000000..275472e4f --- /dev/null +++ b/packages/gooddata-eval/tests/test_agentic_anomaly_detection.py @@ -0,0 +1,481 @@ +# (C) 2026 GoodData Corporation. All rights reserved. +# SPDX-License-Identifier: LicenseRef-GoodData-Enterprise +import json +from contextlib import contextmanager +from unittest.mock import MagicMock, patch + +import pytest +from gooddata_eval.core.agentic.anomaly_detection import ( + AnomalyDetectionAssertionError, + _build_clarification_prompt, + _evaluate_run, + _extract_anomaly_calls, + _inferred_granularity, + _metric_uris, + _point_count, + evaluate_agentic_anomaly_detection, + run_agentic_anomaly_detection, +) +from gooddata_eval.core.models import ChatResult + +_MODULE = "gooddata_eval.core.agentic.anomaly_detection" + +_EXPECTED = {"metric": "metric/spend", "granularity": "MONTH"} + + +def _viz_args(*, metric="metric/spend", dimension="label/process_date.month", filter_granularity="MONTH") -> dict: + """Shaped like the chart observed live for an anomaly question.""" + query: dict = { + "fields": {"d_month": {"using": dimension}, "m_spend": {"using": metric}}, + "filter_by": {}, + } + if filter_granularity: + query["filter_by"] = { + "f_window": { + "to": 0, + "from": -23, + "type": "date_filter", + "using": "dataset/process_date", + "granularity": filter_granularity, + } + } + return { + "visualization": { + "type": "line_chart", + "title": "Monthly Spend - anomaly detection", + "query": query, + "metrics": ["m_spend"], + "view_by": ["d_month"], + } + } + + +_OK_DETECT = json.dumps( + { + "success": True, + "data": { + "attribute": [], + "values": {"m_spend": []}, + "measure_ids": ["m_spend"], + "points": [], + "point_count": 0, + "truncated": False, + }, + "error": None, + } +) + + +def _tc(name: str, args: dict | None = None, result: str | None = None): + tc = MagicMock() + tc.function_name = name + tc.call_ts = None + tc.result_ts = None + tc.index = None + tc.parsed_arguments = lambda a=args: a or {} + tc.result = result + tc.parsed_result = lambda r=result: json.loads(r) if r else None + return tc + + +def _chat(tool_calls, text="No months were flagged.", ended=True) -> ChatResult: + """A real ChatResult, so render_answer_text sees every field it reads.""" + result = ChatResult.model_validate({"textResponse": text, "responseId": "resp-1"}) + result.stream_ended = ended + result.turn_wall_clock_sec = 6.3 + result.tool_call_events = tool_calls + return result + + +def _pair(**kw): + return [ + _tc("create_adhoc_visualization", _viz_args(**kw)), + _tc("execute_anomaly_detection", {"visualization_ref": "viz_1", "max_points": 200}, _OK_DETECT), + ] + + +def _viz(**kw): + return _viz_args(**kw)["visualization"] + + +# ── extraction ────────────────────────────────────────────────────────────── + + +def test_extract_pairs_the_detection_with_the_chart_it_followed(): + """A new chart clears the previous detection: scoring a fresh series against a run + that never touched it would credit work the agent redid.""" + calls = [ + _tc("create_adhoc_visualization", _viz_args(metric="metric/orders")), + _tc("execute_anomaly_detection", {}, _OK_DETECT), + _tc("create_adhoc_visualization", _viz_args(metric="metric/spend")), + ] + viz, result = _extract_anomaly_calls(calls) + + assert viz["query"]["fields"]["m_spend"]["using"] == "metric/spend" + assert result is None + + +def test_point_count_is_read_off_the_result(): + _, result = _extract_anomaly_calls(_pair()) + assert _point_count(result) == 0 + assert _point_count(None) is None + assert _point_count({"data": "not a dict"}) is None + + +# ── granularity inference ─────────────────────────────────────────────────── + + +def test_granularity_comes_from_the_field_token_first(): + """Field references are read before the date filter, as the service reads them.""" + assert _inferred_granularity(_viz(dimension="label/process_date.month")) == "MONTH" + assert _inferred_granularity(_viz(dimension="label/process_date.quarter")) == "QUARTER" + assert _inferred_granularity(_viz(dimension="label/process_date.year")) == "YEAR" + assert _inferred_granularity(_viz(dimension="label/process_date.hour")) == "HOUR" + + +def test_a_reference_naming_two_granularities_resolves_to_its_suffix(): + """A snake_case `first_day_quarter.month` tokenizes to {first, day, quarter, month}, + where "day", "quarter" and "month" all map. gen-ai iterates a set and returns whichever + comes first, so its own answer there is not stable; taking the last token reads the + suffix, which is what a dotted label means, and is the same every run. + + This is rare -- no label in the eval workspace names two granularities -- but a scorer + must not be a coin flip even where the thing it scores is one. + """ + assert _inferred_granularity(_viz(dimension="label/first_day_quarter.month")) == "MONTH" + assert _inferred_granularity(_viz(dimension="label/day_of_week")) == "WEEK" + + +def test_a_date_attribute_without_a_granularity_suffix_names_none(): + """`label/process_date` matches no token: the tool has no "date" key and refuses the + call rather than guessing daily, so guessing DAY here would score a granularity the + service never used.""" + assert _inferred_granularity(_viz(dimension="label/process_date", filter_granularity=None)) is None + + +def test_granularity_falls_back_to_the_date_filter(): + """A dimension with no recognisable token still leaves the filter to read.""" + viz = _viz(dimension="label/some_opaque_label", filter_granularity="QUARTER") + assert _inferred_granularity(viz) == "QUARTER" + + +def test_a_field_token_wins_over_a_disagreeing_filter(): + """The service reads fields first, so the evaluation must too -- otherwise the two + disagree and a correct run scores wrong.""" + viz = _viz(dimension="label/process_date.day", filter_granularity="MONTH") + assert _inferred_granularity(viz) == "DAY" + + +def test_granularity_is_none_when_nothing_names_one(): + viz = _viz(dimension="label/opaque", filter_granularity=None) + assert _inferred_granularity(viz) is None + assert _inferred_granularity(None) is None + + +# ── scoring ───────────────────────────────────────────────────────────────── + + +def _evaluate(viz=None, result=None, expected=None, turn_completed=True): + return _evaluate_run(viz, result, expected if expected is not None else _EXPECTED, turn_completed) + + +def test_a_correct_detection_passes_every_check(): + ev = _evaluate(_viz(), {"success": True}) + assert ev.strict_pass is True + assert ev.asserted == ["metric", "granularity"] + + +def test_detecting_on_the_wrong_measure_fails(): + ev = _evaluate(_viz(metric="metric/orders"), {"success": True}) + assert ev.metric_correct is False + assert ev.granularity_correct is True + + +def test_the_wrong_granularity_fails(): + """Daily anomalies on a question about monthly ones is the wrong series, however well + the detection itself performed.""" + ev = _evaluate(_viz(dimension="label/process_date.day"), {"success": True}) + assert ev.granularity_correct is False + assert ev.metric_correct is True + + +def test_expected_granularity_is_compared_case_insensitively(): + assert _evaluate(_viz(), {"success": True}, expected={"granularity": "month"}).granularity_correct is True + + +def test_an_unstated_expectation_neither_fails_nor_silently_passes(): + ev = _evaluate(_viz(metric="metric/anything"), {"success": True}, expected={}) + assert ev.strict_pass is True + assert ev.asserted == [] + + +def test_finding_no_anomalies_is_still_a_pass(): + """Whether a real series contains anomalies is a property of the data, not the agent -- + the count is reported but never gates the verdict.""" + assert _evaluate(_viz(), json.loads(_OK_DETECT)).strict_pass is True + + +def test_a_failed_detection_is_not_a_success(): + ev = _evaluate(_viz(), {"success": False, "error": "boom"}) + assert ev.executed is True + assert ev.success is False + + +def test_a_chart_alone_is_not_an_executed_detection(): + ev = _evaluate(_viz(), None) + assert ev.triggered is True + assert ev.executed is False + + +# ── run loop ──────────────────────────────────────────────────────────────── + + +def _run(side_effect, *, k=1, expected=None, max_iterations=4): + client = MagicMock() + client.create_conversation.side_effect = [f"conv-{i}" for i in range(1, k + 2)] + client.send_message.side_effect = side_effect + with ( + patch(f"{_MODULE}.ChatClient", return_value=client), + patch(f"{_MODULE}.generate_simulated_anomaly_response", return_value="use spend, monthly"), + ): + return run_agentic_anomaly_detection( + host="http://h", + token="tok", + workspace_id="ws1", + question="Detect anomalies in monthly spend", + expected_output=expected if expected is not None else _EXPECTED, + k=k, + max_iterations=max_iterations, + ) + + +def test_a_single_turn_detection_passes(): + """Observed live: search, build, detect all land in one turn.""" + summary = _run([_chat(_pair())]) + assert summary.pass_at_k is True + assert summary.best.evaluation.disambiguated is False + + +def test_a_clarifying_question_is_answered_and_the_run_continues(): + summary = _run([_chat([], text="Which Spend metric did you mean?"), _chat(_pair())]) + assert summary.pass_at_k is True + assert summary.best.evaluation.disambiguated is True + + +def test_the_loop_stops_at_max_iterations_without_a_detection(): + summary = _run([_chat([], text="Still thinking")] * 4, max_iterations=4) + assert summary.pass_at_k is False + assert summary.best.evaluation.executed is False + + +def test_an_empty_response_ends_the_run_immediately(): + client = MagicMock() + client.create_conversation.return_value = "conv-1" + client.send_message.return_value = _chat([], text="") + with ( + patch(f"{_MODULE}.ChatClient", return_value=client), + patch(f"{_MODULE}.generate_simulated_anomaly_response") as sim, + ): + run_agentic_anomaly_detection( + host="http://h", token="tok", workspace_id="ws1", question="q", expected_output=_EXPECTED + ) + assert client.send_message.call_count == 1 + sim.assert_not_called() + + +def test_a_chat_error_on_a_later_run_does_not_discard_the_earlier_one(): + summary = _run([_chat(_pair()), RuntimeError("boom")], k=2) + assert len(summary.run_results) == 2 + assert summary.pass_at_k is True + + +def test_k_must_be_at_least_one(): + with pytest.raises(ValueError, match="k must be >= 1"): + run_agentic_anomaly_detection(host="h", token="t", workspace_id="w", question="q", expected_output={}, k=0) + + +# ── evaluate_* wrapper ────────────────────────────────────────────────────── + + +def _evaluate_item(calls): + client = MagicMock() + client.create_conversation.return_value = "conv-1" + client.send_message.return_value = _chat(calls) + with patch(f"{_MODULE}.ChatClient", return_value=client): + return evaluate_agentic_anomaly_detection( + host="http://h", + token="tok", + workspace_id="ws1", + question="Detect anomalies in monthly spend", + expected_output=_EXPECTED, + ) + + +def test_detail_reports_the_series_analysed_and_the_flag_count(): + outcome = _evaluate_item(_pair()) + assert outcome.detail["actual_metrics"] == ["metric/spend"] + assert outcome.detail["actual_granularity"] == "MONTH" + assert outcome.detail["anomaly_point_count"] == 0 + assert outcome.detail["asserted"] == ["metric", "granularity"] + assert "latency_breakdown" in outcome.detail + assert outcome.runs_passed == 1 + + +def test_the_wrong_series_raises_naming_what_was_analysed(): + with pytest.raises(AnomalyDetectionAssertionError) as exc_info: + _evaluate_item(_pair(dimension="label/process_date.day")) + + error = exc_info.value + assert "granularity_correct=False" in str(error) + assert "DAY" in str(error) + assert error.runs_passed == 0 + assert error.conversation_id == "conv-1" + + +def test_a_chart_built_on_an_earlier_turn_is_still_the_one_scored(): + """The agent may build the chart on one turn and detect on the next -- reading only the + current turn's calls would drop the series the detection actually ran on and fail a + correct run for having no metric or granularity.""" + summary = _run( + [ + _chat([_tc("create_adhoc_visualization", _viz_args())], text="Building the chart."), + _chat([_tc("execute_anomaly_detection", {"visualization_ref": "viz_1"}, _OK_DETECT)]), + ] + ) + + assert summary.best.evaluation.executed is True + assert summary.best.evaluation.metric_correct is True + assert summary.best.evaluation.granularity_correct is True + assert summary.pass_at_k is True + + +# ── prompt building, string fields, and failure paths ─────────────────────── + + +def test_the_clarification_prompt_mentions_only_the_hints_the_fixture_supplies(): + """An absent hint must be dropped, not asserted as a literal "None" -- that would + answer a question the agent never asked, with a wrong value.""" + full = _build_clarification_prompt("Which one?", {"metric": "metric/spend", "granularity": "MONTH"}) + assert "metric/spend" in full + assert "MONTH" in full + + bare = _build_clarification_prompt("Which one?", {}) + assert "None" not in bare + assert "For reference" not in bare + + +def test_a_bare_uri_field_is_read_for_both_metric_and_granularity(): + """Tool-call arguments are raw JSON, where a field may be a plain string rather than + the object a parsed visualization always has.""" + viz = {"query": {"fields": {"m": "metric/spend", "d": "label/process_date.month"}}, "metrics": ["m"]} + assert _metric_uris(viz) == {"metric/spend"} + assert _inferred_granularity(viz) == "MONTH" + + +def test_a_chat_error_keeps_what_its_partial_result_carried(): + """A stream that broke after the detection ran still has a result worth scoring.""" + error = RuntimeError("stream died") + error.partial_result = _chat(_pair()) + summary = _run([error]) + + assert summary.best.evaluation.executed is True + # The turn did not finish, which strict_pass requires, so this is still a failure. + assert summary.best.evaluation.turn_completed is False + assert summary.pass_at_k is False + + +def test_a_simulated_user_failure_ends_the_run_without_raising(): + client = MagicMock() + client.create_conversation.return_value = "conv-1" + client.send_message.return_value = _chat([], text="Which measure?") + with ( + patch(f"{_MODULE}.ChatClient", return_value=client), + patch(f"{_MODULE}.generate_simulated_anomaly_response", side_effect=RuntimeError("openai down")), + ): + summary = run_agentic_anomaly_detection( + host="http://h", token="tok", workspace_id="ws1", question="q", expected_output=_EXPECTED + ) + + assert summary.pass_at_k is False + assert client.send_message.call_count == 1 # ended rather than looping + + +# ── Langfuse scoring ──────────────────────────────────────────────────────── + + +class _FakeCtx: + """Records what the deferred Langfuse block writes, without a Langfuse.""" + + def __init__(self): + self.scores: dict[str, float] = {} + # Not named `quality`: the method below would overwrite itself on first call. + self.quality_call: dict = {} + + def trace(self, _conversation_id): + return None + + @contextmanager + def observe(self, _trace, _run_idx): + yield "trace-id" + + def score(self, _tid, *, name, value, data_type): + self.scores[name] = value + + def quality(self, _tid, *, strict_checks, latency_sec, cost_usd): + self.quality_call = {"strict_checks": strict_checks, "latency_sec": latency_sec, "cost_usd": cost_usd} + + +def _scored(expected_output, calls=None): + """Run one item with Langfuse on, then execute the deferred block against a fake ctx.""" + client = MagicMock() + client.create_conversation.return_value = "conv-1" + client.send_message.return_value = _chat(calls if calls is not None else _pair()) + captured = {} + + def _capture(_link, _identity, **kwargs): + captured.update(kwargs) + + with ( + patch(f"{_MODULE}.ChatClient", return_value=client), + patch(f"{_MODULE}.submit_trace_scoring", side_effect=_capture), + ): + try: + evaluate_agentic_anomaly_detection( + host="http://h", + token="tok", + workspace_id="ws1", + question="Detect anomalies in monthly spend", + expected_output=expected_output, + langfuse=MagicMock(), + dataset_item_id="ds-1", + ) + except AnomalyDetectionAssertionError: + pass # scores are written before the pass@K raise, which is the point + + ctx = _FakeCtx() + captured["write_scores"](ctx) + return ctx + + +def test_only_the_checks_the_fixture_pinned_are_scored(): + """An unasserted check is True internally so it cannot fail a run. Publishing that as a + BOOLEAN 1 would claim the evaluator verified something it never looked at.""" + ctx = _scored({"metric": "metric/spend"}) # granularity deliberately unpinned + + assert ctx.scores["anomaly_metric_correct"] == 1.0 + assert "anomaly_granularity_correct" not in ctx.scores + # The process checks are unconditional -- they are always actually evaluated. + assert set(ctx.scores) >= {"anomaly_triggered", "anomaly_executed", "anomaly_success"} + + +def test_both_content_checks_are_scored_when_both_are_pinned(): + ctx = _scored(_EXPECTED) + assert ctx.scores["anomaly_metric_correct"] == 1.0 + assert ctx.scores["anomaly_granularity_correct"] == 1.0 + + +def test_cost_is_reported_even_when_the_tool_was_never_reached(): + """A run that answered without detecting anything still spent tokens; gating cost on + ev.triggered hid that and understated what the item cost.""" + ctx = _scored(_EXPECTED, calls=[]) + assert "cost_usd" in ctx.quality_call diff --git a/packages/gooddata-eval/tests/test_agentic_runner.py b/packages/gooddata-eval/tests/test_agentic_runner.py index 77086d101..1a455357b 100644 --- a/packages/gooddata-eval/tests/test_agentic_runner.py +++ b/packages/gooddata-eval/tests/test_agentic_runner.py @@ -85,6 +85,7 @@ def test_dispatch_agentic_omits_agent_id_by_default(): ("agentic_guardrail", "Ignore prior instructions", "evaluate_agentic_guardrail"), ("agentic_kda_skill", {"Measure": {"type": "metric", "id": "revenue"}}, "evaluate_agentic_kda_skill"), ("agentic_conversation", {"fixture": _MIN_CONVERSATION_FIXTURE}, "evaluate_agentic_conversation"), + ("agentic_anomaly_detection", {"metric": "metric/spend"}, "evaluate_agentic_anomaly_detection"), ] diff --git a/packages/gooddata-eval/tests/test_trace_linker.py b/packages/gooddata-eval/tests/test_trace_linker.py index f490effd6..d9cc696a9 100644 --- a/packages/gooddata-eval/tests/test_trace_linker.py +++ b/packages/gooddata-eval/tests/test_trace_linker.py @@ -118,6 +118,7 @@ def test_run_trace_link_inline_runs_the_task_on_the_calling_thread(): ("visualization", "evaluate_agentic_visualization"), ("kda_skill", "evaluate_agentic_kda_skill"), ("conversation", "evaluate_agentic_conversation"), + ("anomaly_detection", "evaluate_agentic_anomaly_detection"), ]