Iden Tumuhirwe
All writing

July 18, 2026 · 18 min read

Building a SQL assistant with a tiny LLM, where the model is the least important part

AI · LLMs · PostgreSQL · Architecture · Deep Dive

Everyone wants the same demo right now: type "how much money did we make today?" into a chat box and get a real answer from the real database. The usual next step is to wire the biggest frontier model you can afford straight to the production PostgreSQL instance and hope for the best.

But for internal reporting questions, the model barely matters. The question "how much income did we receive through Airtel today?" does not require deep reasoning, world knowledge, or long-form generation. It requires understanding a short sentence, picking the right metric, extracting a date and a provider name, and handing off to a system that does the actual work. A model with a few hundred million parameters can do that. What it can't do, and what no model of any size should be trusted to do, is safely freestyle SQL against your operational database.

So this post is the design I'd actually build: a tiny LLM as the natural-language front door, PostgreSQL doing what PostgreSQL does, and a deliberately boring, deterministic pipeline between them doing everything that matters. One rule drives every design decision in this post:

The LLM decides what the user wants. The system decides how the data is accessed.

The shape of the thing

A typical request looks like:

How much income did we receive through Airtel today?

And the answer should come back as:

Airtel processed UGX 28,450,000 in successful collections today.

Between those two sentences sits the whole system:

User question
    ↓
Tiny LLM interprets intent
    ↓
Relevant schema and business context are retrieved
    ↓
LLM produces a structured query plan
    ↓
Validation layers check the request
    ↓
Query runs against a read-only reporting database
    ↓
Structured result comes back
    ↓
LLM or a response template formats the answer

The model is only responsible for the first hop in that diagram, and maybe the last one. Everything in the middle is deterministic code you can test.

Can a model that small actually do this?

Yes, as long as you keep the task narrow. Look at what these requests actually demand:

Check transaction REL460020F48A.
Show failed Airtel payouts yesterday.
How many collections are pending?
Get today's transaction summary.

That's intent classification, parameter extraction, tool selection, and structured output. Small models handle this fine. Where they fall over is general text-to-SQL, because real SQL against a real schema means joins, aggregation semantics, null handling, ambiguous business terminology, and knowing that "income" excludes reversals in your company but not in the next one. No amount of parameters fixes ambiguity that lives in your business, not in the language.

So the strategy is simple: put the complexity in the system and the tools, not in the model. Every section that follows is a different way of doing exactly that.

Two ways to wire it up

There are two broad approaches to connecting a language model to PostgreSQL.

Approach A: predefined business tools. The model picks from a menu:

get_transaction_status(reference)
get_todays_income(provider=None, merchant_id=None)
get_failed_payout_summary(provider, start_date, end_date)
get_pending_reversals(start_date, end_date)

The SQL lives inside the backend, invisible to the model. This is the safest and most predictable option, and if your users ask the same twelve questions all day, stop reading here and go build this.

Approach B: text-to-SQL. The model gets database context, writes SQL, and passes it to an execution tool. It's more flexible, but the edges are sharper: it needs stronger validation and much better database design.

Most real systems want something in between, and the middle path I'll spend most of this post on is: the model emits a structured query plan, and the backend compiles that plan into SQL. You get most of the flexibility of text-to-SQL with most of the safety of predefined tools.

Whatever you do, don't point it at the production schema

This is the design principle I'd defend the hardest. A production database is not a friendly place. It has hundreds of tables, legacy tables, audit tables, retry records, JSON columns, soft-deleted rows, duplicate business concepts, and one-to-many relationships everywhere. A model handed that schema will produce SQL that is syntactically perfect and financially wrong.

The classic failure mode is the careless join:

SELECT SUM(t.amount)
FROM transactions t
JOIN transaction_status_history h
    ON h.transaction_id = t.id;

If each transaction has five history records, every amount is counted five times. The result still looks completely convincing, which is worse than an error, because someone will put it in a board deck. "Do not expose the operational database and expect the LLM to become a database engineer" is the rule. It won't. It'll become a very fast intern with write access to your reputation.

Give it a replica and a curated schema instead

Two pieces of infrastructure fix most of this before the model is even involved.

First, a read-only replica. Not strictly mandatory, but for production I wouldn't skip it:

