Natural Language to SQL: Build an AI-Powered Database Interface

Natural Language to SQL: Build an AI-Powered Database Interface

Business users spend up to 40% of their time on simple SQL queries: "how many sales yesterday", "average check by region", "top 10 products by revenue". Analysts are overloaded with routine, data request queues grow. **Text-to-SQL is the task of translating natural language into SQL (Wikipedia). We develop such interfaces: you ask a question in plain English, get accurate SQL queries and data — without knowing SQL and without distracting analysts.

Why Businesses Need Text-to-SQL, Not Another BI Tool?

BI tools require dashboard setup — that takes days. Text-to-SQL works with any database schema on the fly: ask a question, get SQL and data in seconds. Our clients close 70% of simple queries without analyst involvement. Analysts handle 3–5 times more tasks per day. New employee onboarding time drops from weeks to 1–2 days. Text-to-SQL supports PostgreSQL, MySQL, BigQuery, Snowflake and other popular dialects out of the box. Compare: generating SQL via Text-to-SQL is 5x faster than manual writing and 10x faster than creating a BI dashboard.

How Our Text-to-SQL Architecture Works

The key challenge is not just translating text to SQL, but correctly handling JOINs across 10+ tables, accounting for business logic, and avoiding expensive full-table scans. Our solution uses an LLM (Claude 3.5 Sonnet or GPT-4o) with dynamic database schema context.

from anthropic import Anthropic
import sqlglot
import sqlparse
import pandas as pd
from dataclasses import dataclass

@dataclass
class TableSchema:
    name: str
    columns: list[dict]  # [{name, type, description, example}]
    row_count: int
    sample_rows: list[dict]
    foreign_keys: list[dict]  # [{from_col, to_table, to_col}]

class TextToSQLEngine:
    def __init__(self, db_connection, db_dialect: str = 'postgres'):
        self.db = db_connection
        self.dialect = db_dialect
        self.llm = Anthropic()
        self.schema = self._extract_full_schema()
        self.query_history = []

    def _extract_full_schema(self) -> dict[str, TableSchema]:
        """Automatically extract schema from database"""
        if self.dialect == 'postgres':
            return self._extract_postgres_schema()
        elif self.dialect == 'mysql':
            return self._extract_mysql_schema()
        return {}

    def _extract_postgres_schema(self) -> dict[str, TableSchema]:
        tables = {}

        # Get list of tables
        tables_df = pd.read_sql("""
            SELECT table_name
            FROM information_schema.tables
            WHERE table_schema = 'public'
              AND table_type = 'BASE TABLE'
        """, self.db)

        for table_name in tables_df['table_name']:
            # Columns with types and comments
            cols_df = pd.read_sql(f"""
                SELECT
                    c.column_name,
                    c.data_type,
                    c.is_nullable,
                    col_description('{table_name}'::regclass, c.ordinal_position) as description
                FROM information_schema.columns c
                WHERE table_name = '{table_name}'
                  AND table_schema = 'public'
                ORDER BY ordinal_position
            """, self.db)

            # FK relationships
            fks_df = pd.read_sql(f"""
                SELECT
                    kcu.column_name as from_col,
                    ccu.table_name as to_table,
                    ccu.column_name as to_col
                FROM information_schema.table_constraints tc
                JOIN information_schema.key_column_usage kcu
                    ON tc.constraint_name = kcu.constraint_name
                JOIN information_schema.constraint_column_usage ccu
                    ON ccu.constraint_name = tc.constraint_name
                WHERE tc.constraint_type = 'FOREIGN KEY'
                  AND tc.table_name = '{table_name}'
            """, self.db)

            # Sample data
            sample_df = pd.read_sql(
                f"SELECT * FROM {table_name} LIMIT 3", self.db
            )

            row_count = pd.read_sql(
                f"SELECT COUNT(*) as cnt FROM {table_name}", self.db
            )['cnt'].iloc[0]

            tables[table_name] = TableSchema(
                name=table_name,
                columns=cols_df.to_dict('records'),
                row_count=int(row_count),
                sample_rows=sample_df.to_dict('records'),
                foreign_keys=fks_df.to_dict('records')
            )

        return tables

