Text-to-SQL AI Agents as domain expertise tools

One of the more exciting projects I worked on at AWS was a LangGraph POC for querying data about infrastructure portfolios in our migration portfolio assessment tool. With data scattered between a client portfolio database controlled by the client and migrations databases controlled by our team, it could be hard for customers to answer questions like, “How much money would we save year-over-year with the migration recommended by our assessment?”. To that end, I developed a LangGraph POC that translated queries about the portfolio into SQL statements that were executed against a SQL data store derived from the customer-provided data and our own migration data. The tool showed real potential and I really enjoyed the work, but unfortunately a re-org took me away from the team before we could really dive into the details of that implementation.

With a bit of free time on my hands as I’m exploring the job market, I dove deeper into text-to-SQL agents with a personal LangGraph project. I’m using public data this time, leaning on my geography background by making a tool that queries World Bank’s World Development Indicators. Because cost, availability, and even factors like data sovereignty are becoming more obvious in AI, I used qwen3-coder:30b via Ollama running on my laptop. Because I’m committed to dogfooding my own work and this is a golden opportunity to analyze LLM usage, I’m using my own yallmap LLM proxy.

Just tinkering away at this I’ve been able to greatly improve the quality and consistency of results from my queries to the agent. I have a lot of fun improvements to share and it’s too much for one blog post, so I’ll be following up with a short series that walks through each of them. This first post is the setup piece: what the agent looks like, what I’m asking of it, and — the reason the whole series even works — how every model call the agent makes lands as an observable OTel span in Langfuse. Every improvement in every follow-up post was diagnosed from something the traces told me, and that only works if the wiring is there from day one.

You can find the source code for my sql agent on Github.

The State of Langgraph SQL Agents: What’s out there?

The general idea of Langgraph SQL Agents isn’t new. Even when I was working on my POC in 2024 there was some guidance available on how to wire together a Langgraph SQL Agent to do text-to-SQL using tree-of-thought. At this point there’s a much stronger body of work, though, and I want to call out a few good guides that helped me along in my work.

• LangChain — Build a custom SQL agent. LangChain’s own reference implementation, which describes the state machine architecture, schema-discovery tools, retry with error context, read-only database credentials, and an optional human-in-the-loop interrupt
• Building a Production-Ready SQL Agent with LangGraph (Feb 2026) This one feels like it was written by an LLM, which is a tough read for me, but it does hint at something I will get back to later: a retrieval step, in this case in the form of a table selector. This retrieval step is a bit more basic than what I will do later, as it just gives the names of tables that the LLM can use. One other significant detail is a good description of the importance of streaming for usable UI.
• Text-to-SQL Assistant: From Prompt to Production (Feb 2026) Another LLM-written post, this one mentions a feature that I didn’t implement — conversation memory — but then doesn’t give a description of it, instead showing detailed logs. All the same, it has a good description of a LangGraph state machine for text-to-SQL
• IBM — LangGraph ReAct SQL Agent with Mistral Medium 3. Very good tutorial for building a similar agent using Mistral, with the addition of a guardian node to block harmful messages. It’s also great for going into more detail about the ReAct architecture, which uses a single node in a loop that allows the LLM to choose the appropriate tool for each step.
• LangChain — Data Visualization Agent. This blog mentions using RAG for table retrieval, a concept I used for improving indicator (column) retrieval, but then doesn’t show an example. It does have a very cool extension, using the output as part of data visualization.
• WrenAI Blog post on the early implementation of WrenAI as a text-to-SQL agent, which describes the importance of text-to-SQL in mapping business terms to data schema.

These gave me a solid foundation for jumping back into the text-to-SQL agent problem, so it was pretty trivial to get up and running with an example, especially with AI-assisted development. What was harder, and more rewarding, was refining the agent and making it more useful — the topic of every follow-up post in this series.

Overview of my sql agent

The Data

I wanted a solid piece of publicly-published data in a familiar domain, so I chose the World Bank’s World Development Indicators, a data set of 28 economic and demographic indicators from 217 countries between 1960 and 2023. There are over 249,000 observations, it’s a compact 2.7 MB CSV after compression. It’s public domain, has a non-trivial JOIN surface, the schema is small enough to reason about, and it’s in the social sciences domain, letting me actually use that geography for something domain-specific (rather than the software engineering skills I picked up and have largely used professionally). It’s also a good data block for being able to grade the answers returned by LLM, which will come in handy shortly. I create a SQLite database from the CSV, if it does not already exist.