PostgreSQL Primary
        |
        | Streaming replication
        v
Read-only PostgreSQL Replica
        |
        v
AI Reporting Schema
        |
        v
LLM Query Service

AI queries stop competing with payment traffic, long analytical scans are isolated from the operational workload, the AI service physically cannot modify production data, and you can add reporting indexes without touching the primary. The one caveat is replication lag: "what's the status of the transaction from thirty seconds ago" should keep going through your existing transaction API, and historical or analytical questions go to the replica.

Second, a dedicated schema on that replica:

CREATE SCHEMA ai_reporting;

This holds curated views, materialized views, summary tables, masked data, and explicitly approved metrics. The model sees only this schema:

PostgreSQL Primary
│
├── core
│   ├── transactions
│   ├── transaction_status_history
│   ├── merchants
│   ├── providers
│   ├── reversals
│   └── ... (everything the model never sees)
│
└── Read-only replica
    │
    └── ai_reporting
        ├── transaction_facts
        ├── provider_daily_metrics
        ├── merchant_daily_metrics
        ├── reversal_summary
        ├── settlement_summary
        └── metric_catalog

One fact view beats five operational tables

Suppose the operational truth is spread across payments.transactions, payments.transaction_routes, payments.providers, accounts.merchants, and payments.reversals. Collapse it into one analytical entity:

CREATE VIEW ai_reporting.transaction_facts AS
SELECT
    t.id AS transaction_id,
    t.reference AS transaction_reference,
    t.merchant_id,
    m.name AS merchant_name,
    p.name AS provider,
    t.transaction_type,
    t.amount,
    t.currency,
    t.final_status,
    t.created_at AS transaction_time,
    t.completed_at AS completed_time,

    EXISTS (
        SELECT 1
        FROM payments.reversals r
        WHERE r.transaction_id = t.id
          AND r.status = 'COMPLETED'
    ) AS reversed,

    CASE
        WHEN t.customer_phone IS NULL THEN NULL
        ELSE CONCAT(
            LEFT(t.customer_phone, 4),
            '****',
            RIGHT(t.customer_phone, 3)
        )
    END AS masked_customer_phone

FROM payments.transactions t
JOIN accounts.merchants m
    ON m.id = t.merchant_id
LEFT JOIN payments.transaction_routes tr
    ON tr.id = t.route_id
LEFT JOIN payments.providers p
    ON p.id = tr.provider_id
WHERE t.deleted_at IS NULL
  AND t.is_test = false;

Every dangerous decision has already been made once, correctly: the join path, the soft-delete filter, the test-data exclusion, the phone masking, the "reversed" flag computed as an EXISTS instead of a row-multiplying join. The model gets one clean table-shaped thing to reason about.

Precompute the questions people actually ask

"What was today's income by provider?" "What was the failure rate this week?" "How did Airtel compare with MTN yesterday?" These get asked every day, and they should not scan millions of raw rows every time somebody types them into a chat box. A materialized view precomputes the daily rollup:

CREATE MATERIALIZED VIEW ai_reporting.provider_daily_metrics AS
SELECT
    transaction_time::date AS metric_date,
    provider,
    transaction_type,
    currency,
    COUNT(*) AS total_transactions,
    COUNT(*) FILTER (WHERE final_status = 'SUCCESS') AS successful_transactions,
    COUNT(*) FILTER (WHERE final_status = 'FAILED') AS failed_transactions,
    COUNT(*) FILTER (WHERE final_status = 'PENDING') AS pending_transactions,
    COALESCE(
        SUM(amount) FILTER (WHERE final_status = 'SUCCESS'),
        0
    ) AS successful_amount
FROM ai_reporting.transaction_facts
GROUP BY 1, 2, 3, 4;

Give it a unique index so it can be refreshed without locking readers:

CREATE UNIQUE INDEX ux_provider_daily_metrics
ON ai_reporting.provider_daily_metrics (
    metric_date, provider, transaction_type, currency
);

REFRESH MATERIALIZED VIEW CONCURRENTLY
    ai_reporting.provider_daily_metrics;

Refresh on a schedule; if you actually need near-real-time numbers, maintain incremental summary tables instead of periodic refreshes. Either way, the win is the same: the model's query becomes a cheap scan over a few thousand pre-aggregated rows instead of an adventure across the raw transaction table.