SQL Generation with Context

    def _build_schema_context(self, relevant_tables: list[str]) -> str:
        """Compact schema representation for LLM"""
        lines = []
        for table_name in relevant_tables:
            if table_name not in self.schema:
                continue
            t = self.schema[table_name]
            lines.append(f"Table: {table_name} ({t.row_count:,} rows)")

            for col in t.columns:
                desc = f" -- {col['description']}" if col.get('description') else ""
                lines.append(f"  {col['column_name']} {col['data_type']}{desc}")

            for fk in t.foreign_keys:
                lines.append(f"  FK: {fk['from_col']} → {fk['to_table']}.{fk['to_col']}")

            if t.sample_rows:
                lines.append(f"  Sample: {t.sample_rows[0]}")
            lines.append("")

        return "\n".join(lines)

    def _select_relevant_tables(self, question: str) -> list[str]:
        """Select needed tables via LLM"""
        all_tables_desc = "\n".join([
            f"- {name}: {[c['column_name'] for c in t.columns[:5]]}..."
            for name, t in self.schema.items()
        ])

        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=200,
            messages=[{
                "role": "user",
                "content": f"""Tables available:
{all_tables_desc}

Question: {question}

List only the table names needed, comma-separated."""
            }]
        )
        names = [n.strip() for n in response.content[0].text.split(',')]
        return [n for n in names if n in self.schema]

    def generate_sql(self, question: str) -> dict:
        """Generate SQL from natural language"""
        relevant_tables = self._select_relevant_tables(question)
        schema_context = self._build_schema_context(relevant_tables)

        # Consider history for context queries ("and now by region")
        conversation_context = ""
        if self.query_history:
            last = self.query_history[-1]
            conversation_context = f"\nPrevious question: {last['question']}\nPrevious SQL:\n{last['sql']}\n"

        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=800,
            system=f"""You are a SQL expert for {self.dialect}.
Generate syntactically correct SQL queries.
Return ONLY the SQL query, no explanations.
Use proper {self.dialect} syntax.
Avoid SELECT *. Always use column aliases for aggregates.
Limit results to 1000 rows unless user asks for aggregation.

Schema:
{schema_context}
{conversation_context}""",
            messages=[{"role": "user", "content": question}]
        )

        raw_sql = response.content[0].text.strip()
        # Remove markdown wrapper
        if '```' in raw_sql:
            raw_sql = raw_sql.split('```')[1]
            if raw_sql.startswith('sql\n'):
                raw_sql = raw_sql[4:]

        return {
            'sql': raw_sql,
            'relevant_tables': relevant_tables,
            'question': question
        }

Validation and Safe Execution

    def validate_sql(self, sql: str) -> tuple[bool, str]:
        """Validate SQL before execution"""
        try:
            # Parse via sqlglot
            parsed = sqlglot.parse_one(sql, dialect=self.dialect)
        except Exception as e:
            return False, f"Parse error: {e}"

        # Check for dangerous operations
        sql_upper = sql.upper()
        forbidden = ['DROP', 'DELETE', 'UPDATE', 'INSERT', 'TRUNCATE', 'ALTER', 'CREATE']
        for keyword in forbidden:
            if keyword in sql_upper:
                return False, f"Forbidden operation: {keyword}"

        # Check for LIMIT in non-aggregate queries
        if 'GROUP BY' not in sql_upper and 'LIMIT' not in sql_upper:
            sql += "\nLIMIT 1000"

        return True, sql

    def execute(self, question: str) -> dict:
        """Full pipeline: question to result"""
        generation = self.generate_sql(question)
        sql = generation['sql']

        is_valid, validated_sql = self.validate_sql(sql)
        if not is_valid:
            # Attempt to fix SQL
            sql = self._fix_sql(sql, validated_sql)
            is_valid, validated_sql = self.validate_sql(sql)
            if not is_valid:
                return {'error': validated_sql, 'sql': sql}

        try:
            df = pd.read_sql(validated_sql, self.db)
            self.query_history.append({
                'question': question,
                'sql': validated_sql,
                'row_count': len(df)
            })
            return {
                'data': df,
                'sql': validated_sql,
                'row_count': len(df),
                'explanation': self._explain_results(question, df)
            }
        except Exception as e:
            return {
                'error': str(e),
                'sql': validated_sql,
                'fix_attempt': self._fix_sql(validated_sql, str(e))
            }

    def _fix_sql(self, sql: str, error: str) -> str:
        """Attempt to fix SQL via LLM"""
        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=500,
            messages=[{
                "role": "user",
                "content": f"""Fix this {self.dialect} SQL:
{sql}

Error: {error}

Return only the fixed SQL."""
            }]
        )
        return response.content[0].text.strip()

Generation Quality by Query Type

Self-Correction Loop

