diff --git a/.github/workflows/agent-tests.yml b/.github/workflows/agent-tests.yml new file mode 100644 index 0000000..cfadf10 --- /dev/null +++ b/.github/workflows/agent-tests.yml @@ -0,0 +1,49 @@ +name: agent-tests + +on: + pull_request: + paths: ["agent/**", "core/**", ".github/workflows/agent-tests.yml"] + push: + branches: [main, feature/deep-agent-updated] + paths: ["agent/**", "core/**", ".github/workflows/agent-tests.yml"] + +jobs: + unit: + runs-on: ubuntu-latest + timeout-minutes: 20 + defaults: + run: + working-directory: agent + env: + # Settings that AgentSettings requires; tests never reach these endpoints. + LANGFUSE_PUBLIC_KEY: pk-lf-ci + LANGFUSE_SECRET_KEY: sk-lf-ci + LANGFUSE_BASE_URL: http://127.0.0.1:9 + DEEP_AGENT_LANGFUSE_PROMPTS: "false" + steps: + - uses: actions/checkout@v4 + + - uses: astral-sh/setup-uv@v6 + with: + python-version: "3.12" + enable-cache: true + cache-dependency-glob: agent/uv.lock + + # core depends on the private matzpen-agency/python-core-utils over git+ssh. + # A fine-grained token with read access to that repo is used over HTTPS instead. + - name: Access private python-core-utils + env: + TOKEN: ${{ secrets.PYTHON_CORE_UTILS_TOKEN }} + run: git config --global url."https://x-access-token:${TOKEN}@github.com/".insteadOf "ssh://git@github.com/" + + - name: Install + run: uv sync --frozen + + - name: Lint (deep agent) + run: uv run ruff check --extend-select F src/deep_agent + + - name: Unit tests + run: > + uv run pytest -q + --ignore=tests/refiner/test_refiner_e2e_real.py + --cov=deep_agent --cov-report=term --cov-fail-under=80 diff --git a/agent/README.md b/agent/README.md index 7cf2bcf..4e2075f 100644 --- a/agent/README.md +++ b/agent/README.md @@ -89,3 +89,24 @@ JARVIS_API_KEY=... # sent as "Authorization: Ap The registry service identity needs the `datetime:read`, `text:read`, `geo:read` and `trino:read` scopes, and each tool must be approved in the registry. SQL is never executed through Utils-Server. If the gateway is unreachable the agent starts without these tools and logs a warning. + +### Evaluation (migration go/no-go) + +```bash +uv run deep-agent-eval # deep agent + legacy graph on benchmarks/eval_dataset.json +uv run deep-agent-eval --systems deep --out runs/eval.json +uv run deep-agent-eval --gate --p95-slo 60 # exit 1 unless deep accuracy >= legacy and p95 <= SLO +uv run deep-agent-eval --langfuse-dataset text2sql-eval # Langfuse Experiment: linked dataset run + scores +``` + +Metric: **execution accuracy** — the system's result set vs the gold SQL's (order-insensitive unless the gold query has +a top-level `ORDER BY`; floats to 6 decimals; permuted columns match). Items without `gold_sql` are reported as +`no_gold`. `gold_source: agent_run_unverified` marks gold SQL taken from past agent runs that still needs expert review. +For a Langfuse dataset, items use `input = {"question": ...}` and `expected_output = {"sql": ...}`. + +### CI + +`.github/workflows/agent-tests.yml` runs ruff (deep agent) and the unit tests with an 80% coverage gate on `deep_agent` +for every PR touching `agent/` or `core/`. It needs a repository secret **`PYTHON_CORE_UTILS_TOKEN`**: a fine-grained +token with read access to `matzpen-agency/python-core-utils` (used instead of the `git+ssh` URL). Real-service e2e +tests (`tests/refiner/test_refiner_e2e_real.py`) are excluded. diff --git a/agent/benchmarks/eval_dataset.json b/agent/benchmarks/eval_dataset.json new file mode 100644 index 0000000..0dda28e --- /dev/null +++ b/agent/benchmarks/eval_dataset.json @@ -0,0 +1,117 @@ +{ + "description": "Deep Agent vs legacy graph evaluation set. Metric: execution accuracy against gold_sql result sets. Items with gold_sql=null are reported as no_gold and not scored. gold_source=agent_run_unverified marks gold SQL taken from consistent past agent runs that still needs domain-expert verification.", + "items": [ + { + "id": "q01", + "question": "כמה טיסות המריאו מצרפת", + "tags": [ + "simple_filter", + "categorical" + ], + "gold_sql": "SELECT COUNT(DISTINCT flight_id) AS flight_count FROM \"postgres\".\"public\".\"flights_table\" WHERE launch_country = 'France'", + "gold_source": "agent_run_unverified" + }, + { + "id": "q02", + "question": "what cities are in the Haifa district", + "tags": [ + "large_categorical", + "join" + ], + "gold_sql": null, + "gold_source": null + }, + { + "id": "q03", + "question": "חשב את מספר התראות הטיסה לנפש עבור כל מחוז. לאחר מכן, עבור 3 המחוזות עם כמות ההתראות הגבוהה ביותר, ציין את 3 הערים הגדולות ביותר בכל מחוז על בסיס שטח הפנים שלהן.", + "tags": [ + "per_capita", + "join", + "aggregation", + "top_n" + ], + "gold_sql": null, + "gold_source": null + }, + { + "id": "q04", + "question": "כמה טיסות המריאו מצרפת ונחתו בתל אביב", + "tags": [ + "simple_filter", + "categorical", + "large_categorical" + ], + "gold_sql": "SELECT COUNT(DISTINCT flight_id) AS flight_count FROM \"postgres\".\"public\".\"flights_table\" WHERE launch_country = 'France' AND city_name LIKE 'Tel Aviv%'", + "gold_source": "agent_run_unverified" + }, + { + "id": "q05", + "question": "מצא את אזורי ההנחיה שבהם מדיניות הטיסה אוסרת טיסה בגובה נמוך או סוגרת את המרחב האווירי לחלוטין, ובמקביל מדיניות הנחיתה מחייבת העברה לשדה חלופי. עבור כל אזור הצג את מדיניות הטיסה, את מדיניות הנחיתה ואת מועד העדכון האחרון. מיין מהעדכון החדש ביותר לישן ביותר.", + "tags": [ + "categorical", + "temporal_latest" + ], + "gold_sql": null, + "gold_source": null + }, + { + "id": "q06", + "question": "השווה בין ערים לפי מספר ההתרעות שהתקבלו בהן במהלך יוני 2026 ביחס לגודל האוכלוסייה שלהן. הצג רק ערים שיש להן לפחות 10 התרעות בתקופה, והצג את חמש הערים עם מספר ההתרעות הגבוה ביותר לכל 10,000 תושבים. הצג גם את מספר ההתרעות ואת גודל האוכלוסייה ששימשו לחישוב.", + "tags": [ + "temporal", + "per_capita", + "join", + "aggregation", + "top_n" + ], + "gold_sql": null, + "gold_source": null + }, + { + "id": "q07", + "question": "מצא את הערים שבהן נרשמו לפחות 5 התרעות במהלך השבועיים הראשונים של יולי 2026, אך לא נרשמה בהן אף נחיתת חירום באותה תקופה. הצג את מספר ההתרעות בכל עיר, ומיין מהמספר הגבוה לנמוך.", + "tags": [ + "temporal", + "anti_join", + "aggregation" + ], + "gold_sql": null, + "gold_source": null + }, + { + "id": "q08", + "question": "עבור טיסות שהמריאו במהלך יוני 2026, מצא את אזורי ההנחיה שבהם שיעור הטיסות ארוכות הטווח גבוה מהממוצע של כלל הטיסות בתקופה. הצג את מספר הטיסות בכל אזור, מספר הטיסות ארוכות הטווח, השיעור שלהן והפער בנקודות אחוז מהממוצע הכללי. מיין לפי הפער מהגבוה לנמוך.", + "tags": [ + "temporal", + "join", + "aggregation", + "comparison_to_average" + ], + "gold_sql": null, + "gold_source": null + }, + { + "id": "q09", + "question": "זהה את עשר הרשויות שבהן היחס בין מספר האנשים השוהים במתקני האירוח לבין מספר העובדים הוא הגבוה ביותר, בתנאי שיש בהן לפחות 500 אנשים. הצג גם את מספר החדרים, את מספר האנשים לכל חדר ואת מועד העדכון של נתוני האירוח. אל תכלול רשויות שבהן מספר העובדים הוא אפס.", + "tags": [ + "ratio", + "join", + "aggregation", + "top_n" + ], + "gold_sql": null, + "gold_source": null + }, + { + "id": "q10", + "question": "עבור כל אזור הנחיה, הצג את המדיניות העדכנית ביותר הידועה, את תאריך העדכון שלה ואת מספר הטיסות ששויכו לאזור במהלך יולי 2026. הצג גם אזורים שאין להם טיסות בתקופה, וודא שאזור שמופיעות עבורו כמה גרסאות של מדיניות נספר פעם אחת בלבד.", + "tags": [ + "temporal_latest", + "left_join", + "deduplication" + ], + "gold_sql": null, + "gold_source": null + } + ] +} diff --git a/agent/pyproject.toml b/agent/pyproject.toml index fd83470..18591ea 100644 --- a/agent/pyproject.toml +++ b/agent/pyproject.toml @@ -31,6 +31,7 @@ dependencies = [ [project.scripts] deep-agent = "deep_agent.cli:main" deep-agent-bench = "deep_agent.benchmark:main" +deep-agent-eval = "deep_agent.evaluation:main" [tool.uv.sources] core = { path = "../core", editable = true } diff --git a/agent/src/deep_agent/benchmark.py b/agent/src/deep_agent/benchmark.py index ec874ba..922244d 100644 --- a/agent/src/deep_agent/benchmark.py +++ b/agent/src/deep_agent/benchmark.py @@ -61,6 +61,13 @@ def _token_usage(messages: list) -> tuple[int, int]: async def run_question(agent, item: dict[str, Any], context: DeepAgentContext, timeout_s: float) -> QuestionRun: + return (await run_question_detailed(agent, item, context, timeout_s))[0] + + +async def run_question_detailed( + agent, item: dict[str, Any], context: DeepAgentContext, timeout_s: float +) -> tuple[QuestionRun, dict[str, Any], str | None]: + """Run one question; returns (metrics, final_result with rows, Langfuse trace id).""" thread_id = f"bench_{item['id']}_{uuid.uuid4().hex[:6]}" config = observability.build_run_config(thread_id, connection_id=context.connection_id) start = time.perf_counter() @@ -79,14 +86,16 @@ async def run_question(agent, item: dict[str, Any], context: DeepAgentContext, t messages = state.get("messages") or [] metrics = observability.run_metrics(messages, latency, final_result) - observability.score_run(observability.langfuse_trace_id(config), metrics) + trace_id = observability.langfuse_trace_id(config) + observability.score_run(trace_id, metrics) input_tokens, output_tokens = _token_usage(messages) - return QuestionRun( + run = QuestionRun( id=item["id"], question=item["question"], latency_s=metrics.latency_s, model_calls=metrics.model_calls, tool_calls=metrics.tool_calls, sql_executions=metrics.sql_executions, input_tokens=input_tokens, output_tokens=output_tokens, status=final_result.get("status", "no_answer"), verified=bool(final_result.get("verified")), error=error, ) + return run, final_result, trace_id async def run_benchmark( diff --git a/agent/src/deep_agent/evaluation.py b/agent/src/deep_agent/evaluation.py new file mode 100644 index 0000000..8a45ace --- /dev/null +++ b/agent/src/deep_agent/evaluation.py @@ -0,0 +1,419 @@ +""" +Offline evaluation: the migration go/no-go gate (legacy LangGraph workflow vs Deep Agent). + + uv run deep-agent-eval # benchmarks/eval_dataset.json, both systems + uv run deep-agent-eval --systems deep --out runs/eval.json + uv run deep-agent-eval --gate # exit 1 if the deep agent fails the gate + uv run deep-agent-eval --langfuse-dataset text2sql-eval # Langfuse Experiment (linked dataset run) + +Metric: **execution accuracy** — the system's result set equals the result of the gold SQL +(row order ignored unless the gold SQL has a top-level ORDER BY; floats compared to 6 decimals; +a column permutation counts as a match). Also reported: p50/p95 latency, avg tokens, avg tool +calls, and accuracy per tag. Items without gold SQL are reported as `no_gold`, never scored. +""" + +import argparse +import asyncio +import datetime as dt +import json +import logging +import math +import time +from collections import Counter +from dataclasses import asdict, dataclass, field +from decimal import Decimal +from pathlib import Path +from typing import Any, Protocol + +import sqlglot +from sqlglot import exp + +from agent.config import settings +from agent.utils import sql_guard +from core.trino import execute_query_sync +from deep_agent.benchmark import percentile, run_question_detailed +from deep_agent.context import DeepAgentContext + +logger = logging.getLogger(__name__) + +DEFAULT_DATASET = Path(__file__).resolve().parents[2] / "benchmarks" / "eval_dataset.json" +EVAL_MAX_ROWS = 1000 # gold and predicted result sets are compared up to this many rows + + +# ── dataset ────────────────────────────────────────────────────────────────── + +@dataclass(frozen=True) +class EvalItem: + id: str + question: str + gold_sql: str | None = None + tags: tuple[str, ...] = () + + +def load_dataset(path: Path) -> list[EvalItem]: + """Load `{"items": [...]}` (or a bare list) of {id, question, gold_sql?, tags?}.""" + raw = json.loads(path.read_text(encoding="utf-8")) + rows = raw["items"] if isinstance(raw, dict) else raw + items = [EvalItem(id=r["id"], question=r["question"], gold_sql=r.get("gold_sql"), tags=tuple(r.get("tags", []))) + for r in rows] + if len({i.id for i in items}) != len(items): + raise ValueError(f"duplicate item ids in {path}") + return items + + +# ── result-set comparison ──────────────────────────────────────────────────── + +@dataclass(frozen=True) +class ResultSet: + columns: list[str] + rows: list[list[Any]] + truncated: bool = False + + +def _norm(value: Any) -> Any: + if value is None: + return None + if isinstance(value, bool): + return value + if isinstance(value, (int, float, Decimal)): + f = float(value) + if math.isnan(f): + return "NaN" + return round(f, 6) + if isinstance(value, (dt.date, dt.datetime)): + return value.isoformat() + text = str(value).strip() + try: # numbers that came back as strings (e.g. DECIMAL serialised by JSON) + return round(float(text), 6) + except ValueError: + return text + + +def is_ordered(sql: str) -> bool: + """True if the query's top-level statement has an ORDER BY (row order is then part of the answer).""" + try: + tree = sqlglot.parse_one(sql, read="trino") + except sqlglot.errors.SqlglotError: + return False + return bool(tree.args.get("order")) if isinstance(tree, exp.Query) else False + + +def compare_results(predicted: ResultSet, gold: ResultSet, ordered: bool) -> tuple[bool, str]: + pred_rows = [tuple(_norm(v) for v in r) for r in predicted.rows] + gold_rows = [tuple(_norm(v) for v in r) for r in gold.rows] + if len(pred_rows) != len(gold_rows): + return False, f"row count {len(pred_rows)} != gold {len(gold_rows)}" + if pred_rows and gold_rows and len(pred_rows[0]) != len(gold_rows[0]): + return False, f"column count {len(pred_rows[0])} != gold {len(gold_rows[0])}" + if ordered and pred_rows == gold_rows: + return True, "match" + if not ordered and Counter(pred_rows) == Counter(gold_rows): + return True, "match" + # Same values with columns in a different order still answer the question. + key = (lambda r: tuple(sorted(r, key=repr))) + permuted_pred = [key(r) for r in pred_rows] + permuted_gold = [key(r) for r in gold_rows] + if (ordered and permuted_pred == permuted_gold) or (not ordered and Counter(permuted_pred) == Counter(permuted_gold)): + return True, "match (columns permuted)" + return False, "different values" + (" (order matters)" if ordered else "") + + +def execute_for_eval(sql: str) -> tuple[ResultSet | None, str | None]: + try: + guarded = sql_guard.prepare(sql, EVAL_MAX_ROWS) + except sql_guard.GuardError as e: + return None, f"rejected: {e}" + res = execute_query_sync(guarded.sql, reuse_connection=True) + if not res.success: + return None, res.error_message + rows = res.rows or [] + return ResultSet(res.columns, rows[:EVAL_MAX_ROWS], len(rows) > EVAL_MAX_ROWS), None + + +# ── systems under test ─────────────────────────────────────────────────────── + +@dataclass +class Prediction: + sql: str | None + result: ResultSet | None + latency_s: float + model_calls: int = 0 + tool_calls: int = 0 + input_tokens: int = 0 + output_tokens: int = 0 + status: str = "" + error: str | None = None + trace_id: str | None = None + + +class Runner(Protocol): + name: str + + async def run(self, item: EvalItem) -> Prediction: ... + + +class DeepAgentRunner: + name = "deep" + + def __init__(self, agent, context: DeepAgentContext, timeout_s: float | None = None) -> None: + self.agent, self.context = agent, context + self.timeout_s = timeout_s or settings.DEEP_AGENT_RUN_TIMEOUT_S + + async def run(self, item: EvalItem) -> Prediction: + run, final_result, trace_id = await run_question_detailed( + self.agent, {"id": item.id, "question": item.question}, self.context, self.timeout_s + ) + result = final_result.get("result") + return Prediction( + sql=final_result.get("sql"), + result=ResultSet(result["columns"], result["rows"], result.get("truncated", False)) if result else None, + latency_s=run.latency_s, model_calls=run.model_calls, tool_calls=run.tool_calls, + input_tokens=run.input_tokens, output_tokens=run.output_tokens, status=run.status, error=run.error, + trace_id=trace_id, + ) + + +class LegacyGraphRunner: + """The pre-migration LangGraph workflow, invoked like /chat does (non-interactive).""" + + name = "legacy" + + def __init__(self, graph=None) -> None: + if graph is None: + from agent.graph import agent_graph as graph + self.graph = graph + + async def run(self, item: EvalItem) -> Prediction: + config = {"configurable": {"thread_id": f"eval_legacy_{item.id}_{time.time_ns()}"}} + start = time.perf_counter() + try: + state = await self.graph.ainvoke({ + "user_query": item.question, "allowed_tables": None, "allowed_statuses": None, + "active_extractors": [], "non_interactive": True, + }, config=config) + except Exception as exc: + return Prediction(None, None, time.perf_counter() - start, status="error", error=repr(exc)) + latency = time.perf_counter() - start + sql = state.get("sql_query") or None + result = None + if state.get("inline_result_columns") is not None and state.get("inline_result_rows") is not None: + result = ResultSet(state["inline_result_columns"], state["inline_result_rows"]) + elif sql: # the legacy graph may only return SQL; execute it for a like-for-like comparison + result, _ = execute_for_eval(sql) + messages = state.get("messages") or [] + return Prediction(sql, result, latency, model_calls=sum(1 for m in messages if m.type == "ai"), + status="answered" if sql else "no_sql") + + +# ── scoring ────────────────────────────────────────────────────────────────── + +@dataclass +class ItemScore: + id: str + system: str + status: str # correct | incorrect | no_gold | gold_error | no_result + reason: str + tags: tuple[str, ...] + prediction: dict[str, Any] = field(default_factory=dict) + + +def score_item(item: EvalItem, prediction: Prediction, gold: ResultSet | None, gold_error: str | None, + system: str) -> ItemScore: + pred = {k: v for k, v in asdict(prediction).items() if k != "result"} + pred["row_count"] = len(prediction.result.rows) if prediction.result else None + if not item.gold_sql: + return ItemScore(item.id, system, "no_gold", "item has no gold SQL", item.tags, pred) + if gold is None: + return ItemScore(item.id, system, "gold_error", f"gold SQL failed: {gold_error}", item.tags, pred) + if prediction.result is None: + return ItemScore(item.id, system, "no_result", prediction.error or prediction.status or "no result", item.tags, pred) + ok, reason = compare_results(prediction.result, gold, ordered=is_ordered(item.gold_sql)) + return ItemScore(item.id, system, "correct" if ok else "incorrect", reason, item.tags, pred) + + +def summarize(scores: list[ItemScore]) -> dict[str, Any]: + scored = [s for s in scores if s.status in ("correct", "incorrect", "no_result")] + correct = sum(s.status == "correct" for s in scored) + latencies = [s.prediction.get("latency_s") or 0.0 for s in scores] + n = len(scores) or 1 + by_tag: dict[str, list[int]] = {} + for s in scored: + for tag in s.tags: + by_tag.setdefault(tag, [0, 0]) + by_tag[tag][0] += s.status == "correct" + by_tag[tag][1] += 1 + return { + "items": len(scores), + "scored": len(scored), + "execution_accuracy": round(correct / len(scored), 3) if scored else None, + "latency_p50_s": round(percentile(latencies, 50), 2), + "latency_p95_s": round(percentile(latencies, 95), 2), + "avg_input_tokens": round(sum(s.prediction.get("input_tokens") or 0 for s in scores) / n), + "avg_output_tokens": round(sum(s.prediction.get("output_tokens") or 0 for s in scores) / n), + "avg_tool_calls": round(sum(s.prediction.get("tool_calls") or 0 for s in scores) / n, 2), + "accuracy_by_tag": {t: round(c / total, 3) for t, (c, total) in sorted(by_tag.items())}, + "status_counts": dict(Counter(s.status for s in scores)), + } + + +async def evaluate(items: list[EvalItem], runners: list[Runner], concurrency: int = 1) -> dict[str, list[ItemScore]]: + gold_cache: dict[str, tuple[ResultSet | None, str | None]] = {} + for item in items: + if item.gold_sql: + gold_cache[item.id] = await asyncio.to_thread(execute_for_eval, item.gold_sql) + + sem = asyncio.Semaphore(max(1, concurrency)) + + async def one(runner: Runner, item: EvalItem) -> ItemScore: + async with sem: + prediction = await runner.run(item) + gold, gold_error = gold_cache.get(item.id, (None, None)) + score = score_item(item, prediction, gold, gold_error, runner.name) + if runner.name == "deep" and prediction.trace_id and score.status in ("correct", "incorrect", "no_result"): + _score_trace(prediction.trace_id, score) + return score + + results: dict[str, list[ItemScore]] = {} + for runner in runners: + results[runner.name] = list(await asyncio.gather(*(one(runner, item) for item in items))) + return results + + +def _score_trace(trace_id: str, score: ItemScore) -> None: + try: + from agent.langfuse_client import langfuse_client + + langfuse_client.create_score(trace_id=trace_id, name="deep-agent.execution_accuracy", + value=1.0 if score.status == "correct" else 0.0, data_type="BOOLEAN", + comment=score.reason) + except Exception as exc: + logger.debug("execution_accuracy score skipped: %s", exc) + + +# ── go / no-go ─────────────────────────────────────────────────────────────── + +def gate(summaries: dict[str, dict[str, Any]], p95_slo_s: float) -> tuple[bool, list[str]]: + """Deep agent passes if its accuracy >= legacy's (when legacy ran) and its p95 latency <= the SLO.""" + reasons = [] + deep = summaries.get("deep") + if not deep or deep["execution_accuracy"] is None: + return False, ["deep agent has no scored items"] + legacy = summaries.get("legacy") + if legacy and legacy["execution_accuracy"] is not None and deep["execution_accuracy"] < legacy["execution_accuracy"]: + reasons.append(f"accuracy {deep['execution_accuracy']} < legacy {legacy['execution_accuracy']}") + if deep["latency_p95_s"] > p95_slo_s: + reasons.append(f"p95 latency {deep['latency_p95_s']}s > SLO {p95_slo_s}s") + return not reasons, reasons + + +def report_markdown(summaries: dict[str, dict[str, Any]]) -> str: + keys = ["scored", "execution_accuracy", "latency_p50_s", "latency_p95_s", "avg_tool_calls", + "avg_input_tokens", "avg_output_tokens"] + systems = list(summaries) + lines = ["| metric | " + " | ".join(systems) + " |", "|---|" + "---|" * len(systems)] + lines += [f"| {k} | " + " | ".join(str(summaries[s].get(k)) for s in systems) + " |" for k in keys] + return "\n".join(lines) + + +# ── Langfuse Experiments ───────────────────────────────────────────────────── + +def make_langfuse_task(runner: Runner): + """Experiment task: item.input = {"question": ...} (or a plain string).""" + async def task(*, item, **kwargs): + question = item.input["question"] if isinstance(item.input, dict) else str(item.input) + prediction = await runner.run(EvalItem(id=str(getattr(item, "id", "")), question=question)) + return {"sql": prediction.sql, "status": prediction.status, + "columns": prediction.result.columns if prediction.result else None, + "rows": prediction.result.rows if prediction.result else None} + return task + + +def execution_accuracy_evaluator(*, input, output, expected_output=None, **kwargs): + """Experiment evaluator: expected_output = {"sql": gold_sql} (or the gold SQL string).""" + gold_sql = expected_output.get("sql") if isinstance(expected_output, dict) else expected_output + if not gold_sql: + return {"name": "execution_accuracy", "value": 0.0, "comment": "no gold SQL"} + gold, error = execute_for_eval(gold_sql) + if gold is None: + return {"name": "execution_accuracy", "value": 0.0, "comment": f"gold SQL failed: {error}"} + if not output or output.get("rows") is None: + return {"name": "execution_accuracy", "value": 0.0, "comment": f"no result ({(output or {}).get('status')})"} + ok, reason = compare_results(ResultSet(output["columns"], output["rows"]), gold, ordered=is_ordered(gold_sql)) + return {"name": "execution_accuracy", "value": 1.0 if ok else 0.0, "comment": reason} + + +def run_langfuse_experiment(dataset_name: str, runner: Runner, run_name: str | None = None): + from agent.langfuse_client import langfuse_client + + dataset = langfuse_client.get_dataset(dataset_name) + return dataset.run_experiment( + name=f"{runner.name}-agent", run_name=run_name, task=make_langfuse_task(runner), + evaluators=[execution_accuracy_evaluator], max_concurrency=2, + metadata={"system": runner.name}, + ) + + +# ── CLI ────────────────────────────────────────────────────────────────────── + +def main() -> None: + parser = argparse.ArgumentParser(description="Evaluate the Deep Agent against gold SQL (and the legacy graph).") + parser.add_argument("--dataset", type=Path, default=DEFAULT_DATASET) + parser.add_argument("--langfuse-dataset", default=None, help="Run as a Langfuse Experiment on this dataset instead.") + parser.add_argument("--systems", default="deep,legacy", help="Comma-separated: deep, legacy.") + parser.add_argument("--connection-id", type=int, default=None) + parser.add_argument("--concurrency", type=int, default=1) + parser.add_argument("--out", type=Path, default=None) + parser.add_argument("--gate", action="store_true", help="Exit 1 if the deep agent fails the go/no-go gate.") + parser.add_argument("--p95-slo", type=float, default=60.0, help="p95 latency SLO in seconds for --gate.") + args = parser.parse_args() + + from agent.utils.jeen_metadata_client import close_mcp_sessions + from deep_agent import observability + from deep_agent.agent import acreate_deep_agent + + systems = [s.strip() for s in args.systems.split(",") if s.strip()] + context = DeepAgentContext(connection_id=args.connection_id) + + if args.langfuse_dataset: + async def _deep(): + return DeepAgentRunner(await acreate_deep_agent(), context) + runner = asyncio.run(_deep()) if "deep" in systems else LegacyGraphRunner() + result = run_langfuse_experiment(args.langfuse_dataset, runner) + print(result.format()) + observability.flush() + return + + items = load_dataset(args.dataset) + + async def _run(): + runners: list[Runner] = [] + if "deep" in systems: + runners.append(DeepAgentRunner(await acreate_deep_agent(), context)) + if "legacy" in systems: + runners.append(LegacyGraphRunner()) + try: + return await evaluate(items, runners, args.concurrency) + finally: + await close_mcp_sessions() + + try: + results = asyncio.run(_run()) + finally: + observability.flush() + summaries = {name: summarize(scores) for name, scores in results.items()} + print(report_markdown(summaries)) + print(json.dumps(summaries, ensure_ascii=False, indent=2)) + if args.out: + args.out.parent.mkdir(parents=True, exist_ok=True) + args.out.write_text(json.dumps({ + "summary": summaries, + "items": {name: [asdict(s) for s in scores] for name, scores in results.items()}, + }, ensure_ascii=False, indent=2, default=str), encoding="utf-8") + if args.gate: + passed, reasons = gate(summaries, args.p95_slo) + print("GATE:", "PASS" if passed else "FAIL — " + "; ".join(reasons)) + raise SystemExit(0 if passed else 1) + + +if __name__ == "__main__": + main() diff --git a/agent/tests/deep_agent/test_evaluation.py b/agent/tests/deep_agent/test_evaluation.py new file mode 100644 index 0000000..c0e109b --- /dev/null +++ b/agent/tests/deep_agent/test_evaluation.py @@ -0,0 +1,185 @@ +import json +from decimal import Decimal +from types import SimpleNamespace +from unittest.mock import patch + +import pytest +from langchain_core.messages import AIMessage + +from core.trino import TrinoExecutionResult +from deep_agent import evaluation as ev +from deep_agent import tools +from deep_agent.agent import create_deep_agent +from deep_agent.context import DeepAgentContext +from deep_agent.evaluation import EvalItem, ItemScore, Prediction, ResultSet, compare_results + + +# ── comparison ─────────────────────────────────────────────────────────────── + +@pytest.mark.parametrize("pred, gold, ordered, expected", [ + ([[1, "a"], [2, "b"]], [[2, "b"], [1, "a"]], False, True), # order ignored + ([[1, "a"], [2, "b"]], [[2, "b"], [1, "a"]], True, False), # order matters + ([["a", 1]], [[1, "a"]], False, True), # columns permuted + ([[0.1 + 0.2]], [[0.3]], False, True), # float tolerance + ([[Decimal("466")]], [["466"]], False, True), # decimal vs numeric string + ([[466]], [[466], [466]], False, False), # row count + ([[None]], [[None]], False, True), + ([[1, 2]], [[1]], False, False), # column count + ([[" France "]], [["France"]], False, True), # whitespace +]) +def test_compare_results(pred, gold, ordered, expected): + ok, reason = compare_results(ResultSet(["x"] * len(pred[0]), pred), ResultSet(["y"] * len(gold[0]), gold), ordered) + assert ok is expected, reason + + +def test_duplicates_matter_when_unordered(): + ok, _ = compare_results(ResultSet(["a"], [[1], [1], [2]]), ResultSet(["a"], [[1], [2], [2]]), ordered=False) + assert not ok + + +@pytest.mark.parametrize("sql, ordered", [ + ("SELECT a FROM t ORDER BY a", True), + ("SELECT a FROM (SELECT a FROM t ORDER BY a) x", False), + ("WITH c AS (SELECT a FROM t) SELECT a FROM c ORDER BY a DESC LIMIT 5", True), + ("SELECT count(*) FROM t", False), + ("not sql at all", False), +]) +def test_is_ordered(sql, ordered): + assert ev.is_ordered(sql) is ordered + + +# ── scoring and summaries ──────────────────────────────────────────────────── + +GOLD = ResultSet(["n"], [[466]]) + + +@pytest.mark.parametrize("item, prediction, gold, status", [ + (EvalItem("a", "q", None), Prediction("SELECT 1", GOLD, 1.0), None, "no_gold"), + (EvalItem("a", "q", "SELECT 466"), Prediction("SELECT 1", GOLD, 1.0), None, "gold_error"), + (EvalItem("a", "q", "SELECT 466"), Prediction(None, None, 1.0, status="no_answer"), GOLD, "no_result"), + (EvalItem("a", "q", "SELECT 466"), Prediction("SELECT 466", GOLD, 1.0), GOLD, "correct"), + (EvalItem("a", "q", "SELECT 466"), Prediction("SELECT 1", ResultSet(["n"], [[1]]), 1.0), GOLD, "incorrect"), +]) +def test_score_item(item, prediction, gold, status): + assert ev.score_item(item, prediction, gold, "boom", "deep").status == status + + +def _score(status, latency, tags=("t",), tokens=100, tools=3): + return ItemScore("i", "deep", status, "", tags, + {"latency_s": latency, "input_tokens": tokens, "output_tokens": tokens // 10, "tool_calls": tools}) + + +def test_summarize(): + s = ev.summarize([_score("correct", 10, ("temporal",)), _score("incorrect", 20, ("temporal", "join")), + _score("no_result", 30), _score("no_gold", 40)]) + assert s["scored"] == 3 and s["execution_accuracy"] == 0.333 + assert s["latency_p50_s"] == 20 and s["latency_p95_s"] == 40 + assert s["accuracy_by_tag"] == {"join": 0.0, "t": 0.0, "temporal": 0.5} + assert s["status_counts"] == {"correct": 1, "incorrect": 1, "no_result": 1, "no_gold": 1} + + +@pytest.mark.parametrize("deep, legacy, slo, passed", [ + ({"execution_accuracy": 0.8, "latency_p95_s": 40}, {"execution_accuracy": 0.7, "latency_p95_s": 90}, 60, True), + ({"execution_accuracy": 0.6, "latency_p95_s": 40}, {"execution_accuracy": 0.7, "latency_p95_s": 90}, 60, False), + ({"execution_accuracy": 0.9, "latency_p95_s": 75}, None, 60, False), + ({"execution_accuracy": None, "latency_p95_s": 1}, None, 60, False), +]) +def test_gate(deep, legacy, slo, passed): + summaries = {"deep": deep, **({"legacy": legacy} if legacy else {})} + assert ev.gate(summaries, slo)[0] is passed + + +def test_report_markdown_has_one_column_per_system(): + md = ev.report_markdown({"deep": ev.summarize([_score("correct", 1)]), "legacy": ev.summarize([_score("incorrect", 2)])}) + assert md.splitlines()[0] == "| metric | deep | legacy |" + assert "| execution_accuracy | 1.0 | 0.0 |" in md + + +# ── end to end: both systems on the same items ────────────────────────────── + +ITEMS = [ + EvalItem("q1", "כמה טיסות המריאו מצרפת", "SELECT COUNT(DISTINCT flight_id) FROM flights WHERE launch_country = 'France'", ("simple_filter",)), + EvalItem("q2", "שאלה בלי זהב", None, ("temporal",)), +] + + +def _deep_policy(messages): + if not [m for m in messages if m.type == "tool"]: + return AIMessage("", tool_calls=[{"name": "execute_trino_query", "id": "e", + "args": {"sql": "SELECT COUNT(*) AS n FROM flights WHERE launch_country = 'France'"}}]) + return AIMessage("", tool_calls=[{"name": "DeepAgentAnswer", "id": "f", "args": { + "sql": "SELECT COUNT(*) AS n FROM flights WHERE launch_country = 'France'", "explanation_he": "x", "confidence": "high"}}]) + + +class FakeLegacyGraph: + async def ainvoke(self, state, config=None): + assert state["non_interactive"] is True + return {"sql_query": "SELECT 1", "inline_result_columns": ["n"], "inline_result_rows": [[1]], "messages": []} + + +@pytest.mark.asyncio +async def test_evaluate_both_systems(policy_model): + def trino(sql, **_): + return TrinoExecutionResult(success=True, rows=[[466]], columns=["n"], row_count=1, query_id="q") + + agent = create_deep_agent(model=policy_model(_deep_policy)) + runners = [ev.DeepAgentRunner(agent, DeepAgentContext(), timeout_s=30), ev.LegacyGraphRunner(graph=FakeLegacyGraph())] + with patch.object(ev, "execute_query_sync", side_effect=trino), patch.object(tools, "execute_query_sync", side_effect=trino): + results = await ev.evaluate(ITEMS, runners) + + deep = {s.id: s for s in results["deep"]} + legacy = {s.id: s for s in results["legacy"]} + assert deep["q1"].status == "correct" and deep["q2"].status == "no_gold" + assert legacy["q1"].status == "incorrect" and legacy["q2"].status == "no_gold" + summaries = {k: ev.summarize(v) for k, v in results.items()} + assert summaries["deep"]["execution_accuracy"] == 1.0 and summaries["legacy"]["execution_accuracy"] == 0.0 + assert ev.gate(summaries, p95_slo_s=60)[0] is True + + +@pytest.mark.asyncio +async def test_legacy_sql_only_answers_are_executed(): + class SqlOnlyGraph: + async def ainvoke(self, state, config=None): + return {"sql_query": "SELECT 466 AS n", "messages": []} + + with patch.object(ev, "execute_query_sync", + return_value=TrinoExecutionResult(success=True, rows=[[466]], columns=["n"], row_count=1)): + prediction = await ev.LegacyGraphRunner(graph=SqlOnlyGraph()).run(ITEMS[0]) + assert prediction.result.rows == [[466]] + + +# ── Langfuse experiment adapters ───────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_langfuse_task_and_evaluator(): + class StubRunner: + name = "stub" + + async def run(self, item): + return Prediction("SELECT 466", ResultSet(["n"], [[466]]), 1.0, status="matched") + + output = await ev.make_langfuse_task(StubRunner())(item=SimpleNamespace(id="x", input={"question": "q"})) + assert output["rows"] == [[466]] + with patch.object(ev, "execute_query_sync", + return_value=TrinoExecutionResult(success=True, rows=[[466]], columns=["n"], row_count=1)): + good = ev.execution_accuracy_evaluator(input={}, output=output, expected_output={"sql": "SELECT 466"}) + assert good["value"] == 1.0 + assert ev.execution_accuracy_evaluator(input={}, output=output, expected_output=None)["value"] == 0.0 + + +# ── dataset ────────────────────────────────────────────────────────────────── + +def test_seed_dataset_is_valid(): + items = ev.load_dataset(ev.DEFAULT_DATASET) + assert len(items) >= 10 + for item in items: + assert item.question.strip() and item.tags + if item.gold_sql: + ev.sql_guard.parse_read_statement(item.gold_sql) # gold SQL is read-only and parses + + +def test_duplicate_ids_rejected(tmp_path): + path = tmp_path / "d.json" + path.write_text(json.dumps({"items": [{"id": "a", "question": "q"}, {"id": "a", "question": "r"}]})) + with pytest.raises(ValueError, match="duplicate"): + ev.load_dataset(path)