From e124a8a7d9e3bad480e83a16cf141852cf61e8c3 Mon Sep 17 00:00:00 2001 From: buenosairesam Date: Wed, 3 Jun 2026 05:04:29 -0300 Subject: [PATCH] verbose live UI + tool-level SSE events + Groq default + regression tests --- .dockerignore | 15 + .env.example | 3 + .gitignore | 5 + Makefile | 28 +- api/analyses/__init__.py | 0 api/analyses/base.py | 27 + api/analyses/compare_periods.py | 141 +++++ api/analyses/direct_answer.py | 81 +++ api/analyses/registry.py | 30 ++ api/analyses/types.py | 17 + api/config.py | 18 +- api/datasets/README.md | 23 + api/datasets/__init__.py | 40 ++ api/datasets/financial/metrics.yaml | 50 ++ api/datasets/financial/schema_docs.yaml | 93 ++++ api/evals/__init__.py | 0 api/evals/run_evals.py | 11 + api/langfuse_client.py | 87 ++++ api/llm.py | 91 ++++ api/main.py | 52 +- api/plan/__init__.py | 0 api/plan/planner.py | 52 ++ api/plan/types.py | 37 ++ api/prompts/__init__.py | 31 ++ api/prompts/compare_periods.interpret.txt | 5 + .../compare_periods.interpret.user.txt | 7 + api/prompts/compare_periods.pair.txt | 16 + api/prompts/compare_periods.pair.user.txt | 10 + api/prompts/direct_answer.interpret.txt | 5 + api/prompts/direct_answer.interpret.user.txt | 7 + api/prompts/planner.system.txt | 17 + api/prompts/planner.user.txt | 9 + api/prompts/synthesize.system.txt | 7 + api/prompts/synthesize.user.txt | 4 + api/prompts/text_to_sql.system.txt | 15 + api/prompts/text_to_sql.user.txt | 10 + api/runtime/__init__.py | 0 api/runtime/context.py | 12 + api/runtime/events.py | 133 +++++ api/runtime/runner.py | 139 +++++ api/runtime/state.py | 20 + api/tools/__init__.py | 0 api/tools/db.py | 39 ++ api/tools/execute_sql.py | 55 ++ api/tools/schema.py | 57 ++ api/tools/text_to_sql.py | 70 +++ api/tools/types.py | 128 +++++ ctrl/Dockerfile.api | 1 + ctrl/Tiltfile | 16 +- ctrl/docker-compose.yml | 5 + ctrl/k8s/base/configmap.yaml | 20 +- ctrl/seed/__init__.py | 0 ctrl/seed/download.sh | 44 ++ ctrl/seed/load_bird.py | 147 ++++++ pyproject.toml | 2 + tests/__init__.py | 0 tests/test_events.py | 60 +++ tests/test_imports.py | 42 ++ tests/test_langfuse_client.py | 41 ++ tests/test_llm.py | 15 + tests/test_plan_types.py | 31 ++ tests/test_prompts.py | 30 ++ tests/test_yaml_datasets.py | 46 ++ ui/app/src/composables/useRunStream.ts | 267 ++++++++++ ui/app/src/config.ts | 21 + ui/app/src/main.ts | 3 +- ui/app/src/pages/Ask.vue | 489 ++++++++++++++++-- ui/app/src/pages/RunInspector.vue | 47 -- uv.lock | 143 +++++ 69 files changed, 3030 insertions(+), 137 deletions(-) create mode 100644 .dockerignore create mode 100644 api/analyses/__init__.py create mode 100644 api/analyses/base.py create mode 100644 api/analyses/compare_periods.py create mode 100644 api/analyses/direct_answer.py create mode 100644 api/analyses/registry.py create mode 100644 api/analyses/types.py create mode 100644 api/datasets/README.md create mode 100644 api/datasets/__init__.py create mode 100644 api/datasets/financial/metrics.yaml create mode 100644 api/datasets/financial/schema_docs.yaml create mode 100644 api/evals/__init__.py create mode 100644 api/evals/run_evals.py create mode 100644 api/langfuse_client.py create mode 100644 api/llm.py create mode 100644 api/plan/__init__.py create mode 100644 api/plan/planner.py create mode 100644 api/plan/types.py create mode 100644 api/prompts/__init__.py create mode 100644 api/prompts/compare_periods.interpret.txt create mode 100644 api/prompts/compare_periods.interpret.user.txt create mode 100644 api/prompts/compare_periods.pair.txt create mode 100644 api/prompts/compare_periods.pair.user.txt create mode 100644 api/prompts/direct_answer.interpret.txt create mode 100644 api/prompts/direct_answer.interpret.user.txt create mode 100644 api/prompts/planner.system.txt create mode 100644 api/prompts/planner.user.txt create mode 100644 api/prompts/synthesize.system.txt create mode 100644 api/prompts/synthesize.user.txt create mode 100644 api/prompts/text_to_sql.system.txt create mode 100644 api/prompts/text_to_sql.user.txt create mode 100644 api/runtime/__init__.py create mode 100644 api/runtime/context.py create mode 100644 api/runtime/events.py create mode 100644 api/runtime/runner.py create mode 100644 api/runtime/state.py create mode 100644 api/tools/__init__.py create mode 100644 api/tools/db.py create mode 100644 api/tools/execute_sql.py create mode 100644 api/tools/schema.py create mode 100644 api/tools/text_to_sql.py create mode 100644 api/tools/types.py create mode 100644 ctrl/seed/__init__.py create mode 100755 ctrl/seed/download.sh create mode 100644 ctrl/seed/load_bird.py create mode 100644 tests/__init__.py create mode 100644 tests/test_events.py create mode 100644 tests/test_imports.py create mode 100644 tests/test_langfuse_client.py create mode 100644 tests/test_llm.py create mode 100644 tests/test_plan_types.py create mode 100644 tests/test_prompts.py create mode 100644 tests/test_yaml_datasets.py create mode 100644 ui/app/src/composables/useRunStream.ts create mode 100644 ui/app/src/config.ts delete mode 100644 ui/app/src/pages/RunInspector.vue diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..3e7a0e1 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,15 @@ +.git +.venv +__pycache__ +*.pyc +.pytest_cache +.claude +def +ui/node_modules +ui/framework/node_modules +ui/app/node_modules +ui/app/dist +.data +ctrl/seed/data +api/evals/data +node_modules diff --git a/.env.example b/.env.example index 1a557a8..0fcbad0 100644 --- a/.env.example +++ b/.env.example @@ -4,6 +4,9 @@ # Warehouse DATABASE_URL=postgresql://nvi:nvi@postgres:5432/nvi +# Active dataset (selects api/datasets// + Postgres schema + sqlite file). +NVI_DATASET=financial + # LLM ANTHROPIC_API_KEY= ANTHROPIC_MODEL=claude-sonnet-4-6 diff --git a/.gitignore b/.gitignore index 1d5abe9..28e2df9 100644 --- a/.gitignore +++ b/.gitignore @@ -13,3 +13,8 @@ dist # soleprint-ui framework is propagated by spr; not tracked in nvi. ui/framework + +# Data: BIRD financial SQLite (~70MB) downloaded by ctrl/seed/download.sh, +# and the BIRD golden eval set. Never committed. +ctrl/seed/data +api/evals/data diff --git a/Makefile b/Makefile index 8d238f5..70be936 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: kind tilt-up tilt-down seed evals \ +.PHONY: kind tilt-up tilt-down seed seed-fetch seed-push evals test \ compose-up compose-down compose-clean compose-seed compose-evals \ docs-graphs @@ -17,11 +17,29 @@ tilt-up: tilt-down: cd ctrl && tilt down $(KCTX) -seed: +seed-fetch: + bash ctrl/seed/download.sh + +# Copy the downloaded SQLite into the api pod. The file is excluded from the +# docker build context (.dockerignore) to keep the image small, so we push +# it explicitly the first time. +seed-push: + @POD=$$(kubectl $(KCTX) $(KNS) get pod -l app=api -o jsonpath='{.items[0].metadata.name}'); \ + echo "pushing financial.sqlite → $$POD"; \ + kubectl $(KCTX) $(KNS) exec $$POD -- mkdir -p /app/seed/data; \ + kubectl $(KCTX) $(KNS) cp ctrl/seed/data/financial.sqlite $$POD:/app/seed/data/financial.sqlite + +seed: seed-fetch seed-push kubectl $(KCTX) $(KNS) exec deploy/api -- uv run python -m seed.load_bird evals: - kubectl $(KCTX) $(KNS) exec deploy/api -- uv run python -m evals.run_evals + kubectl $(KCTX) $(KNS) exec deploy/api -- uv run python -m api.evals.run_evals + +# Unit tests run on the host venv (no postgres/llm calls needed). +# Ensures dev extras are installed first; uv is idempotent on no-ops. +test: + uv sync --extra dev + uv run pytest -q # ── docker-compose (alternative path) ────────────────────── @@ -34,11 +52,11 @@ compose-down: compose-clean: $(COMPOSE) down -v -compose-seed: +compose-seed: seed-fetch $(COMPOSE) exec api uv run python -m seed.load_bird compose-evals: - $(COMPOSE) exec api uv run python -m evals.run_evals + $(COMPOSE) exec api uv run python -m api.evals.run_evals # ── docs ─────────────────────────────────────────────────── diff --git a/api/analyses/__init__.py b/api/analyses/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/api/analyses/base.py b/api/analyses/base.py new file mode 100644 index 0000000..c7830a5 --- /dev/null +++ b/api/analyses/base.py @@ -0,0 +1,27 @@ +"""Analysis base class. + +An Analysis is a self-contained mini-agent. It owns its reasoning pattern +(CoT, ReAct, plan-and-execute) and exposes a uniform external contract: +`run(args, question) -> Finding`. The runtime composes Analyses; it doesn't +know how each one thinks internally. + +Public surface is framework-free — no langgraph types leak out. +""" + +from __future__ import annotations + +from typing import Any, ClassVar + +from api.analyses.types import Finding + + +class Analysis: + """Subclass and set `name`, `description`, `args_schema`; then implement run().""" + + name: ClassVar[str] = "" + description: ClassVar[str] = "" + # JSON-schema-ish description of the args dict — the planner reads this. + args_schema: ClassVar[dict[str, Any]] = {} + + async def run(self, args: dict[str, Any], question: str) -> Finding: + raise NotImplementedError diff --git a/api/analyses/compare_periods.py b/api/analyses/compare_periods.py new file mode 100644 index 0000000..14f9769 --- /dev/null +++ b/api/analyses/compare_periods.py @@ -0,0 +1,141 @@ +"""compare_periods Analysis — CoT with two queries. + +Both SQL queries are generated in one LLM call (paired prompt), then both +executed, then a single interpretation pass diffs them. Single round-trip +per LLM step → cheap and bounded latency. +""" + +from __future__ import annotations + +import json +import logging +import re +from typing import Any + +from api import langfuse_client as lf +from api.analyses.base import Analysis +from api.analyses.types import Finding +from api.llm import chat +from api.prompts import load, render +from api.runtime import events +from api.tools.execute_sql import execute_sql +from api.tools.schema import load_schema + +logger = logging.getLogger("nvi.analyses.compare_periods") + + +class ComparePeriods(Analysis): + name = "compare_periods" + description = ( + "Compare a metric across two time windows (e.g. Q2 vs Q3, 1995 vs 1996). " + "Generates two parallel SQL queries and diffs the results." + ) + args_schema = { + "question": {"type": "string", "description": "The metric / scope to compare."}, + "period_a": {"type": "string", "description": "First period label, e.g. '1995' or 'Q2 1996'."}, + "period_b": {"type": "string", "description": "Second period label, e.g. '1996' or 'Q3 1996'."}, + } + + async def run(self, args: dict[str, Any], question: str) -> Finding: + sub_q = args.get("question") or question + period_a = args.get("period_a", "") + period_b = args.get("period_b", "") + + with lf.span("analysis.compare_periods", input=args) as span: + try: + await events.publish_current(events.tool_call_start( + "generate_pair", input={"question": sub_q, "period_a": period_a, "period_b": period_b}, + )) + pair = _generate_pair(sub_q, period_a, period_b) + await events.publish_current(events.tool_call_end( + "generate_pair", + output={"a": {"label": pair["a"]["label"], "sql": pair["a"]["sql"]}, + "b": {"label": pair["b"]["label"], "sql": pair["b"]["sql"]}}, + )) + + await events.publish_current(events.tool_call_start("execute_sql", input={"label": pair["a"]["label"], "sql": pair["a"]["sql"]})) + res_a = execute_sql(pair["a"]["sql"]) + await events.publish_current(events.tool_call_end( + "execute_sql", + output={"label": pair["a"]["label"], "row_count": res_a.row_count, + "preview": res_a.as_dicts()[:5]}, + )) + + await events.publish_current(events.tool_call_start("execute_sql", input={"label": pair["b"]["label"], "sql": pair["b"]["sql"]})) + res_b = execute_sql(pair["b"]["sql"]) + await events.publish_current(events.tool_call_end( + "execute_sql", + output={"label": pair["b"]["label"], "row_count": res_b.row_count, + "preview": res_b.as_dicts()[:5]}, + )) + + interpret_system = load("compare_periods.interpret") + interpret_user = render( + "compare_periods.interpret.user", + question=sub_q, + label_a=pair["a"]["label"], + label_b=pair["b"]["label"], + rows_a=repr(res_a.as_dicts()[:20]), + rows_b=repr(res_b.as_dicts()[:20]), + ) + await events.publish_current(events.llm_call( + "compare_periods.interpret", + system_len=len(interpret_system), + user_len=len(interpret_user), + )) + summary = chat(system=interpret_system, user=interpret_user, max_tokens=512) + except Exception as e: + logger.exception("compare_periods failed") + await events.publish_current(events.tool_call_end("compare_periods", error=str(e))) + span.update(output={"error": str(e)}) + return Finding(analysis=self.name, summary="", error=str(e)) + + span.update(output={"summary": summary}) + return Finding( + analysis=self.name, + summary=summary, + rows=[ + {"period": pair["a"]["label"], **r} for r in res_a.as_dicts()[:20] + ] + [ + {"period": pair["b"]["label"], **r} for r in res_b.as_dicts()[:20] + ], + sql=[pair["a"]["sql"], pair["b"]["sql"]], + metadata={ + "label_a": pair["a"]["label"], + "label_b": pair["b"]["label"], + "row_count_a": res_a.row_count, + "row_count_b": res_b.row_count, + }, + ) + + +def _generate_pair(question: str, period_a: str, period_b: str) -> dict[str, dict[str, str]]: + schema = load_schema() + text = chat( + system=load("compare_periods.pair"), + user=render( + "compare_periods.pair.user", + question=question, + period_a=period_a or "(infer)", + period_b=period_b or "(infer)", + schema_block=schema.render_tables(), + metrics_block=schema.render_metrics(), + ), + max_tokens=1024, + ) + obj = json.loads(_extract_json(text)) + for k in ("a", "b"): + if k not in obj or "sql" not in obj[k] or "label" not in obj[k]: + raise ValueError(f"pair generator returned malformed payload: missing {k}") + obj[k]["sql"] = obj[k]["sql"].strip().rstrip(";").strip() + return obj + + +def _extract_json(text: str) -> str: + m = re.search(r"```(?:json)?\s*(\{.*\})\s*```", text, re.DOTALL) + if m: + return m.group(1) + start, end = text.find("{"), text.rfind("}") + if start >= 0 and end > start: + return text[start:end + 1] + return text diff --git a/api/analyses/direct_answer.py b/api/analyses/direct_answer.py new file mode 100644 index 0000000..ab7d9d0 --- /dev/null +++ b/api/analyses/direct_answer.py @@ -0,0 +1,81 @@ +"""direct_answer Analysis — CoT, one-shot. + +Pattern: text-to-SQL → execute → ask the LLM to interpret the rows. No loop. +For questions that resolve to a single SQL query and a short interpretation. +""" + +from __future__ import annotations + +import logging +from typing import Any + +from api import langfuse_client as lf +from api.analyses.base import Analysis +from api.analyses.types import Finding +from api.llm import chat +from api.prompts import load, render +from api.runtime import events +from api.tools.execute_sql import execute_sql +from api.tools.text_to_sql import text_to_sql + +logger = logging.getLogger("nvi.analyses.direct_answer") + + +class DirectAnswer(Analysis): + name = "direct_answer" + description = ( + "Answer a question that resolves to a single SQL query. Best for " + "lookup / aggregation questions with a single, well-defined metric." + ) + args_schema = { + "question": {"type": "string", "description": "Refined question this Analysis should answer."}, + "hint_tables": {"type": "array", "items": "string", "description": "Tables the planner thinks are relevant (optional)."}, + } + + async def run(self, args: dict[str, Any], question: str) -> Finding: + sub_q = args.get("question") or question + hint_tables = args.get("hint_tables") + + with lf.span("analysis.direct_answer", input={"question": sub_q}) as span: + try: + await events.publish_current(events.tool_call_start("text_to_sql", input={"question": sub_q})) + t2s = text_to_sql(sub_q, hint_tables=hint_tables) + await events.publish_current(events.tool_call_end( + "text_to_sql", output={"sql": t2s.sql, "tables": t2s.used_tables}, + )) + + await events.publish_current(events.tool_call_start("execute_sql", input={"sql": t2s.sql})) + result = execute_sql(t2s.sql) + await events.publish_current(events.tool_call_end( + "execute_sql", + output={"row_count": result.row_count, "truncated": result.truncated, + "preview": result.as_dicts()[:5]}, + )) + + interpret_system = load("direct_answer.interpret") + interpret_user = render( + "direct_answer.interpret.user", + question=sub_q, + sql=t2s.sql, + rows=repr(result.as_dicts()[:20]), + ) + await events.publish_current(events.llm_call( + "direct_answer.interpret", + system_len=len(interpret_system), + user_len=len(interpret_user), + )) + summary = chat(system=interpret_system, user=interpret_user, max_tokens=512) + except Exception as e: + logger.exception("direct_answer failed") + await events.publish_current(events.tool_call_end("direct_answer", error=str(e))) + span.update(output={"error": str(e)}) + return Finding(analysis=self.name, summary="", error=str(e)) + + span.update(output={"summary": summary, "rows": result.row_count}) + return Finding( + analysis=self.name, + summary=summary, + rows=result.as_dicts()[:20], + sql=[t2s.sql], + metadata={"row_count": result.row_count, "truncated": result.truncated}, + ) diff --git a/api/analyses/registry.py b/api/analyses/registry.py new file mode 100644 index 0000000..4a2f272 --- /dev/null +++ b/api/analyses/registry.py @@ -0,0 +1,30 @@ +"""Analysis registry — the planner reads this to know what's available.""" + +from __future__ import annotations + +from typing import Any + +from api.analyses.base import Analysis +from api.analyses.compare_periods import ComparePeriods +from api.analyses.direct_answer import DirectAnswer + +REGISTRY: dict[str, Analysis] = { + DirectAnswer.name: DirectAnswer(), + ComparePeriods.name: ComparePeriods(), +} + + +def get(name: str) -> Analysis | None: + return REGISTRY.get(name) + + +def catalog() -> list[dict[str, Any]]: + """Plain-dict catalog for prompt injection into the planner.""" + return [ + { + "name": a.name, + "description": a.description, + "args_schema": a.args_schema, + } + for a in REGISTRY.values() + ] diff --git a/api/analyses/types.py b/api/analyses/types.py new file mode 100644 index 0000000..decf68f --- /dev/null +++ b/api/analyses/types.py @@ -0,0 +1,17 @@ +"""Public data shapes for the Analyses layer.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + + +@dataclass +class Finding: + """Structured output every Analysis returns.""" + analysis: str + summary: str + rows: list[dict[str, Any]] = field(default_factory=list) + sql: list[str] = field(default_factory=list) + error: str | None = None + metadata: dict[str, Any] = field(default_factory=dict) diff --git a/api/config.py b/api/config.py index aba53d1..c6aa7e0 100644 --- a/api/config.py +++ b/api/config.py @@ -6,10 +6,26 @@ from pydantic_settings import BaseSettings class Settings(BaseSettings): database_url: str = "postgresql://nvi:nvi@postgres:5432/nvi" + # Active dataset. Matches: api/datasets//, the Postgres schema, and + # ctrl/seed/data/.sqlite. Restart the api to switch. + dataset: str = "financial" + + # LLM provider. Supported: "groq", "anthropic", "openai". + llm_provider: str = "groq" + anthropic_api_key: str = "" anthropic_model: str = "claude-sonnet-4-6" - langfuse_host: str = "http://lng.local.ar:3000" + # Groq is OpenAI-API-compatible; we hit it via the openai SDK with a base_url override. + groq_api_key: str = "" + groq_model: str = "llama-3.3-70b-versatile" + groq_base_url: str = "https://api.groq.com/openai/v1" + + openai_api_key: str = "" + openai_model: str = "gpt-4o-mini" + openai_base_url: str = "https://api.openai.com/v1" + + langfuse_host: str = "http://lng.local.ar" langfuse_public_key: str = "" langfuse_secret_key: str = "" diff --git a/api/datasets/README.md b/api/datasets/README.md new file mode 100644 index 0000000..c7eeffe --- /dev/null +++ b/api/datasets/README.md @@ -0,0 +1,23 @@ +# Datasets + +One subdirectory per dataset. The folder name is the dataset's identity — +used as: + +- the `NVI_DATASET` env var value that selects it, +- the Postgres schema where its tables live, +- the SQLite filename in `ctrl/seed/data/.sqlite` that the loader + reads from. + +## Adding a new dataset + +1. Place the source SQLite at `ctrl/seed/data/.sqlite` (or adapt + `ctrl/seed/download.sh` to fetch it). +2. Create `api/datasets//`: + - `schema_docs.yaml` — table and column descriptions. + - `metrics.yaml` — canonical SQL for business terms. +3. Set `NVI_DATASET=` and restart the api. +4. `make seed` will load `.sqlite` into Postgres schema ``. + +The Plan/Analyses/Tools/Runtime code references no dataset name; only the +loader (`api.tools.schema.load_schema`) and the engine (`api.tools.db`) +consult the active-dataset setting. diff --git a/api/datasets/__init__.py b/api/datasets/__init__.py new file mode 100644 index 0000000..431cd3e --- /dev/null +++ b/api/datasets/__init__.py @@ -0,0 +1,40 @@ +"""Dataset registry — the only dataset-specific surface in api/. + +Each subdirectory is one dataset. The folder name doubles as the Postgres +schema name, the SQLite filename under `ctrl/seed/data/.sqlite`, and +the value of the `NVI_DATASET` env var that selects it. + +Inside each dataset folder: + schema_docs.yaml — human descriptions for tables/columns + metrics.yaml — canonical SQL fragments for business terms + +Both files are consumed by `api.tools.schema.load_schema()`. The Plan, +Analyses, Tools, and Runtime code reference no dataset name — they read +the active dataset through `settings.dataset`. +""" + +from __future__ import annotations + +from functools import lru_cache +from pathlib import Path +from typing import Any + +import yaml + +from api.config import get_settings + +DATASETS_DIR = Path(__file__).resolve().parent + + +def active_dir() -> Path: + return DATASETS_DIR / get_settings().dataset + + +def available() -> list[str]: + return sorted(p.name for p in DATASETS_DIR.iterdir() if p.is_dir() and not p.name.startswith("_")) + + +@lru_cache(maxsize=None) +def read_yaml(dataset: str, filename: str) -> dict[str, Any]: + """Load one YAML file from a dataset folder. Cached per (dataset, file).""" + return yaml.safe_load((DATASETS_DIR / dataset / filename).read_text()) or {} diff --git a/api/datasets/financial/metrics.yaml b/api/datasets/financial/metrics.yaml new file mode 100644 index 0000000..af4d5e8 --- /dev/null +++ b/api/datasets/financial/metrics.yaml @@ -0,0 +1,50 @@ +# Domain metrics for the financial schema. The planner reads this so it can +# resolve a business term ("default rate", "loan volume") to a canonical SQL +# expression instead of guessing. + +schema: financial + +metrics: + loan_default_rate: + description: Fraction of finished loans that ended in default (status 'B'). Excludes still-running loans. + sql: "AVG(CASE WHEN status = 'B' THEN 1.0 WHEN status = 'A' THEN 0.0 END)" + from_table: loan + filter: "status IN ('A','B')" + unit: ratio + + at_risk_loan_share: + description: Of currently running loans, the share that are in debt (status 'D'). + sql: "AVG(CASE WHEN status = 'D' THEN 1.0 WHEN status = 'C' THEN 0.0 END)" + from_table: loan + filter: "status IN ('C','D')" + unit: ratio + + loan_volume: + description: Total CZK lent. + sql: "SUM(amount)" + from_table: loan + unit: CZK + + average_loan_amount: + description: Mean loan size in CZK. + sql: "AVG(amount)" + from_table: loan + unit: CZK + + transaction_volume: + description: Total CZK moved through transactions (credits + debits, absolute). + sql: "SUM(amount)" + from_table: trans + unit: CZK + + active_accounts: + description: Distinct accounts with at least one transaction in the period. + sql: "COUNT(DISTINCT account_id)" + from_table: trans + unit: count + + unemployment_rate_96: + description: District unemployment rate in 1996 (already a percentage). + sql: "A13" + from_table: district + unit: percent diff --git a/api/datasets/financial/schema_docs.yaml b/api/datasets/financial/schema_docs.yaml new file mode 100644 index 0000000..88b2dd8 --- /dev/null +++ b/api/datasets/financial/schema_docs.yaml @@ -0,0 +1,93 @@ +# Per-table / per-column human descriptions for the BIRD `financial` schema +# (PKDD'99 Czech bank). The text-to-SQL tool prepends the relevant subset to +# the LLM prompt so it can disambiguate cryptic column names like A2, A3, etc. + +schema: financial + +tables: + account: + description: One row per bank account. The pivot table — most facts hang off account_id. + columns: + account_id: Surrogate primary key for the account. + district_id: Geographic district the account belongs to (joins to district). + frequency: Statement issuance frequency. "POPLATEK MESICNE" = monthly, "POPLATEK TYDNE" = weekly, "POPLATEK PO OBRATU" = on transaction. + date: Date the account was opened (YYMMDD as integer). + + client: + description: One row per bank customer. + columns: + client_id: Surrogate primary key for the client. + gender: '"M" or "F". Derived from birth_number.' + birth_date: Date of birth, normalised. Original birth_number encoded gender by adding 50 to the month field for women. + district_id: District the client lives in (joins to district). + + disp: + description: Disposition — many-to-many link between client and account. Type tells you whether the client is the OWNER or just a USER of the account. + columns: + disp_id: Surrogate key. + client_id: Joins to client. + account_id: Joins to account. + type: '"OWNER" or "DISPONENT" (USER). Only owners can request loans / cards.' + + district: + description: Demographic and economic data for each Czech district. The A-columns are infamous — most analysis questions referencing demographics use these. + columns: + district_id: Surrogate key. + A2: District name. + A3: Region (a grouping of districts). + A4: Number of inhabitants. + A5: Number of municipalities with population < 499. + A6: Number of municipalities with population 500-1999. + A7: Number of municipalities with population 2000-9999. + A8: Number of municipalities with population > 10000. + A9: Number of cities. + A10: Ratio of urban inhabitants (percent). + A11: Average salary (in CZK). + A12: Unemployment rate in 1995 (percent). NULL means missing data. + A13: Unemployment rate in 1996 (percent). + A14: Number of entrepreneurs per 1000 inhabitants. + A15: Number of crimes committed in 1995. NULL means missing data. + A16: Number of crimes committed in 1996. + + loan: + description: One row per loan granted to an account. + columns: + loan_id: Surrogate key. + account_id: The account the loan was granted to. + date: Loan grant date (YYMMDD as integer). + amount: Total amount of the loan (CZK). + duration: Loan duration in months. + payments: Monthly payment (CZK). + status: Loan status code. "A" = contract finished, no problems. "B" = contract finished, loan not paid (DEFAULT). "C" = running contract, payments on time. "D" = running contract, client in debt (AT RISK). Defaults are A,B finished or B,D paid-flag attached. + + card: + description: One row per credit/debit card issued to a disposition. + columns: + card_id: Surrogate key. + disp_id: The disposition (and via it, the account) the card belongs to. + type: '"junior", "classic", or "gold".' + issued: Date the card was issued. + + trans: + description: Transaction history. Largest table by far (~1M rows). Every credit, debit, transfer. + columns: + trans_id: Surrogate key. + account_id: The account the transaction belongs to. + date: Transaction date (YYMMDD as integer). + type: '"PRIJEM" (credit) or "VYDAJ" (debit).' + operation: Mode of operation, e.g. "VKLAD" (cash credit), "VYBER" (cash withdrawal), "PREVOD Z UCTU" (transfer from another bank), etc. + amount: Amount (CZK). + balance: Account balance after the transaction. + k_symbol: Characterisation of the payment. "POJISTNE" = insurance, "SLUZBY" = household, "UROK" = interest credited, "SANKC. UROK" = penalty interest, "SIPO" = household payment, "DUCHOD" = old-age pension, "UVER" = loan payment. + bank: Counterparty bank code (only set for inter-bank transfers). + account: Counterparty account number (only set for inter-bank transfers). + + "order": + description: Standing payment orders (recurring debits). Note the table name collides with a SQL keyword and must be quoted. + columns: + order_id: Surrogate key. + account_id: The originating account. + bank_to: Recipient bank. + account_to: Recipient account. + amount: Order amount (CZK). + k_symbol: Purpose, same vocabulary as trans.k_symbol. diff --git a/api/evals/__init__.py b/api/evals/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/api/evals/run_evals.py b/api/evals/run_evals.py new file mode 100644 index 0000000..d0cb0d6 --- /dev/null +++ b/api/evals/run_evals.py @@ -0,0 +1,11 @@ +"""BIRD `financial` golden-set runner. + +Stub — the eval harness lands once the Plan + Analyses + Tools are in place. +""" + +def main() -> None: + raise SystemExit("api/evals/run_evals.py is a stub; agent stack not yet built") + + +if __name__ == "__main__": + main() diff --git a/api/langfuse_client.py b/api/langfuse_client.py new file mode 100644 index 0000000..dacc936 --- /dev/null +++ b/api/langfuse_client.py @@ -0,0 +1,87 @@ +"""Langfuse SDK wrapper — single source of truth for the client and spans. + +Used by every layer (tools, analyses, plan, runtime, evals), which is why it +lives at the api/ root rather than under any one layer. + +Gracefully returns no-op spans when Langfuse keys aren't configured, so local +dev without Langfuse still works. +""" + +from __future__ import annotations + +import logging +from contextlib import contextmanager +from typing import Any, Iterator + +from api.config import get_settings + +logger = logging.getLogger("nvi.langfuse") + +_client: Any = None + + +def get_client() -> Any | None: + global _client + if _client is not None: + return _client + s = get_settings() + if not s.langfuse_public_key or not s.langfuse_secret_key: + return None + try: + from langfuse import Langfuse + _client = Langfuse( + public_key=s.langfuse_public_key, + secret_key=s.langfuse_secret_key, + host=s.langfuse_host, + ) + return _client + except Exception as e: + logger.warning("langfuse init failed: %s", e) + return None + + +class _NullSpan: + """No-op stand-in when Langfuse isn't configured.""" + def update(self, **_: Any) -> None: ... + def end(self, **_: Any) -> None: ... + def __enter__(self) -> "_NullSpan": return self + def __exit__(self, *_: Any) -> None: ... + + +@contextmanager +def span( + name: str, + *, + as_type: str = "span", + input: Any = None, + metadata: dict[str, Any] | None = None, +) -> Iterator[Any]: + """Open a Langfuse observation; yields the span object or a no-op. + + Exceptions raised by the *body* of the `with` block propagate up + untouched. Only Langfuse's own setup failures fall through to a no-op + span — we never want to mask a real error. + """ + lf = get_client() + if lf is None: + yield _NullSpan() + return + try: + cm = lf.start_as_current_observation( + name=name, as_type=as_type, input=input, metadata=metadata or {} + ) + except Exception as e: + logger.warning("langfuse setup for %s failed: %s", name, e) + yield _NullSpan() + return + with cm as s: + yield s + + +def flush() -> None: + lf = get_client() + if lf is not None: + try: + lf.flush() + except Exception: + pass diff --git a/api/llm.py b/api/llm.py new file mode 100644 index 0000000..3c91d33 --- /dev/null +++ b/api/llm.py @@ -0,0 +1,91 @@ +"""LLM client — single `chat()` entry point that dispatches to the active provider. + +Provider selection is via `settings.llm_provider`. Supported: +- "groq" — hits Groq via the openai SDK (groq_base_url + groq_api_key). +- "anthropic" — native Anthropic SDK. +- "openai" — openai SDK with the standard base_url. + +The Anthropic and OpenAI/Groq clients are cached so HTTP connection pools +are reused across calls. +""" + +from __future__ import annotations + +from functools import lru_cache +from typing import Callable + +from anthropic import Anthropic +from openai import OpenAI + +from api.config import get_settings + + +# ── Provider clients (cached) ── + +@lru_cache(maxsize=1) +def _anthropic() -> Anthropic: + return Anthropic(api_key=get_settings().anthropic_api_key) + + +@lru_cache(maxsize=1) +def _groq() -> OpenAI: + s = get_settings() + return OpenAI(api_key=s.groq_api_key, base_url=s.groq_base_url) + + +@lru_cache(maxsize=1) +def _openai() -> OpenAI: + s = get_settings() + return OpenAI(api_key=s.openai_api_key, base_url=s.openai_base_url) + + +# ── Per-provider chat impls ── + +def _chat_anthropic(system: str, user: str, max_tokens: int) -> str: + s = get_settings() + msg = _anthropic().messages.create( + model=s.anthropic_model, + max_tokens=max_tokens, + system=system, + messages=[{"role": "user", "content": user}], + ) + return "".join(b.text for b in msg.content if getattr(b, "type", None) == "text").strip() + + +def _chat_openai_compat(client: OpenAI, model: str, system: str, user: str, max_tokens: int) -> str: + resp = client.chat.completions.create( + model=model, + max_tokens=max_tokens, + messages=[ + {"role": "system", "content": system}, + {"role": "user", "content": user}, + ], + ) + return (resp.choices[0].message.content or "").strip() + + +def _chat_groq(system: str, user: str, max_tokens: int) -> str: + return _chat_openai_compat(_groq(), get_settings().groq_model, system, user, max_tokens) + + +def _chat_openai(system: str, user: str, max_tokens: int) -> str: + return _chat_openai_compat(_openai(), get_settings().openai_model, system, user, max_tokens) + + +_PROVIDERS: dict[str, Callable[[str, str, int], str]] = { + "groq": _chat_groq, + "anthropic": _chat_anthropic, + "openai": _chat_openai, +} + + +# ── Public surface ── + +def chat(*, system: str, user: str, max_tokens: int = 1024) -> str: + provider = get_settings().llm_provider.lower() + impl = _PROVIDERS.get(provider) + if impl is None: + raise ValueError( + f"unknown llm_provider {provider!r}; supported: {sorted(_PROVIDERS)}" + ) + return impl(system, user, max_tokens) diff --git a/api/main.py b/api/main.py index bb7118f..d9328a6 100644 --- a/api/main.py +++ b/api/main.py @@ -1,14 +1,14 @@ """FastAPI entry point for nvi. Endpoints: - GET /healthz — liveness / readiness probe + GET /healthz — liveness probe POST /ask — submit a question, returns run_id - GET /runs/{run_id}/stream — SSE stream of node-state events + GET /runs/{run_id}/stream — SSE stream of run events GET /runs/{run_id} — final state snapshot - -Stubs only at this point — the Plan/Analyses/Tools/runtime are next. """ +from __future__ import annotations + import asyncio import logging import uuid @@ -20,12 +20,15 @@ from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import StreamingResponse from pydantic import BaseModel +from api.runtime import events +from api.runtime.runner import run as run_graph + logger = logging.getLogger("nvi") logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s") -# ── In-memory run store (single-process; replace with redis if we ever scale) ── - +# In-memory run store. Each entry holds final-state + lifecycle metadata; the +# event queue is owned by api.runtime.events. runs: dict[str, dict] = {} RUN_TTL_SECONDS = 3600 RUN_CLEANUP_INTERVAL = 300 @@ -52,7 +55,6 @@ async def lifespan(_app: FastAPI): app = FastAPI(title="nvi", lifespan=lifespan) - app.add_middleware( CORSMiddleware, allow_origins=["*"], @@ -66,8 +68,6 @@ async def healthz(): return {"status": "ok", "runs_in_memory": len(runs)} -# ── Ask ── - class AskRequest(BaseModel): question: str @@ -77,14 +77,31 @@ async def ask(req: AskRequest): run_id = uuid.uuid4().hex[:8] now = datetime.now(timezone.utc) runs[run_id] = { - "status": "pending", + "status": "running", "question": req.question, "created_at": now.timestamp(), - "events": [], } - # Real runtime invocation lands here once it exists. + events.open_run(run_id) + + async def _drive(): + try: + final = await run_graph(run_id, req.question) + runs[run_id] = { + **runs[run_id], + "status": "error" if final.get("error") else "completed", + "plan_rationale": final.get("plan_rationale"), + "plan_steps": final.get("plan_steps", []), + "findings": final.get("findings", []), + "answer": final.get("answer", ""), + "error": final.get("error"), + } + except Exception as e: + logger.exception("run %s failed at top level", run_id) + runs[run_id] = {**runs[run_id], "status": "error", "error": str(e)} + + asyncio.create_task(_drive()) logger.info("ask_received run_id=%s q=%r", run_id, req.question) - return {"run_id": run_id, "status": "pending"} + return {"run_id": run_id, "status": "running"} @app.get("/runs/{run_id}") @@ -98,11 +115,4 @@ async def get_run(run_id: str): async def stream_run(run_id: str): if run_id not in runs: raise HTTPException(404, detail=f"Run {run_id} not found") - - async def event_stream(): - # Placeholder: real implementation will tail the runtime's event queue. - yield f"event: hello\ndata: {{\"run_id\": \"{run_id}\"}}\n\n" - await asyncio.sleep(0.1) - yield "event: end\ndata: {}\n\n" - - return StreamingResponse(event_stream(), media_type="text/event-stream") + return StreamingResponse(events.stream(run_id), media_type="text/event-stream") diff --git a/api/plan/__init__.py b/api/plan/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/api/plan/planner.py b/api/plan/planner.py new file mode 100644 index 0000000..43864a0 --- /dev/null +++ b/api/plan/planner.py @@ -0,0 +1,52 @@ +"""Plan layer — one LLM call that selects Analyses for a question. + +Public surface is framework-free. Returns a `Plan` that the runtime iterates. +""" + +from __future__ import annotations + +import json +import logging +import re +from typing import Any + +from api import langfuse_client as lf +from api.analyses.registry import catalog +from api.llm import chat +from api.plan.types import Plan +from api.prompts import load, render +from api.tools.schema import load_schema + +logger = logging.getLogger("nvi.plan.planner") + + +def plan(question: str) -> Plan: + schema = load_schema() + user = render( + "planner.user", + question=question, + table_names=", ".join(schema.table_names()), + metrics_block=schema.render_metrics(), + catalog=json.dumps(catalog(), indent=2), + ) + + with lf.span("plan", as_type="generation", input={"question": question}) as span: + result = Plan.from_dict( + _parse(chat(system=load("planner.system"), user=user, max_tokens=1024)) + ) + span.update(output={ + "rationale": result.rationale, + "steps": [s.to_dict() for s in result.steps], + }) + if not result.steps: + raise ValueError("planner returned empty plan") + return result + + +def _parse(text: str) -> dict[str, Any]: + m = re.search(r"```(?:json)?\s*(\{.*\})\s*```", text, re.DOTALL) + raw = m.group(1) if m else text + start, end = raw.find("{"), raw.rfind("}") + if start < 0 or end <= start: + raise ValueError(f"planner returned no JSON: {text[:200]!r}") + return json.loads(raw[start:end + 1]) diff --git a/api/plan/types.py b/api/plan/types.py new file mode 100644 index 0000000..f082844 --- /dev/null +++ b/api/plan/types.py @@ -0,0 +1,37 @@ +"""Public data shapes for the Plan layer.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + + +@dataclass +class PlanStep: + analysis: str + args: dict[str, Any] = field(default_factory=dict) + why: str = "" + + @classmethod + def from_dict(cls, raw: dict[str, Any]) -> "PlanStep": + return cls( + analysis=raw["analysis"], + args=raw.get("args") or {}, + why=raw.get("why", ""), + ) + + def to_dict(self) -> dict[str, Any]: + return {"analysis": self.analysis, "args": self.args, "why": self.why} + + +@dataclass +class Plan: + rationale: str + steps: list[PlanStep] + + @classmethod + def from_dict(cls, raw: dict[str, Any]) -> "Plan": + return cls( + rationale=raw.get("rationale", ""), + steps=[PlanStep.from_dict(s) for s in raw.get("steps", [])], + ) diff --git a/api/prompts/__init__.py b/api/prompts/__init__.py new file mode 100644 index 0000000..7d623e1 --- /dev/null +++ b/api/prompts/__init__.py @@ -0,0 +1,31 @@ +"""Prompt loader + renderer. + +Every prompt — system or user — lives as a `.txt` file in this directory. +Two ways to consume them: + +- `load(name)` returns the file's text unchanged. Use for system prompts + (and any user prompt that has no placeholders). +- `render(name, **vars)` loads then runs `str.format(**vars)` on it. Use + for user prompts that interpolate question/SQL/rows/findings/etc. + Placeholders are standard Python format syntax: `{name}`. Literal braces + must be doubled: `{{` / `}}`. +""" + +from __future__ import annotations + +from functools import lru_cache +from pathlib import Path +from typing import Any + +PROMPTS_DIR = Path(__file__).resolve().parent + + +@lru_cache(maxsize=None) +def load(name: str) -> str: + """Read the named prompt file (no substitution).""" + return (PROMPTS_DIR / f"{name}.txt").read_text(encoding="utf-8").strip() + + +def render(name: str, /, **vars: Any) -> str: + """Load a prompt and substitute `{placeholder}` values from kwargs.""" + return load(name).format(**vars) diff --git a/api/prompts/compare_periods.interpret.txt b/api/prompts/compare_periods.interpret.txt new file mode 100644 index 0000000..1624bf5 --- /dev/null +++ b/api/prompts/compare_periods.interpret.txt @@ -0,0 +1,5 @@ +Task: compare two SQL result sets representing the same metric across two periods. + +Output: two to four sentences. Lead with the most material difference and cite specific numbers from both periods (e.g. "61 in 1995 → 87 in 1996, +43%"). No preamble. + +If the rows look essentially unchanged, say so plainly. If one period is empty, call that out. diff --git a/api/prompts/compare_periods.interpret.user.txt b/api/prompts/compare_periods.interpret.user.txt new file mode 100644 index 0000000..dc37f8e --- /dev/null +++ b/api/prompts/compare_periods.interpret.user.txt @@ -0,0 +1,7 @@ +Question: {question} + +Period A ({label_a}): +{rows_a} + +Period B ({label_b}): +{rows_b} diff --git a/api/prompts/compare_periods.pair.txt b/api/prompts/compare_periods.pair.txt new file mode 100644 index 0000000..f0ef66f --- /dev/null +++ b/api/prompts/compare_periods.pair.txt @@ -0,0 +1,16 @@ +Task: build two parallel Postgres SELECT queries that compute the same metric across two different time windows, so an analyst can compare them. + +Output: a JSON object with exactly this shape, in one fenced ```json block: +{ + "a": {"label": "", "sql": ""} +} + +Constraints: +- The two queries must return result sets with the same columns and shape — otherwise the diff is meaningless. +- Dates are stored as YYMMDD integers; convert with TO_DATE(LPAD(date::text, 6, '0'), 'YYMMDD') when filtering by date. +- Quote `"order"` if you reference it. +- search_path is `financial`; don't qualify table names with the schema. +- Read-only. + +If either period is given as "(infer)", pick a sensible window from the question and reflect it in the label. diff --git a/api/prompts/compare_periods.pair.user.txt b/api/prompts/compare_periods.pair.user.txt new file mode 100644 index 0000000..93b08df --- /dev/null +++ b/api/prompts/compare_periods.pair.user.txt @@ -0,0 +1,10 @@ +Question: {question} + +Period A: {period_a} +Period B: {period_b} + +Schema: +{schema_block} + +Metrics: +{metrics_block} diff --git a/api/prompts/direct_answer.interpret.txt b/api/prompts/direct_answer.interpret.txt new file mode 100644 index 0000000..c59f268 --- /dev/null +++ b/api/prompts/direct_answer.interpret.txt @@ -0,0 +1,5 @@ +Task: answer the analyst's question given the SQL that was run and the result rows. + +Output: one to three sentences that directly answer the question, citing specific numbers from the rows. No preamble ("Here is…", "Based on the results…"), no SQL block, no enumeration of every row. + +If the result is empty or doesn't actually answer the question as asked, say so plainly and stop. diff --git a/api/prompts/direct_answer.interpret.user.txt b/api/prompts/direct_answer.interpret.user.txt new file mode 100644 index 0000000..a8955a9 --- /dev/null +++ b/api/prompts/direct_answer.interpret.user.txt @@ -0,0 +1,7 @@ +Question: {question} + +SQL: +{sql} + +Rows (up to 20): +{rows} diff --git a/api/prompts/planner.system.txt b/api/prompts/planner.system.txt new file mode 100644 index 0000000..0a9443d --- /dev/null +++ b/api/prompts/planner.system.txt @@ -0,0 +1,17 @@ +Task: pick which Analyses to run to answer the analyst's question. + +Given the question, the list of warehouse tables, the metric catalog, and the catalog of available Analyses (each with a name, description, and args schema), choose the smallest set of Analyses whose combined output will produce a defensible answer. + +Output: a JSON object with exactly this shape, in one fenced ```json block: +{ + "rationale": "", + "steps": [ + {"analysis": "", "args": { ... }, "why": ""} + ] +} + +Rules: +- Only use analyses that appear in the catalog. If none fits the question well, pick `direct_answer`. +- The `args` keys must match the analysis's args_schema; additional keys are ignored, missing optional ones are fine. +- Most questions are single-step. Only chain analyses when the second genuinely needs the first's output. +- Comparing two periods is ONE `compare_periods` step, not two `direct_answer` steps. diff --git a/api/prompts/planner.user.txt b/api/prompts/planner.user.txt new file mode 100644 index 0000000..e2982ee --- /dev/null +++ b/api/prompts/planner.user.txt @@ -0,0 +1,9 @@ +Question: {question} + +Warehouse tables: {table_names} + +Metric catalog: +{metrics_block} + +Analysis catalog: +{catalog} diff --git a/api/prompts/synthesize.system.txt b/api/prompts/synthesize.system.txt new file mode 100644 index 0000000..4a658e7 --- /dev/null +++ b/api/prompts/synthesize.system.txt @@ -0,0 +1,7 @@ +Task: write a final answer for the analyst given the original question and a list of Analysis findings. + +Each finding has an Analysis name, its one-line interpretation, and the SQL it ran. + +Output: a single coherent paragraph of two to five sentences that directly answers the question, weaving together the relevant findings. Cite specific numbers. Don't list the SQL, don't enumerate the Analyses by name, don't add any "Summary:" header or preamble. + +If every finding errored, say plainly that the question couldn't be answered and briefly mention why (e.g. "the planner picked an Analysis that couldn't resolve the metric"). diff --git a/api/prompts/synthesize.user.txt b/api/prompts/synthesize.user.txt new file mode 100644 index 0000000..c4171b2 --- /dev/null +++ b/api/prompts/synthesize.user.txt @@ -0,0 +1,4 @@ +Question: {question} + +Findings: +{findings_block} diff --git a/api/prompts/text_to_sql.system.txt b/api/prompts/text_to_sql.system.txt new file mode 100644 index 0000000..f9ba9dc --- /dev/null +++ b/api/prompts/text_to_sql.system.txt @@ -0,0 +1,15 @@ +Task: translate the analyst's question into a single Postgres SELECT (or WITH … SELECT) against the BIRD `financial` schema (PKDD'99 Czech bank). + +Output: exactly one fenced ```sql block, nothing else — no prose, no second block, no trailing semicolon required. + +Constraints: +- search_path is set to `financial`; don't qualify table names with the schema. +- Quote identifiers that collide with SQL keywords (notably `"order"`). +- Dates in the source are YYMMDD integers (e.g. 950315 → 1995-03-15). Convert with TO_DATE(LPAD(date::text, 6, '0'), 'YYMMDD') whenever you need a real date for filtering, comparison, or display. +- Loan `status` values: 'A' finished OK, 'B' finished defaulted, 'C' running OK, 'D' running in debt. +- Transaction `type` values: 'PRIJEM' credit, 'VYDAJ' debit. +- Read-only: never emit DDL or DML. + +When the question maps to a metric in the catalog you're given, copy the metric's SQL fragment verbatim — don't paraphrase it. + +If the question is ambiguous, pick the most defensible interpretation and proceed; don't ask for clarification. diff --git a/api/prompts/text_to_sql.user.txt b/api/prompts/text_to_sql.user.txt new file mode 100644 index 0000000..2aa5244 --- /dev/null +++ b/api/prompts/text_to_sql.user.txt @@ -0,0 +1,10 @@ +Question: +{question} + +Schema (only the tables that look relevant; you may use any of these): +{schema_block} + +Metric catalog (copy the SQL verbatim when the question maps to one): +{metrics_block} + +Return only the SQL in a ```sql fenced block.{retry_hint} diff --git a/api/runtime/__init__.py b/api/runtime/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/api/runtime/context.py b/api/runtime/context.py new file mode 100644 index 0000000..37dd23e --- /dev/null +++ b/api/runtime/context.py @@ -0,0 +1,12 @@ +"""Per-run context — exposes the active run_id to code deep in the call stack +without threading it through every signature. + +Set by the runtime at run entry; read by Analyses and helper functions so they +can publish trace events without needing the run_id passed in explicitly. +""" + +from __future__ import annotations + +from contextvars import ContextVar + +current_run_id: ContextVar[str | None] = ContextVar("nvi.current_run_id", default=None) diff --git a/api/runtime/events.py b/api/runtime/events.py new file mode 100644 index 0000000..9ce1fd7 --- /dev/null +++ b/api/runtime/events.py @@ -0,0 +1,133 @@ +"""Per-run event bus + factories + SSE formatting. + +Tiny in-memory queue keyed by run_id. The runtime publishes on each +transition; the FastAPI SSE endpoint subscribes and forwards. + +Event payloads are constructed via the module-level factory functions below +so the dict shape lives in one place and is grep-able. +""" + +from __future__ import annotations + +import asyncio +import json +from typing import Any + +from api.analyses.types import Finding +from api.runtime.context import current_run_id + +# run_id -> queue of events. Events are plain dicts. +_queues: dict[str, asyncio.Queue[dict[str, Any] | None]] = {} + + +# ── Queue management ── + +def open_run(run_id: str) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue() + _queues[run_id] = q + return q + + +async def publish(run_id: str, event: dict[str, Any]) -> None: + q = _queues.get(run_id) + if q is not None: + await q.put(event) + + +async def publish_current(event: dict[str, Any]) -> None: + """Publish to the run identified by the `current_run_id` contextvar. + No-op when called outside a run.""" + rid = current_run_id.get() + if rid is not None: + await publish(rid, event) + + +async def close(run_id: str) -> None: + q = _queues.get(run_id) + if q is not None: + await q.put(None) + + +def drop(run_id: str) -> None: + _queues.pop(run_id, None) + + +# ── SSE formatting + streaming ── + +def sse_format(event: dict[str, Any]) -> str: + kind = event.get("type", "message") + payload = json.dumps({k: v for k, v in event.items() if k != "type"}, default=str) + return f"event: {kind}\ndata: {payload}\n\n" + + +async def stream(run_id: str): + """Async generator producing SSE-formatted strings until the run ends.""" + q = _queues.get(run_id) + if q is None: + yield sse_format({"type": "error", "message": f"unknown run_id {run_id}"}) + return + try: + while True: + event = await q.get() + if event is None: + yield sse_format({"type": "end"}) + return + yield sse_format(event) + finally: + drop(run_id) + + +# ── Event factories ── +# Each returns the dict payload to pass into `publish()` / +# `publish_current()`. Names mirror the `type` field for grep-ability. + +def run_start(question: str) -> dict[str, Any]: + return {"type": "run_start", "question": question} + + +def run_end(*, answer: str | None = None, error: str | None = None) -> dict[str, Any]: + return {"type": "run_end", "answer": answer or "", "error": error} + + +def plan_ready(rationale: str, steps: list[dict[str, Any]]) -> dict[str, Any]: + return {"type": "plan_ready", "rationale": rationale, "steps": steps} + + +def node_update(node: str, update: Any) -> dict[str, Any]: + return { + "type": "node_update", + "node": node, + "keys": sorted(update.keys()) if isinstance(update, dict) else [], + } + + +def analysis_start(name: str, args: dict[str, Any], why: str) -> dict[str, Any]: + return {"type": "analysis_start", "analysis": name, "args": args, "why": why} + + +def analysis_end(name: str, finding: Finding) -> dict[str, Any]: + return { + "type": "analysis_end", + "analysis": name, + "summary": finding.summary, + "error": finding.error, + "row_count": len(finding.rows), + } + + +def synth_start() -> dict[str, Any]: + return {"type": "synth_start"} + + +def tool_call_start(tool: str, *, input: Any = None) -> dict[str, Any]: + return {"type": "tool_call_start", "tool": tool, "input": input} + + +def tool_call_end(tool: str, *, output: Any = None, error: str | None = None) -> dict[str, Any]: + return {"type": "tool_call_end", "tool": tool, "output": output, "error": error} + + +def llm_call(label: str, *, system_len: int, user_len: int) -> dict[str, Any]: + """Lightweight breadcrumb for an LLM call. Doesn't carry the prompt + contents (those go through Langfuse) — just the fact and rough size.""" + return {"type": "llm_call", "label": label, "system_chars": system_len, "user_chars": user_len} diff --git a/api/runtime/runner.py b/api/runtime/runner.py new file mode 100644 index 0000000..23875ba --- /dev/null +++ b/api/runtime/runner.py @@ -0,0 +1,139 @@ +"""langgraph wiring — the one place the framework appears. + +Graph: START → plan → execute → synthesize → END + +`execute` iterates Plan steps and dispatches each to its Analysis. Sequential +for now; swap to a fan-out with parallel nodes once we want to parallelise. +""" + +from __future__ import annotations + +import dataclasses +import logging +from typing import Any + +from langgraph.graph import END, START, StateGraph + +from api import langfuse_client as lf +from api.analyses.registry import REGISTRY as ANALYSES +from api.analyses.types import Finding +from api.llm import chat +from api.plan.planner import plan as run_planner +from api.prompts import load, render +from api.runtime import events +from api.runtime.context import current_run_id +from api.runtime.state import RunState + +logger = logging.getLogger("nvi.runtime") + + +def _plan_node(state: RunState) -> dict[str, Any]: + plan_obj = run_planner(state["question"]) + return { + "plan_rationale": plan_obj.rationale, + "plan_steps": [s.to_dict() for s in plan_obj.steps], + } + + +async def _execute_node(state: RunState) -> dict[str, Any]: + findings: list[dict[str, Any]] = [] + for step in state.get("plan_steps", []): + name = step["analysis"] + analysis = ANALYSES.get(name) + await events.publish( + state["run_id"], + events.analysis_start(name, step.get("args", {}), step.get("why", "")), + ) + if analysis is None: + f = Finding(analysis=name, summary="", error=f"unknown analysis {name!r}") + else: + try: + f = await analysis.run(step.get("args", {}), state["question"]) + except Exception as e: + logger.exception("analysis %s raised", name) + f = Finding(analysis=name, summary="", error=str(e)) + findings.append(dataclasses.asdict(f)) + await events.publish(state["run_id"], events.analysis_end(name, f)) + return {"findings": findings} + + +async def _synthesize_node(state: RunState) -> dict[str, Any]: + await events.publish(state["run_id"], events.synth_start()) + + findings = state.get("findings", []) + if not findings or all(f.get("error") for f in findings): + return { + "answer": "I couldn't answer this question — see the per-analysis errors above.", + "error": "all_analyses_failed", + } + + findings_block = "\n\n".join( + f"[{f['analysis']}]\n" + f"summary: {f.get('summary') or '(none)'}\n" + f"error: {f.get('error') or '(none)'}\n" + f"sql: {f.get('sql')}" + for f in findings + ) + + with lf.span("synthesize", as_type="generation", input={"findings": len(findings)}) as span: + answer = chat( + system=load("synthesize.system"), + user=render( + "synthesize.user", + question=state["question"], + findings_block=findings_block, + ), + max_tokens=512, + ) + span.update(output={"answer": answer}) + return {"answer": answer} + + +def build_graph(): + g: StateGraph = StateGraph(RunState) + g.add_node("plan", _plan_node) + g.add_node("execute", _execute_node) + g.add_node("synthesize", _synthesize_node) + g.add_edge(START, "plan") + g.add_edge("plan", "execute") + g.add_edge("execute", "synthesize") + g.add_edge("synthesize", END) + return g.compile() + + +GRAPH = build_graph() + + +async def run(run_id: str, question: str) -> RunState: + """Drive the graph end-to-end, publishing SSE events along the way.""" + initial: RunState = {"run_id": run_id, "question": question} + final: RunState = initial + token = current_run_id.set(run_id) + try: + await events.publish(run_id, events.run_start(question)) + async for event in GRAPH.astream(initial, stream_mode="updates"): + for node, update in event.items(): + await events.publish(run_id, events.node_update(node, update)) + if isinstance(update, dict): + final = {**final, **update} + if node == "plan" and isinstance(update, dict): + await events.publish( + run_id, + events.plan_ready( + update.get("plan_rationale", ""), + update.get("plan_steps", []), + ), + ) + await events.publish( + run_id, + events.run_end(answer=final.get("answer", ""), error=final.get("error")), + ) + except Exception as e: + logger.exception("run %s failed", run_id) + await events.publish(run_id, events.run_end(error=str(e))) + final = {**final, "error": str(e)} + finally: + current_run_id.reset(token) + await events.close(run_id) + lf.flush() + return final diff --git a/api/runtime/state.py b/api/runtime/state.py new file mode 100644 index 0000000..d702041 --- /dev/null +++ b/api/runtime/state.py @@ -0,0 +1,20 @@ +"""Shared state passed between runtime nodes. + +This is the one type langgraph sees. Domain code in api/plan, api/analyses, +api/tools never imports it — they take/return their own dataclasses. +""" + +from __future__ import annotations + +from typing import Annotated, Any, TypedDict +from operator import add + + +class RunState(TypedDict, total=False): + run_id: str + question: str + plan_rationale: str + plan_steps: list[dict[str, Any]] # serialised PlanStep + findings: Annotated[list[dict[str, Any]], add] # serialised Finding; concatenated by langgraph + answer: str + error: str diff --git a/api/tools/__init__.py b/api/tools/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/api/tools/db.py b/api/tools/db.py new file mode 100644 index 0000000..0458b3a --- /dev/null +++ b/api/tools/db.py @@ -0,0 +1,39 @@ +"""SQLAlchemy engine for the warehouse. + +Per-connection defaults (search_path, statement_timeout) are baked into the +engine via libpq's `options` connect_arg so we never SET them at the SQL +layer. Read-only mode for query execution is requested via SQLAlchemy's +`execution_options(postgresql_readonly=True)`. +""" + +from __future__ import annotations + +from functools import lru_cache + +from sqlalchemy import create_engine +from sqlalchemy.engine import Engine + +from api.config import get_settings + +SCHEMA = "financial" +DEFAULT_STATEMENT_TIMEOUT_MS = 10_000 + + +@lru_cache(maxsize=1) +def get_engine() -> Engine: + url = get_settings().database_url + if url.startswith("postgresql://"): + url = url.replace("postgresql://", "postgresql+psycopg://", 1) + return create_engine( + url, + pool_pre_ping=True, + future=True, + # libpq -c options run before any SQL the app issues — no need to + # SET search_path / statement_timeout at execution time. + connect_args={ + "options": ( + f"-c search_path={SCHEMA},public " + f"-c statement_timeout={DEFAULT_STATEMENT_TIMEOUT_MS}" + ), + }, + ) diff --git a/api/tools/execute_sql.py b/api/tools/execute_sql.py new file mode 100644 index 0000000..9a8c636 --- /dev/null +++ b/api/tools/execute_sql.py @@ -0,0 +1,55 @@ +"""Execute read-only SQL against the financial warehouse. + +Defence in depth: +- Engine-level `search_path` and `statement_timeout` via libpq options in + `api.tools.db.get_engine` — no SQL-side SETs. +- `execution_options(postgresql_readonly=True, postgresql_deferrable=True)` + — SQLAlchemy emits `SET TRANSACTION READ ONLY DEFERRABLE` internally; we + don't compose a SQL string for it. +- Hard refusal of any statement that doesn't begin with SELECT or WITH. +- Row cap on the returned result set. +- No commit — read-only transactions roll back on context exit. + +The SQL passed in IS a string (it's the LLM's output); nothing on the Python +side concatenates SQL fragments around it. +""" + +from __future__ import annotations + +import logging + +from sqlalchemy import text + +from api import langfuse_client as lf +from api.tools.db import get_engine +from api.tools.types import QueryResult + +logger = logging.getLogger("nvi.tools.execute_sql") + +MAX_ROWS = 1000 + + +def execute_sql(sql: str) -> QueryResult: + sql_clean = sql.strip().rstrip(";").strip() + head = sql_clean.split(None, 1)[0].upper() if sql_clean else "" + if head not in {"SELECT", "WITH"}: + raise ValueError(f"refusing non-SELECT statement (starts with {head!r})") + + engine = get_engine() + with lf.span("execute_sql", input={"sql": sql_clean}) as span: + ro_engine = engine.execution_options( + postgresql_readonly=True, + postgresql_deferrable=True, + ) + with ro_engine.connect() as conn: + res = conn.execute(text(sql_clean)) + columns = list(res.keys()) + fetched = res.fetchmany(MAX_ROWS + 1) + + truncated = len(fetched) > MAX_ROWS + rows = [list(r) for r in fetched[:MAX_ROWS]] + result = QueryResult( + columns=columns, rows=rows, row_count=len(rows), truncated=truncated + ) + span.update(output={"row_count": result.row_count, "truncated": truncated}) + return result diff --git a/api/tools/schema.py b/api/tools/schema.py new file mode 100644 index 0000000..84df945 --- /dev/null +++ b/api/tools/schema.py @@ -0,0 +1,57 @@ +"""Warehouse schema retrieval. + +Physical layer via SQLAlchemy's Inspector against the active dataset's +Postgres schema; semantic layer (table/column descriptions, metric catalog) +from YAML files under `api/datasets//`. + +Cached on first call — the schema is stable for a process's lifetime. +Restart the api to pick up a dataset switch or YAML edits. +""" + +from __future__ import annotations + +from functools import lru_cache +from typing import Any + +from sqlalchemy import inspect + +from api.config import get_settings +from api.datasets import read_yaml +from api.tools.db import get_engine +from api.tools.types import Column, Metric, SchemaContext, Table + + +@lru_cache(maxsize=1) +def load_schema() -> SchemaContext: + dataset = get_settings().dataset + docs = read_yaml(dataset, "schema_docs.yaml") + metrics_yaml = read_yaml(dataset, "metrics.yaml") + + # Postgres schema name comes from the dataset id; the schema_docs file + # can override it, but normally they match. + schema_name: str = docs.get("schema", dataset) + table_descs: dict[str, dict[str, Any]] = docs.get("tables", {}) or {} + + insp = inspect(get_engine()) + tables: dict[str, Table] = { + name: Table( + name=name, + description=(desc := table_descs.get(name, {})).get("description"), + columns=[ + Column.from_inspector(c, (desc.get("columns") or {}).get(c["name"])) + for c in insp.get_columns(name, schema=schema_name) + ], + ) + for name in insp.get_table_names(schema=schema_name) + } + + metrics: dict[str, Metric] = { + name: Metric.from_spec(name, spec) + for name, spec in (metrics_yaml.get("metrics", {}) or {}).items() + } + + return SchemaContext(schema=schema_name, tables=tables, metrics=metrics) + + +def get_metric(name: str) -> Metric | None: + return load_schema().metrics.get(name) diff --git a/api/tools/text_to_sql.py b/api/tools/text_to_sql.py new file mode 100644 index 0000000..ad01ac4 --- /dev/null +++ b/api/tools/text_to_sql.py @@ -0,0 +1,70 @@ +"""LLM-driven NL → SQL with sqlglot validation and one retry.""" + +from __future__ import annotations + +import logging +import re + +import sqlglot + +from api import langfuse_client as lf +from api.llm import chat +from api.prompts import load, render +from api.tools.schema import load_schema +from api.tools.types import T2SResult + +logger = logging.getLogger("nvi.tools.text_to_sql") + + +def text_to_sql(question: str, *, hint_tables: list[str] | None = None) -> T2SResult: + schema = load_schema() + tables = hint_tables or schema.table_names() + system = load("text_to_sql.system") + + def _user(retry_hint: str = "") -> str: + return render( + "text_to_sql.user", + question=question, + schema_block=schema.render_tables(tables), + metrics_block=schema.render_metrics(), + retry_hint=retry_hint, + ) + + with lf.span( + "text_to_sql", + as_type="generation", + input={"question": question, "hint_tables": hint_tables}, + ) as span: + sql = _extract_sql(chat(system=system, user=_user())) + try: + _validate(sql) + except Exception as e: + logger.info("t2s first attempt invalid (%s); retrying once", e) + hint = f"\n\nPrevious attempt failed parsing: {e}. Fix and return only the SQL." + sql = _extract_sql(chat(system=system, user=_user(hint))) + _validate(sql) + + result = T2SResult(sql=sql, used_tables=_extract_tables(sql)) + span.update(output={"sql": sql, "used_tables": result.used_tables}) + return result + + +def _extract_sql(text: str) -> str: + m = re.search(r"```sql\s*(.*?)```", text, re.DOTALL | re.IGNORECASE) + if m: + return m.group(1).strip().rstrip(";").strip() + return text.strip().rstrip(";").strip() + + +def _validate(sql: str) -> None: + parsed = sqlglot.parse_one(sql, dialect="postgres") + if parsed is None: + raise ValueError("sqlglot returned no parse tree") + + +def _extract_tables(sql: str) -> list[str]: + try: + parsed = sqlglot.parse_one(sql, dialect="postgres") + return sorted({t.name for t in parsed.find_all(sqlglot.exp.Table)}) + except Exception: + return [] diff --git a/api/tools/types.py b/api/tools/types.py new file mode 100644 index 0000000..d76790f --- /dev/null +++ b/api/tools/types.py @@ -0,0 +1,128 @@ +"""Public data shapes for the Tools layer. + +Behavior modules in api/tools/ import from here. Keeping types in one file +makes the layer's surface visible at a glance. + +Secondary constructors (`from_inspector`, `from_spec`, …) live as +classmethods so call sites read as one line instead of multi-line kwarg +blocks. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + + +# ── Schema model ── + +@dataclass +class Column: + name: str + sql_type: str + nullable: bool + description: str | None = None + + @classmethod + def from_inspector(cls, info: dict[str, Any], description: str | None = None) -> "Column": + """Build from a SQLAlchemy Inspector column dict.""" + return cls( + name=info["name"], + sql_type=str(info["type"]).upper(), + nullable=bool(info["nullable"]), + description=description, + ) + + +@dataclass +class Table: + name: str + description: str | None + columns: list[Column] + + +@dataclass +class Metric: + name: str + description: str + sql: str + from_table: str + filter: str | None + unit: str | None + + @classmethod + def from_spec(cls, name: str, spec: dict[str, Any]) -> "Metric": + """Build from a metrics.yaml entry.""" + return cls( + name=name, + description=spec.get("description", ""), + sql=spec["sql"], + from_table=spec["from_table"], + filter=spec.get("filter"), + unit=spec.get("unit"), + ) + + +@dataclass +class SchemaContext: + schema: str + tables: dict[str, Table] + metrics: dict[str, Metric] + + def table_names(self) -> list[str]: + return sorted(self.tables) + + def metric_names(self) -> list[str]: + return sorted(self.metrics) + + def render_tables(self, names: list[str] | None = None) -> str: + """CREATE TABLE-like rendering for prompt context.""" + sel = [self.tables[n] for n in (names or self.table_names()) if n in self.tables] + out: list[str] = [] + for t in sel: + header = f"-- {t.description}\n" if t.description else "" + cols: list[str] = [] + for c in t.columns: + line = f' "{c.name}" {c.sql_type}' + if not c.nullable: + line += " NOT NULL" + if c.description: + line += f" -- {c.description}" + cols.append(line) + out.append(header + f'CREATE TABLE "{t.name}" (\n' + ",\n".join(cols) + "\n);") + return "\n\n".join(out) + + def render_metrics(self) -> str: + if not self.metrics: + return "(no metrics defined)" + lines: list[str] = [] + for m in self.metrics.values(): + lines.append( + f"- {m.name} ({m.unit or 'unitless'}): {m.description}\n" + f" sql: {m.sql}\n" + f" from: {m.from_table}" + + (f"\n filter: {m.filter}" if m.filter else "") + ) + return "\n".join(lines) + + +# ── Query execution ── + +@dataclass +class QueryResult: + columns: list[str] + rows: list[list[Any]] + row_count: int + truncated: bool + + def as_dicts(self) -> list[dict[str, Any]]: + return [dict(zip(self.columns, r)) for r in self.rows] + + +# ── Text-to-SQL output ── + +@dataclass +class T2SResult: + sql: str + used_tables: list[str] + explanation: str | None = None diff --git a/ctrl/Dockerfile.api b/ctrl/Dockerfile.api index 9eb9948..c3c86ae 100644 --- a/ctrl/Dockerfile.api +++ b/ctrl/Dockerfile.api @@ -8,5 +8,6 @@ COPY pyproject.toml uv.lock* ./ RUN uv sync --no-dev --no-install-project --frozen 2>/dev/null || uv sync --no-dev --no-install-project COPY api/ api/ +COPY ctrl/seed/ seed/ CMD ["uv", "run", "uvicorn", "api.main:app", "--host", "0.0.0.0", "--port", "8000", "--reload"] diff --git a/ctrl/Tiltfile b/ctrl/Tiltfile index 305d331..522c2ec 100644 --- a/ctrl/Tiltfile +++ b/ctrl/Tiltfile @@ -25,23 +25,23 @@ docker_build( 'nvi-api', context='..', dockerfile='Dockerfile.api', - ignore=['.git', 'def', 'ui', '.claude', 'tests', '.venv', 'node_modules', '.data', 'docs'], + ignore=[ + '.git', 'def', 'ui', '.claude', 'tests', '.venv', 'node_modules', + '.data', 'docs', 'ctrl/seed/data', 'api/evals/data', + ], live_update=[ sync('../api', '/app/api'), + sync('./seed', '/app/seed'), ], ) -# Vue UI — context is project root so the framework symlink resolves +# Vue UI — context is project root so the framework dir is in scope. +# No live_update: the image is multi-stage (vite build → static dist served +# by nginx), so source changes require a full image rebuild. docker_build( 'nvi-ui', context='..', dockerfile='Dockerfile.ui', - live_update=[ - sync('../ui/app/src', '/ui/app/src'), - sync('../ui/app/index.html', '/ui/app/index.html'), - sync('../ui/app/vite.config.ts', '/ui/app/vite.config.ts'), - sync('../ui/framework/src', '/ui/framework/src'), - ], ) # Docs — nginx serving pre-rendered HTML + Graphviz SVGs diff --git a/ctrl/docker-compose.yml b/ctrl/docker-compose.yml index d1625c2..7bcbdd1 100644 --- a/ctrl/docker-compose.yml +++ b/ctrl/docker-compose.yml @@ -32,6 +32,11 @@ services: environment: *common-env ports: - "8000:8000" + volumes: + # Bind-mount data dirs so `make seed-fetch` downloads on the host land + # in the container without rebuilding the image. + - ./seed/data:/app/seed/data + - ../api/evals/data:/app/api/evals/data depends_on: postgres: condition: service_healthy diff --git a/ctrl/k8s/base/configmap.yaml b/ctrl/k8s/base/configmap.yaml index 79a6591..f142d40 100644 --- a/ctrl/k8s/base/configmap.yaml +++ b/ctrl/k8s/base/configmap.yaml @@ -7,13 +7,19 @@ data: # Warehouse (in-cluster postgres) DATABASE_URL: "postgresql://nvi:nvi@postgres:5432/nvi" - # Langfuse (lng cluster, reached over WireGuard from the host). - # Override LANGFUSE_HOST to your lng WireGuard endpoint; keys come from - # the "nvi" project created inside the lng Langfuse UI. - LANGFUSE_HOST: "http://lng.local.ar:3000" - LANGFUSE_PUBLIC_KEY: "" - LANGFUSE_SECRET_KEY: "" + # Active dataset. Selects api/datasets//, the Postgres schema, and + # ctrl/seed/data/.sqlite. Restart the api to switch. + NVI_DATASET: "financial" + + # LLM. Supported providers: groq, anthropic, openai. + LLM_PROVIDER: "groq" + GROQ_API_KEY: "REDACTED_GROQ_API_KEY" + GROQ_MODEL: "llama-3.3-70b-versatile" - # LLM ANTHROPIC_API_KEY: "" ANTHROPIC_MODEL: "claude-sonnet-4-6" + + # Langfuse (nivii project in the lng cluster). + LANGFUSE_HOST: "http://lng.local.ar" + LANGFUSE_PUBLIC_KEY: "REDACTED_LANGFUSE_PUBLIC_KEY" + LANGFUSE_SECRET_KEY: "REDACTED_LANGFUSE_SECRET_KEY" diff --git a/ctrl/seed/__init__.py b/ctrl/seed/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/ctrl/seed/download.sh b/ctrl/seed/download.sh new file mode 100755 index 0000000..e4371d2 --- /dev/null +++ b/ctrl/seed/download.sh @@ -0,0 +1,44 @@ +#!/usr/bin/env bash +# Downloads BIRD financial SQLite into ctrl/seed/data/financial.sqlite and +# the golden eval set into api/evals/data/. Runs on the HOST (the api image +# deliberately doesn't include the data). Override BIRD_URL if upstream +# moves. +# +# Layout inside the outer dev.zip (as of the 2024-06-27 dump): +# dev_/ +# dev.sql, dev.json, dev_tables.json, dev_tied_append.json +# dev_databases.zip → financial/financial.sqlite (and 10 other DBs) +set -euo pipefail +cd "$(dirname "$0")" + +DEST="data/financial.sqlite" +EVAL_DIR="../../api/evals/data" +mkdir -p data "$EVAL_DIR" + +if [ -f "$DEST" ]; then + echo "already have $DEST ($(du -h "$DEST" | cut -f1))" + exit 0 +fi + +URL="${BIRD_URL:-https://bird-bench.oss-cn-beijing.aliyuncs.com/dev.zip}" +TMP="$(mktemp -d)" +trap 'rm -rf "$TMP"' EXIT +OUTER="$TMP/dev.zip" + +echo "fetching $URL (~330MB compressed; full dev set is ~3GB extracted)..." +curl -fL --progress-bar -o "$OUTER" "$URL" + +echo "extracting outer archive..." +unzip -q "$OUTER" -d "$TMP" +ROOT=$(find "$TMP" -maxdepth 1 -type d -name 'dev_*' | head -1) +test -n "$ROOT" || { echo "could not locate dev_* root inside outer zip" >&2; exit 1; } + +echo "extracting financial.sqlite from inner archive..." +unzip -q -j "$ROOT/dev_databases.zip" '*/financial/financial.sqlite' -d data/ + +echo "copying golden eval set to $EVAL_DIR/..." +cp "$ROOT/dev.json" "$ROOT/dev.sql" "$ROOT/dev_tables.json" "$EVAL_DIR/" + +echo "ready:" +echo " $DEST ($(du -h "$DEST" | cut -f1))" +echo " $(ls -1 "$EVAL_DIR/" | wc -l) files in $EVAL_DIR/" diff --git a/ctrl/seed/load_bird.py b/ctrl/seed/load_bird.py new file mode 100644 index 0000000..0b249d8 --- /dev/null +++ b/ctrl/seed/load_bird.py @@ -0,0 +1,147 @@ +"""Load a BIRD SQLite dataset into Postgres. + +Runs inside the api container (`make seed`). Reads which dataset to load +from `settings.dataset` (env var `NVI_DATASET`, default `financial`) and +expects the source SQLite at `ctrl/seed/data/.sqlite` — produced +on the host by `ctrl/seed/download.sh` and reaching the container via +Tilt live_update sync (kind) or the docker-compose bind mount. + +DDL is built from SQLAlchemy `MetaData` + `Table` + `Column` objects, not +string SQL. Bulk insert uses psycopg's COPY protocol via the engine's raw +connection — there's no SQLAlchemy equivalent, and INSERT for ~1M rows +would be orders of magnitude slower. Identifiers in the COPY statement +are rendered through SA's dialect-aware identifier preparer. +""" + +from __future__ import annotations + +import sqlite3 +import sys +from pathlib import Path +from typing import Any + +from sqlalchemy import ( + BigInteger, Boolean, Column, Date, DateTime, Float, Integer, LargeBinary, + MetaData, Numeric, SmallInteger, Table, Text, text, +) +from sqlalchemy.engine import Engine + +from api.config import get_settings +from api.tools.db import get_engine + +# SQLite is type-affinity, not strict — pick a SQLAlchemy type per affinity. +# Anything unrecognised falls through to Text, which is always safe. +TYPE_MAP: dict[str, Any] = { + "INT": BigInteger, + "INTEGER": BigInteger, + "TINYINT": SmallInteger, + "SMALLINT": SmallInteger, + "MEDIUMINT": Integer, + "BIGINT": BigInteger, + "REAL": Float, + "DOUBLE": Float, + "FLOAT": Float, + "NUMERIC": Numeric, + "DECIMAL": Numeric, + "BOOLEAN": Boolean, + "DATE": Date, + "DATETIME": DateTime, + "TIMESTAMP": DateTime, + "TEXT": Text, + "CHAR": Text, + "VARCHAR": Text, + "CLOB": Text, + "BLOB": LargeBinary, +} + + +def _sa_type(sqlite_type: str): + base = sqlite_type.upper().split("(")[0].strip() + return TYPE_MAP.get(base, Text) + + +def _list_tables(src: sqlite3.Connection) -> list[str]: + return [ + r[0] for r in src.execute( + "SELECT name FROM sqlite_master " + "WHERE type='table' AND name NOT LIKE 'sqlite_%' " + "ORDER BY name" + ).fetchall() + ] + + +def _discover_table(src: sqlite3.Connection, metadata: MetaData, + name: str, schema: str) -> Table: + cols = src.execute(f'PRAGMA table_info("{name}")').fetchall() + # PRAGMA returns (cid, name, type, notnull, dflt_value, pk). + sa_cols = [ + Column( + c[1], + _sa_type(c[2]), + nullable=not c[3], + primary_key=bool(c[5]), + ) + for c in cols + ] + return Table(name, metadata, *sa_cols, schema=schema) + + +def _copy_table(engine: Engine, src: sqlite3.Connection, sa_table: Table) -> int: + """Bulk-load via psycopg COPY. Identifiers go through SA's dialect preparer + so we never hand-quote names.""" + prep = engine.dialect.identifier_preparer + table_ident = prep.format_table(sa_table) # ""."" + col_list = ", ".join(prep.quote(c.name) for c in sa_table.columns) + src_ident = '"' + sa_table.name.replace('"', '""') + '"' # SQLite side + + count = 0 + raw = engine.raw_connection() + try: + cur = raw.cursor() + with cur.copy(f"COPY {table_ident} ({col_list}) FROM STDIN") as copy: + for row in src.execute(f"SELECT * FROM {src_ident}"): + copy.write_row(tuple(row)) + count += 1 + if count % 100_000 == 0: + print(f" {sa_table.name}: {count:,} rows...") + raw.commit() + finally: + raw.close() + return count + + +def main() -> None: + dataset = get_settings().dataset + sqlite_path = Path(f"seed/data/{dataset}.sqlite") + + if not sqlite_path.exists(): + sys.exit( + f"{dataset} SQLite not found at {sqlite_path}.\n" + f"Run `bash ctrl/seed/download.sh` on the host first." + ) + + engine = get_engine() + src = sqlite3.connect(sqlite_path) + src.row_factory = sqlite3.Row + + prep = engine.dialect.identifier_preparer + schema_ident = prep.quote_schema(dataset) + with engine.begin() as conn: + conn.execute(text(f"DROP SCHEMA IF EXISTS {schema_ident} CASCADE")) + conn.execute(text(f"CREATE SCHEMA {schema_ident}")) + + metadata = MetaData(schema=dataset) + tables = [_discover_table(src, metadata, t, dataset) for t in _list_tables(src)] + metadata.create_all(engine) + print(f"created {len(tables)} tables in schema {dataset!r}") + + for sa_table in tables: + count = _copy_table(engine, src, sa_table) + print(f" loaded {sa_table.name}: {count:,} rows") + + src.close() + print("done.") + + +if __name__ == "__main__": + main() diff --git a/pyproject.toml b/pyproject.toml index e075eb4..334e656 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -8,9 +8,11 @@ dependencies = [ "pydantic>=2.0", "pydantic-settings", "anthropic", + "openai", "langgraph", "langchain-anthropic", "langfuse", + "sqlalchemy>=2.0", "psycopg[binary]", "sqlglot", "httpx", diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_events.py b/tests/test_events.py new file mode 100644 index 0000000..6158417 --- /dev/null +++ b/tests/test_events.py @@ -0,0 +1,60 @@ +"""Regression: event factory shapes (UI relies on these field names).""" + +from api.analyses.types import Finding +from api.runtime import events + + +def test_run_start_shape(): + e = events.run_start("why?") + assert e == {"type": "run_start", "question": "why?"} + + +def test_run_end_shape(): + assert events.run_end(answer="42") == {"type": "run_end", "answer": "42", "error": None} + assert events.run_end(error="boom") == {"type": "run_end", "answer": "", "error": "boom"} + + +def test_plan_ready_shape(): + e = events.plan_ready("rat", [{"analysis": "x", "args": {}, "why": ""}]) + assert e["type"] == "plan_ready" + assert e["rationale"] == "rat" + assert isinstance(e["steps"], list) + + +def test_analysis_start_shape(): + e = events.analysis_start("direct_answer", {"q": "?"}, "fits") + assert e == { + "type": "analysis_start", + "analysis": "direct_answer", + "args": {"q": "?"}, + "why": "fits", + } + + +def test_analysis_end_uses_finding(): + f = Finding(analysis="direct_answer", summary="ok", rows=[{"a": 1}], sql=["SELECT 1"]) + e = events.analysis_end("direct_answer", f) + assert e["type"] == "analysis_end" + assert e["analysis"] == "direct_answer" + assert e["summary"] == "ok" + assert e["error"] is None + assert e["row_count"] == 1 + + +def test_tool_call_shapes(): + s = events.tool_call_start("text_to_sql", input={"q": "?"}) + assert s == {"type": "tool_call_start", "tool": "text_to_sql", "input": {"q": "?"}} + e = events.tool_call_end("text_to_sql", output={"sql": "..."}) + assert e == {"type": "tool_call_end", "tool": "text_to_sql", "output": {"sql": "..."}, "error": None} + + +def test_synth_start_shape(): + assert events.synth_start() == {"type": "synth_start"} + + +def test_sse_format_round_trip(): + payload = events.tool_call_start("t", input={"x": 1}) + formatted = events.sse_format(payload) + assert formatted.startswith("event: tool_call_start\n") + assert "data: " in formatted + assert formatted.endswith("\n\n") diff --git a/tests/test_imports.py b/tests/test_imports.py new file mode 100644 index 0000000..46eca73 --- /dev/null +++ b/tests/test_imports.py @@ -0,0 +1,42 @@ +"""Regression: every module imports cleanly. + +Catches things like: +- the `api/semantic` → `api/datasets` rename leaving a stale import in schema.py +- the `api/tools/observability` → `api/langfuse_client` move +- the `api/anthropic_client` → `api/llm` move +""" + +import importlib + +import pytest + +MODULES = [ + "api.main", + "api.config", + "api.langfuse_client", + "api.llm", + "api.datasets", + "api.prompts", + "api.tools.db", + "api.tools.schema", + "api.tools.text_to_sql", + "api.tools.execute_sql", + "api.tools.types", + "api.analyses.base", + "api.analyses.types", + "api.analyses.direct_answer", + "api.analyses.compare_periods", + "api.analyses.registry", + "api.plan.planner", + "api.plan.types", + "api.runtime.runner", + "api.runtime.events", + "api.runtime.state", + "api.runtime.context", + "ctrl.seed.load_bird", +] + + +@pytest.mark.parametrize("name", MODULES) +def test_module_imports(name): + importlib.import_module(name) diff --git a/tests/test_langfuse_client.py b/tests/test_langfuse_client.py new file mode 100644 index 0000000..4628093 --- /dev/null +++ b/tests/test_langfuse_client.py @@ -0,0 +1,41 @@ +"""Regression: langfuse_client surface + span context manager behaviour. + +Bugs captured: +1. AttributeError: module 'api.langfuse_client' has no attribute 'span' + — the module was at the old observability.py shape with `langfuse_span`. +2. RuntimeError: generator didn't stop after throw() + — the @contextmanager wrapped the user's `with` body in `try/except` + and yielded a second value on exception, which @contextmanager rejects. +""" + +import pytest + +from api import langfuse_client as lf + + +def test_public_surface(): + assert callable(lf.span) + assert callable(lf.flush) + assert callable(lf.get_client) + + +def test_span_returns_nullspan_without_keys(monkeypatch): + # Ensure no keys → no client → null span. + from api.config import get_settings + get_settings.cache_clear() + monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "") + monkeypatch.setenv("LANGFUSE_SECRET_KEY", "") + # Reset cached client too + lf._client = None + with lf.span("t") as s: + assert isinstance(s, lf._NullSpan) + s.update(output={"x": 1}) # no-op should not raise + get_settings.cache_clear() + + +def test_span_body_exception_propagates(): + """The body's exception MUST propagate — wrapping it inside @contextmanager + causes 'generator didn't stop after throw()' if the except clause yields.""" + with pytest.raises(ValueError, match="kaboom"): + with lf.span("t"): + raise ValueError("kaboom") diff --git a/tests/test_llm.py b/tests/test_llm.py new file mode 100644 index 0000000..f1cbff8 --- /dev/null +++ b/tests/test_llm.py @@ -0,0 +1,15 @@ +"""Regression: LLM dispatcher exposes all three providers and rejects unknowns.""" + +import pytest + +from api import llm + + +def test_providers_registered(): + assert set(llm._PROVIDERS) == {"groq", "anthropic", "openai"} + + +def test_unknown_provider_raises(monkeypatch): + monkeypatch.setattr(llm.get_settings(), "llm_provider", "definitely-not-a-provider", raising=False) + with pytest.raises(ValueError, match="unknown llm_provider"): + llm.chat(system="s", user="u") diff --git a/tests/test_plan_types.py b/tests/test_plan_types.py new file mode 100644 index 0000000..49542cc --- /dev/null +++ b/tests/test_plan_types.py @@ -0,0 +1,31 @@ +"""Regression: Plan/PlanStep round-trip from_dict / to_dict.""" + +from api.plan.types import Plan, PlanStep + + +def test_planstep_roundtrip(): + raw = {"analysis": "direct_answer", "args": {"question": "q"}, "why": "fits"} + step = PlanStep.from_dict(raw) + assert step.analysis == "direct_answer" + assert step.args == {"question": "q"} + assert step.why == "fits" + assert step.to_dict() == raw + + +def test_planstep_defaults_when_keys_missing(): + step = PlanStep.from_dict({"analysis": "direct_answer"}) + assert step.args == {} + assert step.why == "" + + +def test_plan_roundtrip(): + raw = { + "rationale": "single step suffices", + "steps": [ + {"analysis": "direct_answer", "args": {"question": "q"}, "why": "fits"}, + ], + } + plan = Plan.from_dict(raw) + assert plan.rationale == "single step suffices" + assert len(plan.steps) == 1 + assert plan.steps[0].analysis == "direct_answer" diff --git a/tests/test_prompts.py b/tests/test_prompts.py new file mode 100644 index 0000000..0b2d7cc --- /dev/null +++ b/tests/test_prompts.py @@ -0,0 +1,30 @@ +"""Regression: every prompt file loads, render() substitutes placeholders.""" + +from pathlib import Path + +import pytest + +from api.prompts import PROMPTS_DIR, load, render + +PROMPT_NAMES = sorted( + p.stem for p in PROMPTS_DIR.iterdir() if p.suffix == ".txt" +) + + +@pytest.mark.parametrize("name", PROMPT_NAMES) +def test_prompt_loads(name): + text = load(name) + assert text # non-empty + + +def test_render_substitutes(): + out = render("synthesize.user", question="how many?", findings_block="x: 1") + assert "how many?" in out + assert "x: 1" in out + assert "{question}" not in out and "{findings_block}" not in out + + +def test_render_unknown_placeholder_raises(): + # str.format raises KeyError on missing keys + with pytest.raises(KeyError): + render("synthesize.user", question="q") # missing findings_block diff --git a/tests/test_yaml_datasets.py b/tests/test_yaml_datasets.py new file mode 100644 index 0000000..8d1d5d2 --- /dev/null +++ b/tests/test_yaml_datasets.py @@ -0,0 +1,46 @@ +"""Regression: dataset YAML files parse and have the expected shape. + +Bug captured: schema_docs.yaml had values that began with a quote character +(e.g. `gender: 'M' or 'F'.`), which YAML treats as a quoted scalar — anything +after the closing quote is a parse error. +""" + +from pathlib import Path + +import pytest +import yaml + +DATASETS_DIR = Path(__file__).resolve().parent.parent / "api" / "datasets" +DATASETS = [p.name for p in DATASETS_DIR.iterdir() if p.is_dir() and not p.name.startswith("_")] + + +@pytest.mark.parametrize("dataset", DATASETS) +def test_schema_docs_parses(dataset): + docs = yaml.safe_load((DATASETS_DIR / dataset / "schema_docs.yaml").read_text()) + assert docs is not None + assert "tables" in docs + assert isinstance(docs["tables"], dict) and docs["tables"] + + +@pytest.mark.parametrize("dataset", DATASETS) +def test_metrics_parses(dataset): + metrics = yaml.safe_load((DATASETS_DIR / dataset / "metrics.yaml").read_text()) + assert metrics is not None + assert "metrics" in metrics + assert isinstance(metrics["metrics"], dict) and metrics["metrics"] + + +def test_financial_columns_with_quotes_survive_parse(): + """Specific regression: values that contain quoted codes (M, F, OWNER, + PRIJEM, etc.) must parse without truncation.""" + docs = yaml.safe_load( + (DATASETS_DIR / "financial" / "schema_docs.yaml").read_text() + ) + cols = docs["tables"]["client"]["columns"] + assert '"M"' in cols["gender"] and '"F"' in cols["gender"] + + disp = docs["tables"]["disp"]["columns"] + assert '"OWNER"' in disp["type"] and '"DISPONENT"' in disp["type"] + + trans = docs["tables"]["trans"]["columns"] + assert '"PRIJEM"' in trans["type"] and '"VYDAJ"' in trans["type"] diff --git a/ui/app/src/composables/useRunStream.ts b/ui/app/src/composables/useRunStream.ts new file mode 100644 index 0000000..7170b54 --- /dev/null +++ b/ui/app/src/composables/useRunStream.ts @@ -0,0 +1,267 @@ +import { ref, onUnmounted, computed } from 'vue' +import type { LogEntry } from 'soleprint-ui' +import { apiUrl } from '../config' + +/* Mirror of the backend factories in api/runtime/events.py */ +export type RunStatus = 'idle' | 'connecting' | 'streaming' | 'done' | 'error' + +export interface PlanStep { + analysis: string + args: Record + why: string +} + +export interface GraphNode { + id: string // e.g. "plan", "direct_answer", "synthesize" + kind: 'plan' | 'analysis' | 'synthesize' + label: string + status: 'pending' | 'running' | 'done' | 'error' + detail?: string +} + +export interface ToolCall { + tool: string + input?: any + output?: any + error?: string | null + /** ms since the run started */ + t_started: number + t_ended?: number +} + +export function useRunStream() { + const status = ref('idle') + const runId = ref(null) + const question = ref('') + const rationale = ref('') + const planSteps = ref([]) + const graphNodes = ref([]) + const log = ref([]) + const toolCalls = ref([]) + const answer = ref('') + const error = ref(null) + + let es: EventSource | null = null + let t0 = 0 + + const elapsed = (): number => Math.round(performance.now() - t0) + + function reset() { + runId.value = null + question.value = '' + rationale.value = '' + planSteps.value = [] + graphNodes.value = [] + log.value = [] + toolCalls.value = [] + answer.value = '' + error.value = null + status.value = 'idle' + } + + function push(level: string, stage: string, msg: string) { + log.value.push({ + level, + stage, + msg, + ts: new Date().toISOString().split('T')[1]!.split('.')[0]!, + }) + } + + function nodeBy(id: string): GraphNode | undefined { + return graphNodes.value.find(n => n.id === id) + } + + function setNodeStatus(id: string, s: GraphNode['status'], detail?: string) { + const n = nodeBy(id) + if (n) { + n.status = s + if (detail !== undefined) n.detail = detail + } + } + + async function ask(q: string): Promise { + reset() + status.value = 'connecting' + question.value = q + t0 = performance.now() + + let res: Response + try { + res = await fetch(apiUrl('/ask'), { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ question: q }), + }) + } catch (e: any) { + status.value = 'error' + error.value = String(e) + push('error', 'system', `network: ${e}`) + return null + } + if (!res.ok) { + status.value = 'error' + error.value = `${res.status} ${res.statusText}` + push('error', 'system', `submit: ${error.value}`) + return null + } + const { run_id } = await res.json() + runId.value = run_id + + // Seed graph with the meta-nodes we know about; analysis nodes are added + // once the plan arrives. + graphNodes.value = [ + { id: 'plan', kind: 'plan', label: 'Plan', status: 'running' }, + { id: 'synthesize', kind: 'synthesize', label: 'Synthesize', status: 'pending' }, + ] + + push('info', 'system', `ask submitted → run ${run_id}`) + openStream(run_id) + return run_id + } + + function openStream(id: string) { + if (es) es.close() + es = new EventSource(apiUrl(`/runs/${id}/stream`)) + status.value = 'streaming' + + es.addEventListener('run_start', (e: MessageEvent) => { + const d = JSON.parse(e.data) + push('info', 'run', `start: ${d.question}`) + }) + + es.addEventListener('node_update', (e: MessageEvent) => { + const d = JSON.parse(e.data) + push('debug', 'graph', `${d.node} → ${d.keys?.join(', ')}`) + }) + + es.addEventListener('plan_ready', (e: MessageEvent) => { + const d = JSON.parse(e.data) as { rationale: string; steps: PlanStep[] } + rationale.value = d.rationale + planSteps.value = d.steps + + // Replace placeholder analysis nodes with real ones from the plan. + const planNode = nodeBy('plan')! + planNode.status = 'done' + planNode.detail = d.rationale + const synth = graphNodes.value.find(n => n.id === 'synthesize') + graphNodes.value = [ + planNode, + ...d.steps.map((s, i): GraphNode => ({ + id: `${s.analysis}#${i}`, + kind: 'analysis', + label: s.analysis, + status: 'pending', + detail: s.why, + })), + synth || { id: 'synthesize', kind: 'synthesize', label: 'Synthesize', status: 'pending' }, + ] + push('info', 'plan', `${d.steps.length} step(s): ${d.steps.map(s => s.analysis).join(' → ')}`) + }) + + es.addEventListener('analysis_start', (e: MessageEvent) => { + const d = JSON.parse(e.data) as { analysis: string; args: any; why: string } + // Mark the first matching pending analysis node as running. + const n = graphNodes.value.find(g => g.kind === 'analysis' && g.label === d.analysis && g.status === 'pending') + if (n) n.status = 'running' + push('info', d.analysis, `start: ${d.why || JSON.stringify(d.args)}`) + }) + + es.addEventListener('analysis_end', (e: MessageEvent) => { + const d = JSON.parse(e.data) as { analysis: string; summary: string; error: string | null; row_count: number } + const n = graphNodes.value.find(g => g.kind === 'analysis' && g.label === d.analysis && g.status === 'running') + if (n) { + n.status = d.error ? 'error' : 'done' + n.detail = d.error || d.summary + } + const level = d.error ? 'error' : 'info' + const msg = d.error ? `error: ${d.error}` : `done — ${d.row_count} rows · ${d.summary}` + push(level, d.analysis, msg) + }) + + es.addEventListener('tool_call_start', (e: MessageEvent) => { + const d = JSON.parse(e.data) as { tool: string; input: any } + toolCalls.value.push({ tool: d.tool, input: d.input, t_started: elapsed() }) + const detail = d.tool === 'execute_sql' ? (d.input?.sql || '') : + d.tool === 'text_to_sql' ? (d.input?.question || '') : + JSON.stringify(d.input) + push('debug', d.tool, `→ ${detail}`) + }) + + es.addEventListener('tool_call_end', (e: MessageEvent) => { + const d = JSON.parse(e.data) as { tool: string; output: any; error: string | null } + const call = [...toolCalls.value].reverse().find(c => c.tool === d.tool && c.t_ended === undefined) + if (call) { + call.output = d.output + call.error = d.error + call.t_ended = elapsed() + } + if (d.error) { + push('error', d.tool, `✗ ${d.error}`) + return + } + if (d.tool === 'text_to_sql' && d.output?.sql) { + push('info', d.tool, `✓ ${d.output.sql}`) + } else if (d.tool === 'execute_sql') { + const label = d.output?.label ? `${d.output.label}: ` : '' + push('info', d.tool, `✓ ${label}${d.output?.row_count} rows`) + } else if (d.tool === 'generate_pair') { + push('info', d.tool, `✓ a=${d.output?.a?.label} · b=${d.output?.b?.label}`) + } else { + push('info', d.tool, '✓') + } + }) + + es.addEventListener('llm_call', (e: MessageEvent) => { + const d = JSON.parse(e.data) as { label: string; system_chars: number; user_chars: number } + push('debug', d.label, `llm ← sys:${d.system_chars} usr:${d.user_chars}`) + }) + + es.addEventListener('synth_start', () => { + setNodeStatus('synthesize', 'running') + push('info', 'synthesize', '→ composing final answer') + }) + + es.addEventListener('run_end', (e: MessageEvent) => { + const d = JSON.parse(e.data) as { answer: string; error: string | null } + setNodeStatus('synthesize', d.error ? 'error' : 'done') + answer.value = d.answer || '' + if (d.error) { + error.value = d.error + push('error', 'run', `end: ${d.error}`) + } else { + push('info', 'run', `end · ${elapsed()}ms`) + } + }) + + es.addEventListener('end', () => { + status.value = error.value ? 'error' : 'done' + es?.close() + es = null + }) + + es.addEventListener('error', (e: MessageEvent) => { + // The browser also fires a generic `error` event on disconnect; only + // surface it when the run hasn't already ended. + if (status.value === 'streaming') { + status.value = 'error' + const msg = e?.data ? JSON.parse(e.data)?.message : 'stream error' + push('error', 'stream', msg) + } + }) + } + + function disconnect() { + if (es) { es.close(); es = null } + } + + onUnmounted(disconnect) + + const isRunning = computed(() => status.value === 'connecting' || status.value === 'streaming') + + return { + status, runId, question, rationale, planSteps, graphNodes, + log, toolCalls, answer, error, isRunning, + ask, disconnect, + } +} diff --git a/ui/app/src/config.ts b/ui/app/src/config.ts new file mode 100644 index 0000000..b2feaac --- /dev/null +++ b/ui/app/src/config.ts @@ -0,0 +1,21 @@ +/* API URL resolver. + * + * The SPA is served from `nvi.local.ar` (or `nivii.mcrn.ar` in AWS, or + * `localhost:5173` in `vite dev`). The API lives at the `api.` subdomain in + * production-style deploys; in dev the vite proxy forwards relative paths. + */ + +export function apiUrl(path: string): string { + const { protocol, hostname } = window.location + + if (hostname.endsWith('.local.ar')) { + return `${protocol}//api.${hostname}${path}` + } + + if (hostname === 'nivii.mcrn.ar' || hostname === 'docs.nivii.mcrn.ar') { + return `https://api.nivii.mcrn.ar${path}` + } + + // localhost (vite dev) or unrecognised host — rely on same-origin / proxy. + return path +} diff --git a/ui/app/src/main.ts b/ui/app/src/main.ts index da5e9e4..9f5dc5c 100644 --- a/ui/app/src/main.ts +++ b/ui/app/src/main.ts @@ -3,13 +3,12 @@ import { createRouter, createWebHistory } from 'vue-router' import 'soleprint-ui/src/tokens.css' import App from './App.vue' import Ask from './pages/Ask.vue' -import RunInspector from './pages/RunInspector.vue' const router = createRouter({ history: createWebHistory(), routes: [ { path: '/', component: Ask }, - { path: '/runs/:id', component: RunInspector, props: true }, + { path: '/:catchAll(.*)', redirect: '/' }, ], }) diff --git a/ui/app/src/pages/Ask.vue b/ui/app/src/pages/Ask.vue index 8b9abe0..a2d149b 100644 --- a/ui/app/src/pages/Ask.vue +++ b/ui/app/src/pages/Ask.vue @@ -1,80 +1,477 @@