On execution error, the system automatically starts a second generation cycle with the error text in context. 85% of errors are fixed on the first attempt. Critical errors (wrong table names, missing columns) are less common when using the full schema in the prompt. For complex queries, we add few-shot examples from your database — this improves accuracy for JOIN 3+ tables to 85%.

Step-by-Step Implementation Plan for Text-to-SQL

1. Schema analysis and data profiling. Extract metadata, identify frequently asked questions.
2. Pipeline setup. Select LLM, calibrate prompts for your DBMS.
3. Testing. Generate 100+ questions on your data, measure accuracy.
4. Optimization. Fix errors, add few-shot examples.
5. Deployment. Set up REST API or chat interface, train users.

What’s Included: Deliverables and Timelines

Total timeline — 2 to 6 weeks depending on schema complexity and number of tables. Pricing is calculated individually and includes unlimited query licensing. Average cost savings of $50,000 per year for enterprises.

How We Do It: Detailed Case Study

For a large retail chain (45 tables, 12 FK, part of data in BigQuery, part in PostgreSQL), we implemented Text-to-SQL that processes questions in Russian and English. After two weeks of calibration, accuracy on top-20 queries (sales totals, warehouse summaries, return analytics) reached 97%. Within a month of use, requests to analysts dropped by 60%, and data response time went from 2 hours to 10 seconds. The key was adding window function examples in the prompt — without that, accuracy was 15% lower. Analytics budget savings reached 40%, which with an average department payroll of 1 million rubles per month gives 400k rubles monthly savings.

Experience and Guarantees

Our team has over 5 years of experience developing AI data solutions and has completed 15+ successful Text-to-SQL implementation projects for retail, fintech, and logistics companies. We guarantee accuracy of at least 85% on typical business queries in your domain, and if not achieved, we refine for free until acceptable. Results are documented in a transparent metrics report. All data stays on your servers — we don't transfer it to third parties and use it only within the session for SQL generation. Proof of Concept available for $2,500.

Data Engineering for ML: Pipelines, Labeling, and Data Quality

“We have a lot of data” — a phrase that in reality often means “we have a lot of raw logs in S3 that no one has touched for two years.” Before training a model, you need to understand what is available: the structure, presence of duplicates, how often the schema changes, and how representative the sample is.

Data Engineering for ML is not just ETL. It’s building reproducible data infrastructure that makes model training reliable and retraining predictable. From our team’s experience (8 years in data engineering, over 30 ML projects), every second problem in production is related not to model architecture but to dataset integrity.

How Are ETL Pipelines for ML Different from BI?

ETL for analytics and ETL for ML are different tasks. Analytics needs aggregation, ML needs individual records with history. Analytics doesn’t require train/val/test split, ML does. Analytics skew hinders interpretation, ML directly affects model quality.

Tools. Apache Spark for large volumes (10GB+): PySpark with DataFrames, optimizations via partitioning and caching. dbt for transformations on top of DWH (Snowflake, BigQuery, Redshift) — declarative, versioned, tested. Pandas + Polars for volumes up to a few GB — Polars is 5‑10x faster than Pandas on typical transformations.

Temporal splits. For ML it’s important that the split is by time, not random. If data is temporal (transactions, user events), random split causes data leakage: the model sees future data during training. Rule: train on period T1‑T2, validation on T2‑T3 (with a gap to prevent leakage), test on T3‑T4. An incorrect split can cost 10–15% of model quality on validation.

Incremental pipelines. The model is retrained weekly on new data. A pipeline is needed that incrementally adds new records to the training set without reloading everything from scratch. Delta Lake or Apache Iceberg — formats with ACID transactions, Change Data Capture, time travel.

What Causes Training‑Serving Skew and How to Avoid It?

Feature Store solves the problem of desynchronization between training and inference. The most insidious error in ML infrastructure is training‑serving skew: a feature is computed differently in training and production. The model learns on correct data, but inference gets different values.

Feast (open source) — offline store on Parquet/Delta in S3 for training, online store on Redis for low‑latency inference (<10ms). Feature definitions as Python code:

from feast import FeatureView, Field
from feast.types import Float32, Int64

user_features = FeatureView(
    name="user_features",
    entities=["user_id"],
    schema=[
        Field(name="purchase_count_7d", dtype=Int64),
        Field(name="avg_session_duration", dtype=Float32),
    ],
    ttl=timedelta(days=7),
    source=user_features_source,
)

One definition, used everywhere. No discrepancies. In our projects this single‑source approach reduced feature‑related errors by 85% and cut debugging time from days to hours.

