Deterministic Guardrails for LLM Agents: Restricting SQL and Pydantic Schemas without Open-Ended Code Execution

1. The Flaw of Arbitrary Code Execution in Agents

Many early LLM analytics frameworks popularized an architectural pattern: prompt the LLM to write arbitrary Python code (using Pandas or exec()), run the code in a local subprocess, and display the output dataframe or plot.

While flexible in simple notebook demos, deploying open-ended code generation into production applications introduces critical vulnerabilities:

1. Remote Code Execution (RCE) & Sandbox Escapes: An adversarial user prompt (or indirect prompt injection from malicious CSV column names) can trick the model into executing system commands:

   # Dangerous payload generated by unconstrained agent
   import os; os.system("rm -rf /") # or exfiltrating environment variables

2. Non-Deterministic Execution Paths: Language models frequently invent non-existent library methods, invoke deprecated keyword arguments, or generate unbounded while loops that consume 100% CPU and exhaust server memory.
3. Unverifiable Data Claims: When models summarize output tables, they frequently misread rows or hallucinate percentages that contradict the dataframe itself.

To build reliable data systems, we must replace arbitrary code execution with restricted tool dispatch, SQL statement validation, and automated numerical verification.

2. The Four-Tool Deterministic Dispatch Pattern

Instead of allowing the agent to generate free-form Python scripts, restrict the agent's action space to a small set of fixed tools. In the Analyst project the four tools are:

1. query_data: Runs a SQL query against an in-process DuckDB table, after the SQL guard described below.
2. plot_chart: Builds a chart from structured parameters (chart type, columns, aggregation) instead of model-written plotting code.
3. summary_stats: Returns summary statistics computed by code, not by the model.
4. clarify: Asks the user a question when the request is ambiguous.

By constraining the model to choosing a tool and filling in its parameters, the model no longer writes executable code, which removes the code-execution risk described above. It does not remove every risk: the SQL the model chooses to run, and the numbers it writes in its answer, still need their own checks.

3. SQL Safety Validation: Keyword Filters and AST Parsing

There are two common ways to restrict the SQL an agent can run.

Keyword filtering. The Analyst project's guard is a regex keyword filter, not a parser. It strips SQL comments, requires the query to start with SELECT or WITH, allows a single statement, and rejects 35 blocked keywords (mutation, file-system and administrative commands). It is simple and fast, but it has limits: it matches keywords inside string literals, so a harmless string containing a blocked word is rejected, and it only blocks what the list names, so a file-reading form the list does not name can still pass. It is not engine-enforced read-only access.

AST parsing. A stronger approach is to parse the query with a SQL parser such as sqlglot and inspect the syntax tree. The sketch below is illustrative; it is not code from the Analyst project:

import sqlglot
from sqlglot import exp

class SQLSafetyValidator:
    ALLOWED_STATEMENT_TYPES = (exp.Select, exp.Union)
    FORBIDDEN_EXPRESSIONS = (
        exp.Drop, exp.Delete, exp.Insert, exp.Update, 
        exp.Alter, exp.Create, exp.Command, exp.Transaction
    )

    @classmethod
    def validate_query(cls, sql_str: str) -> str:
        """
        Parses a SQL query and rejects anything that is not a SELECT or UNION, or that contains mutation syntax.
        """
        try:
            parsed = sqlglot.parse_one(sql_str, read="duckdb")
        except Exception as e:
            raise ValueError(f"Malformed SQL syntax: {str(e)}")

        # Enforce that root expression is a SELECT or UNION
        if not isinstance(parsed, cls.ALLOWED_STATEMENT_TYPES):
            raise PermissionError(f"Prohibited query type: {type(parsed).__name__}. Only SELECT queries are permitted.")

        # Traverse entire AST to prevent mutation commands in subqueries or CTEs
        for expression in parsed.find_all(cls.FORBIDDEN_EXPRESSIONS):
            raise PermissionError(f"Forbidden operation detected: {type(expression).__name__}")

        # Return sanitized normalized SQL
        return parsed.sql(dialect="duckdb")

Even a parsed SELECT can call functions that read files, so a parser-based validator should also allowlist the functions and table sources it accepts.

