Генерація 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%, що при середньому ФОП відділу в 1 млн рублів на місяць дає 400 тис. рублів економії щомісяця.

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

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

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

Чому дата-інжиніринг визначає успіх ML-моделі

Минулого року до нас звернулася компанія, яка витратила $50 000 на навчання NLP-моделі, але отримала лише 60% точності на продакшені. Причина — data leakage через випадковий split часових даних. Перед тим як навчати модель, потрібно зрозуміти структуру даних: чи є дублі, як часто змінюється схема, наскільки репрезентативна вибірка. Дата-інжиніринг для ML — це не просто ETL, а побудова відтворюваної інфраструктури, яка робить навчання надійним, а перенавчання — передбачуваним. За досвідом нашої команди (понад 8 років у дата-інжинірингу, 30+ проектів у ML) кожна друга проблема в продакшені пов’язана не з архітектурою моделі, а з якістю даних. Замовте аудит ваших даних — оцінимо поточний пайплайн безкоштовно.

Як ETL-пайплайни для ML відрізняються від BI

ETL для аналітики та ETL для ML — різні завдання. В аналітиці важлива агрегація, у ML — індивідуальні записи з історією. В аналітиці train/val/test split не потрібен, у ML — критичний. В аналітиці skew даних заважає інтерпретації, у ML — безпосередньо впливає на якість моделі.

Інструменти. Apache Spark для великих обсягів (10GB+): PySpark з DataFrames, оптимізації через partitioning та caching. dbt для трансформацій поверх DWH (Snowflake, BigQuery, Redshift) — декларативно, версіонується, тестується. Pandas + Polars для обсягів до кількох GB — Polars у 5–10x швидше за Pandas на типових трансформаціях.

Temporal splits. Для ML важливо, що split за часом, а не випадковий. Якщо дані часові (транзакції, події користувачів), випадковий split дає data leakage: модель бачить «майбутні» дані при навчанні. Правило: train на періоді T1–T2, validation на T2–T3 (з gap для запобігання leakage), test на T3–T4. Неправильний split може коштувати 10–15% якості моделі на валідації. Temporal split best practices (scikit-learn docs)

Інкрементальні пайплайни. Модель перенавчається щотижня на нових даних. Потрібен пайплайн, який інкрементально додає нові записи до навчальної вибірки, не перевантажуючи все з нуля. Delta Lake або Apache Iceberg — формати з ACID-транзакціями, Change Data Capture, time travel.

Як уникнути training-serving skew за допомогою Feature Store

Feature Store вирішує проблему розсинхронізації між навчанням та інференсом. Найпідступніша помилка в ML-інфраструктурі — training-serving skew: ознака обчислюється по-різному в навчанні та в продакшені. Модель вчиться на «правильних» даних, а інференс отримує інші.

Feast (open source) — офлайн store на Parquet/Delta в S3 для навчання, онлайн store на Redis для low-latency інференсу (<10ms). Feature definitions як Python-код:

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,
)

Один definition використовується всюди — немає розбіжностей.

Потокові ознаки. Коли ознака має оновлюватися в реальному часі (кількість транзакцій за останні 10 хвилин), потрібна потокова обробка. Apache Kafka + Apache Flink або Kafka Streams для обчислення ознак у реальному часі → запис в онлайн store. Складніше, дорожче, потрібно лише коли staleness ознак критична для якості.

Розмітка даних: як не витратити бюджет даремно

Розмітка — найтрудомісткіша та недооцінювана частина ML-проекту. Погано розмічені дані не виправить жодна архітектура.

Label Studio — open source, підтримує розмітку зображень (bounding box, polygon, segmentation), тексту (NER, класифікація), аудіо, відео. Піднімається за 10 хвилин через Docker. Для невеликих команд — перший вибір.

Оцінка якості розмітки. Inter-annotator agreement — наскільки згодні розмітники між собою. Cohen's Kappa > 0.8 — добре, 0.6–0.8 — прийнятно, < 0.6 — завдання неоднозначне або інструкція погана. Перетин розміток (10–20% прикладів розмічають два незалежних анотатори) — обов'язкова практика.