Streaming features. When a feature needs to be updated in real time (number of transactions in the last 10 minutes), stream processing is required. Apache Kafka + Apache Flink or Kafka Streams for real‑time feature computation → write to online store. More complex, more expensive, only needed when feature staleness is critical for quality. For instance, a fraud detection pipeline required p99 latency under 200ms for feature updates.

Data Labeling: How Not to Waste Budget

Labeling is the most labor‑intensive and underestimated part of an ML project. Poorly labeled data cannot be fixed by any architecture.

Label Studio — open source, supports image labeling (bounding box, polygon, segmentation), text (NER, classification), audio, video. Deploys in 10 minutes via Docker. For small teams — first choice.

Labeling quality assessment. Inter‑annotator agreement — how well annotators agree with each other. Cohen’s Kappa > 0.8 — good, 0.6‑0.8 — acceptable, < 0.6 — task ambiguous or instructions poor. Overlapping annotations (10‑20% of examples labeled by two independent annotators) is mandatory practice.

Active learning prevents budget waste. Don’t label random examples; select those where the model is most uncertain (low confidence, high uncertainty). Allows achieving the same quality with 50‑70% of the labeling volume. Modals, Prodigy, Label Studio support active learning workflows. In one NLP project, we reduced the labeling budget by 2.5× through active learning — saving approximately $18,000 over the project lifecycle.

Synthetic data. When real data is scarce or expensive to obtain. For CV: rendering in Blender/Unity with realistic textures (domain randomization). For NLP: paraphrase via LLM, backtranslation. Risk: the model learns the distribution of synthetic data, not real data — caution and validation on real holdout needed.

Data Quality: Validation and Monitoring

Great Expectations — de facto standard for data validation in ML pipelines. Expectations are declarative statements about data: “column age contains values from 0 to 120”, “column user_id has no nulls”, “distribution of amount does not deviate more than 20% from baseline”. Runs in the pipeline, on failure blocks progression. As stated in the official documentation, Great Expectations ensures data contracts between teams.

Pandera — Pythonic alternative for pandas/polars DataFrames. Schema‑based validation with type hints:

import pandera as pa

schema = pa.DataFrameSchema({
    "user_id": pa.Column(int, nullable=False),
    "score": pa.Column(float, pa.Check.between(0, 1)),
    "label": pa.Column(str, pa.Check.isin(["positive", "negative", "neutral"])),
})

Data freshness. The model expects data from the last N days. ETL fails, data is not updated — the model uses stale features. Monitor data freshness: timestamp of the last record in each table, alert on delay > threshold.

Deduplication. Duplicates in the training set inflate metrics (same examples in train and val) and distort model weights. MinHash LSH for approximate deduplication of large datasets. For exact — hash by normalized content.

Validation Tools Comparison

What Does a Data Engineering Project for ML Include?

We provide the full cycle:

• Audit of existing data and pipelines (1 week).
• Architecture design: selection of tools, formats, labeling methods.
• Implementation of ETL/ELT pipeline with validation and monitoring.
• Documentation of code and processes (model card, data card).
• Training your team on pipeline operation.
• Post‑deployment support for 3 months.
• Access to code repository and all pipeline definitions.

How We Build a Pipeline: Step by Step

1. Audit existing data. Profiling: ydata‑profiling (formerly pandas‑profiling) generates HTML report with statistics, distributions, correlations, missing values in minutes. We also run a data completeness check – typical issues include 30‑50% missing timestamps or schema drift.
2. Pipeline design. Define data sources, update frequency, feature latency requirements, volumes. Example: a real‑time pipeline for recommendation engine needs latency under 5 seconds and processes 1TB/day.
3. Implementation and testing. Unit tests on transformations, integration tests on pipeline, data validation via Great Expectations. We target 95% test coverage for transformation logic.
4. Deployment and monitoring. Alerts on freshness, quality checks, anomalies in data volumes. Typical alert threshold: no new data for 2 hours.

Storage and Formats

For most ML projects at start: Parquet in S3 + DVC for versioning. Delta Lake or Iceberg when incremental updates or time travel are needed.

Why Trust Us

We have been working in data engineering and ML for over 8 years. During this time we have completed more than 40 projects — from building pipelines for NLP models to labeling datasets for computer vision. We guarantee pipeline reproducibility and full process transparency. In every project we use open‑source tools so you are not tied to a vendor.

Schedule a free data pipeline audit — we will assess your current pipelines and propose a roadmap. Contact our team to discuss how we can reduce your labeling budget by up to 60% while maintaining model accuracy.