diff --git a/packages/gooddata-eval/src/gooddata_eval/core/agentic/__init__.py b/packages/gooddata-eval/src/gooddata_eval/core/agentic/__init__.py index 639bee5b7..89e93dde8 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/agentic/__init__.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/agentic/__init__.py @@ -30,6 +30,14 @@ evaluate_agentic_guardrail, run_agentic_guardrail, ) +from gooddata_eval.core.agentic.kda_skill import ( + AgenticKdaSummary, + KdaEvaluation, + KdaRunResult, + KdaSkillAssertionError, + evaluate_agentic_kda_skill, + run_agentic_kda_skill, +) from gooddata_eval.core.agentic.metric_skill import ( AgenticMetricSummary, MetricRunResult, @@ -56,6 +64,7 @@ "AgenticAlertSummary", "AgenticGeneralQuestionSummary", "AgenticGuardrailSummary", + "AgenticKdaSummary", "AgenticMetricSummary", "AgenticSearchSummary", "AgenticRunSummary", @@ -69,6 +78,9 @@ "GeneralQuestionResult", "GuardrailAssertionError", "GuardrailResult", + "KdaEvaluation", + "KdaRunResult", + "KdaSkillAssertionError", "MetricRunResult", "MetricSkillAssertionError", "RunResult", @@ -81,6 +93,7 @@ "evaluate_agentic_conversation", "evaluate_agentic_general_question", "evaluate_agentic_guardrail", + "evaluate_agentic_kda_skill", "evaluate_agentic_metric_skill", "evaluate_agentic_search_tool", "evaluate_agentic_visualization", @@ -88,6 +101,7 @@ "run_agentic_conversation", "run_agentic_general_question", "run_agentic_guardrail", + "run_agentic_kda_skill", "run_agentic_metric_skill", "run_agentic_search_tool", "run_agentic_visualization", diff --git a/packages/gooddata-eval/src/gooddata_eval/core/agentic/_langfuse.py b/packages/gooddata-eval/src/gooddata_eval/core/agentic/_langfuse.py index 67630ce2f..34bca4da6 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/agentic/_langfuse.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/agentic/_langfuse.py @@ -8,7 +8,7 @@ import os import time import uuid -from collections.abc import Iterator +from collections.abc import Callable, Iterator from contextlib import contextmanager from datetime import datetime, timedelta, timezone from typing import Any @@ -55,6 +55,40 @@ def _ts(v: Any) -> str: resp.raise_for_status() return _TraceListResult([_TraceObj(t) for t in resp.json().get("data", [])]) + def get(self, trace_id: str) -> _TraceObj: + resp = self._client.get(f"/api/public/traces/{trace_id}") + resp.raise_for_status() + return _TraceObj(resp.json()) + + +class _ObservationObj: + """Duck-type wrapper around a raw Langfuse observation dict.""" + + def __init__(self, raw: dict) -> None: + self.id: str = raw.get("id", "") + self.name: str | None = raw.get("name") + self.output: Any = raw.get("output") + + +class _ObservationListResult: + def __init__(self, data: list[_ObservationObj], total_pages: int) -> None: + self.data = data + self.total_pages = total_pages + + +class _ObservationsAPI: + def __init__(self, client: httpx.Client) -> None: + self._client = client + + def list(self, trace_id: str, page: int = 1, limit: int = 100) -> _ObservationListResult: + resp = self._client.get("/api/public/observations", params={"traceId": trace_id, "page": page, "limit": limit}) + resp.raise_for_status() + body = resp.json() + meta = body.get("meta") or {} + return _ObservationListResult( + [_ObservationObj(o) for o in body.get("data", [])], total_pages=int(meta.get("totalPages") or 1) + ) + class _DatasetRunItemsAPI: def __init__(self, client: httpx.Client) -> None: @@ -83,6 +117,7 @@ def create( class _LangfuseAPI: def __init__(self, client: httpx.Client) -> None: self.trace = _TraceAPI(client) + self.observations = _ObservationsAPI(client) self.dataset_run_items = _DatasetRunItemsAPI(client) @@ -247,11 +282,16 @@ def find_traces_per_conversation( langfuse: Any, conversation_ids: list[str], window_start: datetime, + select: Callable[[list[Any]], Any | None] | None = None, ) -> dict[str, Any]: - """Poll Langfuse until traces matching all conversation_ids are found or retries exhaust.""" + """Check Langfuse for the trace(s) matching each conversation_id; get latency, picking + the right turn via ``select`` (default: largest latency) -- e.g. KDA passes a selector + that picks the turn that actually made the KDA tool call. + """ if bool(os.environ.get(SKIP_ENV_VAR)): return dict.fromkeys(conversation_ids) + select = select or (lambda found: max(found, key=lambda t: getattr(t, "latency", None) or 0.0)) by_conv: dict[str, Any] = dict.fromkeys(conversation_ids) window_end = datetime.now(timezone.utc) pad = timedelta(seconds=_WINDOW_PADDING_SEC) @@ -269,7 +309,7 @@ def find_traces_per_conversation( break delay *= _BACKOFF if found: - by_conv[cid] = max(found, key=lambda t: getattr(t, "latency", None) or 0.0) + by_conv[cid] = select(found) else: _log.warning( "[langfuse] No trace found for conversation %s in window [%s, %s]", cid, window_start, window_end diff --git a/packages/gooddata-eval/src/gooddata_eval/core/agentic/kda_skill.py b/packages/gooddata-eval/src/gooddata_eval/core/agentic/kda_skill.py new file mode 100644 index 000000000..7db661c65 --- /dev/null +++ b/packages/gooddata-eval/src/gooddata_eval/core/agentic/kda_skill.py @@ -0,0 +1,540 @@ +# (C) 2026 GoodData Corporation. All rights reserved. +"""Agentic KDA (Key Driver Analysis)-skill evaluation runner.""" + +from __future__ import annotations + +import json +import logging +import os +import time +from dataclasses import dataclass +from typing import Any + +from gooddata_eval.core.chat.sse_client import ChatClient +from gooddata_eval.core.config import ReasoningEffort +from gooddata_eval.core.models import ToolCallEvent + +_log = logging.getLogger(__name__) + +_DEFAULT_K = 1 +# KDA cases are designed to resolve in one turn (unlike alert/metric skills), so this is +# only a safety net for the rare disambiguation turn -- a title collision (see the +# handoff's known-collision cases) or a metric-vs-fact form choice -- not a general +# multi-turn budget. +_DEFAULT_MAX_ITERATIONS = 2 + + +def _to_number(value: object) -> float | int | None: + """Convert string/number to int or float, None on failure. Mirrors alert_skill._to_number + -- the API is contractually numeric here, but this guards against a malformed response + raising a raw ValueError instead of failing the check cleanly.""" + if value is None: + return None + try: + f = float(str(value)) + return int(f) if f == int(f) else f + except (ValueError, TypeError): + return None + + +def _normalize_measure(m: dict) -> tuple[Any, Any, Any]: + return (m.get("type"), m.get("id"), m.get("aggregation")) + + +def _measure_matches(actual: object, expected: dict | list[dict] | None) -> bool: + """expected may be a single candidate dict or a list of candidate dicts (mirrors + metric_skill's expected_output: dict | list -- e.g. case 1 accepts either the + catalog metric id or the mathematically equivalent ad-hoc fact+SUM). + + ``actual`` is typed ``object``, not ``dict``, and checked with ``isinstance`` (mirroring + alert_skill._deep_subset) because it comes from a tool call the LLM constructed -- + a malformed call could put a non-dict value there. + """ + if not isinstance(actual, dict) or expected is None: + return False + candidates = expected if isinstance(expected, list) else [expected] + actual_norm = _normalize_measure(actual) + return any(actual_norm == _normalize_measure(c) for c in candidates if isinstance(c, dict)) + + +def _filters_match(actual: object, expected: list) -> bool: + actual = actual or [] + try: + return json.dumps(actual, sort_keys=True) == json.dumps(expected, sort_keys=True) + except TypeError: + return False + + +def _within_tolerance(actual: object, expected: object, tolerance: float) -> bool: + a, e = _to_number(actual), _to_number(expected) + if a is None or e is None: + return False + return abs(a - e) <= tolerance + + +def _is_asking_clarification(text: str) -> bool: + if not text: + return False + t = text.lower() + return "?" in t or "could you" in t or "please provide" in t or "clarif" in t + + +def generate_simulated_kda_response(agent_message: str, measure_candidates: dict | list[dict] | None) -> str: + """Generate a user reply to keep the KDA-skill conversation going (gpt-4o-mini). + + Used only when the agent asks a clarifying question instead of triggering KDA + directly (e.g. a title collision between two metrics). Picks *any* candidate from + ``measure_candidates`` -- not necessarily the one an eventual correctness ticket + would require -- because the current scope only needs KDA to trigger, not the + resulting measure to be exactly right (see KdaEvaluation docstring). + """ + try: + from openai import OpenAI # noqa: PLC0415 + except ImportError as exc: + raise RuntimeError("openai package is required for generate_simulated_kda_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) + candidates = measure_candidates if isinstance(measure_candidates, list) else [measure_candidates or {}] + candidate_desc = "; or ".join( + f"{c.get('type')} '{c.get('id')}'" + (f" (aggregation {c['aggregation']})" if c.get("aggregation") else "") + for c in candidates + ) + prompt = ( + f"You are simulating a user in a conversation with a BI assistant that runs key driver " + f"analysis. The assistant said: '{agent_message}'. " + f"The user is happy to proceed with any of the following: {candidate_desc}. " + f"Reply briefly as the user, picking whichever of those the assistant offered." + ) + response = client.chat.completions.create( + model="gpt-4o-mini", + messages=[{"role": "user", "content": prompt}], + max_tokens=150, + ) + return response.choices[0].message.content or "Please proceed with either option." + + +def _extract_kda_calls(tool_call_events: list[ToolCallEvent]) -> tuple[dict | None, dict | None]: + """Return (create_args, execute_result): the arguments of the LAST + `create_key_driver_analysis` call and the parsed result of the LAST + `execute_key_driver_analysis` call. Taking the last (not first) attempt matches + the observed retry-loop behaviour (kda_1 fails, kda_2 retries) -- the last + attempt is what actually determined the answer the chatbot gave. + """ + create_args: dict | None = None + execute_result: dict | None = None + for tc in tool_call_events: + if tc.function_name == "create_key_driver_analysis": + create_args = tc.parsed_arguments() + elif tc.function_name == "execute_key_driver_analysis" and tc.result: + execute_result = tc.parsed_result() + return create_args, execute_result + + +_KDA_TOOL_NAMES = frozenset({"create_key_driver_analysis", "execute_key_driver_analysis"}) +_OBSERVATIONS_PAGE_CAP = 10 # 10 pages * 100/page = 1000 observations; real traces run ~20-30 + +# Langfuse ingests a trace's observations asynchronously -- the trace returned by +# find_traces_per_conversation right after the chat response lands can still be +# mid-ingestion, so its .latency reads low. Verified empirically against real +# staging traces (light and full-KDA cases): the value stabilizes within ~20s of +# the response and does not move again afterwards; 25s adds a safety margin. +_LATENCY_SETTLE_DELAY_SEC = 25.0 + + +def _observation_has_kda_call(obs: Any) -> bool: + output = obs.output + items = output if isinstance(output, list) else [output] if output else [] + return any( + isinstance(item, dict) and item.get("type") == "function_call" and item.get("name") in _KDA_TOOL_NAMES + for item in items + ) + + +def _trace_has_kda_call(langfuse: Any, trace_id: str) -> bool: + """True if any observation on this trace contains a KDA tool call. + + The tool call itself isn't its own named observation -- it shows up as a + ``function_call`` item inside an ``OpenAI-generation`` observation's ``output``. + Pages through all observations (bounded by _OBSERVATIONS_PAGE_CAP) rather than just + the first page, so a KDA call on a later page isn't missed. + """ + try: + page = 1 + while page <= _OBSERVATIONS_PAGE_CAP: + result = langfuse.api.observations.list(trace_id=trace_id, page=page) + if any(_observation_has_kda_call(obs) for obs in result.data): + return True + if page >= result.total_pages: + return False + page += 1 + except Exception as exc: # noqa: BLE001 -- best-effort selection, never block scoring on it + _log.debug("Failed to fetch observations for trace %s: %s", trace_id, exc) + return False + + +def _select_kda_trace(langfuse: Any, candidates: list[Any]) -> Any | None: + """Pick the candidate trace that made the KDA tool call. Falls back to max-latency + only so scores still attach to some real trace instead of being orphaned -- callers + must not treat that fallback trace's latency as a real KDA duration.""" + for candidate in candidates: + if _trace_has_kda_call(langfuse, candidate.id): + return candidate + return max(candidates, key=lambda t: getattr(t, "latency", None) or 0.0) + + +def _settle_trace_latency(langfuse: Any, trace: Any) -> Any: + """Re-fetch ``trace`` after letting Langfuse's async ingestion catch up (see + _LATENCY_SETTLE_DELAY_SEC). Falls back to the original (possibly-early) trace if the + re-fetch fails -- best-effort, never block scoring on it.""" + time.sleep(_LATENCY_SETTLE_DELAY_SEC) + try: + return langfuse.api.trace.get(trace.id) + except Exception as exc: # noqa: BLE001 + _log.debug("Failed to re-fetch trace %s for latency settle: %s", trace.id, exc) + return trace + + +@dataclass +class KdaEvaluation: + """Evaluation scores for a single KDA-skill run. + + Scope: this suite currently asserts only that the KDA process runs to completion -- + the tool chain triggers, executes successfully, and the chatbot delivers a final + answer. Per-field correctness (Measure/Date Attribute/Periods/Filters/Summary + matching the expected values) is computed and logged for visibility but + intentionally excluded from ``strict_pass`` -- that verification is scoped to a + follow-up ticket, not this one. + """ + + # Core: gates strict_pass. + kda_triggered: bool + executed: bool + success: bool + turn_completed: bool + + # Informational only: computed and logged, but not required for strict_pass. + measure_correct: bool + date_attribute_correct: bool + analyzed_period_correct: bool + reference_period_correct: bool + filters_correct: bool + summary_correct: bool + + @property + def strict_pass(self) -> bool: + return all([self.kda_triggered, self.executed, self.success, self.turn_completed]) + + +_REPORT_LATENCY_THRESHOLD_SEC = 60.0 + + +def classify_kda_report_bucket(ev: KdaEvaluation, latency_sec: float | None) -> str: + """Classify a run for the daily latency report: 'pass' | 'failed' | 'error'. + + Distinct from ``strict_pass`` (which gates the CI assertion, completion-only, no + timing). 'error' = the process didn't complete; among completed runs, 'pass' if + within ``_REPORT_LATENCY_THRESHOLD_SEC``, else 'failed'. A completed run with no + latency value (trace-linking failed) falls to 'failed' rather than 'pass' -- treat + unknown timing as not-within-target, not as a free pass. + """ + if not ev.strict_pass: + return "error" + if latency_sec is not None and latency_sec <= _REPORT_LATENCY_THRESHOLD_SEC: + return "pass" + return "failed" + + +@dataclass +class KdaRunResult: + """Outcome of one run (one conversation, one message) for a KDA case.""" + + conversation_id: str + eval: KdaEvaluation + actual_create_args: dict | None + actual_execute_result: dict | None + + +@dataclass +class AgenticKdaSummary: + """Aggregated outcome of K runs for a KDA case.""" + + run_results: list[KdaRunResult] + pass_at_k: bool + pass_power_k: bool + best: KdaRunResult + + +def _evaluate_run( + create_args: dict | None, + execute_result: dict | None, + turn_completed: bool, + expected: dict, +) -> KdaEvaluation: + kda_triggered = create_args is not None + executed = execute_result is not None + # Checked against the tool's own result, not compared to expected_output -- this + # scope only cares whether KDA itself reported success, not input/output correctness. + success = executed and execute_result.get("success") is True + + # Informational only (see KdaEvaluation docstring) -- still computed so a follow-up + # ticket can promote these to strict_pass without redoing the extraction logic. + measure_correct = kda_triggered and _measure_matches(create_args.get("measure"), expected.get("Measure")) + date_attribute_correct = kda_triggered and create_args.get("date_attribute_id") == expected.get("Date Attribute") + analyzed_period_correct = kda_triggered and create_args.get("analyzed_period") == expected.get("Analyzed Period") + reference_period_correct = kda_triggered and create_args.get("reference_period") == expected.get("Reference Period") + filters_correct = kda_triggered and _filters_match(create_args.get("filters"), expected.get("Filters", [])) + + summary_correct = False + if executed and success: + data = execute_result.get("data") or {} + actual_summary = data.get("summary") or {} + expected_summary = expected.get("Summary") or {} + tolerance = expected_summary.get("absolute_tolerance", 0.01) + summary_correct = ( + _within_tolerance(actual_summary.get("reference_value"), expected_summary.get("reference_value"), tolerance) + and _within_tolerance( + actual_summary.get("analyzed_value"), expected_summary.get("analyzed_value"), tolerance + ) + and _within_tolerance(actual_summary.get("change"), expected_summary.get("change"), tolerance) + ) + + return KdaEvaluation( + kda_triggered=kda_triggered, + executed=executed, + success=success, + turn_completed=turn_completed, + measure_correct=measure_correct, + date_attribute_correct=date_attribute_correct, + analyzed_period_correct=analyzed_period_correct, + reference_period_correct=reference_period_correct, + filters_correct=filters_correct, + summary_correct=summary_correct, + ) + + +def run_agentic_kda_skill( + 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, +) -> AgenticKdaSummary: + """Run the KDA-skill agentic evaluation K times and return a summary. + + Each run is normally a single message in a single turn -- the agent_kda_skill + dataset is designed so every question resolves unambiguously -- but if the agent + asks a clarifying question instead of triggering KDA (a title collision, or a + metric-vs-fact form choice), a simulated user reply nudges it forward for up to + ``max_iterations`` turns, so a disambiguation turn doesn't block measuring whether + KDA itself triggers and completes. + """ + run_results: list[KdaRunResult] = [] + client = ChatClient(host=host, token=token, workspace_id=workspace_id, reasoning_effort=reasoning_effort) + + def _run_once(conv_id: str) -> KdaRunResult: + create_args: dict | None = None + execute_result: dict | None = None + turn_completed = False + current_question = question + + for iteration in range(max_iterations): + chat_result = client.send_message(conv_id, current_question) + c_args, e_result = _extract_kda_calls(chat_result.tool_call_events or []) + turn_completed = bool((chat_result.text_response or "").strip()) + if c_args is not None: + create_args, execute_result = c_args, e_result + break + response_text = (chat_result.text_response or "").strip() + if iteration >= max_iterations - 1 or not _is_asking_clarification(response_text): + break + try: + current_question = generate_simulated_kda_response(response_text, expected_output.get("Measure")) + except Exception as exc: # noqa: BLE001 -- safety net, not the assertion; end only this run + _log.warning("Simulated KDA user reply failed for conversation %s: %s", conv_id, exc) + break + + ev = _evaluate_run(create_args, execute_result, turn_completed, expected_output) + return KdaRunResult( + conversation_id=conv_id, + eval=ev, + actual_create_args=create_args, + actual_execute_result=execute_result, + ) + + 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.eval.strict_pass for r in run_results) + pass_power_k = all(r.eval.strict_pass for r in run_results) + best = max( + run_results, + key=lambda r: sum([r.eval.kda_triggered, r.eval.executed, r.eval.success, r.eval.turn_completed]), + ) + return AgenticKdaSummary( + run_results=run_results, + pass_at_k=pass_at_k, + pass_power_k=pass_power_k, + best=best, + ) + + +class KdaSkillAssertionError(AssertionError): + """Raised when a KDA-skill evaluation fails.""" + + __tracebackhide__ = True + + +def evaluate_agentic_kda_skill( + 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, + langfuse: object | None = None, + dataset_item_id: str = "", + dataset_name: str = "kda_skill", + run_timestamp: str | None = None, + model_version_override: str | None = None, + run_metadata_extra: dict | None = None, + reasoning_effort: ReasoningEffort | None = None, +) -> None: + """Run KDA-skill evaluation, log to Langfuse, and raise KdaSkillAssertionError on failure.""" + from datetime import datetime as _dt # noqa: PLC0415 + from datetime import timezone as _tz # noqa: PLC0415 + + from gooddata_eval.core.agentic._langfuse import try_make_langfuse_client # noqa: PLC0415 + + if langfuse is None: + langfuse = try_make_langfuse_client() + window_start = _dt.now(_tz.utc) + summary = run_agentic_kda_skill( + 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, + ) + + if langfuse is not None and dataset_item_id: + from gooddata_eval.core.agentic._langfuse import ( # noqa: PLC0415 + build_run_context, + find_traces_per_conversation, + log_quality_and_value_scores, + observe, + score_safe, + ) + + run_name_base, run_metadata = build_run_context( + host, + token, + workspace_id, + dataset_name, + run_timestamp, + model_version_override, + run_metadata_extra, + reasoning_effort, + ) + traces_by_conv = find_traces_per_conversation( + langfuse, + [r.conversation_id for r in summary.run_results], + window_start, + select=lambda found: _select_kda_trace(langfuse, found), + ) + suffix_needed = len(summary.run_results) > 1 + for run_idx, run in enumerate(summary.run_results): + pt = traces_by_conv.get(run.conversation_id) + run_name = f"{run_name_base}_run{run_idx}" if suffix_needed else run_name_base + ev = run.eval + # Gates strict_pass -- current scope is completion only (see KdaEvaluation docstring). + strict_checks = { + "kda_triggered": ev.kda_triggered, + "executed": ev.executed, + "success": ev.success, + "turn_completed": ev.turn_completed, + } + # Informational only -- logged for visibility / a future correctness ticket, + # NOT part of strict_checks/strict_pass. See KdaEvaluation docstring. + informational_checks = { + "measure_correct": ev.measure_correct, + "date_attribute_correct": ev.date_attribute_correct, + "analyzed_period_correct": ev.analyzed_period_correct, + "reference_period_correct": ev.reference_period_correct, + "filters_correct": ev.filters_correct, + "summary_correct": ev.summary_correct, + } + # pt can be a fallback (non-KDA) trace when kda_triggered is False -- its + # latency/cost aren't real KDA numbers, so don't treat them as such. Only + # settle (wait + re-fetch) when we're actually going to trust pt's numbers. + if pt is not None and ev.kda_triggered: + pt = _settle_trace_latency(langfuse, pt) + kda_latency_sec = pt.latency if pt and ev.kda_triggered else None + report_bucket = classify_kda_report_bucket(ev, kda_latency_sec) + print( + f"[kda-report] {run_name}: bucket={report_bucket} strict_pass={ev.strict_pass} " + f"latency_sec={kda_latency_sec}", + flush=True, + ) + with observe(langfuse, pt.id if pt else None, dataset_item_id, run_name, run_metadata) as tid: + for score_name, value in {**strict_checks, **informational_checks}.items(): + score_safe(langfuse, tid, name=score_name, value=float(value), data_type="BOOLEAN") + for bucket in ("pass", "failed", "error"): + score_safe( + langfuse, + tid, + name=f"kda_report_{bucket}", + value=float(report_bucket == bucket), + data_type="BOOLEAN", + ) + log_quality_and_value_scores( + langfuse, + tid, + strict_checks=strict_checks, + latency_sec=kda_latency_sec, + cost_usd=pt.total_cost if pt and ev.kda_triggered else None, + ) + + if not summary.pass_at_k: + best = summary.best + ev = best.eval + message = ( + f"KDA skill assertion failed. strict_pass={ev.strict_pass} " + f"(kda_triggered={ev.kda_triggered}, executed={ev.executed}, " + f"success={ev.success}, turn_completed={ev.turn_completed}). " + f"Informational only, not part of strict_pass: " + f"measure_correct={ev.measure_correct}, date_attribute_correct={ev.date_attribute_correct}, " + f"analyzed_period_correct={ev.analyzed_period_correct}, " + f"reference_period_correct={ev.reference_period_correct}, " + f"filters_correct={ev.filters_correct}, summary_correct={ev.summary_correct}. " + f"Actual create args: {best.actual_create_args}. " + f"Actual execute result: {best.actual_execute_result}." + ) + raise KdaSkillAssertionError(message) diff --git a/packages/gooddata-eval/tests/test_agentic_kda_skill.py b/packages/gooddata-eval/tests/test_agentic_kda_skill.py new file mode 100644 index 000000000..01c4d3cd6 --- /dev/null +++ b/packages/gooddata-eval/tests/test_agentic_kda_skill.py @@ -0,0 +1,628 @@ +# (C) 2026 GoodData Corporation. All rights reserved. +# SPDX-License-Identifier: LicenseRef-GoodData-Enterprise +import json +from unittest.mock import MagicMock, patch + +import pytest +from gooddata_eval.core.agentic.kda_skill import ( + KdaEvaluation, + KdaSkillAssertionError, + _extract_kda_calls, + _filters_match, + _is_asking_clarification, + _measure_matches, + _normalize_measure, + _observation_has_kda_call, + _select_kda_trace, + _settle_trace_latency, + _to_number, + _trace_has_kda_call, + _within_tolerance, + classify_kda_report_bucket, + evaluate_agentic_kda_skill, + run_agentic_kda_skill, +) +from gooddata_eval.core.models import ChatResult + +_EXPECTED = {"Measure": {"type": "metric", "id": "revenue"}} + + +def _tool_call(name: str, result: dict | None = None, arguments: dict | None = None): + return { + "functionName": name, + "functionArguments": "{}" if arguments is None else json.dumps(arguments), + "result": None if result is None else json.dumps(result), + } + + +def _kda_chat_result(*, success: bool = True, text: str = "Here is the analysis.") -> ChatResult: + return ChatResult.model_validate( + { + "textResponse": text, + "toolCallEvents": [ + _tool_call("create_key_driver_analysis", arguments={"measure": {"type": "metric", "id": "revenue"}}), + _tool_call("execute_key_driver_analysis", result={"success": success, "data": {"summary": {}}}), + ], + "reasoningStepCount": 1, + } + ) + + +def _no_kda_chat_result(text: str = "I could not find that metric.") -> ChatResult: + return ChatResult.model_validate({"textResponse": text, "toolCallEvents": [], "reasoningStepCount": 1}) + + +# --------------------------------------------------------------------------- # +# Pure helpers +# --------------------------------------------------------------------------- # +def test_to_number_int(): + assert _to_number("42") == 42 + + +def test_to_number_float(): + assert _to_number("4.5") == 4.5 + + +def test_to_number_none_on_garbage(): + assert _to_number("not-a-number") is None + assert _to_number(None) is None + + +def test_normalize_measure(): + assert _normalize_measure({"type": "metric", "id": "revenue", "aggregation": "SUM"}) == ( + "metric", + "revenue", + "SUM", + ) + + +def test_measure_matches_single_candidate(): + assert _measure_matches({"type": "metric", "id": "revenue"}, {"type": "metric", "id": "revenue"}) is True + + +def test_measure_matches_list_of_candidates(): + actual = {"type": "fact", "id": "order_value", "aggregation": "SUM"} + expected = [{"type": "metric", "id": "revenue"}, {"type": "fact", "id": "order_value", "aggregation": "SUM"}] + assert _measure_matches(actual, expected) is True + + +def test_measure_matches_false_when_actual_not_a_dict(): + assert _measure_matches("revenue", {"type": "metric", "id": "revenue"}) is False + + +def test_measure_matches_false_when_expected_none(): + assert _measure_matches({"type": "metric", "id": "revenue"}, None) is False + + +def test_filters_match_equal_ignores_key_order(): + assert _filters_match([{"b": 2, "a": 1}], [{"a": 1, "b": 2}]) is True + + +def test_filters_match_false_on_mismatch(): + assert _filters_match([{"a": 1}], [{"a": 2}]) is False + + +def test_filters_match_treats_none_actual_as_empty_list(): + assert _filters_match(None, []) is True + + +def test_filters_match_false_on_non_serializable_value(): + assert _filters_match([{"a", "not json serializable"}], []) is False + + +def test_within_tolerance_true(): + assert _within_tolerance(100.0, 100.5, 1.0) is True + + +def test_within_tolerance_false_when_exceeds(): + assert _within_tolerance(100.0, 105.0, 1.0) is False + + +def test_within_tolerance_false_on_non_numeric(): + assert _within_tolerance("n/a", 100.0, 1.0) is False + + +@pytest.mark.parametrize( + "text", + ["Could you clarify which metric?", "Please provide the date range.", "Did you mean revenue?"], +) +def test_is_asking_clarification_true(text): + assert _is_asking_clarification(text) is True + + +def test_is_asking_clarification_false_on_plain_statement(): + assert _is_asking_clarification("Here is the key driver analysis result.") is False + + +def test_is_asking_clarification_false_on_empty(): + assert _is_asking_clarification("") is False + + +def test_extract_kda_calls_takes_last_execute_on_retry(): + events = ( + _kda_chat_result(success=False).tool_call_events + + ChatResult.model_validate( + { + "toolCallEvents": [ + _tool_call("execute_key_driver_analysis", result={"success": True, "data": {"summary": {}}}), + ], + } + ).tool_call_events + ) + create_args, execute_result = _extract_kda_calls(events) + assert create_args == {"measure": {"type": "metric", "id": "revenue"}} + assert execute_result == {"success": True, "data": {"summary": {}}} + + +def test_extract_kda_calls_none_when_no_tool_calls(): + create_args, execute_result = _extract_kda_calls([]) + assert create_args is None + assert execute_result is None + + +def test_extract_kda_calls_ignores_execute_call_with_no_result(): + events = ChatResult.model_validate( + {"toolCallEvents": [_tool_call("execute_key_driver_analysis", result=None)]} + ).tool_call_events + _, execute_result = _extract_kda_calls(events) + assert execute_result is None + + +# --------------------------------------------------------------------------- # +# KdaEvaluation.strict_pass / classify_kda_report_bucket +# --------------------------------------------------------------------------- # +def _evaluation(**overrides) -> KdaEvaluation: + fields = { + "kda_triggered": True, + "executed": True, + "success": True, + "turn_completed": True, + "measure_correct": True, + "date_attribute_correct": True, + "analyzed_period_correct": True, + "reference_period_correct": True, + "filters_correct": True, + "summary_correct": True, + } + fields.update(overrides) + return KdaEvaluation(**fields) + + +def test_strict_pass_true_when_all_core_checks_pass(): + assert _evaluation().strict_pass is True + + +def test_strict_pass_false_when_any_core_check_fails(): + assert _evaluation(success=False).strict_pass is False + + +def test_classify_kda_report_bucket_error_when_strict_pass_false(): + assert classify_kda_report_bucket(_evaluation(executed=False), latency_sec=10.0) == "error" + + +def test_classify_kda_report_bucket_pass_within_threshold(): + assert classify_kda_report_bucket(_evaluation(), latency_sec=59.9) == "pass" + + +def test_classify_kda_report_bucket_failed_over_threshold(): + assert classify_kda_report_bucket(_evaluation(), latency_sec=60.1) == "failed" + + +def test_classify_kda_report_bucket_failed_when_latency_unknown(): + # Unknown timing on a completed run must not be treated as a free pass. + assert classify_kda_report_bucket(_evaluation(), latency_sec=None) == "failed" + + +# --------------------------------------------------------------------------- # +# KDA-trace selection (observations pagination, containment check, fallback) +# --------------------------------------------------------------------------- # +def _observation(output): + return MagicMock(output=output) + + +def test_observation_has_kda_call_true(): + obs = _observation([{"type": "function_call", "name": "execute_key_driver_analysis"}]) + assert _observation_has_kda_call(obs) is True + + +def test_observation_has_kda_call_false_for_unrelated_function(): + obs = _observation([{"type": "function_call", "name": "create_metric"}]) + assert _observation_has_kda_call(obs) is False + + +def test_observation_has_kda_call_false_when_output_is_none(): + assert _observation_has_kda_call(_observation(None)) is False + + +def test_observation_has_kda_call_handles_non_list_output(): + obs = _observation({"type": "function_call", "name": "create_key_driver_analysis"}) + assert _observation_has_kda_call(obs) is True + + +def _observations_api(pages: list[list]): + """Fake langfuse.api.observations.list -- pages[i] is page i+1's observation list.""" + langfuse = MagicMock() + + def _list(trace_id, page=1): + data = pages[page - 1] + return MagicMock(data=data, total_pages=len(pages)) + + langfuse.api.observations.list.side_effect = _list + return langfuse + + +def test_trace_has_kda_call_found_on_first_page(): + langfuse = _observations_api([[_observation([{"type": "function_call", "name": "create_key_driver_analysis"}])]]) + assert _trace_has_kda_call(langfuse, "trace-1") is True + + +def test_trace_has_kda_call_found_on_later_page(): + langfuse = _observations_api( + [ + [_observation([{"type": "function_call", "name": "create_metric"}])], + [_observation([{"type": "function_call", "name": "execute_key_driver_analysis"}])], + ] + ) + assert _trace_has_kda_call(langfuse, "trace-1") is True + + +def test_trace_has_kda_call_false_when_exhausted(): + langfuse = _observations_api([[_observation([{"type": "function_call", "name": "create_metric"}])]]) + assert _trace_has_kda_call(langfuse, "trace-1") is False + + +def test_trace_has_kda_call_false_on_api_error(): + langfuse = MagicMock() + langfuse.api.observations.list.side_effect = RuntimeError("boom") + assert _trace_has_kda_call(langfuse, "trace-1") is False + + +def test_select_kda_trace_picks_the_one_with_the_call_over_higher_latency(): + with_call = MagicMock(id="t-with-call", latency=10.0) + without_call = MagicMock(id="t-without-call", latency=999.0) + langfuse = MagicMock() + langfuse.api.observations.list.side_effect = lambda trace_id, page=1: MagicMock( + data=[_observation([{"type": "function_call", "name": "create_key_driver_analysis"}])] + if trace_id == "t-with-call" + else [], + total_pages=1, + ) + assert _select_kda_trace(langfuse, [without_call, with_call]) is with_call + + +def test_select_kda_trace_falls_back_to_max_latency_when_none_have_the_call(): + low = MagicMock(id="low", latency=1.0) + high = MagicMock(id="high", latency=50.0) + langfuse = MagicMock() + langfuse.api.observations.list.side_effect = lambda trace_id, page=1: MagicMock(data=[], total_pages=1) + assert _select_kda_trace(langfuse, [low, high]) is high + + +def test_settle_trace_latency_waits_then_returns_the_refetched_trace(): + # The trace found right after the chat response can still be mid-ingestion (Langfuse + # writes observations asynchronously) -- the re-fetch after the settle delay must win. + early_trace = MagicMock(id="t1", latency=12.5) + settled_trace = MagicMock(id="t1", latency=28.7) + langfuse = MagicMock() + langfuse.api.trace.get.return_value = settled_trace + + with patch("gooddata_eval.core.agentic.kda_skill.time.sleep") as mock_sleep: + result = _settle_trace_latency(langfuse, early_trace) + + mock_sleep.assert_called_once_with(25.0) + langfuse.api.trace.get.assert_called_once_with("t1") + assert result is settled_trace + + +def test_settle_trace_latency_falls_back_to_original_on_fetch_error(): + early_trace = MagicMock(id="t1", latency=12.5) + langfuse = MagicMock() + langfuse.api.trace.get.side_effect = RuntimeError("boom") + + with patch("gooddata_eval.core.agentic.kda_skill.time.sleep"): + result = _settle_trace_latency(langfuse, early_trace) + + assert result is early_trace + + +# --------------------------------------------------------------------------- # +# run_agentic_kda_skill +# --------------------------------------------------------------------------- # +def test_run_agentic_kda_skill_triggers_and_succeeds(): + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.return_value = _kda_chat_result(success=True) + + with patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client): + summary = run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=1, + ) + + assert summary.pass_at_k is True + assert summary.best.eval.kda_triggered is True + assert summary.best.eval.executed is True + assert summary.best.eval.success is True + mock_client.close.assert_called_once() + + +def test_run_agentic_kda_skill_no_tool_call(): + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.return_value = _no_kda_chat_result() + + with patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client): + summary = run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=1, + ) + + assert summary.pass_at_k is False + assert summary.best.eval.kda_triggered is False + + +def test_run_agentic_kda_skill_resolves_after_clarification_turn(): + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.side_effect = [ + _no_kda_chat_result("Could you clarify which revenue measure you mean?"), + _kda_chat_result(success=True), + ] + + with ( + patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client), + patch( + "gooddata_eval.core.agentic.kda_skill.generate_simulated_kda_response", + return_value="The revenue metric is fine.", + ) as mock_simulate, + ): + summary = run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=2, + ) + + mock_simulate.assert_called_once() + assert summary.pass_at_k is True + assert mock_client.send_message.call_count == 2 + + +def test_run_agentic_kda_skill_gives_up_after_max_iterations_of_clarification(): + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.return_value = _no_kda_chat_result("Could you clarify which measure?") + + with ( + patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client), + patch( + "gooddata_eval.core.agentic.kda_skill.generate_simulated_kda_response", + return_value="Please use revenue.", + ), + ): + summary = run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=2, + ) + + assert summary.pass_at_k is False + assert mock_client.send_message.call_count == 2 + + +def test_run_agentic_kda_skill_survives_simulated_reply_failure(): + # The simulated-user helper is a safety net, not the assertion under test -- if it + # raises, only the current run ends early; earlier completed runs are preserved. + mock_client = MagicMock() + mock_client.create_conversation.side_effect = ["conv-1", "conv-2"] + mock_client.send_message.side_effect = [ + _kda_chat_result(success=True), # run 0: triggers KDA immediately + _no_kda_chat_result("Could you clarify which measure?"), # run 1: asks, then helper blows up + ] + + with ( + patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client), + patch( + "gooddata_eval.core.agentic.kda_skill.generate_simulated_kda_response", + side_effect=RuntimeError("openai down"), + ), + ): + summary = run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=2, + max_iterations=2, + ) + + assert len(summary.run_results) == 2 + assert summary.run_results[0].eval.kda_triggered is True + assert summary.run_results[1].eval.kda_triggered is False + assert summary.pass_at_k is True # run 0 still counts + + +def test_run_agentic_kda_skill_uses_initial_conversation_for_run_0(): + mock_client = MagicMock() + mock_client.send_message.return_value = _kda_chat_result(success=True) + with patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client): + run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=1, + initial_conversation_id="existing-conv", + ) + mock_client.create_conversation.assert_not_called() + mock_client.delete_conversation.assert_not_called() + + +def test_run_agentic_kda_skill_creates_fresh_conversations_for_remaining_runs(): + mock_client = MagicMock() + mock_client.create_conversation.side_effect = ["fresh-1", "fresh-2"] + mock_client.send_message.return_value = _kda_chat_result(success=True) + with patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client): + run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=3, + max_iterations=1, + initial_conversation_id="existing-conv", + ) + assert mock_client.create_conversation.call_count == 2 + assert mock_client.delete_conversation.call_count == 2 + + +# --------------------------------------------------------------------------- # +# evaluate_agentic_kda_skill +# --------------------------------------------------------------------------- # +def test_evaluate_agentic_kda_skill_raises_on_failure(): + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.return_value = _no_kda_chat_result() + + with ( + patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client), + patch("gooddata_eval.core.agentic._langfuse.try_make_langfuse_client", return_value=None), + pytest.raises(KdaSkillAssertionError), + ): + evaluate_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=1, + langfuse=None, + ) + + +def test_evaluate_agentic_kda_skill_does_not_raise_on_success(): + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.return_value = _kda_chat_result(success=True) + + with ( + patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client), + patch("gooddata_eval.core.agentic._langfuse.try_make_langfuse_client", return_value=None), + ): + evaluate_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=1, + langfuse=None, + ) + + +def test_evaluate_agentic_kda_skill_never_treats_fallback_trace_latency_as_kda_latency(): + # Regression test: when KDA never triggered, the trace picked by find_traces_per_conversation + # is _select_kda_trace's max-latency FALLBACK, not a real KDA turn -- its latency/cost must + # not be logged as the KDA run's own value_score inputs. + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.return_value = _no_kda_chat_result() + + fallback_trace = MagicMock(id="fallback-trace", latency=999.0, total_cost=5.0) + mock_langfuse = MagicMock() + + with ( + patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client), + patch("gooddata_eval.core.agentic._langfuse.build_run_context", return_value=("run-base", {})), + patch( + "gooddata_eval.core.agentic._langfuse.find_traces_per_conversation", + return_value={"conv-1": fallback_trace}, + ), + patch("gooddata_eval.core.agentic._langfuse.observe") as mock_observe, + patch("gooddata_eval.core.agentic._langfuse.score_safe"), + patch("gooddata_eval.core.agentic._langfuse.log_quality_and_value_scores") as mock_log_scores, + pytest.raises(KdaSkillAssertionError), + ): + mock_observe.return_value.__enter__.return_value = "fallback-trace" + evaluate_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=1, + langfuse=mock_langfuse, + dataset_item_id="item-1", + ) + + mock_log_scores.assert_called_once() + assert mock_log_scores.call_args.kwargs["latency_sec"] is None + assert mock_log_scores.call_args.kwargs["cost_usd"] is None + + +def test_evaluate_agentic_kda_skill_uses_settled_latency_when_kda_triggered(): + # When KDA does trigger, the early trace from find_traces_per_conversation must be + # re-fetched (after the settle delay) before its latency is trusted for the report. + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.return_value = _kda_chat_result(success=True) + + early_trace = MagicMock(id="trace-1", latency=12.5, total_cost=0.01) + settled_trace = MagicMock(id="trace-1", latency=76.0, total_cost=0.02) + mock_langfuse = MagicMock() + mock_langfuse.api.trace.get.return_value = settled_trace + + with ( + patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client), + patch("gooddata_eval.core.agentic.kda_skill.time.sleep") as mock_sleep, + patch("gooddata_eval.core.agentic._langfuse.build_run_context", return_value=("run-base", {})), + patch( + "gooddata_eval.core.agentic._langfuse.find_traces_per_conversation", + return_value={"conv-1": early_trace}, + ), + patch("gooddata_eval.core.agentic._langfuse.observe") as mock_observe, + patch("gooddata_eval.core.agentic._langfuse.score_safe"), + patch("gooddata_eval.core.agentic._langfuse.log_quality_and_value_scores") as mock_log_scores, + ): + mock_observe.return_value.__enter__.return_value = "trace-1" + evaluate_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=1, + langfuse=mock_langfuse, + dataset_item_id="item-1", + ) + + mock_sleep.assert_called_once_with(25.0) + mock_langfuse.api.trace.get.assert_called_once_with("trace-1") + mock_log_scores.assert_called_once() + assert mock_log_scores.call_args.kwargs["latency_sec"] == 76.0 + assert mock_log_scores.call_args.kwargs["cost_usd"] == 0.02