Active learning. Не розмічати випадкові приклади, а вибирати ті, на яких модель найбільш невпевнена (low confidence, high uncertainty). Дозволяє досягти тієї ж якості при 50–70% обсягу розмітки. Modals, Prodigy, Label Studio підтримують active learning workflows. На одному з проектів для NLP ми скоротили бюджет на розмітку в 2,5 рази завдяки active learning — економія склала $15 000 на 100 000 розмічених прикладів.

Синтетичні дані. Коли реальних даних мало або отримати їх дорого. Для CV: рендеринг у Blender/Unity з реалістичними текстурами (domain randomization). Для NLP: parafrase через LLM, backtranslation. Ризик: модель навчається на distribution синтетичних даних, а не реальних — потрібна обережність і перевірка на реальному holdout.

Якість даних: валідація та моніторинг

Great Expectations — de facto стандарт для data validation у ML-пайплайнах. Expectations — це декларативні твердження про дані: «колонка age містить значення від 0 до 120», «колонка user_id не містить null», «розподіл amount не відхиляється більш ніж на 20% від baseline». Запускається в пайплайні, при провалі — блокує проходження.

Pandera — Pythonic alternative для pandas/polars DataFrames. Schema-based validation з 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. Модель очікує дані за останні N днів. ETL впав, дані не оновилися — модель використовує застарілі ознаки. Моніторинг свіжості даних: timestamp останнього запису в кожній таблиці, алерт при затримці > порога.

Дедуплікація. Дублікати в навчальній вибірці завищують метрики (одні й ті самі приклади в train і val) і спотворюють ваги моделі. MinHash LSH для наближеної дедуплікації великих датасетів. Для точної — хеш за нормалізованим контентом.

Сховища та формати

Для більшості ML-проектів на старті: Parquet в S3 + DVC для версіонування. Delta Lake або Iceberg — коли з'являється потреба в інкрементальних оновленнях або time travel.

Типові помилки при побудові пайплайнів

• Пропуск перевірки свіжості даних. Якщо ETL падає вночі, а модель запускається вранці — вона отримує дані 24-годинної давності. Рішення: алерт при затримці > 30 хвилин.
• Відсутність версіонування даних. Не можна відтворити експеримент, бо дані змінилися. DVC або Delta Lake time travel виправляють це.
• Забувають про schema evolution. Нове поле з’являється, а пайплайн падає. Автоматичне виявлення змін схеми через Great Expectations.

Active learning дозволяє скоротити бюджет на розмітку до 50–70%. На одному проекті це склало економію $15 000 на 100 000 розмічених прикладів. Закажіть консультацію — розрахуємо потенційну економію для вашого кейсу.

Що входить у проект з дата-інжинірингу для ML

Ми надаємо повний цикл:

• Аудит існуючих даних та пайплайнів (1 тиждень).
• Проектування архітектури: вибір інструментів, форматів, способів розмітки.
• Реалізація ETL/ELT пайплайну з валідацією та моніторингом.
• Документація коду та процесів (model card, data card).
• Навчання вашої команди роботі з пайплайном.
• SLA на супровід та підтримку.

Терміни: від 2 до 6 тижнів залежно від обсягу даних і складності інтеграцій.

Як ми будуємо пайплайн: покроково

1. Аудит існуючих даних. Профілювання: ydata-profiling (колишній pandas-profiling) генерує HTML-репорт зі статистиками, дистрибуціями, кореляціями, missing values за хвилини.
2. Проектування пайплайну. Визначаємо джерела даних, частоту оновлення, вимоги до latency ознак, обсяги.
3. Реалізація та тестування. Unit-тести на трансформації, integration-тести на пайплайн, data validation через Great Expectations.
4. Деплой та моніторинг. Алерти на freshness, quality checks, аномалії в обсягах даних.

Чому варто довірити це нам

Ми займаємося дата-інжинірингом та ML з понад 8-річним досвідом. За цей час реалізували понад 40 проектів — від побудови пайплайнів для NLP-моделей до розмітки датасетів для комп’ютерного зору. Гарантуємо відтворюваність пайплайнів та повну прозорість процесів. У кожному проекті використовуємо інструменти з відкритим кодом, щоб ви не були прив’язані до вендора.

Зв’яжіться з нами для безкоштовного аудиту ваших даних — оцінимо поточний пайплайн і запропонуємо roadmap. Замовте побудову ML-пайплайну під ключ.