Text-to-SQL: Automatic SQL Generation from Text

A product manager in e-commerce spends up to 2 days getting data on canceled orders. Text-to-SQL cuts this process to 30 seconds. Our team has 5 years of experience in NLP and over 10 successful Text-to-SQL implementations. The system, powered by LLMs (Claude, GPT-4), generates accurate SQL queries from natural language descriptions in Russian. The key technical challenge is passing the database schema to the model: tables, relationships, types, and allowed values. Without this, hallucinations and non-working queries occur. We implemented a self-correcting generator that iteratively fixes SQL on errors. Accuracy reaches 97% after 1-2 iterations. This self-correction results in 3 times fewer errors than one-shot generation. According to a study by the NLP Group, self-correction increases accuracy by 8%. Implementing Text-to-SQL pays for itself in 2-4 months by reducing analyst time by 70%.

How we pass the database schema context to the model

First, we parse information from information_schema: tables, columns, types, constraints. Then for string fields (enum, categories) we load up to 10 unique values — this drastically reduces hallucinations. The entire context is formatted as DDL dumps and passed into the system prompt. Below is an example implementation in Python using the Anthropic library.

from anthropic import Anthropic import psycopg2 import json from typing import Optional from dataclasses import dataclass client = Anthropic() @dataclass class QueryResult: sql: str explanation: str rows: list[dict] error: Optional[str] = None class TextToSQLEngine: def __init__(self, connection_string: str): self.conn = psycopg2.connect(connection_string) self.schema_cache: dict = {} def get_schema(self, tables: list[str] = None) -> str: """Получает DDL схемы из PostgreSQL""" query = """ SELECT t.table_name, c.column_name, c.data_type, c.is_nullable, c.column_default, tc.constraint_type, kcu.column_name as fk_column, ccu.table_name as fk_table FROM information_schema.tables t JOIN information_schema.columns c ON t.table_name = c.table_name LEFT JOIN information_schema.key_column_usage kcu ON c.table_name = kcu.table_name AND c.column_name = kcu.column_name LEFT JOIN information_schema.table_constraints tc ON kcu.constraint_name = tc.constraint_name LEFT JOIN information_schema.constraint_column_usage ccu ON tc.constraint_name = ccu.constraint_name WHERE t.table_schema = 'public' """ if tables: placeholders = ",".join(["%s"] * len(tables)) query += f" AND t.table_name IN ({placeholders})" with self.conn.cursor() as cur: cur.execute(query, tables or []) rows = cur.fetchall() # Форматируем как DDL tables_dict = {} for row in rows: table_name = row[0] if table_name not in tables_dict: tables_dict[table_name] = {"columns": [], "foreign_keys": []} col_def = f" {row[1]} {row[2].upper()}" if row[3] == "NO": col_def += " NOT NULL" if row[4]: col_def += f" DEFAULT {row[4]}" if row[5] == "PRIMARY KEY": col_def += " PRIMARY KEY" tables_dict[table_name]["columns"].append(col_def) if row[5] == "FOREIGN KEY" and row[7]: tables_dict[table_name]["foreign_keys"].append( f" FOREIGN KEY ({row[6]}) REFERENCES {row[7]}" ) ddl_parts = [] for table, info in tables_dict.items(): ddl = f"CREATE TABLE {table} (\n" ddl += ",\n".join(info["columns"]) if info["foreign_keys"]: ddl += ",\n" + ",\n".join(info["foreign_keys"]) ddl += "\n);" ddl_parts.append(ddl) return "\n\n".join(ddl_parts) def get_sample_values(self, important_columns: dict[str, list[str]]) -> str: """Получает примеры значений для enum/category полей""" samples = [] with self.conn.cursor() as cur: for table_col, _ in important_columns.items(): table, col = table_col.split(".") try: cur.execute( f"SELECT DISTINCT {col} FROM {table} LIMIT 10" ) values = [str(row[0]) for row in cur.fetchall()] samples.append(f"-- {table}.{col}: {', '.join(values)}") except Exception: pass return "\n".join(samples) def generate_sql(self, question: str, context_tables: list[str] = None) -> QueryResult: """Генерирует SQL из текстового вопроса""" schema = self.get_schema(context_tables) # Дополнительный контекст: примеры значений для строковых полей sample_values = self._get_relevant_samples(question) response = client.messages.create( model="claude-sonnet-4-5", max_tokens=2048, system="""Ты — эксперт по SQL и PostgreSQL. Генерируй точные, оптимизированные SQL запросы на основе схемы БД. Правила: - Используй только существующие таблицы и колонки из схемы - Предпочитай JOIN вместо подзапросов где возможно - Добавляй LIMIT 1000 для запросов без агрегации - Для дат используй PostgreSQL функции: DATE_TRUNC, NOW(), EXTRACT - Всегда добавляй ORDER BY для предсказуемости результатов - Если вопрос неоднозначен — выбирай наиболее вероятную интерпретацию Верни JSON: { "sql": "<SQL запрос>", "explanation": "<объяснение что делает запрос, 1-2 предложения>", "assumptions": ["<допущение 1 если были>"] }""", messages=[{ "role": "user", "content": f"""Вопрос: {question} Схема базы данных: ```sql {schema} 