The schema tells you structure. It doesn't tell you meaning.

Here's where text-to-SQL projects quietly die. The database says transactions.status, transactions.amount, transactions.type. It does not say whether "income" means gross or net, whether it includes fees, whether reversals are excluded, whether the failure rate counts pending transactions, or which timezone "today" is in. Those answers live in people's heads, which is exactly where a language model can't reach.

So write them down. A semantic catalog defines each metric explicitly:

{
  "metric": "successful_collection_income",
  "display_name": "Successful collection income",
  "description": "Total principal amount of final successful collection transactions.",
  "source": "ai_reporting.transaction_facts",
  "aggregation": "SUM(amount)",
  "mandatory_filters": {
    "transaction_type": "COLLECTION",
    "final_status": "SUCCESS",
    "is_test": false
  },
  "time_field": "transaction_time",
  "allowed_dimensions": [
    "merchant_id", "merchant_name", "provider", "currency", "transaction_date"
  ],
  "synonyms": [
    "income", "collections income", "money received", "successful collections"
  ]
}

Failure rate gets its own entry that spells out the numerator, the denominator, and the fact that PENDING, INITIATED, and PROCESSING are excluded from both. The definitions live in a plain table:

CREATE TABLE ai_metadata.metric_catalog (
    metric_key text PRIMARY KEY,
    display_name text NOT NULL,
    description text NOT NULL,
    source_relation text NOT NULL,
    aggregation_expression text NOT NULL,
    mandatory_filters jsonb NOT NULL DEFAULT '{}',
    allowed_dimensions text[] NOT NULL DEFAULT '{}',
    synonyms text[] NOT NULL DEFAULT '{}',
    sensitivity_level text NOT NULL DEFAULT 'internal'
);

The bonus nobody mentions: writing this catalog forces your organization to agree on what "income" means. That argument was going to happen eventually. Better now, in a JSON file, than later, in two dashboards showing different numbers.

Retrieve context, don't dump it

Don't paste the entire schema into every prompt. Give the model a search tool instead:

search_semantic_schema(
    concepts: list[str],
    user_role: str,
    maximum_objects: int = 8
)

For "Compare Airtel and MTN payout failures today", the system searches for ["provider", "payout", "failed transactions", "today"] and gets back exactly what's needed:

{
  "recommended_source": "ai_reporting.provider_daily_metrics",
  "columns": [
    { "name": "metric_date", "type": "date" },
    { "name": "provider", "type": "text" },
    {
      "name": "transaction_type",
      "type": "text",
      "allowed_values": ["COLLECTION", "PAYOUT"]
    },
    { "name": "failed_transactions", "type": "bigint" }
  ],
  "business_rules": [
    "Use transaction_type = 'PAYOUT'.",
    "Today means the Africa/Kampala calendar date.",
    "Provider names are Airtel and MTN."
  ]
}

Less context means less to get wrong, and for a tiny model that's the difference between reliable and random.

Make the model emit a query plan, not SQL

This is the single most important decision in the design. Instead of asking the model for SQL, ask it for a structured plan:

{
  "source": "ai_reporting.provider_daily_metrics",
  "metrics": [
    {
      "field": "failed_transactions",
      "aggregation": "sum",
      "alias": "failed_payouts"
    }
  ],
  "dimensions": ["provider"],
  "filters": [
    { "field": "provider", "operator": "in", "value": ["Airtel", "MTN"] },
    { "field": "transaction_type", "operator": "=", "value": "PAYOUT" },
    { "field": "metric_date", "operator": "=", "value": "2026-07-18" }
  ],
  "sort": [{ "field": "failed_payouts", "direction": "desc" }],
  "limit": 10
}

Your backend compiles that into parameterized SQL:

SELECT
    provider,
    SUM(failed_transactions) AS failed_payouts
FROM ai_reporting.provider_daily_metrics
WHERE provider = ANY($1)
  AND transaction_type = $2
  AND metric_date = $3
GROUP BY provider
ORDER BY failed_payouts DESC
LIMIT 10;

The compiler itself is about a page of allowlist checks and string assembly:

from dataclasses import dataclass
from typing import Any, Literal

