Cost per sub-question in a LangChain SQL analyst
A planner, a SQL writer, a result checker and a narrator in LangChain 1.x on Claude, then a blocking budget on the role whose call count the planner chose.
- Framework
- LangChain 1.x
- Provider
- anthropic
- Python
- 3.11+
- Capsera
- pip install capsera
This builds a SQL analyst agent in LangChain 1.x against Anthropic: a planner
turns one business question into sub-questions, a SQL writer writes one SELECT
for each, Python runs them against a read-only database, a checker reads the rows
back, and a narrator writes the answer from tables that have already been
computed. One question is four model calls at the floor and twenty-two at the
cap, and the role that decides where in that range a run lands makes exactly one
of them. The last third of this page adds Capsera: one init() call to capture
every request, four decorator lines to say which role made it, and a blocking
budget on the role that multiplies.
What you will build
analyst.py, a single-file agent with four roles and a supervisor, answering
questions about a small accounts-and-tickets database:
- planner — one call per question, on the stronger model. Turns the question into at most five sub-questions, each answerable by a single SELECT. It decides the fan-out and then takes no further part in it.
- sql_writer — one call per sub-question, plus one per rewrite, on the stronger model, with the whole schema in every prompt. The multiplier.
- checker — one call per sub-question whose query returned rows, on the fast model. Says whether those rows answer the sub-question that was asked.
- narrator — one call. Writes the answer from the result tables, naming which sub-question each figure came from.
Everything that touches the database is Python. Deciding whether a statement may
run, capping the rows it may return, formatting money out of minor units and
noticing that a query came back empty are a guard, a fetchmany, a Decimal and
a len. An analyst agent that asks a model whether a query is safe to run has
put the security boundary in a prompt, and one that asks a model to divide by a
hundred has put the arithmetic somewhere it cannot be checked.
Each role is its own function. That matters later — it is the seam the instrumentation attaches to — and it is the better shape regardless, because a pipeline that inlines four model calls has nothing you can name, budget, or move to a different model.
Written against LangChain 1.x with langchain-anthropic, calling Anthropic's API
directly rather than through a gateway. Models: claude-sonnet-5-5 for the
planner and the SQL writer; claude-haiku-5-5 for the checker and the narrator.
The database is SQLite from the standard library, built from a fixture, so the
whole thing runs with no service to stand up.
Prerequisites
Python 3.11 or newer.
pip install "langchain>=1.0,<2" langchain-anthropic
The major is pinned so this page stays true as LangChain moves.
langchain-anthropic is left unpinned on purpose: install the one that matches
your langchain major. SQLite ships with Python; there is nothing to install for
it.
You need an Anthropic API key. ChatAnthropic reads ANTHROPIC_API_KEY from the
environment, so export it and no code has to touch it:
export ANTHROPIC_API_KEY=...
Step 1 — The database, and the facts that are not in the schema
A model can read a schema. It cannot read what a column means at your company, and nearly every wrong answer an analyst agent gives comes from the gap between those two things. Start by writing both down.
"""analyst.py — the database this agent answers from, and what it may do to it."""
import sqlite3
import tempfile
from pathlib import Path
DB_PATH = Path(tempfile.gettempdir()) / "analyst-demo.sqlite"
SCHEMA = """
CREATE TABLE accounts (
account_id TEXT PRIMARY KEY,
name TEXT NOT NULL,
region TEXT NOT NULL,
plan TEXT NOT NULL,
started_on TEXT NOT NULL,
churned_on TEXT
);
CREATE TABLE invoices (
invoice_id TEXT PRIMARY KEY,
account_id TEXT NOT NULL REFERENCES accounts(account_id),
issued_on TEXT NOT NULL,
amount_cents INTEGER NOT NULL,
status TEXT NOT NULL
);
CREATE TABLE tickets (
ticket_id TEXT PRIMARY KEY,
account_id TEXT NOT NULL REFERENCES accounts(account_id),
opened_on TEXT NOT NULL,
closed_on TEXT,
severity TEXT NOT NULL,
channel TEXT NOT NULL
);
"""
SCHEMA_NOTES = """
Dates are ISO strings, YYYY-MM-DD, and compare correctly with <, > and BETWEEN.
accounts.churned_on is NULL while an account is still a customer. An account was
active on a date if started_on <= that date and churned_on IS NULL or
churned_on > that date.
invoices.amount_cents is an integer in minor units. It is not a currency amount
and it is never divided.
invoices.status is one of paid, open, void. A void invoice was cancelled and was
never owed by anyone; revenue counts paid invoices only.
tickets.closed_on is NULL while a ticket is still open.
tickets.severity is one of sev1, sev2, sev3, and sev1 is the most severe.
There is no revenue column, no churn flag and no account health score anywhere
in this database. Every one of those is computed from the rows above.
"""
ACCOUNTS = [
# account_id, name, region, plan, started_on, churned_on
("acc-01", "Halversen Labs", "emea", "scale", "2025-02-11", None),
("acc-02", "Northgate Freight", "emea", "team", "2025-06-03", "2026-04-30"),
("acc-03", "Pellier Tooling", "emea", "team", "2026-01-19", None),
("acc-04", "Tamsin Print", "amer", "team", "2025-09-24", None),
("acc-05", "Brightwater Mutual", "amer", "scale", "2024-11-05", None),
("acc-06", "Castellan Foods", "amer", "scale", "2025-03-17", "2026-07-31"),
("acc-07", "Okonjo Mobility", "apac", "team", "2026-02-02", None),
("acc-08", "Serra Analytics", "apac", "scale", "2025-08-08", "2026-02-28"),
]
INVOICES = [
# invoice_id, account_id, issued_on, amount_cents, status
("inv-1001", "acc-01", "2026-01-05", 480_000, "paid"),
("inv-1002", "acc-01", "2026-04-05", 480_000, "paid"),
("inv-1003", "acc-01", "2026-07-05", 520_000, "paid"),
("inv-1004", "acc-01", "2026-10-05", 520_000, "open"),
("inv-1005", "acc-02", "2026-01-05", 90_000, "paid"),
("inv-1006", "acc-02", "2026-04-05", 90_000, "void"),
("inv-1007", "acc-03", "2026-02-01", 75_000, "paid"),
("inv-1008", "acc-03", "2026-05-01", 75_000, "paid"),
("inv-1009", "acc-03", "2026-08-01", 75_000, "paid"),
("inv-1010", "acc-04", "2026-01-12", 110_000, "paid"),
("inv-1011", "acc-04", "2026-04-12", 110_000, "paid"),
("inv-1012", "acc-04", "2026-07-12", 110_000, "paid"),
("inv-1013", "acc-04", "2026-10-12", 110_000, "open"),
("inv-1014", "acc-05", "2026-01-03", 640_000, "paid"),
("inv-1015", "acc-05", "2026-04-03", 640_000, "paid"),
("inv-1016", "acc-05", "2026-07-03", 700_000, "paid"),
("inv-1017", "acc-05", "2026-10-03", 700_000, "open"),
("inv-1018", "acc-06", "2026-01-20", 410_000, "paid"),
("inv-1019", "acc-06", "2026-04-20", 410_000, "paid"),
("inv-1020", "acc-06", "2026-07-20", 410_000, "void"),
("inv-1021", "acc-07", "2026-03-02", 68_000, "paid"),
("inv-1022", "acc-07", "2026-06-02", 68_000, "paid"),
("inv-1023", "acc-07", "2026-09-02", 68_000, "open"),
("inv-1024", "acc-08", "2026-01-08", 455_000, "paid"),
("inv-1025", "acc-08", "2026-04-08", 455_000, "void"),
]
TICKETS = [
# ticket_id, account_id, opened_on, closed_on, severity, channel
("tkt-5001", "acc-01", "2026-02-14", "2026-02-15", "sev3", "email"),
("tkt-5002", "acc-02", "2026-02-20", "2026-03-04", "sev1", "phone"),
("tkt-5003", "acc-02", "2026-03-11", "2026-04-02", "sev1", "phone"),
("tkt-5004", "acc-02", "2026-04-06", None, "sev2", "email"),
("tkt-5005", "acc-04", "2026-05-09", "2026-05-12", "sev3", "chat"),
("tkt-5006", "acc-05", "2026-06-01", "2026-06-02", "sev2", "email"),
("tkt-5007", "acc-06", "2026-05-18", "2026-06-30", "sev1", "phone"),
("tkt-5008", "acc-06", "2026-06-22", "2026-07-28", "sev1", "chat"),
("tkt-5009", "acc-06", "2026-07-15", None, "sev2", "email"),
("tkt-5010", "acc-07", "2026-07-07", "2026-07-08", "sev3", "chat"),
("tkt-5011", "acc-08", "2026-01-22", "2026-02-19", "sev1", "email"),
("tkt-5012", "acc-03", "2026-09-30", None, "sev3", "chat"),
]
def build_db(path: Path) -> None:
"""Rebuild the fixture database from scratch, so every run starts equal."""
path.unlink(missing_ok=True)
conn = sqlite3.connect(path)
try:
conn.executescript(SCHEMA)
conn.executemany("INSERT INTO accounts VALUES (?, ?, ?, ?, ?, ?)", ACCOUNTS)
conn.executemany("INSERT INTO invoices VALUES (?, ?, ?, ?, ?)", INVOICES)
conn.executemany("INSERT INTO tickets VALUES (?, ?, ?, ?, ?, ?)", TICKETS)
conn.commit()
finally:
conn.close()
def read_only(path: Path) -> sqlite3.Connection:
"""A connection that cannot write, whatever statement reaches it."""
return sqlite3.connect(f"file:{path}?mode=ro", uri=True)
if __name__ == "__main__":
build_db(DB_PATH)
conn = read_only(DB_PATH)
for table in ("accounts", "invoices", "tickets"):
count = conn.execute(f"SELECT count(*) FROM {table}").fetchone()[0]
print(f"{table}: {count} rows")
active = conn.execute(
"SELECT count(*) FROM accounts WHERE churned_on IS NULL"
).fetchone()[0]
print(f"accounts still active: {active}")
conn.close()
SCHEMA_NOTES is the part that does the work. Three of the eight accounts have a
churned_on date and the rest are NULL, so a sub-question about active customers
answered with churned_on = '' returns nothing and a sub-question answered with
WHERE churned_on IS NOT NULL counts exactly the customers you lost. Three of
the invoices are void — and all three belong to accounts that churned, because
a cancelled final invoice is what churn looks like in a billing table. Sum
amount_cents without filtering on status and every churned account keeps a
quarter of revenue it never paid, which flatters precisely the regions you are
asking about.
None of that is inferable from column names. It is also not something to keep discovering in the prompt: written once, it goes into the planner's prompt and the SQL writer's prompt as the same block, which matters later for what gets billed twice.
read_only opens the same file through a URI with mode=ro. A write that
reaches this connection raises rather than committing. That is the boundary; the
guard in the next step exists to produce a readable error before SQLite produces
an unreadable one, not to be the thing standing between a model and your data.
Step 2 — Run a stranger's SQL the way you would run a stranger's SQL
Every statement that reaches this function was written by a model from a prompt that will one day contain text someone else wrote. Treat it accordingly.
# fragment
import re
import sqlite3
from dataclasses import dataclass
from decimal import Decimal
MAX_ROWS = 50
FORBIDDEN = re.compile(
r"\b(attach|detach|pragma|insert|update|delete|drop|alter|create|replace|vacuum)\b",
re.IGNORECASE,
)
@dataclass
class QueryResult:
sql: str
columns: list[str]
rows: list[tuple]
truncated: bool
error: str | None
def extract_sql(raw: str) -> str:
"""The statement the model wrote, with any code-fence lines dropped."""
kept = [
line
for line in raw.strip().splitlines()
if not line.lstrip().startswith("`")
]
return "\n".join(kept).strip().rstrip(";").strip()
def check_sql(sql: str) -> str | None:
"""Why this statement may not run, or None if it may."""
if not sql:
return "no statement was returned"
if ";" in sql:
return "more than one statement was returned; send exactly one SELECT"
head = sql.lstrip("( \n\t").split(None, 1)
if not head or head[0].lower() not in {"select", "with"}:
return "only a single SELECT or WITH statement may be run"
found = FORBIDDEN.search(sql)
if found:
return f"the statement contains {found.group(0).upper()}, which may not be run"
return None
def run_query(conn: sqlite3.Connection, sql: str) -> QueryResult:
problem = check_sql(sql)
if problem is not None:
return QueryResult(sql, [], [], False, problem)
try:
cursor = conn.execute(sql)
fetched = cursor.fetchmany(MAX_ROWS + 1)
except sqlite3.Error as exc:
return QueryResult(sql, [], [], False, f"sqlite rejected the query: {exc}")
columns = [description[0] for description in cursor.description or []]
rows = [tuple(row) for row in fetched[:MAX_ROWS]]
return QueryResult(sql, columns, rows, len(fetched) > MAX_ROWS, None)
def cell(column: str, value: object) -> str:
if value is None:
return "null"
if column.endswith("_cents") and isinstance(value, int):
return f"{Decimal(value).scaleb(-2):.2f}"
return str(value)
def render(result: QueryResult) -> str:
if result.error is not None:
return f"error: {result.error}"
if not result.rows:
return "no rows"
body = "\n".join(
" | ".join(cell(column, value) for column, value in zip(result.columns, row))
for row in result.rows
)
tail = f"\n...capped at {MAX_ROWS} rows" if result.truncated else ""
return f"{' | '.join(result.columns)}\n{body}{tail}"
extract_sql drops any line that starts with a backtick before anything else
looks at the text. A model told not to use a code fence will use one eventually,
and a statement that fails because its first line is three backticks produces a
syntax error that sends the rewrite in entirely the wrong direction — you pay a
call on the stronger model to fix something a list comprehension already solved.
It is a line filter rather than a parser on purpose: it leaves prose alone, so a
reply that explains itself instead of answering fails the guard below and
collects a note saying so, which is the thing worth telling the model about.
The semicolon rule reads oddly until you notice the order: extract_sql removes a
trailing ;, so check_sql only ever sees a semicolon that is inside the
statement, which means a second statement is riding along behind the first. One
SELECT goes through; SELECT 1; DROP TABLE accounts does not get as far as
SQLite.
FORBIDDEN will reject some perfectly harmless queries — anything with the word
create in a string literal, for one. Against a read-only connection that is the
cheap direction to be wrong in: a false positive costs one rewrite, and the
alternative is a denylist that has to be right about every statement SQLite will
ever accept.
fetchmany(MAX_ROWS + 1) is the row cap and the truncation flag in a single call.
The SQL writer is asked for a LIMIT in the next step, but a LIMIT in a prompt
is a request. This is the enforcement, and the extra row is how the renderer knows
to say so. Without a cap, one SELECT * against a table you grow next year puts a
few hundred thousand tokens into the checker's prompt.
cell is where minor units become currency. The SQL writer is told to keep money
as an integer and to name the column with a _cents suffix, so the division
happens exactly once, here, in Python, after the database has done the summing.
A model asked to produce a tidy currency figure will produce one, and the rounding
it did will not be visible in the row it hands back.
Step 3 — Decompose the question, once
# fragment
from langchain_anthropic import ChatAnthropic
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
PLAN_MODEL = "claude-sonnet-5-5"
MAX_SUBQUESTIONS = 5
PLAN_PROMPT = ChatPromptTemplate.from_messages(
[
(
"system",
"You turn one business question into the smallest set of "
"sub-questions that together answer it. Each sub-question must be "
"answerable by a single SELECT against the schema below. Write one "
"per line, in this exact form, and nothing else:\n"
"q :: the sub-question, as a sentence\n"
"At most five, and fewer is better: each one costs a query, and two "
"that would return the same table are one sub-question. Ask only "
"for facts these tables can supply. Never ask for a judgement, a "
"recommendation, a cause or a forecast — those are not in the "
"database and a query cannot return one.\n"
"Schema:\n{schema}\n"
"What the schema does not say:\n{notes}",
),
("human", "{question}"),
]
)
planner_llm = ChatAnthropic(model=PLAN_MODEL, max_tokens=512, temperature=0)
plan_chain = PLAN_PROMPT | planner_llm | StrOutputParser()
def _subquestions(raw: str) -> list[str]:
out: list[str] = []
for row in raw.splitlines():
key, sep, value = row.partition("::")
text = value.strip()
if sep and key.strip().lower() == "q" and text and text not in out:
out.append(text)
return out[:MAX_SUBQUESTIONS]
def plan(question: str) -> list[str]:
raw = plan_chain.invoke(
{"question": question, "schema": SCHEMA, "notes": SCHEMA_NOTES}
)
return _subquestions(raw)
MAX_SUBQUESTIONS is a cost control before it is anything else. The planner makes
one call and chooses a number that every downstream role is then multiplied by,
so the cap on that number is the cap on the run. Five is in the prompt and sliced
again in _subquestions, because a limit stated to a model is a preference and a
limit applied in Python is a limit.
Duplicates are dropped in the same pass. "What was revenue by region?" asked twice in slightly different words is two calls on the stronger model returning the same table, and the planner has no memory of its own earlier lines to stop it.
Telling the planner it cannot ask for a cause or a recommendation is the other half of the decomposition. "Why is emea declining?" is not a sub-question; it has no SELECT. Left in the list, it produces a query that returns something — SQL almost always returns something — and a result table that looks like evidence for a claim the rows do not support. The judgement belongs to the person reading the answer, and the agent's job is to put the rows in front of them.
Step 4 — One SELECT per sub-question
# fragment
SQL_MODEL = "claude-sonnet-5-5"
SQL_CONCURRENCY = 3
SQL_PROMPT = ChatPromptTemplate.from_messages(
[
(
"system",
"You write exactly one SQLite SELECT that answers the sub-question. "
"Reply with the statement and nothing else: no explanation, no code "
"fence, no second statement.\n"
"Rules that are not in the schema:\n"
"Keep money in minor units as an integer, and give any money column "
"a name ending in _cents. Never divide an amount and never round "
"one.\n"
"Name every computed column with AS.\n"
"Put the most important row first and LIMIT the result to 50 rows "
"or fewer.\n"
"Return the smallest set of columns that answers the "
"sub-question.\n"
"Schema:\n{schema}\n"
"What the schema does not say:\n{notes}\n"
"What went wrong last time, if anything: {note}",
),
("human", "{subquestion}"),
]
)
sql_llm = ChatAnthropic(model=SQL_MODEL, max_tokens=600, temperature=0)
sql_chain = SQL_PROMPT | sql_llm | StrOutputParser()
def write_sql(subquestions: list[str], notes: dict[str, str]) -> dict[str, str]:
replies = sql_chain.batch(
[
{
"subquestion": question,
"schema": SCHEMA,
"notes": SCHEMA_NOTES,
"note": notes.get(question, "nothing, this is the first attempt"),
}
for question in subquestions
],
config={"max_concurrency": SQL_CONCURRENCY},
)
return {
question: extract_sql(reply)
for question, reply in zip(subquestions, replies, strict=True)
}
The rewrite note is a field in the same prompt rather than a second prompt. The first attempt sends a sentence saying there was no previous attempt; a rewrite sends the SQLite error, the guard's refusal or the checker's verdict. Two prompts that differ by one field are two prompts to keep in step, and they drift.
SCHEMA and SCHEMA_NOTES go into every one of these calls, which is the other
reason this role is the expensive one. Five sub-questions is five copies of the
same several hundred tokens in front of five one-sentence questions, on the
stronger model. Note the order they are in: the block that never changes sits
ahead of the part that does, which is the arrangement that gives prompt caching
anything to work with. Whether it earned anything is a number to read later
rather than a claim to make here.
max_concurrency bounds how many of these are in flight at once. It is also, as
the budget section gets to, the number that decides how far a cap can overshoot.
Step 5 — Ask whether the rows answer the question that was asked
# fragment
CHECK_MODEL = "claude-haiku-5-5"
CHECK_CONCURRENCY = 4
VERDICTS = ("answers", "wrong-shape", "wrong-filter")
CHECK_PROMPT = ChatPromptTemplate.from_messages(
[
(
"system",
"You are given a sub-question and the first rows a query returned "
"for it. Say whether those rows are the thing the sub-question "
"asked for. Exactly two lines, in this form:\n"
"verdict :: one of answers, wrong-shape, wrong-filter\n"
"reason :: one sentence under twenty words\n"
"wrong-shape means the rows are not the thing asked for: the wrong "
"grain, the wrong columns, a single total where a breakdown was "
"asked for. wrong-filter means the right kind of rows counted over "
"the wrong set, such as a revenue figure that includes void "
"invoices or a customer count that includes churned accounts. "
"Everything else is answers.\n"
"You are not being asked whether the numbers are right and you "
"cannot check them. Judge the shape and the filter only.",
),
(
"human",
"Sub-question: {subquestion}\nQuery:\n{sql}\nRows:\n{rows}",
),
]
)
checker_llm = ChatAnthropic(model=CHECK_MODEL, max_tokens=160, temperature=0)
check_chain = CHECK_PROMPT | checker_llm | StrOutputParser()
def _verdict(raw: str) -> tuple[str, str]:
verdict, reason = "answers", ""
for row in raw.splitlines():
key, _, value = row.partition("::")
key, value = key.strip().lower(), value.strip()
if key == "verdict" and value.lower() in VERDICTS:
verdict = value.lower()
elif key == "reason" and value:
reason = value
return verdict, reason
def check(cases: list[tuple[str, QueryResult]]) -> dict[str, tuple[str, str]]:
replies = check_chain.batch(
[
{"subquestion": question, "sql": result.sql, "rows": render(result)}
for question, result in cases
],
config={"max_concurrency": CHECK_CONCURRENCY},
)
return {
question: _verdict(reply)
for (question, _), reply in zip(cases, replies, strict=True)
}
"You cannot check the numbers" is in the prompt because it is true, and a checker that believes otherwise is worse than no checker. It is looking at up to fifty rows of output; it has no access to the database, nothing to recompute against, and the one thing it could do with a total is guess. What it can do is notice that a sub-question asking for a breakdown by region came back as a single row, or that a revenue figure was computed over every invoice status. Both of those are visible in the output. Scoping the role to what is visible is what makes its verdicts worth a call.
An unparseable reply defaults to answers. That is deliberate and it is a cost
decision: the opposite default turns every malformed checker response into a
rewrite on the stronger model, so a cheap role failing quietly would start
spending money in an expensive one. Failing towards "accept" means a bad result
reaches the narrator, which is visible in the answer; failing towards "retry"
means a bill, which is not.
Note which role is not here. Nothing asks a model whether a result set is empty —
len(rows) does that, and an empty result goes straight to a rewrite note without
a checker call. The cheapest call in any agent is the one some ordinary Python
made unnecessary.
Step 6 — Write the answer from the tables, not from the database
# fragment
ANSWER_PROMPT = ChatPromptTemplate.from_messages(
[
(
"system",
"You answer one business question from result tables that have "
"already been computed. Use only the figures in front of you and "
"never calculate a new one — not a total, not a share, not a "
"difference between two rows. Money is already formatted; copy it "
"as printed. Name the sub-question each figure came from. If any "
"sub-question went unanswered, say so in the first sentence and "
"say what is therefore not covered. Under 150 words, plain "
"sentences, no bullet list, and no recommendation the rows do not "
"support.",
),
("human", "Question: {question}\n\n{evidence}"),
]
)
narrator_llm = ChatAnthropic(model=CHECK_MODEL, max_tokens=400, temperature=0)
answer_chain = ANSWER_PROMPT | narrator_llm | StrOutputParser()
def narrate(question: str, evidence: str) -> str:
return answer_chain.invoke({"question": question, "evidence": evidence})
The narrator never sees the database and never sees a query result that was not checked. Its whole input is tables Python rendered from rows SQLite computed, which is the difference between a summary that is true by construction and a second, unchecked reading of the same data.
The instruction about unanswered sub-questions is written now, before there is anything to be unanswered, because it is the hook the budget section hangs on. An analyst agent's failure mode is not a crash; it is a fluent paragraph about four fifths of the evidence.
Step 7 — The supervisor
One function, and every branch the agent can take is visible in it.
# fragment
from typing import Any
MAX_ROUNDS = 2
def run_analysis(question: str) -> dict[str, Any]:
conn = read_only(DB_PATH)
try:
subquestions = plan(question) or [question]
pending = list(subquestions)
answered: dict[str, QueryResult] = {}
notes: dict[str, str] = {}
rounds = 0
while pending and rounds < MAX_ROUNDS:
rounds += 1
written = write_sql(pending, notes)
executed: list[tuple[str, QueryResult]] = []
for item in pending:
result = run_query(conn, written[item])
if result.error is not None:
notes[item] = result.error
elif not result.rows:
notes[item] = (
"the query ran and returned no rows; check the filter "
"and the join before narrowing anything further"
)
else:
executed.append((item, result))
verdicts = check(executed) if executed else {}
for item, result in executed:
verdict, reason = verdicts.get(item, ("answers", ""))
if verdict == "answers":
answered[item] = result
else:
notes[item] = f"{verdict}: {reason}"
pending = [item for item in pending if item not in answered]
blocks = [
f"Sub-question: {item}\nSQL: {answered[item].sql}\n{render(answered[item])}"
for item in subquestions
if item in answered
]
evidence = "\n\n".join(blocks) or "No sub-question could be answered."
if pending:
evidence += "\n\nUnanswered sub-questions:\n" + "\n".join(
f"- {item} ({notes.get(item, 'no reason was recorded')})"
for item in pending
)
return {
"question": question,
"answer": narrate(question, evidence),
"answered": [
{"subquestion": item, "sql": answered[item].sql}
for item in subquestions
if item in answered
],
"unanswered": [
{"subquestion": item, "reason": notes.get(item, "")}
for item in pending
],
"rounds": rounds,
}
finally:
conn.close()
MAX_ROUNDS bounds the obvious loop. A sub-question whose query keeps coming back
wrong-shaped can be rewritten forever, and each lap is a call on the stronger
model carrying the whole schema. Two is defensible because the second attempt has
something the first did not — the error or the verdict, in the prompt — and a third
at temperature zero is unlikely to be different. What the third attempt would
really need is a better sub-question, which is the planner's job, not the writer's.
plan(question) or [question] is the fallback when the planner returns nothing
parseable. The original question becomes the single sub-question and the run
continues at the floor cost rather than ending with an empty result and a model
call already paid for.
Unanswered sub-questions travel in the result and in the evidence block. The first is for your code; the second is for the narrator, which is the only one of the two a person reads.
Count the calls: one plan, one to five SQL writes, up to one check per sub-question that returned rows, the same again on a second round, and one answer. Four at the floor, twenty-two at the cap. The one and the one come from your traffic. Everything between is a decision another model made.
The whole file
"""analyst.py — a SQL analyst agent: plan, write SQL, check the rows, answer."""
import re
import sqlite3
import tempfile
from dataclasses import dataclass
from decimal import Decimal
from pathlib import Path
from typing import Any
from langchain_anthropic import ChatAnthropic
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
PLAN_MODEL = "claude-sonnet-5-5"
SQL_MODEL = "claude-sonnet-5-5"
CHECK_MODEL = "claude-haiku-5-5"
MAX_SUBQUESTIONS = 5
MAX_ROUNDS = 2
MAX_ROWS = 50
SQL_CONCURRENCY = 3
CHECK_CONCURRENCY = 4
VERDICTS = ("answers", "wrong-shape", "wrong-filter")
FORBIDDEN = re.compile(
r"\b(attach|detach|pragma|insert|update|delete|drop|alter|create|replace|vacuum)\b",
re.IGNORECASE,
)
DB_PATH = Path(tempfile.gettempdir()) / "analyst-demo.sqlite"
SCHEMA = """
CREATE TABLE accounts (
account_id TEXT PRIMARY KEY,
name TEXT NOT NULL,
region TEXT NOT NULL,
plan TEXT NOT NULL,
started_on TEXT NOT NULL,
churned_on TEXT
);
CREATE TABLE invoices (
invoice_id TEXT PRIMARY KEY,
account_id TEXT NOT NULL REFERENCES accounts(account_id),
issued_on TEXT NOT NULL,
amount_cents INTEGER NOT NULL,
status TEXT NOT NULL
);
CREATE TABLE tickets (
ticket_id TEXT PRIMARY KEY,
account_id TEXT NOT NULL REFERENCES accounts(account_id),
opened_on TEXT NOT NULL,
closed_on TEXT,
severity TEXT NOT NULL,
channel TEXT NOT NULL
);
"""
SCHEMA_NOTES = """
Dates are ISO strings, YYYY-MM-DD, and compare correctly with <, > and BETWEEN.
accounts.churned_on is NULL while an account is still a customer. An account was
active on a date if started_on <= that date and churned_on IS NULL or
churned_on > that date.
invoices.amount_cents is an integer in minor units. It is not a currency amount
and it is never divided.
invoices.status is one of paid, open, void. A void invoice was cancelled and was
never owed by anyone; revenue counts paid invoices only.
tickets.closed_on is NULL while a ticket is still open.
tickets.severity is one of sev1, sev2, sev3, and sev1 is the most severe.
There is no revenue column, no churn flag and no account health score anywhere
in this database. Every one of those is computed from the rows above.
"""
ACCOUNTS = [
("acc-01", "Halversen Labs", "emea", "scale", "2025-02-11", None),
("acc-02", "Northgate Freight", "emea", "team", "2025-06-03", "2026-04-30"),
("acc-03", "Pellier Tooling", "emea", "team", "2026-01-19", None),
("acc-04", "Tamsin Print", "amer", "team", "2025-09-24", None),
("acc-05", "Brightwater Mutual", "amer", "scale", "2024-11-05", None),
("acc-06", "Castellan Foods", "amer", "scale", "2025-03-17", "2026-07-31"),
("acc-07", "Okonjo Mobility", "apac", "team", "2026-02-02", None),
("acc-08", "Serra Analytics", "apac", "scale", "2025-08-08", "2026-02-28"),
]
INVOICES = [
("inv-1001", "acc-01", "2026-01-05", 480_000, "paid"),
("inv-1002", "acc-01", "2026-04-05", 480_000, "paid"),
("inv-1003", "acc-01", "2026-07-05", 520_000, "paid"),
("inv-1004", "acc-01", "2026-10-05", 520_000, "open"),
("inv-1005", "acc-02", "2026-01-05", 90_000, "paid"),
("inv-1006", "acc-02", "2026-04-05", 90_000, "void"),
("inv-1007", "acc-03", "2026-02-01", 75_000, "paid"),
("inv-1008", "acc-03", "2026-05-01", 75_000, "paid"),
("inv-1009", "acc-03", "2026-08-01", 75_000, "paid"),
("inv-1010", "acc-04", "2026-01-12", 110_000, "paid"),
("inv-1011", "acc-04", "2026-04-12", 110_000, "paid"),
("inv-1012", "acc-04", "2026-07-12", 110_000, "paid"),
("inv-1013", "acc-04", "2026-10-12", 110_000, "open"),
("inv-1014", "acc-05", "2026-01-03", 640_000, "paid"),
("inv-1015", "acc-05", "2026-04-03", 640_000, "paid"),
("inv-1016", "acc-05", "2026-07-03", 700_000, "paid"),
("inv-1017", "acc-05", "2026-10-03", 700_000, "open"),
("inv-1018", "acc-06", "2026-01-20", 410_000, "paid"),
("inv-1019", "acc-06", "2026-04-20", 410_000, "paid"),
("inv-1020", "acc-06", "2026-07-20", 410_000, "void"),
("inv-1021", "acc-07", "2026-03-02", 68_000, "paid"),
("inv-1022", "acc-07", "2026-06-02", 68_000, "paid"),
("inv-1023", "acc-07", "2026-09-02", 68_000, "open"),
("inv-1024", "acc-08", "2026-01-08", 455_000, "paid"),
("inv-1025", "acc-08", "2026-04-08", 455_000, "void"),
]
TICKETS = [
("tkt-5001", "acc-01", "2026-02-14", "2026-02-15", "sev3", "email"),
("tkt-5002", "acc-02", "2026-02-20", "2026-03-04", "sev1", "phone"),
("tkt-5003", "acc-02", "2026-03-11", "2026-04-02", "sev1", "phone"),
("tkt-5004", "acc-02", "2026-04-06", None, "sev2", "email"),
("tkt-5005", "acc-04", "2026-05-09", "2026-05-12", "sev3", "chat"),
("tkt-5006", "acc-05", "2026-06-01", "2026-06-02", "sev2", "email"),
("tkt-5007", "acc-06", "2026-05-18", "2026-06-30", "sev1", "phone"),
("tkt-5008", "acc-06", "2026-06-22", "2026-07-28", "sev1", "chat"),
("tkt-5009", "acc-06", "2026-07-15", None, "sev2", "email"),
("tkt-5010", "acc-07", "2026-07-07", "2026-07-08", "sev3", "chat"),
("tkt-5011", "acc-08", "2026-01-22", "2026-02-19", "sev1", "email"),
("tkt-5012", "acc-03", "2026-09-30", None, "sev3", "chat"),
]
QUESTION = (
"Which region should we be most worried about going into 2027, "
"and what in the data says so?"
)
PLAN_PROMPT = ChatPromptTemplate.from_messages(
[
(
"system",
"You turn one business question into the smallest set of "
"sub-questions that together answer it. Each sub-question must be "
"answerable by a single SELECT against the schema below. Write one "
"per line, in this exact form, and nothing else:\n"
"q :: the sub-question, as a sentence\n"
"At most five, and fewer is better: each one costs a query, and two "
"that would return the same table are one sub-question. Ask only "
"for facts these tables can supply. Never ask for a judgement, a "
"recommendation, a cause or a forecast — those are not in the "
"database and a query cannot return one.\n"
"Schema:\n{schema}\n"
"What the schema does not say:\n{notes}",
),
("human", "{question}"),
]
)
SQL_PROMPT = ChatPromptTemplate.from_messages(
[
(
"system",
"You write exactly one SQLite SELECT that answers the sub-question. "
"Reply with the statement and nothing else: no explanation, no code "
"fence, no second statement.\n"
"Rules that are not in the schema:\n"
"Keep money in minor units as an integer, and give any money column "
"a name ending in _cents. Never divide an amount and never round "
"one.\n"
"Name every computed column with AS.\n"
"Put the most important row first and LIMIT the result to 50 rows "
"or fewer.\n"
"Return the smallest set of columns that answers the "
"sub-question.\n"
"Schema:\n{schema}\n"
"What the schema does not say:\n{notes}\n"
"What went wrong last time, if anything: {note}",
),
("human", "{subquestion}"),
]
)
CHECK_PROMPT = ChatPromptTemplate.from_messages(
[
(
"system",
"You are given a sub-question and the first rows a query returned "
"for it. Say whether those rows are the thing the sub-question "
"asked for. Exactly two lines, in this form:\n"
"verdict :: one of answers, wrong-shape, wrong-filter\n"
"reason :: one sentence under twenty words\n"
"wrong-shape means the rows are not the thing asked for: the wrong "
"grain, the wrong columns, a single total where a breakdown was "
"asked for. wrong-filter means the right kind of rows counted over "
"the wrong set, such as a revenue figure that includes void "
"invoices or a customer count that includes churned accounts. "
"Everything else is answers.\n"
"You are not being asked whether the numbers are right and you "
"cannot check them. Judge the shape and the filter only.",
),
(
"human",
"Sub-question: {subquestion}\nQuery:\n{sql}\nRows:\n{rows}",
),
]
)
ANSWER_PROMPT = ChatPromptTemplate.from_messages(
[
(
"system",
"You answer one business question from result tables that have "
"already been computed. Use only the figures in front of you and "
"never calculate a new one — not a total, not a share, not a "
"difference between two rows. Money is already formatted; copy it "
"as printed. Name the sub-question each figure came from. If any "
"sub-question went unanswered, say so in the first sentence and "
"say what is therefore not covered. Under 150 words, plain "
"sentences, no bullet list, and no recommendation the rows do not "
"support.",
),
("human", "Question: {question}\n\n{evidence}"),
]
)
planner_llm = ChatAnthropic(model=PLAN_MODEL, max_tokens=512, temperature=0)
sql_llm = ChatAnthropic(model=SQL_MODEL, max_tokens=600, temperature=0)
checker_llm = ChatAnthropic(model=CHECK_MODEL, max_tokens=160, temperature=0)
narrator_llm = ChatAnthropic(model=CHECK_MODEL, max_tokens=400, temperature=0)
plan_chain = PLAN_PROMPT | planner_llm | StrOutputParser()
sql_chain = SQL_PROMPT | sql_llm | StrOutputParser()
check_chain = CHECK_PROMPT | checker_llm | StrOutputParser()
answer_chain = ANSWER_PROMPT | narrator_llm | StrOutputParser()
@dataclass
class QueryResult:
sql: str
columns: list[str]
rows: list[tuple]
truncated: bool
error: str | None
def build_db(path: Path) -> None:
path.unlink(missing_ok=True)
conn = sqlite3.connect(path)
try:
conn.executescript(SCHEMA)
conn.executemany("INSERT INTO accounts VALUES (?, ?, ?, ?, ?, ?)", ACCOUNTS)
conn.executemany("INSERT INTO invoices VALUES (?, ?, ?, ?, ?)", INVOICES)
conn.executemany("INSERT INTO tickets VALUES (?, ?, ?, ?, ?, ?)", TICKETS)
conn.commit()
finally:
conn.close()
def read_only(path: Path) -> sqlite3.Connection:
return sqlite3.connect(f"file:{path}?mode=ro", uri=True)
def extract_sql(raw: str) -> str:
kept = [
line
for line in raw.strip().splitlines()
if not line.lstrip().startswith("`")
]
return "\n".join(kept).strip().rstrip(";").strip()
def check_sql(sql: str) -> str | None:
if not sql:
return "no statement was returned"
if ";" in sql:
return "more than one statement was returned; send exactly one SELECT"
head = sql.lstrip("( \n\t").split(None, 1)
if not head or head[0].lower() not in {"select", "with"}:
return "only a single SELECT or WITH statement may be run"
found = FORBIDDEN.search(sql)
if found:
return f"the statement contains {found.group(0).upper()}, which may not be run"
return None
def run_query(conn: sqlite3.Connection, sql: str) -> QueryResult:
problem = check_sql(sql)
if problem is not None:
return QueryResult(sql, [], [], False, problem)
try:
cursor = conn.execute(sql)
fetched = cursor.fetchmany(MAX_ROWS + 1)
except sqlite3.Error as exc:
return QueryResult(sql, [], [], False, f"sqlite rejected the query: {exc}")
columns = [description[0] for description in cursor.description or []]
rows = [tuple(row) for row in fetched[:MAX_ROWS]]
return QueryResult(sql, columns, rows, len(fetched) > MAX_ROWS, None)
def cell(column: str, value: object) -> str:
if value is None:
return "null"
if column.endswith("_cents") and isinstance(value, int):
return f"{Decimal(value).scaleb(-2):.2f}"
return str(value)
def render(result: QueryResult) -> str:
if result.error is not None:
return f"error: {result.error}"
if not result.rows:
return "no rows"
body = "\n".join(
" | ".join(cell(column, value) for column, value in zip(result.columns, row))
for row in result.rows
)
tail = f"\n...capped at {MAX_ROWS} rows" if result.truncated else ""
return f"{' | '.join(result.columns)}\n{body}{tail}"
def _subquestions(raw: str) -> list[str]:
out: list[str] = []
for row in raw.splitlines():
key, sep, value = row.partition("::")
text = value.strip()
if sep and key.strip().lower() == "q" and text and text not in out:
out.append(text)
return out[:MAX_SUBQUESTIONS]
def _verdict(raw: str) -> tuple[str, str]:
verdict, reason = "answers", ""
for row in raw.splitlines():
key, _, value = row.partition("::")
key, value = key.strip().lower(), value.strip()
if key == "verdict" and value.lower() in VERDICTS:
verdict = value.lower()
elif key == "reason" and value:
reason = value
return verdict, reason
def plan(question: str) -> list[str]:
raw = plan_chain.invoke(
{"question": question, "schema": SCHEMA, "notes": SCHEMA_NOTES}
)
return _subquestions(raw)
def write_sql(subquestions: list[str], notes: dict[str, str]) -> dict[str, str]:
replies = sql_chain.batch(
[
{
"subquestion": question,
"schema": SCHEMA,
"notes": SCHEMA_NOTES,
"note": notes.get(question, "nothing, this is the first attempt"),
}
for question in subquestions
],
config={"max_concurrency": SQL_CONCURRENCY},
)
return {
question: extract_sql(reply)
for question, reply in zip(subquestions, replies, strict=True)
}
def check(cases: list[tuple[str, QueryResult]]) -> dict[str, tuple[str, str]]:
replies = check_chain.batch(
[
{"subquestion": question, "sql": result.sql, "rows": render(result)}
for question, result in cases
],
config={"max_concurrency": CHECK_CONCURRENCY},
)
return {
question: _verdict(reply)
for (question, _), reply in zip(cases, replies, strict=True)
}
def narrate(question: str, evidence: str) -> str:
return answer_chain.invoke({"question": question, "evidence": evidence})
def run_analysis(question: str) -> dict[str, Any]:
conn = read_only(DB_PATH)
try:
subquestions = plan(question) or [question]
pending = list(subquestions)
answered: dict[str, QueryResult] = {}
notes: dict[str, str] = {}
rounds = 0
while pending and rounds < MAX_ROUNDS:
rounds += 1
written = write_sql(pending, notes)
executed: list[tuple[str, QueryResult]] = []
for item in pending:
result = run_query(conn, written[item])
if result.error is not None:
notes[item] = result.error
elif not result.rows:
notes[item] = (
"the query ran and returned no rows; check the filter "
"and the join before narrowing anything further"
)
else:
executed.append((item, result))
verdicts = check(executed) if executed else {}
for item, result in executed:
verdict, reason = verdicts.get(item, ("answers", ""))
if verdict == "answers":
answered[item] = result
else:
notes[item] = f"{verdict}: {reason}"
pending = [item for item in pending if item not in answered]
blocks = [
f"Sub-question: {item}\nSQL: {answered[item].sql}\n{render(answered[item])}"
for item in subquestions
if item in answered
]
evidence = "\n\n".join(blocks) or "No sub-question could be answered."
if pending:
evidence += "\n\nUnanswered sub-questions:\n" + "\n".join(
f"- {item} ({notes.get(item, 'no reason was recorded')})"
for item in pending
)
return {
"question": question,
"answer": narrate(question, evidence),
"answered": [
{"subquestion": item, "sql": answered[item].sql}
for item in subquestions
if item in answered
],
"unanswered": [
{"subquestion": item, "reason": notes.get(item, "")}
for item in pending
],
"rounds": rounds,
}
finally:
conn.close()
if __name__ == "__main__":
build_db(DB_PATH)
result = run_analysis(QUESTION)
print(result["answer"])
print()
for item in result["answered"]:
print(f"answered {item['subquestion']}")
print(f" {item['sql']}")
for item in result["unanswered"]:
print(f"unanswered {item['subquestion']} — {item['reason']}")
print(f"rounds: {result['rounds']}")
python analyst.py
You get a written answer, the sub-questions that were answered with the SQL that
answered them, and anything that was not. The interesting output is the SQL list:
it is the agent showing its working, and it is the only part of the result a
person can check. Read the WHERE clauses — whether status = 'paid' appears in
the revenue queries, and whether churned_on IS NULL appears in the ones about
customers, is the difference between an answer and a paragraph.
The agent is finished. Nothing below changes what it does.
Seeing what it cost
What you do not have is any idea which of the four roles spent the money. The provider's invoice has one line for the API key, and the four to twenty-two calls behind it are indistinguishable — same key, same account, and the two roles on the stronger model are the two whose costs differ most.
pip install capsera
At the top of analyst.py, above everything else:
# fragment
import os
import capsera
capsera.init(api_key=os.environ["CAPSERA_API_KEY"], endpoint="https://api.capsera.ai")
That is the capture step, and it is already complete. capsera.init() patches the
provider client that ChatAnthropic constructs underneath, so from that call on
every request the agent makes is recorded: the plan call, each SQL write inside
batch(), each check, the answer. The prompts are untouched, the chains are
untouched, batch() is untouched, and no traffic changed route — the SDK runs
inside your process rather than in front of it, so there is no proxy, no
base_url to swap and nothing new in the request path.
One addition for a script: a background thread ships events on an interval, so a process that exits immediately can exit before the queue drains.
# fragment
if __name__ == "__main__":
build_db(DB_PATH)
result = run_analysis(QUESTION)
print(result["answer"])
capsera.shutdown()
capsera.shutdown() flushes pending events and stops the worker. A long-running
service does not need it; a script, a notebook cell or a question asked from a
terminal does, and omitting it is the usual reason a first run reports nothing.
Run it now and you have a total, a per-model split and token counts per call. You
also have every one of those events attributed to unknown, because nothing has
said which role made it. unknown is a real row, not a dropped one — an
unattributed cost is still a cost — but it is one row where you wanted four.
Name the roles
The calls are already captured. The decorator only says who:
# fragment
@capsera.agent("planner", team="analytics", task_type="question-decomposition")
def plan(question: str) -> list[str]:
... # body unchanged
@capsera.agent("sql_writer", team="analytics", task_type="sql-generation")
def write_sql(subquestions: list[str], notes: dict[str, str]) -> dict[str, str]:
... # body unchanged
@capsera.agent("checker", team="analytics", task_type="result-check")
def check(cases: list[tuple[str, QueryResult]]) -> dict[str, tuple[str, str]]:
... # body unchanged
@capsera.agent("narrator", team="analytics", task_type="answer-writing")
def narrate(question: str, evidence: str) -> str:
... # body unchanged
Four added lines, and that is the entire diff. No function body changed, no call site changed, no model construction changed, no argument threaded through the supervisor. Every LLM call made while a decorated function is on the stack is attributed to that agent, including calls made by the functions it calls and by the framework it invokes.
The fan-out keeps its identity. write_sql and check both do their work in
.batch(...), which dispatches to worker threads rather than calling on the
thread you were on, and the decorated scope is still what gets recorded. Capsera
holds the attribution stack in a ContextVar and LangChain's batch() dispatches
through a context-copying executor, so the scope the caller was in is the scope
the worker sees — verified empirically on LangChain 1.3 with capsera 0.3.0.
batch() also produces one event per underlying call rather than one per batch,
so five sub-questions are five sql_writer events
(coverage table).
The credit there belongs to LangChain, and it is worth knowing where the guarantee
stops. A ContextVar does not cross a thread boundary on its own — it crosses
when the code starting the thread copies the context, which batch() does and a
hand-rolled pool does not. Fan the sub-questions out yourself with a bare
ThreadPoolExecutor and the same decorator records unknown instead of
sql_writer, with no error to tell you. Measured on the same versions:
# fragment
import contextvars
from concurrent.futures import ThreadPoolExecutor
# Loses the scope: workers start with a fresh context.
with ThreadPoolExecutor(max_workers=SQL_CONCURRENCY) as pool:
written = list(pool.map(write_one, subquestions))
# Keeps it: hand each worker a copy of the caller's context.
ctx = contextvars.copy_context()
with ThreadPoolExecutor(max_workers=SQL_CONCURRENCY) as pool:
written = list(pool.map(lambda item: ctx.run(write_one, item), subquestions))
Async needs nothing extra — a ContextVar is snapshotted per task, so roles
awaited concurrently under asyncio.gather keep their own attribution.
Leave the supervisor undecorated. run_analysis makes no LLM call of its own;
every call inside it happens while one of the four roles is on the stack, and the
innermost scope is the one recorded. Decorating it would add a row that can never
have spend on it. For a per-question total, use the team field — which is what
team="analytics" on all four decorators is for.
Cost the rewrites separately
The second round is the part of this agent whose cost you cannot see from the role
rows alone. A question whose five queries all landed first time and a question
where three of them had to be rewritten both report as sql_writer, on the
stronger model, with the whole schema in the prompt, and the row says nothing
about which. capsera.tag() opens a sub-scope inside a function you have already
decorated, which is exactly this case:
# fragment
import capsera
@capsera.agent("sql_writer", team="analytics", task_type="sql-generation")
def write_sql(subquestions: list[str], notes: dict[str, str]) -> dict[str, str]:
first = [item for item in subquestions if item not in notes]
again = [item for item in subquestions if item in notes]
written = _write_sql(first, notes) if first else {}
if again:
# A rewrite is the same role writing SQL for the same sub-question
# because the last statement failed, returned nothing, or returned the
# wrong thing. Cost the second attempt separately from the first.
with capsera.tag("sql_writer", team="analytics", task_type="sql-rewrite"):
written.update(_write_sql(again, notes))
return written
_write_sql is the original body, moved down one level and otherwise unchanged.
In this supervisor the two lists are never both populated — a round-two
sub-question is by definition one that collected a note in round one — so the split
costs no extra wall-clock here. Write it this way anyway: the day a rewrite and a
fresh sub-question share a round, the code is already right.
The rewrite keeps the agent name, so a budget scoped to sql_writer still counts
both passes, and it changes the task type, so the share of SQL spend that came
from statements that did not work first time is a number you can read instead of a
suspicion. That number is also a schema-quality signal: a rewrite rate that climbs
is usually a table whose meaning is not in SCHEMA_NOTES yet.
Use tag() for a genuine sub-scope like this one. It is the wrong tool for naming
an iteration: tagging each sub-question with its own text would give you one row
per sentence, none of which appear in tomorrow's questions, and an agent budget
cannot accumulate against a name that lived for one run. The per-invocation axis
has its own fields on the same event — customer_id and cost_center on
@capsera.agent, session_id on capsera.tag — so you can slice by which team or
which cost centre asked the question without fragmenting the agent dimension that
budgets and cost attribution depend on.
Put a budget on the role that multiplies
Three of the four roles have a call count you already know. The planner runs once per question and the narrator runs once per question, so both of those rows are your traffic multiplied by one. Double the questions and you have doubled two small, predictable rows.
The SQL writer's count is not in your file. It is the number of sub-questions the
planner invented, plus the number of those that have to be written again, capped
at MAX_ROUNDS. One question where the planner asks for one sub-question is one
call; the same question where it asks for five, three of which need a rewrite, is
eight. Nothing you wrote decides which. And the SQL writer is also the role on the
stronger model carrying the schema and the column notes in every prompt — so the
call count and the cost per call grow together. That is
multi-agent amplification in one sentence,
and it is why the cap belongs on the SQL writer rather than on the question.
The checker is the useful comparison. In round one it runs exactly as often as the SQL writer, and it costs a fraction as much, because it is on the fast model with one sentence and at most fifty rows in front of it. Two roles, the same call count, bills that are not close. A per-model split cannot show you that — the planner and the SQL writer share a model — and the provider's invoice cannot show you any of it.
Create the budget in the dashboard with scope agent and the agent set to
sql_writer — the same string as the decorator. That string is the whole join
between the two halves of this page: a budget can only act on a boundary the
events carry, and spend attributed to unknown cannot be governed by a
per-agent budget. Pick the amount from a week of
your own questions rather than a figure from a tutorial, and set the action to
block.
Enforcement is on by default (enable_budget_enforcement=True in init()) and
the check runs before the provider call. When the SQL writer is over its
budget, BudgetExceededError is raised inside your process and the request never
leaves it, so no tokens are spent on it. This is
pre-call enforcement: the check happens where
your code is, not in front of the provider, and a blocked call costs you the check
rather than the completion.
Then decide what a blocked sub-question means. batch() raises on the first
failure and abandons the remaining inputs unless you tell it otherwise, so this is
a decision you make rather than one you inherit:
# fragment
import capsera
def _write_sql(
subquestions: list[str], notes: dict[str, str]
) -> tuple[dict[str, str], list[str]]:
"""The SQL written, and the sub-questions a budget stopped."""
replies = sql_chain.batch(
[
{
"subquestion": question,
"schema": SCHEMA,
"notes": SCHEMA_NOTES,
"note": notes.get(question, "nothing, this is the first attempt"),
}
for question in subquestions
],
config={"max_concurrency": SQL_CONCURRENCY},
return_exceptions=True,
)
written: dict[str, str] = {}
blocked: list[str] = []
for question, reply in zip(subquestions, replies, strict=True):
if isinstance(reply, capsera.BudgetExceededError):
blocked.append(question)
continue
if isinstance(reply, BaseException):
raise RuntimeError(f"writing SQL for {question!r} failed: {reply}")
written[question] = extract_sql(reply)
return written, blocked
return_exceptions=True asks batch() to hand failures back as items in the
result list instead of raising on the first one, which is what lets you tell a
budget block from a flaky call and name the sub-questions that were never written.
Without it, one blocked sub-question ends the round and the rest are never
attempted.
_write_sql now returns two values, so the decorated write_sql above returns
them too — it merges the two blocked lists the same way it merges the two
dictionaries, and its annotation becomes tuple[dict[str, str], list[str]]. The
decorator and the tag() scope do not change.
Here is the trap, and in an analyst agent it is worse than a crash. A budget that blocks two of five sub-questions does not produce a short answer. It produces a confident one. The narrator is handed whatever evidence exists and writes fluent prose over three result tables, and nothing in that paragraph says the other two were never asked. Somebody takes it into a meeting. The blocked sub-questions have to reach the evidence block, not just the return value:
# fragment
from typing import Any
def run_analysis(question: str) -> dict[str, Any]:
conn = read_only(DB_PATH)
try:
subquestions = plan(question) or [question]
pending = list(subquestions)
answered: dict[str, QueryResult] = {}
notes: dict[str, str] = {}
blocked: list[str] = []
rounds = 0
while pending and rounds < MAX_ROUNDS:
rounds += 1
written, stopped = write_sql(pending, notes)
blocked.extend(stopped)
# A sub-question nobody could afford to write SQL for is not a
# failed sub-question. There is nothing to rewrite and nothing for
# the checker to look at; it was never asked.
pending = [item for item in pending if item in written]
... # execute, check and sort the rest exactly as before
evidence = "\n\n".join(
f"Sub-question: {item}\nSQL: {answered[item].sql}\n{render(answered[item])}"
for item in subquestions
if item in answered
)
if blocked:
evidence += "\n\nNot asked at all, because the query budget was "
evidence += "reached:\n" + "\n".join(f"- {item}" for item in blocked)
return {
"question": question,
"answer": narrate(question, evidence),
# Not a finished analysis. Do not quote this without reading it.
"partial": bool(blocked),
"blocked": blocked,
"rounds": rounds,
}
finally:
conn.close()
The narrator still runs, and the instruction it was given in Step 6 — say so in the first sentence, and say what is therefore not covered — is what turns a blocked budget into a sentence a reader sees. One cheap call on the fast model is a good price for the difference between a partial answer that announces itself and one that does not.
Blocked sub-questions are kept out of pending on purpose. A sub-question with no
SQL has nothing to execute, nothing for the checker to read and no note to rewrite
from; leaving it in the loop buys a second round that cannot do anything, and if
the budget has cleared by then it buys the expensive call you just declined to
make.
Three limits to know before you rely on this as a hard ceiling.
The blast radius is the role, not the question. A budget on sql_writer stops
SQL generation. The planner is a different agent with no budget on it, and it runs
first — on the same stronger model, with the same schema in its prompt — so a
question that is blocked before a single query is written has still paid for its
own decomposition. That is one call, and it is not zero. If what you want capped
is the whole agent, that is a team-scoped budget on analytics, not a side effect
of this one.
A parallel pass can overshoot. The check reads budget state computed from
delivered events, and events are delivered in batches on an interval
(flush_interval_ms, 500 ms by default), so a burst of concurrent calls can each
pass a check taken before any of them was recorded. Here the burst is
SQL_CONCURRENCY — three as shipped, and whatever you raise it to when questions
start arriving from more than one person. Set the budget below a figure you cannot
exceed, or lower the concurrency; it is the concurrency that decides the
overshoot, not the number of sub-questions.
It fails open, on purpose. If the pre-call check does not complete within
budget_check_timeout (1 second by default), the call is allowed. A network
problem between your service and Capsera should not halt production traffic. Of
the three actions, block and margin downgrade act at runtime; throttle records
the decision but does not delay the call. The full behaviour is in
budgets and enforcement.
What this run actually cost
No dollar figures on this page: this agent has not been run against a real key, and a number invented here would not be yours anyway — it depends on your schema, how many sub-questions your questions decompose into, and today's prices. What is worth predicting is the shape of the result.
Four agent rows — planner, sql_writer, checker, narrator — under one team,
analytics, with a cost per run for the question as a
whole, and the sql_writer row split by task type into sql-generation and
sql-rewrite.
Read each row as call count and cost per call, not as a total. Three things move
on this agent and each has a different fix. The planner's call count moving means
people are asking more questions, which is the agent working. The SQL writer's
count moving while the question count stayed flat means the planner is
decomposing more finely, which costs the stronger model every time and is fixed in
the planner's prompt or in MAX_SUBQUESTIONS, not in the SQL prompt. And the
sql-rewrite share moving means more statements are failing or coming back
wrong-shaped, which usually means the schema changed and SCHEMA_NOTES did not.
Expect sql_writer to dominate the cost and planner to have the smallest call
count of anything that matters. The pair to watch is sql_writer against
checker: in round one they run the same number of times, so the gap between
those two rows is a pure read on what a model choice and a prompt size are worth.
If the gap ever narrows, something has put more into the checker's prompt than one
sub-question and fifty rows — which is what a raised MAX_ROWS does.
Both of the stronger-model roles also carry an identical block in front of their
variable part: the schema and the column notes, byte for byte the same on every
call in a run. The cache read share column is what tells you whether any of it was
billed five times instead of once. It is a number to read rather than one to
predict, because a shared prefix has to be long enough and identical enough to be
worth anything, and putting the schema before the sub-question is what earns it.
Turning on enable_prompt_analysis=True in init() puts a figure on how often one
system-prompt hash repeats, which is the evidence for whether that restructuring
is worth doing: prompt caching that never
engaged.
Each event also carries the file that made the call, which points at analyst.py
rather than into langchain_core, so a row you did not expect traces back to code
rather than to a model name.
Questions this page answers
How do I get per-agent cost for a LangChain SQL analyst agent? Call
capsera.init() once, which patches the provider client ChatAnthropic
constructs underneath, so every call the agent makes is already recorded. Then put
@capsera.agent("<role>") on each role function — the planner, the SQL writer, the
checker, the narrator. The decorator says who made the call; it does not change
whether the call is captured, and no function body, call site or model
construction changes.
Which role in a SQL analyst agent should carry the budget? The SQL writer. The planner runs exactly once per question, so its call count is your traffic — a number you already have. The SQL writer runs once per sub-question the planner invented, again for every sub-question that has to be rewritten, on the stronger model, with the whole schema in every prompt. Its count is a decision another model made and its prompt is the largest in the agent.
What happens to the answer when a budget blocks some of the queries? Nothing, unless you make it. The narrator is handed whatever result tables exist and will write a confident answer over three of five sub-questions, with no sentence saying which two are missing. Carry the blocked sub-questions into the evidence block the narrator reads, and mark the result partial, so a short run reads as a short run rather than as an answer.
Why do the SQL writer and the result checker cost different amounts when they run the same number of times? Because a call's price is its model and its tokens, not its position in the pipeline. In the first round both roles run once per sub-question. The SQL writer sends the full schema and the column notes to the stronger model every time; the checker sends one sentence and up to fifty rows to the fast model. Identical call counts, different bills — which is why a per-model split cannot stand in for a per-agent one.
Questions this page answers
- How do I get per-agent cost for a LangChain SQL analyst agent?
- Call capsera.init() once, which patches the provider client ChatAnthropic constructs underneath, so every call the agent makes is already recorded. Then put @capsera.agent("<role>") on each role function — the planner, the SQL writer, the checker, the narrator. The decorator says who made the call; it does not change whether the call is captured, and no function body, call site or model construction changes.
- Which role in a SQL analyst agent should carry the budget?
- The SQL writer. The planner runs exactly once per question, so its call count is your traffic — a number you already have. The SQL writer runs once per sub-question the planner invented, again for every sub-question that has to be rewritten, on the stronger model, with the whole schema in every prompt. Its count is a decision another model made and its prompt is the largest in the agent.
- What happens to the answer when a budget blocks some of the queries?
- Nothing, unless you make it. The narrator is handed whatever result tables exist and will write a confident answer over three of five sub-questions, with no sentence saying which two are missing. Carry the blocked sub-questions into the evidence block the narrator reads, and mark the result partial, so a short run reads as a short run rather than as an answer.
- Why do the SQL writer and the result checker cost different amounts when they run the same number of times?
- Because a call's price is its model and its tokens, not its position in the pipeline. In the first round both roles run once per sub-question. The SQL writer sends the full schema and the column notes to the stronger model every time; the checker sends one sentence and up to fifty rows to the fast model. Identical call counts, different bills — which is why a per-model split cannot stand in for a per-agent one.
Give every agent an identity, a budget, and hard limits.
One line of code. Anthropic, OpenAI, and Google Gemini.
See pricing