diff --git a/packages/gooddata-eval/src/gooddata_eval/cli/agentic_runner.py b/packages/gooddata-eval/src/gooddata_eval/cli/agentic_runner.py index 0bd6f5cf8..d592853b6 100644 --- a/packages/gooddata-eval/src/gooddata_eval/cli/agentic_runner.py +++ b/packages/gooddata-eval/src/gooddata_eval/cli/agentic_runner.py @@ -18,6 +18,7 @@ from gooddata_eval.core.agentic.metric_skill import evaluate_agentic_metric_skill from gooddata_eval.core.agentic.search_tool import evaluate_agentic_search_tool from gooddata_eval.core.agentic.visualization import evaluate_agentic_visualization +from gooddata_eval.core.agentic.what_if import evaluate_agentic_what_if from gooddata_eval.core.config import ReasoningEffort from gooddata_eval.core.models import AgenticEvalOutcome, CreatedVisualization, DatasetItem from gooddata_eval.core.runner import EvalReport, ItemReport @@ -44,6 +45,7 @@ class _LfKw(TypedDict, total=False): "agentic_guardrail", "agentic_conversation", "agentic_kda_skill", + "agentic_what_if", } ) @@ -228,6 +230,17 @@ def _dispatch_agentic( agent_id=agent_id, **lf_kw, ) + elif kind == "agentic_what_if": + return evaluate_agentic_what_if( + host=host, + token=token, + workspace_id=workspace_id, + question=item.question, + expected_output=eo if isinstance(eo, dict) else {}, + k=k, + agent_id=agent_id, + **lf_kw, + ) elif kind == "agentic_conversation": fixture_data = eo.get("fixture") or eo if isinstance(eo, dict) else {} return evaluate_agentic_conversation( diff --git a/packages/gooddata-eval/src/gooddata_eval/core/agentic/what_if.py b/packages/gooddata-eval/src/gooddata_eval/core/agentic/what_if.py new file mode 100644 index 000000000..9a01b1575 --- /dev/null +++ b/packages/gooddata-eval/src/gooddata_eval/core/agentic/what_if.py @@ -0,0 +1,593 @@ +# (C) 2026 GoodData Corporation. All rights reserved. +"""Agentic what-if-analysis skill evaluation runner. + +The skill builds a scenario spec and executes it: + + create_what_if_scenario(visualization_ref, scenarios[], include_baseline) + execute_what_if_scenario(scenario_ref) -> one result per scenario, plus the baseline + +Each scenario carries adjustments of the form ``{metric_id, metric_type, scenario_maql}``, +where ``scenario_maql`` is the adjusted expression -- a 10% uplift on a revenue metric +defined as ``SELECT SUM({fact/price} * {fact/quantity})`` becomes +``SELECT SUM({fact/price} * 1.10 * {fact/quantity})``. + +That makes this the most checkable of the analysis skills: the adjustment is MAQL, and +MAQL already has a comparator here (``evaluators._maql.normalize_maql``, used by +metric_skill), so "did it apply the right adjustment" is answerable without a judge. What +the adjustment produced is not checked -- that is the platform's arithmetic, not the +agent's -- only that the agent asked for the right thing and the execution succeeded. +""" + +from __future__ import annotations + +import logging +import os +from dataclasses import dataclass, field +from typing import Any + +from gooddata_eval.core.agentic._trace_linker import ( + RunIdentity, + RunTraceContext, + SubmitTraceLink, + open_trace_window, + run_trace_link_inline, + submit_trace_scoring, + utc_now, +) +from gooddata_eval.core.chat.render import render_answer_text +from gooddata_eval.core.chat.sse_client import ChatClient +from gooddata_eval.core.config import ReasoningEffort +from gooddata_eval.core.evaluators._maql import normalize_maql +from gooddata_eval.core.models import ( + AgenticAssertionError, + AgenticEvalOutcome, + ChatResult, + ReasoningStepEvent, + ToolCallEvent, + build_latency_breakdown, + shift_and_index_events, +) + +_log = logging.getLogger(__name__) + +_DEFAULT_K = 1 +# The agent asks which measure to adjust before building anything (observed live: "I need +# to confirm which 'Spend' calculation you want to adjust"), so the budget covers a couple +# of disambiguation rounds plus slack. +_DEFAULT_MAX_ITERATIONS = 4 + + +def _build_clarification_prompt(agent_message: str, expected_output: dict) -> str: + """The simulated-user reply, mentioning only the hints the fixture actually supplies.""" + hints: list[str] = [] + metric = expected_output.get("metric_id") + if metric: + hints.append(f"the measure to adjust is '{metric}'") + change = expected_output.get("change") + if change: + hints.append(f"the adjustment is {change}") + period = expected_output.get("period") + if period: + hints.append(f"the time period is {period}") + reference = "; ".join(hints) + return ( + f"You are simulating a user in a conversation with a BI assistant that runs what-if " + f"scenario analysis. The assistant asked: '{agent_message}'. " + + (f"For reference, {reference}. " if reference else "") + + "Reply briefly as the user, answering whichever of those the assistant actually asked about." + ) + + +def generate_simulated_what_if_response(agent_message: str, expected_output: dict) -> str: + """Generate a user reply to keep the what-if conversation going (gpt-4o-mini). + + Always OpenAI regardless of the workspace's own model: harness plumbing, not the system + under test. + """ + try: + from openai import OpenAI # noqa: PLC0415 + except ImportError as exc: + raise RuntimeError("openai package is required for generate_simulated_what_if_response") from exc + + api_key = os.environ.get("OPENAI_API_KEY") + if not api_key: + raise OSError("OPENAI_API_KEY environment variable is not set") + + client = OpenAI(api_key=api_key) + response = client.chat.completions.create( + model="gpt-4o-mini", + messages=[{"role": "user", "content": _build_clarification_prompt(agent_message, expected_output)}], + max_tokens=150, + temperature=0, + timeout=30, + ) + return response.choices[0].message.content or "Please proceed with the most complete option." + + +def _extract_what_if_calls(tool_call_events: list[ToolCallEvent]) -> tuple[dict | None, dict | None]: + """Return (create_args, execute_result) for the LAST create/execute pair. + + A new create_what_if_scenario clears any earlier execute result: that result belongs to + the spec it followed. Picking the last of each independently would score a fresh + scenario against a stale execution. + """ + create_args: dict | None = None + execute_result: dict | None = None + for tc in tool_call_events: + if tc.function_name == "create_what_if_scenario": + create_args = tc.parsed_arguments() + execute_result = None + elif tc.function_name == "execute_what_if_scenario" and tc.result: + execute_result = tc.parsed_result() + return create_args, execute_result + + +def _adjustments(create_args: dict | None) -> list[dict]: + """Every adjustment across every scenario, flattened. + + Scenario grouping does not matter to the checks below -- a fixture asserts that the + right measure was adjusted the right way, not which scenario label it landed under. + """ + scenarios = (create_args or {}).get("scenarios") + if not isinstance(scenarios, list): + return [] + out: list[dict] = [] + for scenario in scenarios: + if not isinstance(scenario, dict): + continue + out.extend(a for a in scenario.get("adjustments") or [] if isinstance(a, dict)) + return out + + +def _maql_matches(actual_maql: str, expected: str | list[str]) -> bool: + """Whether the adjustment matches any accepted expression, compared as MAQL. + + Uses metric_skill's normalizer, so whitespace and casing differences do not decide a + verdict. A list is a candidate set: several expressions can be equally correct + adjustments (``* 1.1`` and ``* 1.10``, or a rewrite that reaches the same value). + """ + candidates = [expected] if isinstance(expected, str) else list(expected) + normalized = normalize_maql(actual_maql) + return any(normalized == normalize_maql(c) for c in candidates) + + +@dataclass +class WhatIfEvaluation: + """Scores for a single what-if run. + + ``triggered``/``executed``/``success``/``turn_completed`` are the shared process checks. + ``metric_correct``, ``maql_correct``, ``scenario_count_correct`` and ``baseline_correct`` + are content checks, each True when the fixture did not pin it; ``asserted`` records + which ones it did, so a run that verified nothing is not reported as a full pass. + """ + + triggered: bool + executed: bool + success: bool + turn_completed: bool + metric_correct: bool + maql_correct: bool + scenario_count_correct: bool + baseline_correct: bool + asserted: list[str] = field(default_factory=list) + disambiguated: bool = False + + @property + def strict_pass(self) -> bool: + return all( + [ + self.triggered, + self.executed, + self.success, + self.turn_completed, + self.metric_correct, + self.maql_correct, + self.scenario_count_correct, + self.baseline_correct, + ] + ) + + +@dataclass +class WhatIfRunResult: + """Outcome of one run (one conversation, up to max_iterations messages).""" + + conversation_id: str + evaluation: WhatIfEvaluation + actual_create_args: dict | None + actual_execute_result: dict | None + turn_wall_clock_sec: float | None = None + reasoning_steps: list[str] = field(default_factory=list) + response_id: str | None = None + tool_call_events: list[ToolCallEvent] = field(default_factory=list) + reasoning_step_events: list[ReasoningStepEvent] = field(default_factory=list) + + +@dataclass +class AgenticWhatIfSummary: + """Aggregated outcome of K runs for one what-if item.""" + + run_results: list[WhatIfRunResult] + pass_at_k: bool + pass_power_k: bool + best: WhatIfRunResult + + +def _evaluate_run( + create_args: dict | None, + execute_result: dict | None, + expected_output: dict, + turn_completed: bool, + disambiguated: bool = False, +) -> WhatIfEvaluation: + triggered = create_args is not None + executed = execute_result is not None + success = executed and execute_result.get("success") is True + adjustments = _adjustments(create_args) + asserted: list[str] = [] + + expected_metric = expected_output.get("metric_id") + if not expected_metric: + wanted: set[str] | None = None + metric_correct = True + else: + asserted.append("metric_id") + wanted = {expected_metric} if isinstance(expected_metric, str) else set(expected_metric) + metric_correct = any(a.get("metric_id") in wanted for a in adjustments) + + expected_maql = expected_output.get("scenario_maql") + if not expected_maql: + maql_correct = True + else: + asserted.append("scenario_maql") + # Only adjustments on the expected measure. Searching every adjustment independently + # lets a wrong adjustment on the right metric and a right adjustment on the wrong + # metric satisfy the two checks between them -- two failures scoring as a pass. + candidates = adjustments if wanted is None else [a for a in adjustments if a.get("metric_id") in wanted] + maql_correct = any( + isinstance(a.get("scenario_maql"), str) and _maql_matches(a["scenario_maql"], expected_maql) + for a in candidates + ) + + expected_scenarios = expected_output.get("scenarios") + if expected_scenarios is None: + scenario_count_correct = True + else: + asserted.append("scenarios") + actual = (create_args or {}).get("scenarios") + scenario_count_correct = isinstance(actual, list) and len(actual) == expected_scenarios + + expected_baseline = expected_output.get("include_baseline") + if expected_baseline is None: + baseline_correct = True + else: + asserted.append("include_baseline") + # The tool defaults include_baseline to true, so an absent argument means true -- + # `.get(..., True)` would be wrong only if the agent sent an explicit null, which + # the `is None` fallback below also treats as the default. + actual_baseline = (create_args or {}).get("include_baseline") + baseline_correct = (True if actual_baseline is None else bool(actual_baseline)) == bool(expected_baseline) + + return WhatIfEvaluation( + triggered=triggered, + executed=executed, + success=success, + turn_completed=turn_completed, + metric_correct=metric_correct, + maql_correct=maql_correct, + scenario_count_correct=scenario_count_correct, + baseline_correct=baseline_correct, + asserted=asserted, + disambiguated=disambiguated, + ) + + +def run_agentic_what_if( + host: str, + token: str, + workspace_id: str, + question: str, + expected_output: dict, + k: int = _DEFAULT_K, + max_iterations: int = _DEFAULT_MAX_ITERATIONS, + initial_conversation_id: str | None = None, + reasoning_effort: ReasoningEffort | None = None, + agent_id: str | None = None, +) -> AgenticWhatIfSummary: + """Run the what-if agentic evaluation K times and return a summary. + + A run ends when execute_what_if_scenario returns. Short of that it keeps sending + simulated replies up to ``max_iterations``, without trying to classify whether the + agent's text was a question: missing a genuine one hard-fails the run, while answering + a final answer costs one harmless extra turn. + """ + if k < 1: + raise ValueError(f"k must be >= 1, got {k}") + run_results: list[WhatIfRunResult] = [] + client = ChatClient( + host=host, token=token, workspace_id=workspace_id, reasoning_effort=reasoning_effort, agent_id=agent_id + ) + + def _run_once(conv_id: str) -> WhatIfRunResult: + create_args: dict | None = None + execute_result: dict | None = None + turn_wall_clock_sec: float | None = None + turn_completed = False + disambiguated = False + current_question = question + reasoning_steps: list[str] = [] + response_id: str | None = None + all_tool_call_events: list[ToolCallEvent] = [] + all_reasoning_step_events: list[ReasoningStepEvent] = [] + turn_offset = 0.0 # each turn's call_ts/ts restarts near 0 -- shift by prior turns' wall time + tool_index_offset = 0 + reasoning_index_offset = 0 + + def _accumulate(result: ChatResult) -> None: + nonlocal turn_offset, tool_index_offset, reasoning_index_offset + turn_offset, tool_index_offset, reasoning_index_offset = shift_and_index_events( + result, + turn_offset=turn_offset, + tool_index_offset=tool_index_offset, + reasoning_index_offset=reasoning_index_offset, + ) + all_tool_call_events.extend(result.tool_call_events or []) + all_reasoning_step_events.extend(result.reasoning_step_events or []) + + for iteration in range(max_iterations): + try: + chat_result = client.send_message(conv_id, current_question) + except Exception as exc: # noqa: BLE001 -- end this run, not the whole item + _log.warning("What-if send_message failed for conversation %s: %s", conv_id, exc) + partial = getattr(exc, "partial_result", None) + if partial is not None: + reasoning_steps.extend(partial.reasoning_steps or []) + response_id = partial.response_id or response_id + _accumulate(partial) + create_args, execute_result = _extract_what_if_calls(all_tool_call_events) + turn_completed = False + break + reasoning_steps.extend(chat_result.reasoning_steps or []) + response_id = chat_result.response_id or response_id + _accumulate(chat_result) + # Over every turn so far, not just this one: the agent may build the spec on + # one turn and execute it on the next, and reading a single turn would drop the + # scenario the execution actually ran. + create_args, execute_result = _extract_what_if_calls(all_tool_call_events) + response_text = render_answer_text(chat_result) + turn_completed = chat_result.stream_ended and bool(response_text) + if execute_result is not None: + # The turn that ran the scenario, not an earlier disambiguation turn. + turn_wall_clock_sec = chat_result.turn_wall_clock_sec + break + if not response_text: + break + if iteration >= max_iterations - 1: + break + try: + current_question = generate_simulated_what_if_response(response_text, expected_output) + disambiguated = True + except Exception as exc: # noqa: BLE001 -- harness-side fault; end only this run + _log.warning("Simulated what-if user reply failed for conversation %s: %s", conv_id, exc) + break + + return WhatIfRunResult( + conversation_id=conv_id, + evaluation=_evaluate_run(create_args, execute_result, expected_output, turn_completed, disambiguated), + actual_create_args=create_args, + actual_execute_result=execute_result, + turn_wall_clock_sec=turn_wall_clock_sec, + reasoning_steps=reasoning_steps, + response_id=response_id, + tool_call_events=all_tool_call_events, + reasoning_step_events=all_reasoning_step_events, + ) + + try: + conv_id_0 = initial_conversation_id if initial_conversation_id is not None else client.create_conversation() + try: + run_results.append(_run_once(conv_id_0)) + finally: + if initial_conversation_id is None: # only delete conversations we created + client.delete_conversation(conv_id_0) + + for _ in range(1, k): + conv_id = client.create_conversation() + try: + run_results.append(_run_once(conv_id)) + finally: + client.delete_conversation(conv_id) + finally: + client.close() + + pass_at_k = any(r.evaluation.strict_pass for r in run_results) + pass_power_k = all(r.evaluation.strict_pass for r in run_results) + best = max( + run_results, + key=lambda r: sum( + [ + r.evaluation.triggered, + r.evaluation.executed, + r.evaluation.success, + r.evaluation.turn_completed, + r.evaluation.metric_correct, + r.evaluation.maql_correct, + r.evaluation.scenario_count_correct, + r.evaluation.baseline_correct, + ] + ), + ) + return AgenticWhatIfSummary( + run_results=run_results, + pass_at_k=pass_at_k, + pass_power_k=pass_power_k, + best=best, + ) + + +class WhatIfAssertionError(AgenticAssertionError): + """Raised when a what-if evaluation fails.""" + + +def _detail(best: WhatIfRunResult) -> dict[str, Any]: + ev = best.evaluation + adjustments = _adjustments(best.actual_create_args) + return { + "triggered": ev.triggered, + "executed": ev.executed, + "success": ev.success, + "turn_completed": ev.turn_completed, + "metric_correct": ev.metric_correct, + "maql_correct": ev.maql_correct, + "scenario_count_correct": ev.scenario_count_correct, + "baseline_correct": ev.baseline_correct, + # Which content checks the fixture pinned -- without it a run that verified nothing + # reads the same as one where everything matched. + "asserted": ev.asserted, + "disambiguated": ev.disambiguated, + "actual_adjustments": adjustments, + "actual_scenario_labels": [ + s.get("label") for s in ((best.actual_create_args or {}).get("scenarios") or []) if isinstance(s, dict) + ], + "actual_execute_result": best.actual_execute_result, + "latency_breakdown": build_latency_breakdown(best.tool_call_events, best.reasoning_step_events), + } + + +def evaluate_agentic_what_if( + host: str, + token: str, + workspace_id: str, + question: str, + expected_output: dict, + k: int = _DEFAULT_K, + max_iterations: int = _DEFAULT_MAX_ITERATIONS, + initial_conversation_id: str | None = None, + agent_id: str | None = None, + langfuse: object | None = None, + dataset_item_id: str = "", + dataset_name: str = "what_if_analysis", + run_timestamp: str | None = None, + model_version_override: str | None = None, + run_metadata_extra: dict | None = None, + reasoning_effort: ReasoningEffort | None = None, + submit_trace_link: SubmitTraceLink = run_trace_link_inline, +) -> AgenticEvalOutcome: + """Run what-if evaluation, log to Langfuse, and raise WhatIfAssertionError on failure.""" + langfuse, window_start = open_trace_window(langfuse) + summary = run_agentic_what_if( + host=host, + token=token, + workspace_id=workspace_id, + question=question, + expected_output=expected_output, + k=k, + max_iterations=max_iterations, + initial_conversation_id=initial_conversation_id, + reasoning_effort=reasoning_effort, + agent_id=agent_id, + ) + + if langfuse is not None and dataset_item_id: + # Pinned on the calling thread: a deferred poll must not widen its query window. + window_end = utc_now() + + def _write_scores(ctx: RunTraceContext) -> None: + for run_idx, run in enumerate(summary.run_results): + pt = ctx.trace(run.conversation_id) + ev = run.evaluation + strict_checks = { + "what_if_triggered": ev.triggered, + "what_if_executed": ev.executed, + "what_if_success": ev.success, + "what_if_turn_completed": ev.turn_completed, + } + # Only the content checks the fixture actually pinned. An unasserted check + # is True internally so it cannot fail a run, but publishing that as a + # BOOLEAN 1 would claim the evaluator verified something it never looked at. + strict_checks.update( + { + key: value + for name, key, value in ( + ("metric_id", "what_if_metric_correct", ev.metric_correct), + ("scenario_maql", "what_if_maql_correct", ev.maql_correct), + ("scenarios", "what_if_scenario_count_correct", ev.scenario_count_correct), + ("include_baseline", "what_if_baseline_correct", ev.baseline_correct), + ) + if name in ev.asserted + } + ) + with ctx.observe(pt, run_idx) as tid: + for score_name, value in strict_checks.items(): + ctx.score(tid, name=score_name, value=float(value), data_type="BOOLEAN") + ctx.quality( + tid, + strict_checks=strict_checks, + # pt.latency covers the whole conversation, which is the item's real + # elapsed cost when the agent needed clarification turns to get + # there; turn_wall_clock_sec (the goal turn alone) is the fallback. + # This is what 7 of the 8 existing kinds do -- kda_skill is the + # outlier and documents its own reason. Cost is not gated on + # ev.triggered: a run that answered without ever reaching the tool + # still spent tokens, and hiding that understates what the item cost. + latency_sec=pt.latency if pt else run.turn_wall_clock_sec, + cost_usd=pt.total_cost if pt else None, + ) + + # Before the pass@K raise: a failing item's scores are the ones worth having. + submit_trace_scoring( + submit_trace_link, + RunIdentity( + host, + token, + workspace_id, + dataset_name, + run_timestamp, + model_version_override, + run_metadata_extra, + reasoning_effort, + ), + langfuse=langfuse, + dataset_item_id=dataset_item_id, + conversation_ids=[r.conversation_id for r in summary.run_results], + window_start=window_start, + window_end=window_end, + suffix_runs=len(summary.run_results) > 1, + write_scores=_write_scores, + # The question this run answered, so a score is readable without resolving the + # conversation back to its item. + item_input=question, + ) + + best = summary.best + ev = best.evaluation + detail = _detail(best) + runs_passed = sum(1 for r in summary.run_results if r.evaluation.strict_pass) + + if not summary.pass_at_k: + message = ( + f"What-if assertion failed. strict_pass={ev.strict_pass} " + f"(triggered={ev.triggered}, executed={ev.executed}, success={ev.success}, " + f"turn_completed={ev.turn_completed}, metric_correct={ev.metric_correct}, " + f"maql_correct={ev.maql_correct}, scenario_count_correct={ev.scenario_count_correct}, " + f"baseline_correct={ev.baseline_correct}). " + f"Actual adjustments: {detail['actual_adjustments']}. " + f"Actual execute result: {best.actual_execute_result}." + ) + exc = WhatIfAssertionError(message) + exc.reasoning_steps = best.reasoning_steps + exc.conversation_id = best.conversation_id + exc.response_id = best.response_id + exc.detail = detail + exc.runs_passed = runs_passed + exc.runs_effective = len(summary.run_results) + raise exc + + return AgenticEvalOutcome( + runs_passed=runs_passed, + runs_effective=len(summary.run_results), + reasoning_steps=best.reasoning_steps, + conversation_id=best.conversation_id, + response_id=best.response_id, + detail=detail, + ) diff --git a/packages/gooddata-eval/tests/test_agentic_runner.py b/packages/gooddata-eval/tests/test_agentic_runner.py index 77086d101..24a264c90 100644 --- a/packages/gooddata-eval/tests/test_agentic_runner.py +++ b/packages/gooddata-eval/tests/test_agentic_runner.py @@ -85,6 +85,7 @@ def test_dispatch_agentic_omits_agent_id_by_default(): ("agentic_guardrail", "Ignore prior instructions", "evaluate_agentic_guardrail"), ("agentic_kda_skill", {"Measure": {"type": "metric", "id": "revenue"}}, "evaluate_agentic_kda_skill"), ("agentic_conversation", {"fixture": _MIN_CONVERSATION_FIXTURE}, "evaluate_agentic_conversation"), + ("agentic_what_if", {"metric_id": "spend"}, "evaluate_agentic_what_if"), ] diff --git a/packages/gooddata-eval/tests/test_agentic_what_if.py b/packages/gooddata-eval/tests/test_agentic_what_if.py new file mode 100644 index 000000000..07da79ca8 --- /dev/null +++ b/packages/gooddata-eval/tests/test_agentic_what_if.py @@ -0,0 +1,439 @@ +# (C) 2026 GoodData Corporation. All rights reserved. +# SPDX-License-Identifier: LicenseRef-GoodData-Enterprise +import json +from contextlib import contextmanager +from unittest.mock import MagicMock, patch + +import pytest +from gooddata_eval.core.agentic.what_if import ( + WhatIfAssertionError, + _adjustments, + _evaluate_run, + _extract_what_if_calls, + evaluate_agentic_what_if, + run_agentic_what_if, +) +from gooddata_eval.core.models import ChatResult + +_MODULE = "gooddata_eval.core.agentic.what_if" + +_BASE_MAQL = "SELECT SUM({fact/price} * 1.10 * {fact/quantity})" +_EXPECTED = {"metric_id": "revenue", "scenario_maql": _BASE_MAQL} + + +def _create_args(*, metric="revenue", maql=_BASE_MAQL, scenarios=1, baseline=True, label="Scenario A") -> dict: + return { + "visualization_ref": "viz_1", + "include_baseline": baseline, + "scenarios": [ + { + "label": f"{label} {i}" if scenarios > 1 else label, + "adjustments": [{"metric_id": metric, "metric_type": "metric", "scenario_maql": maql}], + } + for i in range(scenarios) + ], + } + + +_OK_EXECUTE = json.dumps( + { + "success": True, + "scenario_results": [ + {"label": "Baseline", "data": {"rows": [[100]]}, "error": None}, + {"label": "Scenario A", "data": {"rows": [[110]]}, "error": None}, + ], + "error": None, + } +) + + +def _tc(name: str, args: dict | None = None, result: str | None = None): + tc = MagicMock() + tc.function_name = name + tc.call_ts = None + tc.result_ts = None + tc.index = None + tc.parsed_arguments = lambda a=args: a or {} + tc.result = result + tc.parsed_result = lambda r=result: json.loads(r) if r else None + return tc + + +def _chat(tool_calls, text="Here is the scenario.", ended=True) -> ChatResult: + """A real ChatResult, so render_answer_text sees every field it reads.""" + result = ChatResult.model_validate({"textResponse": text, "responseId": "resp-1"}) + result.stream_ended = ended + result.turn_wall_clock_sec = 5.1 + result.tool_call_events = tool_calls + return result + + +def _pair(**kw): + return [ + _tc("create_what_if_scenario", _create_args(**kw)), + _tc("execute_what_if_scenario", {"scenario_ref": "wia_1"}, _OK_EXECUTE), + ] + + +# ── extraction ────────────────────────────────────────────────────────────── + + +def test_extract_pairs_the_execute_with_the_spec_it_followed(): + """A new spec clears the previous execution: scoring a fresh scenario against a stale + result would credit work the agent redid.""" + calls = [ + _tc("create_what_if_scenario", _create_args(metric="orders")), + _tc("execute_what_if_scenario", {}, _OK_EXECUTE), + _tc("create_what_if_scenario", _create_args(metric="revenue")), + ] + create, result = _extract_what_if_calls(calls) + + assert create["scenarios"][0]["adjustments"][0]["metric_id"] == "revenue" + assert result is None + + +def test_adjustments_flatten_across_scenarios(): + """Scenario grouping is not asserted -- a fixture cares that the right measure was + adjusted the right way, not which label it landed under.""" + args = _create_args(scenarios=3) + assert len(_adjustments(args)) == 3 + + +def test_adjustments_tolerate_a_malformed_scenario_list(): + assert _adjustments({"scenarios": "not a list"}) == [] + assert _adjustments({"scenarios": [None, {"adjustments": None}]}) == [] + assert _adjustments(None) == [] + + +# ── scoring ───────────────────────────────────────────────────────────────── + + +def _evaluate(create=None, result=None, expected=None, turn_completed=True): + return _evaluate_run(create, result, expected if expected is not None else _EXPECTED, turn_completed) + + +def test_a_correct_scenario_passes_every_check(): + ev = _evaluate(_create_args(), {"success": True}) + assert ev.strict_pass is True + assert ev.asserted == ["metric_id", "scenario_maql"] + + +def test_adjusting_the_wrong_measure_fails_both_metric_and_maql(): + """The MAQL check is scoped to adjustments on the expected measure, so adjusting the + wrong one leaves it nothing valid to match. Both checks failing is the honest reading: + the expected measure was not adjusted at all, correctly or otherwise.""" + ev = _evaluate(_create_args(metric="orders"), {"success": True}) + assert ev.metric_correct is False + assert ev.maql_correct is False + + +def test_the_wrong_adjustment_fails_on_maql(): + ev = _evaluate(_create_args(maql="SELECT SUM({fact/price} * 2.0 * {fact/quantity})"), {"success": True}) + assert ev.maql_correct is False + assert ev.metric_correct is True + + +def test_maql_is_compared_normalized_not_literally(): + """metric_skill's normalizer decides this, so whitespace and casing do not.""" + noisy = "select SUM( {fact/price} * 1.10 * {fact/quantity} )" + assert _evaluate(_create_args(maql=noisy), {"success": True}).maql_correct is True + + +def test_several_expressions_can_be_equally_correct(): + """`* 1.1` and `* 1.10` are the same uplift, so a fixture may accept either.""" + expected = { + "scenario_maql": [ + "SELECT SUM({fact/price} * 1.1 * {fact/quantity})", + "SELECT SUM({fact/price} * 1.10 * {fact/quantity})", + ] + } + assert _evaluate(_create_args(), {"success": True}, expected=expected).maql_correct is True + + +def test_scenario_count_is_checked_when_pinned(): + expected = {**_EXPECTED, "scenarios": 2} + assert _evaluate(_create_args(scenarios=2), {"success": True}, expected=expected).scenario_count_correct is True + assert _evaluate(_create_args(scenarios=1), {"success": True}, expected=expected).scenario_count_correct is False + + +def test_an_absent_include_baseline_counts_as_the_tools_default(): + """The tool defaults include_baseline to true, so an agent that omits it has still + asked for a baseline -- failing that would penalise correct behaviour.""" + expected = {"include_baseline": True} + args = _create_args() + del args["include_baseline"] + assert _evaluate(args, {"success": True}, expected=expected).baseline_correct is True + + +def test_explicitly_dropping_the_baseline_is_caught_when_pinned(): + expected = {"include_baseline": True} + assert _evaluate(_create_args(baseline=False), {"success": True}, expected=expected).baseline_correct is False + + +def test_an_unstated_expectation_neither_fails_nor_silently_passes(): + ev = _evaluate(_create_args(metric="anything", maql="SELECT 1"), {"success": True}, expected={}) + assert ev.strict_pass is True + assert ev.asserted == [] + + +def test_a_failed_execution_is_not_a_success(): + ev = _evaluate(_create_args(), {"success": False, "error": "boom"}) + assert ev.executed is True + assert ev.success is False + assert ev.strict_pass is False + + +def test_a_spec_that_never_executed_is_triggered_but_not_executed(): + ev = _evaluate(_create_args(), None) + assert ev.triggered is True + assert ev.executed is False + + +# ── run loop ──────────────────────────────────────────────────────────────── + + +def _run(side_effect, *, k=1, expected=None, max_iterations=4): + client = MagicMock() + client.create_conversation.side_effect = [f"conv-{i}" for i in range(1, k + 2)] + client.send_message.side_effect = side_effect + with ( + patch(f"{_MODULE}.ChatClient", return_value=client), + patch(f"{_MODULE}.generate_simulated_what_if_response", return_value="use revenue, +10%"), + ): + return run_agentic_what_if( + host="http://h", + token="tok", + workspace_id="ws1", + question="What if revenue rose 10%?", + expected_output=expected if expected is not None else _EXPECTED, + k=k, + max_iterations=max_iterations, + ) + + +def test_a_single_turn_scenario_passes(): + summary = _run([_chat(_pair())]) + assert summary.pass_at_k is True + assert summary.best.evaluation.disambiguated is False + + +def test_a_clarifying_question_is_answered_and_the_run_continues(): + """Observed live: 'I need to confirm which Spend calculation you want to adjust'.""" + summary = _run([_chat([], text="Which Spend metric did you mean?"), _chat(_pair())]) + assert summary.pass_at_k is True + assert summary.best.evaluation.disambiguated is True + + +def test_the_loop_stops_at_max_iterations_without_a_scenario(): + summary = _run([_chat([], text="Still thinking")] * 4, max_iterations=4) + assert summary.pass_at_k is False + assert summary.best.evaluation.triggered is False + + +def test_an_empty_response_ends_the_run_immediately(): + client = MagicMock() + client.create_conversation.return_value = "conv-1" + client.send_message.return_value = _chat([], text="") + with ( + patch(f"{_MODULE}.ChatClient", return_value=client), + patch(f"{_MODULE}.generate_simulated_what_if_response") as sim, + ): + run_agentic_what_if(host="http://h", token="tok", workspace_id="ws1", question="q", expected_output=_EXPECTED) + assert client.send_message.call_count == 1 + sim.assert_not_called() + + +def test_a_chat_error_on_a_later_run_does_not_discard_the_earlier_one(): + summary = _run([_chat(_pair()), RuntimeError("boom")], k=2) + assert len(summary.run_results) == 2 + assert summary.pass_at_k is True + + +def test_k_must_be_at_least_one(): + with pytest.raises(ValueError, match="k must be >= 1"): + run_agentic_what_if(host="h", token="t", workspace_id="w", question="q", expected_output={}, k=0) + + +# ── evaluate_* wrapper ────────────────────────────────────────────────────── + + +def _evaluate_item(calls): + client = MagicMock() + client.create_conversation.return_value = "conv-1" + client.send_message.return_value = _chat(calls) + with patch(f"{_MODULE}.ChatClient", return_value=client): + return evaluate_agentic_what_if( + host="http://h", + token="tok", + workspace_id="ws1", + question="What if revenue rose 10%?", + expected_output=_EXPECTED, + ) + + +def test_detail_reports_the_adjustments_the_agent_asked_for(): + outcome = _evaluate_item(_pair()) + assert outcome.detail["actual_adjustments"][0]["metric_id"] == "revenue" + assert outcome.detail["actual_scenario_labels"] == ["Scenario A"] + assert outcome.detail["asserted"] == ["metric_id", "scenario_maql"] + assert "latency_breakdown" in outcome.detail + assert outcome.runs_passed == 1 + + +def test_a_wrong_adjustment_raises_naming_what_the_agent_actually_did(): + with pytest.raises(WhatIfAssertionError) as exc_info: + _evaluate_item(_pair(maql="SELECT SUM({fact/price} * 3 * {fact/quantity})")) + + error = exc_info.value + assert "maql_correct=False" in str(error) + assert "* 3 *" in str(error) + assert error.runs_passed == 0 + assert error.conversation_id == "conv-1" + + +def test_a_spec_built_on_an_earlier_turn_is_still_the_one_scored(): + """The agent may build the spec on one turn and execute it on the next -- reading only + the current turn's calls would drop the scenario the execution actually ran and fail a + correct run for having no adjustments.""" + summary = _run( + [ + _chat([_tc("create_what_if_scenario", _create_args())], text="Preparing the scenario."), + _chat([_tc("execute_what_if_scenario", {"scenario_ref": "wia_1"}, _OK_EXECUTE)]), + ] + ) + + assert summary.best.evaluation.executed is True + assert summary.best.evaluation.metric_correct is True + assert summary.best.evaluation.maql_correct is True + assert summary.pass_at_k is True + + +def test_the_maql_must_match_on_the_adjustment_that_matched_the_metric(): + """Checking the two independently lets two failures score as a pass: a wrong adjustment + on the right measure and a right adjustment on the wrong measure would satisfy one + check each.""" + create = { + "visualization_ref": "viz_1", + "include_baseline": True, + "scenarios": [ + { + "label": "Scenario A", + "adjustments": [ + # Right metric, wrong adjustment. + {"metric_id": "revenue", "metric_type": "metric", "scenario_maql": "SELECT 0"}, + # Right adjustment, wrong metric. + {"metric_id": "orders", "metric_type": "metric", "scenario_maql": _BASE_MAQL}, + ], + } + ], + } + ev = _evaluate(create, {"success": True}) + + assert ev.metric_correct is True # revenue was adjusted + assert ev.maql_correct is False # but not with the expected expression + assert ev.strict_pass is False + + +def test_the_maql_still_matches_when_the_right_adjustment_is_on_the_right_metric(): + """The pairing must not reject a correct spec that also adjusts something else.""" + create = { + "visualization_ref": "viz_1", + "include_baseline": True, + "scenarios": [ + { + "label": "Scenario A", + "adjustments": [ + {"metric_id": "orders", "metric_type": "metric", "scenario_maql": "SELECT 0"}, + {"metric_id": "revenue", "metric_type": "metric", "scenario_maql": _BASE_MAQL}, + ], + } + ], + } + assert _evaluate(create, {"success": True}).strict_pass is True + + +# ── Langfuse scoring ──────────────────────────────────────────────────────── + + +class _FakeCtx: + """Records what the deferred Langfuse block writes, without a Langfuse.""" + + def __init__(self): + self.scores: dict[str, float] = {} + # Not named `quality`: the method below would overwrite itself on first call. + self.quality_call: dict = {} + + def trace(self, _conversation_id): + return None + + @contextmanager + def observe(self, _trace, _run_idx): + yield "trace-id" + + def score(self, _tid, *, name, value, data_type): + self.scores[name] = value + + def quality(self, _tid, *, strict_checks, latency_sec, cost_usd): + self.quality_call = {"strict_checks": strict_checks, "latency_sec": latency_sec, "cost_usd": cost_usd} + + +def _scored(expected_output, calls=None): + """Run one item with Langfuse on, then execute the deferred block against a fake ctx.""" + client = MagicMock() + client.create_conversation.return_value = "conv-1" + client.send_message.return_value = _chat(calls if calls is not None else _pair()) + captured = {} + + def _capture(_link, _identity, **kwargs): + captured.update(kwargs) + + with ( + patch(f"{_MODULE}.ChatClient", return_value=client), + patch(f"{_MODULE}.submit_trace_scoring", side_effect=_capture), + ): + try: + evaluate_agentic_what_if( + host="http://h", + token="tok", + workspace_id="ws1", + question="What if revenue rose 10%?", + expected_output=expected_output, + langfuse=MagicMock(), + dataset_item_id="ds-1", + ) + except WhatIfAssertionError: + pass # scores are written before the pass@K raise, which is the point + + ctx = _FakeCtx() + captured["write_scores"](ctx) + return ctx + + +def test_only_the_checks_the_fixture_pinned_are_scored(): + """An unasserted check is True internally so it cannot fail a run. Publishing that as a + BOOLEAN 1 would claim the evaluator verified something it never looked at.""" + ctx = _scored({"metric_id": "revenue"}) # maql, scenario count, baseline unpinned + + assert ctx.scores["what_if_metric_correct"] == 1.0 + for absent in ("what_if_maql_correct", "what_if_scenario_count_correct", "what_if_baseline_correct"): + assert absent not in ctx.scores + # The process checks are unconditional -- they are always actually evaluated. + assert set(ctx.scores) >= {"what_if_triggered", "what_if_executed", "what_if_success"} + + +def test_every_pinned_check_is_scored(): + ctx = _scored({**_EXPECTED, "scenarios": 1, "include_baseline": True}) + for name in ( + "what_if_metric_correct", + "what_if_maql_correct", + "what_if_scenario_count_correct", + "what_if_baseline_correct", + ): + assert ctx.scores[name] == 1.0 + + +def test_cost_is_reported_even_when_the_tool_was_never_reached(): + """A run that answered without building a scenario still spent tokens; gating cost on + ev.triggered hid that and understated what the item cost.""" + ctx = _scored(_EXPECTED, calls=[]) + assert "cost_usd" in ctx.quality_call diff --git a/packages/gooddata-eval/tests/test_trace_linker.py b/packages/gooddata-eval/tests/test_trace_linker.py index f490effd6..089f39cc2 100644 --- a/packages/gooddata-eval/tests/test_trace_linker.py +++ b/packages/gooddata-eval/tests/test_trace_linker.py @@ -118,6 +118,7 @@ def test_run_trace_link_inline_runs_the_task_on_the_calling_thread(): ("visualization", "evaluate_agentic_visualization"), ("kda_skill", "evaluate_agentic_kda_skill"), ("conversation", "evaluate_agentic_conversation"), + ("what_if", "evaluate_agentic_what_if"), ]