4. In-Process DuckDB: What It Does and Does Not Isolate

DuckDB is designed specifically for fast analytical queries (OLAP) directly within the application process. Unlike SQLite, DuckDB is columnar and vectorized; unlike PostgreSQL, it requires no external server process or network sockets.

An in-process, in-memory database keeps the working table inside the application and exposes no database server. It is not a security sandbox: the query runs with the application's own privileges, and an in-memory connection is not the same as engine-enforced read-only access.

Useful practical limits:

• Attach datasets in memory: con = duckdb.connect(database=':memory:')
• Set memory and thread limits:

  con.execute("SET max_memory = '1GB'")
  con.execute("SET threads = 4")

• Enforce query timeouts using background thread watchers or connection interrupt handlers.

The Analyst project uses an in-memory connection and a 500-row result cap, but it does not set an engine-level read-only mode, so its SQL guard is the only restriction on what a query may do.

5. Validating Numerical Claims Against Source Data

A major failure mode in LLM analytics is hallucinated summary text. A query might return a sum of 452,100, but the generative model writes: "The total revenue for Q3 was 542,000."

To reduce this, implement an automated Claim Verifier:

import re
from typing import List, Dict, Any

def verify_numerical_claims(
    generated_text: str, 
    query_result_table: List[Dict[str, Any]], 
    tolerance: float = 0.01
) -> bool:
    """
    Extracts numerical figures from generated summary text and verifies
    that every extracted number exists in the query result table.
    """
    # Extract floating-point and integer numbers from text
    raw_numbers = re.findall(r'[-+]?(?:\d*\.\d+|\d+)', generated_text.replace(',', ''))
    extracted_values = [float(n) for n in raw_numbers]
    
    # Flatten all numerical values from query result
    table_numbers = []
    for row in query_result_table:
        for val in row.values():
            if isinstance(val, (int, float)):
                table_numbers.append(float(val))
                
    # Check that each significant extracted number is backed by the table
    for num in extracted_values:
        # Ignore common non-data numbers (e.g. 1, 2 for lists, or years like 2025 if filtered)
        if num in [0.0, 1.0, 2.0]:
            continue
        
        # Verify proximity to at least one cell in the result table
        is_supported = any(abs(num - tbl_val) <= (tolerance * abs(tbl_val) + 1e-5) for tbl_val in table_numbers)
        if not is_supported:
            return False # Flag as unsupported claim
            
    return True

A check like this reduces unsupported figures but does not eliminate them, and it covers numbers only, not wording. In the Analyst project's 20-question benchmark, 16 of 20 answers had every number match the tool output.

6. Pydantic v2 Structured Output Contracts

To reduce unparsed or malformed model output, enforce strict JSON schemas using Pydantic:

from pydantic import BaseModel, Field
from typing import Optional, Literal, List

class ChartSpecification(BaseModel):
    chart_type: Literal['bar', 'line', 'scatter', 'pie'] = Field(description="Visual chart representation")
    x_axis: str = Field(description="Column name for horizontal dimension")
    y_axis: str = Field(description="Column name for vertical numerical metric")
    title: str = Field(description="Clean descriptive chart title")

class AnalyticalResponse(BaseModel):
    sql_query: str = Field(description="Validated SELECT query")
    reasoning: str = Field(description="Brief explanation of analytical approach")
    key_findings: List[str] = Field(description="Bullet points describing verified results")
    chart: Optional[ChartSpecification] = Field(default=None, description="Optional chart specification if visual is warranted")
    confidence_score: float = Field(ge=0.0, le=1.0, description="Self-assessed evidence confidence score")

7. Production Security & Reliability Checklist

1. No eval() or exec(): Ban all dynamic string execution in application source code.
2. Memory Quotas: Restrict DuckDB process allocations to reduce the risk of out-of-memory container crashes.
3. Statement Restriction: Accept only SELECT or WITH queries and reject DDL and DML statements. AST-based validation is more robust than keyword matching, which is what the Analyst project uses; neither is the same as engine-enforced read-only access.
4. Data Verification: Block LLM responses whose numerical statements fail cross-validation against raw query outputs.