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

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

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 and validates that a SQL query is strictly read-only and free of 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")

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

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

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 read-only SQL 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")