ALLOWED_SOURCES = {
    "ai_reporting.provider_daily_metrics": {
        "metric_date",
        "provider",
        "transaction_type",
        "currency",
        "total_transactions",
        "successful_transactions",
        "failed_transactions",
        "pending_transactions",
        "successful_amount",
    }
}

ALLOWED_AGGREGATIONS = {
    "sum": "SUM",
    "count": "COUNT",
    "avg": "AVG",
    "min": "MIN",
    "max": "MAX",
}


@dataclass
class QueryMetric:
    field: str
    aggregation: Literal["sum", "count", "avg", "min", "max"]
    alias: str


def compile_query_plan(plan: dict[str, Any]) -> tuple[str, list[Any]]:
    source = plan.get("source")
    if source not in ALLOWED_SOURCES:
        raise ValueError("Unapproved reporting source")

    allowed_fields = ALLOWED_SOURCES[source]
    select_parts: list[str] = []
    group_fields: list[str] = []
    parameters: list[Any] = []
    where_parts: list[str] = []

    for dimension in plan.get("dimensions", []):
        if dimension not in allowed_fields:
            raise ValueError(f"Unapproved dimension: {dimension}")
        select_parts.append(dimension)
        group_fields.append(dimension)

    for metric_data in plan.get("metrics", []):
        metric = QueryMetric(**metric_data)

        if metric.field not in allowed_fields:
            raise ValueError(f"Unapproved metric field: {metric.field}")

        sql_function = ALLOWED_AGGREGATIONS.get(metric.aggregation)
        if not sql_function:
            raise ValueError("Unsupported aggregation")

        if not metric.alias.replace("_", "").isalnum():
            raise ValueError("Invalid alias")

        select_parts.append(f"{sql_function}({metric.field}) AS {metric.alias}")

    for condition in plan.get("filters", []):
        field = condition["field"]
        operator = condition["operator"].lower()
        value = condition["value"]

        if field not in allowed_fields:
            raise ValueError(f"Unapproved filter field: {field}")

        if operator == "=":
            parameters.append(value)
            where_parts.append(f"{field} = ${len(parameters)}")
        elif operator == "in":
            if not isinstance(value, list) or len(value) > 50:
                raise ValueError("Invalid IN filter")
            parameters.append(value)
            where_parts.append(f"{field} = ANY(${len(parameters)})")
        elif operator in {">", ">=", "<", "<="}:
            parameters.append(value)
            where_parts.append(f"{field} {operator} ${len(parameters)}")
        else:
            raise ValueError(f"Unsupported operator: {operator}")

    if not select_parts:
        raise ValueError("No fields selected")

    sql = f"SELECT {', '.join(select_parts)} FROM {source}"
    if where_parts:
        sql += " WHERE " + " AND ".join(where_parts)
    if group_fields:
        sql += " GROUP BY " + ", ".join(group_fields)

    limit = min(int(plan.get("limit", 100)), 500)
    sql += f" LIMIT {limit}"

    return sql, parameters

The model never touches parameter numbering, injection prevention, or identifier allowlists. Those are the compiler's problem, and the compiler has unit tests.

The validation gauntlet

Even with structured plans, never send anything model-shaped toward PostgreSQL without making it run the gauntlet:

LLM query plan
      │
      ▼
1. JSON schema validation
      ▼
2. User permission validation
      ▼
3. Metric and field allowlist
      ▼
4. Mandatory business filters
      ▼
5. SQL compilation
      ▼
6. SQL AST validation
      ▼
7. EXPLAIN cost validation
      ▼
8. Read-only execution
      ▼
9. Result validation and masking

Each layer catches a different class of failure, and none of them trusts the layers before it. A few of these deserve a closer look.

Permissions belong to the application, not the prompt. No amount of "you are a helpful assistant that respects access control" in a system prompt is access control:

ROLE_ACCESS = {
    "customer_support": {
        "ai_reporting.transaction_facts",
        "ai_reporting.provider_daily_metrics",
    },
    "finance": {
        "ai_reporting.transaction_facts",
        "ai_reporting.provider_daily_metrics",
        "ai_reporting.merchant_daily_metrics",
        "ai_reporting.settlement_summary",
    },
}


def verify_source_access(role: str, source: str) -> None:
    if source not in ROLE_ACCESS.get(role, set()):
        raise PermissionError("You are not permitted to query this dataset.")