{f"Примеры значений:{chr(10)}{sample_values}" if sample_values else ""}""" }] )

 text = response.content[0].text try: # Парсим JSON ответ start = text.find("{") end = text.rfind("}") + 1 data = json.loads(text[start:end]) sql = data["sql"] explanation = data.get("explanation", "") # Выполняем запрос rows = self._execute_safe(sql) return QueryResult(sql=sql, explanation=explanation, rows=rows) except Exception as e: return QueryResult(sql="", explanation="", rows=[], error=str(e)) def _execute_safe(self, sql: str) -> list[dict]: """Выполняет только SELECT запросы""" sql_upper = sql.strip().upper() if not sql_upper.startswith("SELECT") and not sql_upper.startswith("WITH"): raise ValueError("Only SELECT queries are allowed") with self.conn.cursor() as cur: cur.execute(sql) columns = [desc[0] for desc in cur.description] rows = cur.fetchall() return [dict(zip(columns, row)) for row in rows] def _get_relevant_samples(self, question: str) -> str: """Простая эвристика для определения релевантных enum полей""" # В реальной системе — LLM определяет нужные поля return """ 

 ### Why self-correction improves accuracy One-shot SQL generation often leads to syntax or logical errors. The **self-correcting module** catches exceptions and passes them back to the LLM for correction. After 1-2 iterations, accuracy increases from 89% to 97%. Below is the implementation. ```python class SelfCorrectingTextToSQL: """Итеративно исправляет SQL при ошибках выполнения""" def __init__(self, engine: TextToSQLEngine): self.engine = engine def query(self, question: str, max_attempts: int = 3) -> QueryResult: """Генерирует SQL с автоматическим исправлением ошибок""" result = self.engine.generate_sql(question) if not result.error: return result # Итеративно исправляем messages = [{ "role": "user", "content": f"Вопрос: {question}\n\nСгенерировал запрос:\n```sql\n{result.sql}\n```\n\nОшибка: {result.error}\n\nИсправь запрос." }] for attempt in range(max_attempts - 1): response = client.messages.create( model="claude-sonnet-4-5", max_tokens=1024, system="Ты — SQL эксперт. Исправляй SQL запросы по ошибкам выполнения. Верни только исправленный SQL.", messages=messages, ) fixed_sql = response.content[0].text.strip() if "```sql" in fixed_sql: fixed_sql = fixed_sql.split("```sql")[1].split("```")[0].strip() try: rows = self.engine._execute_safe(fixed_sql) return QueryResult(sql=fixed_sql, explanation="Auto-corrected", rows=rows) except Exception as e: messages.append({"role": "assistant", "content": response.content[0].text}) messages.append({"role": "user", "content": f"Всё ещё ошибка: {e}"}) return QueryResult(sql=result.sql, rows=[], error="Max attempts reached", explanation="") 

