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/kda_skill.py b/packages/gooddata-eval/src/gooddata_eval/core/agentic/kda_skill.py new file mode 100644 index 000000000..a4b67c25f --- /dev/null +++ b/packages/gooddata-eval/src/gooddata_eval/core/agentic/kda_skill.py @@ -0,0 +1,444 @@ +# (C) 2026 GoodData Corporation. All rights reserved. +"""Agentic KDA (Key Driver Analysis)-skill evaluation runner.""" + +from __future__ import annotations + +import json +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.models import ToolCallEvent + +_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 + + +@dataclass +class KdaEvaluation: + """Evaluation scores for a single KDA-skill run. + + Scope (QA-28800): 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]) + + +@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 + # Wall-clock duration of the single send_message call that triggered KDA (None if + # never triggered). Measured directly, not via Langfuse trace lookup: a session can + # contain more than one turn (disambiguation, or a transient-retry the client SDK + # does internally) and there is no reliable way to infer from trace metadata alone + # which turn is "the" KDA one -- but _run_once's loop breaks the instant a turn + # triggers KDA, so timing that exact turn is always correct by construction. + kda_turn_latency_sec: float | 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, +) -> 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) + + def _run_once(conv_id: str) -> KdaRunResult: + create_args: dict | None = None + execute_result: dict | None = None + turn_completed = False + kda_turn_latency_sec: float | None = None + current_question = question + + for iteration in range(max_iterations): + turn_start = time.monotonic() + chat_result = client.send_message(conv_id, current_question) + turn_elapsed = time.monotonic() - turn_start + 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 + kda_turn_latency_sec = turn_elapsed + break + response_text = (chat_result.text_response or "").strip() + if iteration >= max_iterations - 1 or not _is_asking_clarification(response_text): + break + current_question = generate_simulated_kda_response(response_text, expected_output.get("Measure")) + + 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, + kda_turn_latency_sec=kda_turn_latency_sec, + ) + + 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, +) -> 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, + ) + + 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, + ) + traces_by_conv = find_traces_per_conversation( + langfuse, + [r.conversation_id for r in summary.run_results], + window_start, + ) + 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 QA-28800 scope (process ran to completion). + 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, + } + 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") + # kda_turn_latency_sec is measured directly around the triggering send_message + # call (see KdaRunResult docstring) -- deliberately NOT pt.latency, which reflects + # whichever trace in the session happens to have the largest Langfuse-reported + # latency and can pick a disambiguation/retry turn instead of the KDA one. + if run.kda_turn_latency_sec is not None: + score_safe( + langfuse, tid, name="kda_turn_latency_sec", value=run.kda_turn_latency_sec, data_type="NUMERIC" + ) + log_quality_and_value_scores( + langfuse, + tid, + strict_checks=strict_checks, + latency_sec=run.kda_turn_latency_sec, + cost_usd=pt.total_cost if pt 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)