Security filters get injected, not requested. If a merchant user may only see their own records, don't trust the model to remember the filter. The backend appends it unconditionally:

def inject_security_scope(plan: dict, user_context: dict) -> dict:
    if user_context["role"] == "merchant_user":
        plan.setdefault("filters", []).append({
            "field": "merchant_id",
            "operator": "=",
            "value": user_context["merchant_id"],
            "system_enforced": True,
        })
    return plan

PostgreSQL Row-Level Security backs the same rule up at the database layer, for the day the application check has a bug:

ALTER TABLE ai_reporting.transaction_facts_table
ENABLE ROW LEVEL SECURITY;

CREATE POLICY merchant_scope_policy
ON ai_reporting.transaction_facts_table
FOR SELECT
TO ai_merchant_user
USING (
    merchant_id = current_setting('app.current_merchant_id')::bigint
);

with SET LOCAL app.current_merchant_id = '105' before each query.

Validate SQL with a parser, not string checks. sql.strip().lower().startswith("select") is a vibe check, not validation. Parse the statement into an AST and enforce real rules: exactly one statement, SELECT (or WITH containing only SELECTs) at the root, only approved schemas, tables, columns, and functions, no data-modifying CTEs, no cartesian joins, a result limit, and a date filter on large sources. Everything else (INSERT, UPDATE, DELETE, TRUNCATE, DDL, COPY, CALL, DO, multiple statements) gets rejected at the tree level, where it can't hide inside a comment or a clever string.

Gate on EXPLAIN before executing. Ask PostgreSQL what it would do:

EXPLAIN (FORMAT JSON)
SELECT provider, SUM(failed_transactions)
FROM ai_reporting.provider_daily_metrics
WHERE metric_date = $1
GROUP BY provider;

Then reject plans that are too expensive before they run:

def validate_explain_plan(plan: dict) -> None:
    root = plan["Plan"]

    if root["Total Cost"] > 100_000:
        raise ValueError("Estimated query cost is too high")
    if root["Plan Rows"] > 1_000_000:
        raise ValueError("Query may process too many rows")

    inspect_nodes(root)


def inspect_nodes(node: dict) -> None:
    if (
        node.get("Node Type") == "Seq Scan"
        and node.get("Plan Rows", 0) > 500_000
    ):
        raise ValueError("Large sequential scan rejected")

    for child in node.get("Plans", []):
        inspect_nodes(child)

Use plain EXPLAIN, not EXPLAIN ANALYZE. EXPLAIN ANALYZE executes the query, which defeats the point of checking before execution.

Execute inside a restricted transaction, as a restricted role.

def execute_safe_query(connection, sql, params):
    with connection.transaction():
        connection.execute("SET LOCAL statement_timeout = '5s'")
        connection.execute("SET LOCAL lock_timeout = '1s'")
        connection.execute("SET TRANSACTION READ ONLY")
        return connection.fetch(sql, *params)

And the database role gets CONNECT, USAGE on ai_reporting, and SELECT on its tables. Nothing else:

CREATE ROLE ai_query_user LOGIN;

GRANT CONNECT ON DATABASE reporting_database TO ai_query_user;
GRANT USAGE ON SCHEMA ai_reporting TO ai_query_user;
GRANT SELECT ON ALL TABLES IN SCHEMA ai_reporting TO ai_query_user;

ALTER DEFAULT PRIVILEGES IN SCHEMA ai_reporting
GRANT SELECT ON TABLES TO ai_query_user;

If every other layer fails simultaneously, the worst possible outcome is a slow SELECT that times out after five seconds.

Joins, indexes, and the rest of the Postgres homework

A few database-side jobs are left, and none of them belong to the model.

Joins come from an approved-relationship catalog, not the model's imagination. The semantic layer stores which joins exist, their cardinality, and their traps:

{
  "left_source": "transaction_facts",
  "right_source": "reversal_events",
  "join_type": "one_to_many",
  "condition": "transaction_facts.transaction_id = reversal_events.transaction_id",
  "approved": true,
  "aggregation_warning": "Aggregate reversal events before joining to prevent duplicated transaction amounts."
}

