Генерація SQL природною мовою: Text-to-SQL інтерфейс

Інтелектуальний інтерфейс генерації SQL природною мовою

Бізнес-користувачі витрачають до 40% часу на прості SQL-запити: «скільки продажів учора», «середній чек по регіонах», «топ-10 товарів за виручкою». Аналітики перевантажені рутиною, черги на дані зростають. Text-to-SQL — це задача перекладу природної мови в SQL (Wikipedia). Ми розробляємо такі інтерфейси: ви ставите запитання українською, отримуєте точні SQL-запити та дані — без знання SQL і без відволікання аналітиків.

Чому бізнесу потрібен Text-to-SQL, а не черговий BI-інструмент?

BI-інструменти вимагають налаштування дашбордів — це займає дні. Text-to-SQL працює з будь-якою схемою БД на льоту: ставите запитання, отримуєте SQL та дані за секунди. Наші клієнти закривають 70% простих запитів без участі аналітиків. Аналітики обробляють у 3–5 разів більше завдань на день. Час онбордингу нових співробітників скорочується з кількох тижнів до 1–2 днів. Text-to-SQL підтримує PostgreSQL, MySQL, BigQuery, Snowflake та інші популярні діалекти з коробки. Порівняймо: генерація SQL через Text-to-SQL в 5 разів швидша за ручне написання і в 10 разів швидша за створення дашборду в BI.

Як влаштована наша архітектура Text-to-SQL

Ключова складність — не просто перекласти текст у SQL, а коректно обробити JOIN між 10+ таблицями, врахувати бізнес-логіку та уникнути дорогих full-table scans. Наше рішення використовує LLM (Claude 3.5 Sonnet або GPT-4o) із динамічним контекстом схеми БД.

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]: """Автоматичне вилучення схеми з БД""" 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 = {} # Отримання списку таблиць 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']: # Колонки з типами та коментарями 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 зв'язки 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_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 з контекстом

 def _build_schema_context(self, relevant_tables: list[str]) -> str: """Компактне представлення схеми для 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]: """Вибір потрібних таблиць через 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: """Генерація SQL з природної мови""" relevant_tables = self._select_relevant_tables(question) schema_context = self._build_schema_context(relevant_tables) # Враховуємо історію для контекстних запитів ("а тепер по регіонах") conversation_context = "" if self.query_history: last = self.query_history[-1] conversation_context = f"\nПопереднє запитання: {last['question']}\nПопередній 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() # Прибрати markdown-обгортку 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 } 

Валідація та безпечне виконання

 def validate_sql(self, sql: str) -> tuple[bool, str]: """Перевірка SQL перед виконанням""" try: # Парсинг через sqlglot parsed = sqlglot.parse_one(sql, dialect=self.dialect) except Exception as e: return False, f"Parse error: {e}" # Перевірка на небезпечні операції 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}" # Перевірка наявності LIMIT для non-aggregate запитів 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: """Повний pipeline: питання → результат""" generation = self.generate_sql(question) sql = generation['sql'] is_valid, validated_sql = self.validate_sql(sql) if not is_valid: # Спроба виправити 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: """Спроба виправити SQL через 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() 

Якість генерації за типом запиту

Self-correction loop

При помилці виконання система автоматично запускає другий цикл генерації з текстом помилки в контексті. 85% помилок виправляються з першої спроби. Критичні помилки (неправильні імена таблиць, відсутні колонки) трапляються рідше при використанні повної схеми в промпті. Для складних запитів ми додаємо few-shot приклади з вашої БД — це підвищує точність JOIN 3+ таблиць до 85%.

Покроковий план впровадження Text-to-SQL

1. Аналіз схеми та профілювання даних. Витягуємо метадані, виявляємо часто задавані питання.
2. Налаштування пайплайну. Вибираємо LLM, калібруємо промпти під вашу СУБД.
3. Тестування. Генеруємо 100+ питань за вашими даними, заміряємо точність.
4. Оптимізація. Виправляємо помилки, додаємо few-shot приклади.
5. Розгортання. Встановлюємо REST API або чат-інтерфейс, навчаємо користувачів.

Що входить в роботу: deliverables та строки

Загальний строк — від 2 до 6 тижнів залежно від складності схеми та кількості таблиць. Вартість розраховується індивідуально та включає ліцензію на використання без обмежень за кількістю запитів.

Як ми це робимо: розгорнутий кейс

Для однієї великої ритейл-мережі (схема на 45 таблиць, 12 FK, частина даних у BigQuery, частина в PostgreSQL) ми реалізували Text-to-SQL, який обробляє питання українською та англійською. Після двох тижнів калібрування точність на топ-20 запитах (суми продажів, зведення по складах, аналітика повернень) досягла 97%. За місяць використання кількість звернень до аналітиків скоротилася на 60%, а швидкість відповіді на дані знизилася з 2 годин до 10 секунд. Ключовим стало додавання в промпт прикладів запитів із віконними функціями — без цього точність була на 15% нижчою. Економія бюджету на аналітику досягла 40%, що при середньому ФОП відділу в $9k–13kів на місяць дає 400 тис. рублів економії щомісяця.

Досвід та гарантії

Наша команда розробляє AI-рішення для роботи з даними понад 5 років. Ми виконали 15+ проєктів із впровадження Text-to-SQL для компаній із ритейлу, фінтеху та логістики. Гарантуємо точність не нижче 85% на типових запитах вашої предметної області, а при недосягненні безкоштовно доопрацьовуємо до прийнятного рівня. Результати фіксуємо в прозорому звіті з метриками. Всі дані залишаються на ваших серверах — ми не передаємо їх третім особам і використовуємо лише в рамках сесії для генерації SQL.

Замовте Proof of Concept за 2 дні — наші інженери оцінять схему та підготують робочий прототип. Отримайте консультацію щодо впровадження Text-to-SQL для вашої БД.