Evaluate dbt Semantic Layer queries¶
Check AI-generated dbt Semantic Layer (MetricFlow) queries against gold answers. evaldata reads
the metrics and dimensions your project defines, uses the warehouse connection from the project's
dbt profile, and scores each answer with three checks: resolved query equivalence, result-set
equivalence, and an LLM judge. Scoring stops as soon as one check reaches a verdict.
Prerequisites¶
The dbt-sl extra pulls in dbt-metricflow, the toolchain that resolves and runs metric queries;
litellm backs the solver and the judge when you pass a model id.
You also need a built dbt project with a semantic layer (semantic models and metrics). Parse it so
target/ holds the semantic manifest, and build it so the warehouse has the data to query:
dbt build # materialise the models and the time spine
dbt parse # writes target/semantic_manifest.json
Write the cases file¶
A metric cases file pairs each question with a gold metric query. The query specifies the metrics to compute and how to slice, filter, and limit them:
# metric_cases.yml
- question: What is total revenue by month?
metrics: [revenue]
group_by: [metric_time__month]
- question: What is the average order value for large orders?
metrics: [average_order_value]
where: ["{{ Dimension('order_id__is_large_order') }} = true"]
id: aov-large-orders
Each entry needs a question and at least one metric. group_by items are dimensions, entities,
or a time dimension with a grain (metric_time__month); where holds MetricFlow filter
expressions; order_by, limit, and id are optional.
Run it¶
evaldata gives the model the project's metrics and dimensions, asks it for a metric query per
question, and scores each against the gold query. It reports the fraction that match:
--model is any litellm model id. Other options:
--grader-model ID: the model for the judge tier; defaults to--model.--no-judge: score with the deterministic tiers only (resolved query equivalence and result-set equivalence); no LLM judge, so the result is reproducible in CI.--temperature FLOAT: sampling temperature for the solver and judge (default0.0); reasoning models such as the GPT-5 family often accept only1.0.--target-dir DIR: where the artifacts live, if not<project>/target.--profiles-dir DIR/--target NAME: find and select the dbt profile target.--limit N: run only the firstNquestions.--json PATH: also write the scores and every result to a JSON file.
How it's scored¶
Each question runs through up to three checks in order:
- Resolve and compare. MetricFlow resolves both the candidate and gold queries against the semantic manifest, filling in default time grains and entity paths. Queries that resolve to the same form are equivalent, decided without running anything.
- Run and compare. When the resolved forms differ, both queries run through
mfand their result rows are compared. The verdict comes from the data the warehouse returns. - LLM judge. A grader model reads the candidate and gold queries and decides whether they answer the question the same way. It doesn't run either query.
The LLM judge runs only if the first two checks are inconclusive.
Run it in pytest¶
Run Semantic Layer evals as pytest tests by loading the cases yourself:
import pytest
from evaldata.dbt import (
DbtError,
load_dbt_metrics,
metric_layer_equivalence,
MetricLayerSolver,
assert_metric_eval,
platform_from_profile,
)
platform = platform_from_profile("path/to/dbt_project")
if isinstance(platform, DbtError):
raise RuntimeError(platform.message)
cases = load_dbt_metrics("path/to/dbt_project/target", platform=platform, cases="metric_cases.yml")
if isinstance(cases, DbtError):
raise RuntimeError(cases.message)
@pytest.mark.parametrize("case", cases, ids=lambda case: case.id)
def test_sl_question(case):
solver = MetricLayerSolver("openai/gpt-4o-mini")
assert_metric_eval(case, solver, scorers=[metric_layer_equivalence("openai/gpt-4o-mini")])
load_dbt_metrics and platform_from_profile return a DbtError when the project can't be read.
To compose the cascade yourself, use MetricSpecEquivalence,
MetricResultEquivalence, and MetricLayerJudge with MetricFirstDecisive.
Bundled example¶
The bundled example runs a jaffle-shop Semantic Layer project on DuckDB with fixed responses for the solver and judge.
"""dbt Semantic Layer (MetricFlow) evals against the jaffle-shop project, run locally on DuckDB.
`load_dbt_metrics` builds one `MetricCase` per question from a metric cases file; each gold is a
`MetricQuery` — the metrics to compute and how to group, filter, and order them. Candidates are
scored by three checks, cheapest first, stopping at the first verdict:
- `MetricSpecEquivalence` resolves both queries through MetricFlow and confirms a match from their
resolved form, without running anything. It proves, for example, that grouping by `metric_time`
(whose default grain is a day) equals grouping by `metric_time__day`.
- `MetricResultEquivalence` runs both queries through `mf` and compares the result rows, so a
reformulation that returns the same rows is confirmed even when the resolved forms differ.
- `metric_layer_equivalence(...)` is the cost-ordered cascade of both plus an LLM judge. The judge
here is a `StubLlm`, so the file runs offline with no model or warehouse service.
Running `mf` needs the `dbt-metricflow` toolchain with a DuckDB adapter (the `dbt-sl` extra plus
`dbt-duckdb`); resolving a query needs only the semantic manifest.
"""
import shutil
from pathlib import Path
import pytest
from evaldata.dbt import (
DbtError,
MetricCase,
MetricFirstDecisive,
MetricLayerSolver,
MetricQuery,
MetricResultEquivalence,
MetricSpecEquivalence,
load_dbt_metrics,
metric_layer_equivalence,
run_metric_benchmark,
)
from evaldata.llm import StubLlm
from evaldata.platforms import duckdb_platform
from evaldata.scorers.llm_judge import JudgeReply
_PROJECT = Path(__file__).parent / "jaffle"
_CASES = Path(__file__).parent / "metric_cases.yml"
_PLATFORM = duckdb_platform(name="jaffle")
_ARTIFACTS = ("manifest.json", "catalog.json", "semantic_manifest.json")
@pytest.fixture
def cases(tmp_path: Path) -> dict[str, MetricCase]:
"""Copy the project to a temp dir, seed `target/` with the artifacts, and load the cases by id.
Args:
tmp_path: A temp directory the project is copied into so its files are never mutated.
Returns:
The loaded Semantic Layer cases, keyed by case id.
"""
project = tmp_path / "jaffle"
shutil.copytree(_PROJECT, project)
target = project / "target"
target.mkdir()
for name in _ARTIFACTS:
shutil.copy(project / "artifacts" / name, target / name)
loaded = load_dbt_metrics(target, platform=_PLATFORM, cases=_CASES)
assert not isinstance(loaded, DbtError)
return {case.id: case for case in loaded}
def test_default_grain_proven_without_running(cases: dict[str, MetricCase]) -> None:
"""Grouping by `metric_time` resolves to its default grain, so `metric_time__day` is proven equal — no query runs."""
case = cases["revenue-by-day"]
score = MetricSpecEquivalence().score(case, MetricQuery(metrics=["revenue"], group_by=["metric_time__day"]))
assert score.verdict == "pass"
assert score.basis == "proven"
@pytest.mark.timeout(300)
def test_reformulation_confirmed_by_results(cases: dict[str, MetricCase]) -> None:
"""Adding an ordering the resolver treats as different still passes: the run tier confirms the same rows."""
case = cases["revenue-by-region"]
candidate = MetricQuery(metrics=["revenue"], group_by=["customer__region"], order_by=["-revenue"])
cascade = MetricFirstDecisive([MetricSpecEquivalence(), MetricResultEquivalence()])
score = cascade.score(case, candidate)
assert score.verdict == "pass"
assert score.basis == "observed"
def test_large_order_filter_resolves(cases: dict[str, MetricCase]) -> None:
"""A filter on `order_id__is_large_order` is a valid query the spec tier resolves and confirms."""
case = cases["revenue-large-orders"]
score = MetricSpecEquivalence().score(case, case.gold)
assert score.verdict == "pass"
assert score.basis == "proven"
def _answer(prompt: str, _response_format: object) -> MetricQuery:
"""Answer three of the questions correctly and deliberately misread the fourth.
Args:
prompt: The solver prompt, which carries the question text.
_response_format: The requested structured-output type (unused).
Returns:
The metric query the stubbed solver answers the question with.
"""
if "each month" in prompt:
return MetricQuery(metrics=["revenue"], group_by=["metric_time__month"])
if "customer region" in prompt:
return MetricQuery(metrics=["revenue"], group_by=["customer__region"])
if "large orders" in prompt:
return MetricQuery(metrics=["revenue"], where=["{{ Dimension('order_id__is_large_order') }} = true"])
# Misreads the daily question as monthly — the one miss the benchmark catches.
return MetricQuery(metrics=["revenue"], group_by=["metric_time__month"])
@pytest.mark.timeout(300)
def test_benchmark_reports_accuracy(cases: dict[str, MetricCase]) -> None:
"""Three questions answered correctly and one deliberately misread aggregate to 0.75."""
solver = MetricLayerSolver(model=StubLlm(_answer))
# The judge is the last tier; here spec or run decides every case, so the stub grader is never reached.
scorers = [metric_layer_equivalence(StubLlm(JudgeReply(reason="unused", score=0.0)))]
summary = run_metric_benchmark(list(cases.values()), solver, scorers=scorers)
assert summary.total == 4
assert summary.passed == 3
assert summary.accuracy == 0.75
How it works¶
- MetricFlow resolves and runs every query, so a verdict matches what the semantic layer would
return;
evaldatadoesn't reimplement MetricFlow's logic. - Resolving a query needs only
target/semantic_manifest.json; running one needs the built warehouse the project's dbt profile points at. - The resolved query check confirms equivalence but never rejects on resolved form alone: when the resolved forms differ, it defers to running the queries rather than call them unequal.
Next steps¶
- Evaluate against a dbt project: text-to-SQL evals on the same project.
- Score with an LLM judge: the judge tier in depth.
- dbt reference: the Semantic Layer types, loaders, and scorers.