Better still, if the same joins keep showing up, bake them into pre-joined reporting views (reversal_summary, settlement_summary) so the model never joins at all. Every join the model doesn't write is a duplicated-amount bug that never ships.

Indexes come from approved query patterns: common filters, join columns, date ranges, and sort/group-by shapes. The model gets no say here. Partial indexes work well because the reporting workload is so predictable:

CREATE INDEX idx_successful_collections_created
ON core.transactions (created_at, provider_id)
INCLUDE (amount, currency)
WHERE status = 'SUCCESS'
  AND transaction_type = 'COLLECTION';

Keep planner statistics current, because the EXPLAIN gate is only as honest as the statistics behind it. Run ANALYZE regularly, and use extended statistics for correlated columns like provider_id + transaction_type so PostgreSQL's row estimates aren't fiction:

CREATE STATISTICS transaction_provider_status_stats
    (dependencies, ndistinct)
ON provider_id, status, transaction_type
FROM core.transactions;

Partition huge tables by date, and make the query compiler emit explicit ranges, created_at >= $1 AND created_at < $2, never DATE(created_at) = CURRENT_DATE. Wrapping the column in a function throws away both the index and partition pruning, and it's precisely the kind of SQL a model writes when left unsupervised, which is one more small argument for not letting it write SQL.

When it still goes wrong

It will. Two mechanisms keep the failures contained.

Validate results before display. A query that passed every gate can still return something misleading. Check the shape (expected columns, types, row count) and check plausibility, a "rate" of 3.7 is not a rate:

def validate_result(rows: list[dict], expected_schema: dict) -> list[dict]:
    if len(rows) > 500:
        raise ValueError("Result exceeds row limit")

    for row in rows:
        for field in expected_schema:
            if field not in row:
                raise ValueError(f"Missing expected field: {field}")

    return mask_sensitive_fields(rows)


def validate_rate(value: float) -> float:
    if not 0 <= value <= 1:
        raise ValueError("Invalid rate returned")
    return value

Give the model a repair loop with a short leash. When a plan fails validation, return a structured, actionable error and let the model try again, at most twice, then fail safely:

{
  "status": "rejected",
  "reason": "ONE_TO_MANY_AGGREGATION_RISK",
  "message": "Joining reversal_events directly may duplicate transaction amounts. Aggregate reversal_events by transaction_id first.",
  "repairable": true
}

The error message teaches the fix. A small model can act on "aggregate by transaction_id first"; it cannot act on a raw PostgreSQL stack trace, and your users should never see one either.

One request, end to end

Put it all together for "How much did each provider successfully collect yesterday?":

Intent extraction
    metric = successful collection amount
    dimension = provider, period = yesterday
      ↓
Semantic retrieval
    metric definition, approved source,
    mandatory filters, timezone
      ↓
Structured query plan
    source = provider_daily_metrics
    metric = sum(successful_amount)
    dimension = provider
    type = COLLECTION, date = 2026-07-17
      ↓
Permission check → SQL compilation → AST validation
      ↓
EXPLAIN gate → read-only execution on the replica
      ↓
Result
    Airtel: UGX 18,400,000
    MTN: UGX 27,500,000
      ↓
"On 17 July 2026, MTN collected UGX 27.5 million,
 while Airtel collected UGX 18.4 million."

The model made two contributions: it understood the question, and it phrased the answer. Every number in between came from deterministic, tested, permissioned code.

The version I'd actually build first

A practical first version is smaller than this post makes it sound:

  1. A read-only reporting replica.
  2. An ai_reporting schema with five to ten curated views, a few of them materialized.
  3. A metric catalog and a schema retrieval tool.
  4. A structured query-plan format, a backend compiler, and an AST validator.
  5. An EXPLAIN cost gate and a SELECT-only database role.
  6. Role-based access, audit logs, and deterministic response templates.

The tiny LLM does exactly three language tasks: interpret the question, pick the metric and dimensions, and produce a query plan. Everything involving security, joins, permissions, SQL construction, and business correctness stays in code you can review.

Which is why the model is the least important part. The LLM decides what the user wants; the system decides how the data is accessed safely and correctly. Get that boundary right and a model small enough to run on a modest box will answer real questions about real money and get them right. Get it wrong and no frontier model will save you, it'll just be confidently wrong with better grammar.