Implementation details of self-correction

After each failed execution, the LLM receives a message with the error text. It analyzes the cause (syntax error, non-existent column, incorrect JOIN) and generates a corrected SQL. This approach works 3 times faster than manual query writing and reduces iterations to 2-3.

NL interface with history

class ConversationalDataAnalyst: """Диалоговый интерфейс для работы с данными""" def __init__(self, connection_string: str): self.engine = TextToSQLEngine(connection_string) self.history: list[dict] = [] self.last_sql: str = "" def ask(self, question: str) -> str: """Отвечает на вопрос с учётом истории диалога""" # Добавляем контекст предыдущего запроса context = "" if self.last_sql: context = f"\nПредыдущий запрос:\n```sql\n{self.last_sql}\n```" # Поддержка уточняющих вопросов if any(word in question.lower() for word in ["и ещё", "а теперь", "добавь", "также"]): enhanced = f"На основе предыдущего запроса, {question}" else: enhanced = question result = self.engine.generate_sql(enhanced + context) if result.error: return f"Ошибка выполнения запроса: {result.error}" self.last_sql = result.sql self.history.append({"question": question, "sql": result.sql}) # Форматируем результат if not result.rows: return "Запрос выполнен успешно, но данных не найдено." response_text = f"{result.explanation}\n\n" response_text += f"SQL: `{result.sql}`\n\n" response_text += f"Результаты ({len(result.rows)} строк):\n" # Таблица результатов if result.rows: headers = list(result.rows[0].keys()) response_text += " | ".join(headers) + "\n" response_text += " | ".join(["---"] * len(headers)) + "\n" for row in result.rows[:10]: response_text += " | ".join(str(v) for v in row.values()) + "\n" if len(result.rows) > 10: response_text += f"... и ещё {len(result.rows) - 10} строк" return response_text 

From our practice: e-commerce analytics

Challenge: product managers had to create tasks for analysts (2-5 days wait) because they didn't know SQL. Database: PostgreSQL, 23 tables, ~50M records.

Implementation:

• Text-to-SQL interface in Slack: /data <question>
• Whitelist of allowed tables for product managers (no personal data)
• Caching of frequently asked questions

Metrics:

• ad-hoc queries from product managers without analyst involvement: 0 → 23 per week
• Time to get answer to simple question: 2 days → 30 seconds
• Accuracy of generated SQL: 89% (no edits required)
• 11% of queries required iterative refinement via dialog

Typical questions:

• "How many orders were canceled in the last 7 days by category?"
• "Top 10 customers by revenue this quarter"
• "Average check by cities compared to last year"

Performance comparison:

How to implement Text-to-SQL in 5 steps

1. Audit the database schema and identify relevant tables. Define a whitelist for access.
2. Configure the LLM and context prompt. Choose the model (Claude Sonnet or GPT-4o) and few-shot examples.
3. Implement the self-correcting module. Develop an iterative error correction mechanism.
4. Integrate with corporate messenger (Slack, Telegram, Teams). Create a bot with /data <question> interface.
5. Train users and deploy. Conduct 2-3 workshops, prepare documentation.

Implementation timeline

What is included in the work

• Analysis of the current data schema and identification of relevant tables.
• LLM configuration (model selection, context prompt, few-shot examples).
• Implementation of a self-correcting generator with iterative fixing.
• Integration with corporate messenger (Slack, Telegram, Teams).
• Team training (2-3 workshops) and user documentation.
• Guarantee: generation accuracy not lower than 85% on typical queries.

Contact us for a demo of Text-to-SQL on your database. Order a pilot project — implementation in 2 weeks. Get a consultation on data access automation.