From 19bfdfb37f8737a0f9d44b4a927ac37d05aa6772 Mon Sep 17 00:00:00 2001 From: Frank Huynh Date: Wed, 5 Aug 2026 21:26:30 +0700 Subject: [PATCH 1/2] feat(gooddata-eval): add KDA-skill agentic evaluator Adds run_agentic_kda_skill/evaluate_agentic_kda_skill, mirroring the metric_skill/alert_skill evaluators: runs the agentic chat flow against a KDA question, asserts the tool chain triggers/executes/succeeds and the chatbot delivers a final answer (per-field correctness is computed and logged for visibility but scoped out of strict_pass for now), and logs Langfuse scores including a daily-report pass/failed/error bucket keyed off latency. Latency is measured by picking the trace whose observations actually contain the KDA tool call (not just the longest one, since a conversation can have multiple turns) and by re-fetching it after a 25s settle delay -- verified empirically that Langfuse's async ingestion can otherwise report a trace's latency well under its true, settled value. JIRA: QA-28800 risk: nonprod --- .../gooddata_eval/core/agentic/__init__.py | 14 + .../gooddata_eval/core/agentic/_langfuse.py | 46 +- .../gooddata_eval/core/agentic/kda_skill.py | 540 +++++++++++++++ .../tests/test_agentic_kda_skill.py | 628 ++++++++++++++++++ 4 files changed, 1225 insertions(+), 3 deletions(-) create mode 100644 packages/gooddata-eval/src/gooddata_eval/core/agentic/kda_skill.py create mode 100644 packages/gooddata-eval/tests/test_agentic_kda_skill.py 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 From 8557c38198a482a467503ba5d9e6e319264ae1d4 Mon Sep 17 00:00:00 2001 From: Frank Huynh Date: Thu, 6 Aug 2026 11:31:08 +0700 Subject: [PATCH 2/2] fix(gooddata-eval): stop treating unknown KDA latency as a free pass Two review findings, both about the same root cause -- reading Langfuse trace data before it's had time to fully ingest: - _TraceObj.latency coerced a missing/null latency (trace not fully ingested yet) to 0.0. Since 0.0 <= threshold, an unknown-duration run was classified as the fastest possible "pass" instead of "failed" -- exactly backwards from classify_kda_report_bucket's own documented intent. Fixed by preserving None through _TraceObj (shared by every skill, so this also fixes the same silent-zero bug in their value_score speed calculation). - _select_kda_trace checked candidates' observations for the KDA tool call BEFORE the settle delay, i.e. at the point ingestion is least complete -- risking a wrong pick in the one case with more than one candidate (disambiguation). The caller then trusted a fallback guess's latency based on kda_triggered (an unrelated SSE signal), not on whether the selector actually confirmed the match. Fixed by moving the wait before the observation check and having the selector report back whether it matched, so the caller only trusts a trace's numbers when both signals agree. JIRA: QA-28800 risk: nonprod --- .../gooddata_eval/core/agentic/_langfuse.py | 6 +- .../gooddata_eval/core/agentic/kda_skill.py | 75 +++++++---- .../tests/test_agentic_kda_skill.py | 125 ++++++++++++++---- .../tests/test_agentic_langfuse_trace.py | 19 +++ 4 files changed, 175 insertions(+), 50 deletions(-) create mode 100644 packages/gooddata-eval/tests/test_agentic_langfuse_trace.py 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 34bca4da6..31ae08c80 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/agentic/_langfuse.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/agentic/_langfuse.py @@ -31,7 +31,11 @@ def __init__(self, raw: dict) -> None: self.id: str = raw.get("id", "") self.metadata: dict = raw.get("metadata") or {} self.session_id: str | None = raw.get("sessionId") or raw.get("session_id") - self.latency: float = float(raw.get("latency") or 0.0) + # None (missing/null) is preserved, not coerced to 0.0 -- a trace that hasn't + # finished ingesting has UNKNOWN latency, not zero latency, and callers (e.g. + # classify_kda_report_bucket) rely on that distinction to not treat "unknown" as + # the best possible outcome. + self.latency: float | None = float(raw["latency"]) if raw.get("latency") is not None else None self.total_cost: float = float(raw.get("totalCost") or raw.get("total_cost") or 0.0) 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 index 7db661c65..80871f61e 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/agentic/kda_skill.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/agentic/kda_skill.py @@ -176,28 +176,36 @@ def _trace_has_kda_call(langfuse: Any, trace_id: str) -> bool: 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) +def _refetch_trace(langfuse: Any, trace: Any) -> Any: + """Re-fetch ``trace`` by id -- best-effort, falls back to the original if it fails.""" 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) + _log.debug("Failed to re-fetch trace %s: %s", trace.id, exc) return trace +def _select_kda_trace(langfuse: Any, candidates: list[Any]) -> tuple[Any | None, bool]: + """Pick the candidate trace that made the KDA tool call. + + Waits _LATENCY_SETTLE_DELAY_SEC *before* checking observations, not after picking a + candidate -- a trace's observations can be just as mid-ingestion as its latency, so + checking too early risks missing the real KDA call and picking the wrong candidate + in a disambiguation case (the one scenario with more than one candidate to choose + from). The matched candidate is then re-fetched so its latency reflects the same + settled state. + + Returns (trace, matched): matched is False when no candidate's observations + confirmed the KDA call and a max-latency guess was used instead -- callers must not + treat that guess's latency as a real KDA duration. + """ + time.sleep(_LATENCY_SETTLE_DELAY_SEC) + for candidate in candidates: + if _trace_has_kda_call(langfuse, candidate.id): + return _refetch_trace(langfuse, candidate), True + return max(candidates, key=lambda t: getattr(t, "latency", None) or 0.0), False + + @dataclass class KdaEvaluation: """Evaluation scores for a single KDA-skill run. @@ -240,6 +248,11 @@ def classify_kda_report_bucket(ev: KdaEvaluation, latency_sec: float | None) -> 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. + + ``latency_sec`` is the whole conversational turn (LLM planning/orchestration plus + every tool call, not isolated KDA-backend execution time) -- intentional, since what + this report tracks is how long the *user* waits, not which component of the turn is + slow. A slow turn here does not by itself pin the slowness on the KDA backend. """ if not ev.strict_pass: return "error" @@ -463,11 +476,23 @@ def evaluate_agentic_kda_skill( run_metadata_extra, reasoning_effort, ) + # find_traces_per_conversation's select must return a single trace, so the + # "was this candidate actually confirmed to have made the KDA call" bit rides + # along in this closure-captured dict instead (keyed by session_id, which every + # candidate for a conversation shares). + kda_matched_by_conv: dict[str, bool] = {} + + def _select_and_record(found: list[Any]) -> Any | None: + trace, matched = _select_kda_trace(langfuse, found) + if trace is not None and trace.session_id: + kda_matched_by_conv[trace.session_id] = matched + return trace + 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), + select=_select_and_record, ) suffix_needed = len(summary.run_results) > 1 for run_idx, run in enumerate(summary.run_results): @@ -491,12 +516,14 @@ def evaluate_agentic_kda_skill( "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 + # pt can be a fallback (non-KDA) trace -- either because kda_triggered is + # False, or because the selector itself couldn't confirm which candidate + # made the KDA call (kda_matched False, e.g. a disambiguation case where + # observations weren't ready yet). Its latency/cost aren't real KDA numbers + # in either case, so don't treat them as such. + kda_matched = kda_matched_by_conv.get(run.conversation_id, False) + kda_trusted = pt is not None and ev.kda_triggered and kda_matched + kda_latency_sec = pt.latency if kda_trusted 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} " @@ -519,7 +546,7 @@ def evaluate_agentic_kda_skill( tid, strict_checks=strict_checks, latency_sec=kda_latency_sec, - cost_usd=pt.total_cost if pt and ev.kda_triggered else None, + cost_usd=pt.total_cost if kda_trusted else None, ) if not summary.pass_at_k: diff --git a/packages/gooddata-eval/tests/test_agentic_kda_skill.py b/packages/gooddata-eval/tests/test_agentic_kda_skill.py index 01c4d3cd6..55306f28c 100644 --- a/packages/gooddata-eval/tests/test_agentic_kda_skill.py +++ b/packages/gooddata-eval/tests/test_agentic_kda_skill.py @@ -13,8 +13,8 @@ _measure_matches, _normalize_measure, _observation_has_kda_call, + _refetch_trace, _select_kda_trace, - _settle_trace_latency, _to_number, _trace_has_kda_call, _within_tolerance, @@ -280,6 +280,7 @@ def test_trace_has_kda_call_false_on_api_error(): 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) + settled = MagicMock(id="t-with-call", latency=11.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"}])] @@ -287,7 +288,14 @@ def test_select_kda_trace_picks_the_one_with_the_call_over_higher_latency(): else [], total_pages=1, ) - assert _select_kda_trace(langfuse, [without_call, with_call]) is with_call + langfuse.api.trace.get.return_value = settled + + with patch("gooddata_eval.core.agentic.kda_skill.time.sleep") as mock_sleep: + trace, matched = _select_kda_trace(langfuse, [without_call, with_call]) + + mock_sleep.assert_called_once_with(25.0) + assert matched is True + assert trace is settled # re-fetched after the match, not the stale candidate object def test_select_kda_trace_falls_back_to_max_latency_when_none_have_the_call(): @@ -295,34 +303,48 @@ def test_select_kda_trace_falls_back_to_max_latency_when_none_have_the_call(): 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 + + with patch("gooddata_eval.core.agentic.kda_skill.time.sleep"): + trace, matched = _select_kda_trace(langfuse, [low, high]) + + assert matched is False + assert trace is high # fallback guess only, caller must not trust its latency + langfuse.api.trace.get.assert_not_called() # no point re-fetching a trace we don't trust -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) +def test_select_kda_trace_waits_before_checking_observations(): + # The observation check needs ingested data just as much as the latency field does -- + # checking before the settle delay is exactly the bug this ordering avoids. + with_call = MagicMock(id="t1", latency=10.0) langfuse = MagicMock() - langfuse.api.trace.get.return_value = settled_trace + call_order = [] + langfuse.api.observations.list.side_effect = lambda trace_id, page=1: ( + call_order.append("observations.list"), + MagicMock( + data=[_observation([{"type": "function_call", "name": "create_key_driver_analysis"}])], total_pages=1 + ), + )[1] - with patch("gooddata_eval.core.agentic.kda_skill.time.sleep") as mock_sleep: - result = _settle_trace_latency(langfuse, early_trace) + with patch("gooddata_eval.core.agentic.kda_skill.time.sleep", side_effect=lambda _: call_order.append("sleep")): + _select_kda_trace(langfuse, [with_call]) - mock_sleep.assert_called_once_with(25.0) - langfuse.api.trace.get.assert_called_once_with("t1") - assert result is settled_trace + assert call_order == ["sleep", "observations.list"] -def test_settle_trace_latency_falls_back_to_original_on_fetch_error(): - early_trace = MagicMock(id="t1", latency=12.5) +def test_refetch_trace_returns_the_fresh_trace(): + stale = MagicMock(id="t1") + fresh = MagicMock(id="t1") langfuse = MagicMock() - langfuse.api.trace.get.side_effect = RuntimeError("boom") + langfuse.api.trace.get.return_value = fresh + assert _refetch_trace(langfuse, stale) is fresh + langfuse.api.trace.get.assert_called_once_with("t1") - with patch("gooddata_eval.core.agentic.kda_skill.time.sleep"): - result = _settle_trace_latency(langfuse, early_trace) - assert result is early_trace +def test_refetch_trace_falls_back_to_original_on_error(): + stale = MagicMock(id="t1") + langfuse = MagicMock() + langfuse.api.trace.get.side_effect = RuntimeError("boom") + assert _refetch_trace(langfuse, stale) is stale # --------------------------------------------------------------------------- # @@ -585,24 +607,31 @@ def test_evaluate_agentic_kda_skill_never_treats_fallback_trace_latency_as_kda_l 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. + # When the selector actually confirms the KDA call (matched=True), its re-fetched, + # settled latency/cost must be what gets reported -- not the stale candidate's. 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) + early_trace = MagicMock(id="trace-1", session_id="conv-1", latency=12.5, total_cost=0.01) + settled_trace = MagicMock(id="trace-1", session_id="conv-1", latency=76.0, total_cost=0.02) mock_langfuse = MagicMock() + mock_langfuse.api.observations.list.return_value = MagicMock( + data=[_observation([{"type": "function_call", "name": "create_key_driver_analysis"}])], + total_pages=1, + ) mock_langfuse.api.trace.get.return_value = settled_trace + def fake_find_traces(langfuse, conversation_ids, window_start, select=None): + return {cid: select([early_trace]) for cid in conversation_ids} + 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}, + side_effect=fake_find_traces, ), patch("gooddata_eval.core.agentic._langfuse.observe") as mock_observe, patch("gooddata_eval.core.agentic._langfuse.score_safe"), @@ -626,3 +655,49 @@ def test_evaluate_agentic_kda_skill_uses_settled_latency_when_kda_triggered(): 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 + + +def test_evaluate_agentic_kda_skill_does_not_trust_latency_when_selector_could_not_match(): + # ev.kda_triggered=True (SSE saw the tool call) but the selector still fell back + # (matched=False, e.g. observations weren't ingested yet) -- must not trust pt's + # latency/cost just because the SSE signal alone looked fine. + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.return_value = _kda_chat_result(success=True) + + fallback_trace = MagicMock(id="trace-1", session_id="conv-1", latency=999.0, total_cost=5.0) + mock_langfuse = MagicMock() + mock_langfuse.api.observations.list.return_value = MagicMock(data=[], total_pages=1) + + def fake_find_traces(langfuse, conversation_ids, window_start, select=None): + return {cid: select([fallback_trace]) for cid in conversation_ids} + + with ( + patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client), + patch("gooddata_eval.core.agentic.kda_skill.time.sleep"), + patch("gooddata_eval.core.agentic._langfuse.build_run_context", return_value=("run-base", {})), + patch( + "gooddata_eval.core.agentic._langfuse.find_traces_per_conversation", + side_effect=fake_find_traces, + ), + 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_langfuse.api.trace.get.assert_not_called() + 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 diff --git a/packages/gooddata-eval/tests/test_agentic_langfuse_trace.py b/packages/gooddata-eval/tests/test_agentic_langfuse_trace.py new file mode 100644 index 000000000..fe95f3aa1 --- /dev/null +++ b/packages/gooddata-eval/tests/test_agentic_langfuse_trace.py @@ -0,0 +1,19 @@ +# (C) 2026 GoodData Corporation +from gooddata_eval.core.agentic._langfuse import _TraceObj + + +def test_trace_obj_latency_none_when_missing(): + assert _TraceObj({"id": "t1"}).latency is None + + +def test_trace_obj_latency_none_when_explicitly_null(): + assert _TraceObj({"id": "t1", "latency": None}).latency is None + + +def test_trace_obj_latency_preserves_real_zero(): + # A real 0.0 (start == end) must stay 0.0, not be confused with "unknown". + assert _TraceObj({"id": "t1", "latency": 0.0}).latency == 0.0 + + +def test_trace_obj_latency_preserves_real_value(): + assert _TraceObj({"id": "t1", "latency": 45.3}).latency == 45.3