def build_database(
    db_path: str | Path,
    seed_long_df: Optional[pd.DataFrame] = None,
) -> Path:
    """Build (or rebuild) the WDI SQLite DB at db_path.

    Idempotent: drops and recreates all tables. If seed_long_df is provided,
    uses it instead of reading the bundled CSV."""
    db_path = Path(db_path)
    db_path.parent.mkdir(parents=True, exist_ok=True)

    df = seed_long_df.copy() if seed_long_df is not None else _load_long_df()
    missing = set(LONG_COLUMNS) - set(df.columns)
    if missing:
        raise ValueError(f"seed_long_df missing columns: {missing}")

    engine = create_engine(f"sqlite:///{db_path}")
    with engine.begin() as conn:
        for stmt in [s.strip() for s in SCHEMA_SQL.split(";") if s.strip()]:
            conn.execute(text(stmt))

        regions = _enum_table(df["region"])
        regions.to_sql("regions", conn, if_exists="append", index=False)

        incomes = _enum_table(df["income_group"])
        incomes.to_sql("income_groups", conn, if_exists="append", index=False)

        countries = (
            df[["iso3", "country", "region", "income_group"]]
            .drop_duplicates()
            .sort_values("iso3")
            .reset_index(drop=True)
        )
        region_id = dict(zip(regions["name"], regions["id"]))
        income_id = dict(zip(incomes["name"], incomes["id"]))
        countries_out = pd.DataFrame({
            "iso3": countries["iso3"],
            "name": countries["country"],
            "region_id": countries["region"].map(region_id),
            "income_group_id": countries["income_group"].map(income_id),
        })
        countries_out.index += 1
        countries_out = countries_out.rename_axis("id").reset_index()
        countries_out.to_sql("countries", conn, if_exists="append", index=False)

        indicators = (
            df[["indicator_code", "indicator_name", "indicator_topic", "indicator_unit"]]
            .drop_duplicates(subset=["indicator_code"])
            .sort_values("indicator_code")
            .reset_index(drop=True)
        )
        indicators_out = pd.DataFrame({
            "code": indicators["indicator_code"],
            "name": indicators["indicator_name"],
            "topic": indicators["indicator_topic"],
            "unit": indicators["indicator_unit"],
        })
        indicators_out.index += 1
        indicators_out = indicators_out.rename_axis("id").reset_index()
        indicators_out.to_sql("indicators", conn, if_exists="append", index=False)

        iso_to_id = dict(zip(countries_out["iso3"], countries_out["id"]))
        ind_to_id = dict(zip(indicators_out["code"], indicators_out["id"]))
        obs = pd.DataFrame({
            "country_id": df["iso3"].map(iso_to_id),
            "indicator_id": df["indicator_code"].map(ind_to_id),
            "year": df["year"].astype(int),
            "value": df["value"].astype(float),
        }).dropna(subset=["country_id", "indicator_id", "value"])
        obs["country_id"] = obs["country_id"].astype(int)
        obs["indicator_id"] = obs["indicator_id"].astype(int)
        obs.to_sql("observations", conn, if_exists="append", index=False)

    return db_path

Define the tree-of-thought graph as a finite state machine using LangGraph

LangGraph allows you to define a tree-of-thought graph using a finite state machine. While the chain-of-thought process tells the LLM to think step-by-step, you don’t get a good look inside the states the LLM moves through, and certainly you don’t get control over those states. LangGraph’s state machine approach gives you explicit access to these states. With that, you can define separate prompts and tools for each state, specific data types for input and output of transitions, and testability for each node.

The state machine for my sql agent is very simple:

        ┌─────────┐
START ──► think   │   Schema introspection (sql_db_list_tables / sql_db_schema)
        │         │   + planning. Reasoning accumulates in a scratchpad.
        └────┬────┘
             ▼
        ┌─────────┐
        │ query   │   Bounded tool-call sub-loop:
        │         │     probe (≤MAX_PROBES) → check (optional) → commit
        │         │   Probes verify data shape via read-only SELECTs;
        │         │   commit (sql_db_query) is what reaches the answer node.
        └────┬────┘
             │
   ┌─────────┴────────────┐
   ▼ (error / empty rows,  ▼ (rows returned)
     retries<2)
 (back to think with      ┌─────────┐
  failure context;        │ answer  │   Natural language reply. Only cites values
  hint points at          │         │   present in <query_result>;
  sql_db_probe)           │         │   no fabricated absolutes.
                          └────┬────┘
                               ▼
                              END

There are 3 states:

• think: Retrieves the indicators, builds the glossary of domain-specific terms to be used in the remainder of the prompts, and runs the sql_db_list_tables and sql_db_schema tools in a loop until it emits a plain text description of what it is looking for as a plan
• query: Takes the indicators, schema, and the plan, then uses the sql_db_probe tool to understand details about the database, a loop of sql_db_query and sql_db_query_checker to build a safe query string, and the execute_safely tool to retrieve the data from the database
• answer: Takes the original question, the SQL statement run by the query node, and the row result or error from the query node and uses a structured LLM call to render a plain language result to the end user.

Using LangGraph, the graph construction looks like this:

g: StateGraph = StateGraph(AgentState)
    g.add_node("think", lambda s: _think(s, llm, db, retriever))
    g.add_node("query", lambda s: _query(s, llm, db))
    g.add_node("answer", lambda s: _answer(s, answer_llm))

    g.add_edge(START, "think")
    g.add_edge("think", "query")
    g.add_conditional_edges("query", _should_retry, {"think": "think", "answer": "answer"})
    g.add_edge("answer", END)

    return g.compile()

