Разработка AI-интерфейса генерации 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 для вашей БД.

Data Engineering для ML: пайплайны, разметка и качество данных

«У нас много данных» — фраза, которая на деле часто означает «у нас много сырых логов в S3, которые никто не трогал два года». Перед тем как обучить модель, нужно понять, что вообще есть: какова структура, есть ли дубли, как часто меняется схема, насколько репрезентативна выборка.

Data Engineering для ML — не просто ETL. Это построение воспроизводимой инфраструктуры данных, которая делает обучение моделей надёжным, а переобучение — предсказуемым. По опыту нашей команды (8 лет в дата-инжиниринге, более 30 проектов в ML) каждая вторая проблема в продакшене связана не с архитектурой модели, а с качеством данных.

ETЛ-пайплайны для ML: чем отличаются от BI

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

Инструменты. Apache Spark (Wikipedia) для больших объёмов (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% качества модели на валидации.

Инкрементальные пайплайны. Модель переобучается еженедельно на новых данных. Нужен пайплайн, который инкрементально добавляет новые записи к обучающей выборке, не перегружая всё с нуля. 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.

Синтетические данные. Когда реальных данных мало или получить их дорого. Для 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.

Что входит в проект по дата-инжинирингу для ML

Мы предоставляем полный цикл:

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

Как мы строим пайплайн: пошагово

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

Почему стоит доверить это нам

Мы занимаемся дата-инжинирингом и ML с 2016 года. За это время реализовали более 40 проектов — от построения пайплайнов для NLP-моделей до разметки датасетов для компьютерного зрения. Гарантируем воспроизводимость пайплайнов и полную прозрачность процессов. В каждом проекте используем инструменты с открытым исходным кодом, чтобы вы не были привязаны к вендору.

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