The details of the probe, empty-rows retry, and anti-fabrication cues in the diagram all came from watching the agent fail against real questions — I’ll pick those up in the follow-up posts. For now the important thing is the shape: three named nodes, a bounded retry loop, and every LLM call inside those nodes going to the same place.

Route LLM calls through yallmap

By default I’m routing the LLM calls through a local yallmap proxy, using my own LLM proxy to get telemetry on LLM usage during the agent calls. By setting the ANTHROPIC_BASE_URL environment variable to http://localhost:3001, I’m directing the Claude-shaped LLM calls instead to the local proxy. This also allows me to run the agent using local models by setting the optional environment variable MODEL without changing the shape of my LLM calls, in my case to ollama/qwen3-coder:30b.

yallmap translates between Anthropic’s Messages API and a backend that you choose, either an Ollama OpenAI-compatible Chat Completions endpoint or Anthropic itself. From the perspective of the ChatAnthropic class nothing changes, but yallmap adds observability by emitting an OTel span for every request. Each span includes gen_ai.request.model, gen_ai.usage.input_tokens, gen_ai.usage.output_tokens, gen_ai.usage.cost_usd, and latency, which can then be viewed in Langfuse.

The client further extends observability by emitting a parent span per question (agent.ask) with per-node children (agent.think, agent.query, and agent.answer, correlating with the 3 nodes in the state machine). W3C trace context propagation makes the yallmap-emitted LLM spans grandchildren of these node spans. HTTPXClientInstrumentor injects a traceparent header on the anthropic SDK’s outbound HTTP calls, and yallmap’s HTTP server reads it. The trace tree assembles itself.

The result is that a single question produces a trace tree that looks like the graph:

A single question expanded: the agent.ask parent span, three child spans for think / query / answer, and inside each node span the underlying LLM calls emitted by yallmap — with token counts, cost, and latency on each. You can read the whole run from top to bottom.

yallmap also forwards telemetry that tracks LLM usage, as seen in the Langfuse dashboard.

The Langfuse dashboard for a single set of experiments

An example question, end to end

To show how the agent works, here’s what a real question looks like from the REPL:

Which country had the highest life expectancy in 2020?
[think] tool: sql_db_list_tables({})
[think] tool: sql_db_schema({'table_names': 'observations, indicators, countries'})
[think] [{'text': 'I need to find the country with the highest life expectancy in 2020. 
Let me plan this step by step: I need to use the indicator code for "Life expectancy at 
birth, total (years)" which is SP.DYN.LE00.IN. I''ll filter observations for year 2020. 
I''ll join with countries table to get country names. I''ll order by value descending 
and limit to 1 to get the highest  This requires a simple query joining observations, 
indicators, and countries tables.', 'type': 'text', 'index': 0}]
[probe] sql_db_probe({'query': "SELECT COUNT(*) FROM observations o JOIN indicators i ON i.id = o.indicator_id WHERE i.code = 'SP.DYN.LE00.IN' AND o.year = 2020"}) → [(217,)]
[query] SELECT c.name, o.value as life_expectancy
FROM observations o
JOIN indicators i ON i.id = o.indicator_id
JOIN countries c ON c.id = o.country_id
WHERE i.code = 'SP.DYN.LE00.IN' 
AND o.year = 2020
ORDER BY o.value DESC
LIMIT 1
[result] [('Monaco', 86.089)]

In 2020, Monaco had the highest life expectancy at 86.089 years.

Six LLM calls were made in the three child spans of agent.ask above: three to plan (think), two to write and commit SQL (query), one to render the natural-language answer (answer). Because each span carries the token count and cost for that specific call, I can see exactly where I spend my tokens (and, if I have per-token pricing, my money) on this question.

What’s next

Having built a sql agent with best practices and wired it up to my observability tools, I moved on to improving the performance of the tool. In coming posts I’ll describe some of the improvements you can observe in the source code for the sql agent.

• Reducing ambiguity in a Text-to-SQL LLM Agent by improving the prompt (coming soon) — I noticed some issues where the model picked a semantically adjacent but wrong indicator (particularly, it used data for maternal mortality when I asked about under-5 mortality). Improvements in the glossary and top-k catalog retrieval reduced the likelihood of failure, which I tested using a test harness that included a disambiguation-heavy question.
• Reducing ambiguity in a Text-to-SQL LLM Agent using runtime tools (coming soon) — I used the sql_db_probe tool, a probe budget cap, an empty-rows retry hint, and an anti-fabrication guard on the answer node to reduce the likelihood of a failed query or hallucinated results.
• Safe execution of model-emitted SQL (coming soon) — Three-layers of defense (read-only URI + sqlglot allowlist + row/timeout caps + cell sanitization) between the model’s SQL and the database to reduce the attack surface. Includes the CTE-write smuggling test case (WITH x AS (DELETE … RETURNING *) SELECT * FROM x) that can be missed by top-level statement-type check.

Once I publish these I’ll link back, so it’s easy to follow along in the series. I’ll probably add on more, too, if I keep having fun with this project.

Comments

View & reply on Bluesky →

No comments yet. Be the first to reply on Bluesky!