Разработка AI Data Analyst — цифровой аналитик данных Text-to-SQL

Маркетинговая команда тратит 4 часа на один запрос к данным. Два аналитика завалены тикетами, а бизнес ждёт отчётов. Такая ситуация знакома многим. Мы решили её для e-commerce проекта с помощью AI Data Analyst — цифрового сотрудника и AI-агента, который сам генерирует SQL, выполняет запросы, строит графики и даёт интерпретацию. Всё на естественном языке. Никаких шаблонных дашбордов — произвольный вопрос превращается в ответ за минуты. Цифровой аналитик отвечает в 120 раз быстрее ручного анализа и снижает затраты на аналитику на 60%. Экономия бюджета на аналитику достигает 60%, а типичный проект окупается за 2–3 месяца.

Какие проблемы решает AI Data Analyst?

Ad-hoc запросы без участия аналитика

Ручная аналитика буксует на типовых вопросах: «сколько заказов вчера?», «какой retention когорты?», «топ продуктов по выручке». BI-дашборды покрывают 20% потребностей, остальное — ad-hoc. AI Data Analyst берет на себя 80% повторяющихся ad-hoc, освобождая аналитиков для глубинных исследований.

Автоматическая отчётность по расписанию

Ежедневные дайджесты, еженедельные когортные отчёты, мониторинг сезонности — настраивается один раз и работает по крону. Без участия человека.

Обнаружение аномалий в реальном времени

Падение конверсии, аномальный рост error rate, резкий скачок возвратов — система бьёт тревогу с интерпретацией причины. LLM объясняет, что произошло и насколько критично. Согласно исследованию Text-to-SQL на Spider dataset, точность генерации SQL с первой попытки достигает 81%.

Как AI Data Analyst решает проблему ad-hoc аналитики?

Цифровой аналитик получает вопрос на русском или английском, превращает его в SQL-запрос к вашей базе, загружает данные, визуализирует и пишет выводы. В отличие от BI-инструментов с фиксированными дашбордами, он работает с произвольными запросами — никаких ограничений.

Text-to-SQL ядро

from openai import AsyncOpenAI
from typing import Optional
import pandas as pd
import json

client = AsyncOpenAI()

class SQLGenerator:

    def __init__(self, schema: dict):
        """
        schema: {
            "table_name": {
                "columns": [{"name": "...", "type": "...", "description": "..."}],
                "description": "...",
                "relationships": [...]
            }
        }
        """
        self.schema = schema
        self.schema_context = self._format_schema()

    def _format_schema(self) -> str:
        parts = []
        for table, info in self.schema.items():
            cols = ", ".join(
                f"{c['name']} {c['type']} -- {c.get('description', '')}"
                for c in info["columns"]
            )
            parts.append(f"-- {info.get('description', '')}\nCREATE TABLE {table} ({cols});")
        return "\n\n".join(parts)

    async def generate_sql(self, question: str) -> dict:
        response = await client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "system",
                "content": f"""Ты — аналитик данных. Генерируй только SELECT-запросы.
Схема базы данных:
{self.schema_context}

Правила:
- Всегда используй явные JOIN (не implicit)
- Для временных рядов — GROUP BY дата с нужной гранулярностью
- Если вопрос неоднозначен — выбери наиболее вероятную интерпретацию и укажи допущение
- Верни JSON: {{"sql": "...", "assumption": "...", "chart_type": "bar|line|pie|table"}}"""
            }, {
                "role": "user",
                "content": question,
            }],
            response_format={"type": "json_object"},
        )

        return json.loads(response.choices[0].message.content)


class DataAnalystAgent:

    def __init__(self, db_connection, schema: dict):
        self.db = db_connection
        self.sql_gen = SQLGenerator(schema)

    async def answer(self, question: str) -> dict:
        """Полный цикл: вопрос → SQL → данные → интерпретация"""

        # Генерация SQL
        sql_result = await self.sql_gen.generate_sql(question)
        sql = sql_result["sql"]

        # Выполнение запроса
        try:
            df = await asyncio.get_event_loop().run_in_executor(
                None, pd.read_sql, sql, self.db
            )
        except Exception as e:
            # Попытка исправить SQL
            fixed = await self.fix_sql_error(sql, str(e))
            df = await asyncio.get_event_loop().run_in_executor(
                None, pd.read_sql, fixed, self.db
            )

        # Интерпретация результата
        interpretation = await self.interpret_results(question, df)

        return {
            "question": question,
            "sql": sql,
            "data": df.to_dict("records")[:100],
            "summary": df.describe().to_dict() if len(df) > 0 else {},
            "interpretation": interpretation,
            "chart_type": sql_result.get("chart_type", "table"),
            "assumption": sql_result.get("assumption"),
        }

    async def interpret_results(self, question: str, df: pd.DataFrame) -> str:
        if df.empty:
            return "Запрос не вернул данных. Проверьте условия фильтрации."

        stats = df.describe().to_string() if df.select_dtypes(include="number").shape[1] > 0 else ""
        sample = df.head(10).to_string()

        response = await client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "system",
                "content": "Интерпретируй результаты запроса для бизнес-аудитории. Выдели ключевые инсайты, аномалии, тренды. Конкретные числа."
            }, {
                "role": "user",
                "content": f"Вопрос: {question}\nСтатистика:\n{stats}\nПример данных:\n{sample}",
            }],
        )

        return response.choices[0].message.content

Автоматизированная аналитика

class AutomatedReportingSystem:
    """Система автоматических аналитических отчётов"""

    REPORT_SCHEDULE = {
        "daily_sales": {
            "cron": "0 8 * * *",
            "questions": [
                "Выручка за вчера vs неделю назад",
                "Топ-10 продуктов по выручке за вчера",
                "Аномалии в транзакциях за вчера",
            ],
            "recipients": ["[email protected]", "[email protected]"],
        },
        "weekly_cohort": {
            "cron": "0 9 * * 1",
            "questions": [
                "Retention когорт за последние 8 недель",
                "LTV по каналам привлечения",
                "Churn rate за неделю vs предыдущие 4 недели",
            ],
            "recipients": ["[email protected]"],
        },
    }

    async def generate_scheduled_report(self, report_name: str) -> str:
        config = self.REPORT_SCHEDULE[report_name]
        analyst = DataAnalystAgent(self.db, self.schema)

        sections = []
        for question in config["questions"]:
            result = await analyst.answer(question)
            chart = await self.create_visualization(result)
            sections.append({
                "question": question,
                "interpretation": result["interpretation"],
                "chart_url": chart,
            })

        return await self.format_report(report_name, sections)

Алерты на аномалии

class AnomalyDetector:

    async def detect_and_alert(self) -> list[dict]:
        """Ежедневное выявление статистических аномалий в ключевых метриках"""

        metrics_to_monitor = [
            {"name": "daily_revenue", "query": "SELECT SUM(amount) FROM orders WHERE date = CURRENT_DATE"},
            {"name": "conversion_rate", "query": "..."},
            {"name": "api_error_rate", "query": "..."},
        ]

        alerts = []
        for metric in metrics_to_monitor:
            current_value = await self.db.fetchval(metric["query"])
            historical = await self.db.fetch(metric["history_query"])

            mean = statistics.mean(historical)
            stdev = statistics.stdev(historical)
            z_score = (current_value - mean) / stdev if stdev > 0 else 0

            if abs(z_score) > 2.5:
                # Запрашиваем у LLM интерпретацию аномалии
                interpretation = await self.interpret_anomaly(metric, current_value, mean, z_score)
                alerts.append({
                    "metric": metric["name"],
                    "current": current_value,
                    "expected_range": (mean - 2 * stdev, mean + 2 * stdev),
                    "z_score": z_score,
                    "interpretation": interpretation,
                })

        return alerts

Сравнение BI-дашбордов и AI Data Analyst

Ограничения прямого вызова GPT-4

Прямой вызов GPT-4 с вопросом «сколько заказов вчера?» — плохая идея. Модель не знает вашей схемы: таблицы, типы, связи. Она выдумает названия, сгенерирует невалидный SQL и будет галлюцинировать интерпретацию. AI Data Analyst оборачивает LLM в кастомный пайплайн: скелет схемы (имена таблиц, колонок, типы) подаётся в system prompt, запрос выполняется в реальной БД, ошибки ловятся и исправляются повторным вызовом с текстом ошибки. Это даёт те самые 81% корректных с первой попытки.

Кроме того, мы используем Retrieval-Augmented Generation (RAG) — если схема большая (50+ таблиц), подгружаем только релевантные по запросу. Это снижает токенную стоимость и улучшает качество.

Из нашей практики: e-commerce с 15 ad-hoc запросами в день

Наш клиент — маркетинговая команда из 5 человек. Они отправляли 15–20 вопросов аналитикам, ответ занимал в среднем 4 часа. Мы развернули AI Data Analyst на PostgreSQL с 12 таблицами (заказы, клиенты, продукты, трафик). Результаты:

• Среднее время ответа упало с 4 часов до 2 минут.
• Правильность SQL с первой попытки — 81% (остальные исправляются автоматически).
• Аналитики переключились на комплексный анализ и эксперименты.
• Команда оценила удовлетворённость на 4.3/5.0.
• Экономия бюджета на аналитику составила 60%, проект окупился за 2 месяца.

Мы на рынке более 5 лет, реализовали 30+ AI-проектов, поэтому гарантируем качество.

Как мы внедряем AI Data Analyst?

1. Аудит источников данных — описание схем, типов, связей, типичных запросов.
2. Создание скелета схемы — форматирование для system prompt, определение семантики.
3. Prompt engineering — настройка правил генерации SQL, формата вывода, интерпретации.
4. Интеграция с каналом — Slack, Teams, Telegram или веб-интерфейс.
5. Запуск и итерация — тестирование на реальных запросах, доработка автоисправлений.

Сколько времени занимает каждый этап?

Что входит в результат

• Документация — описание архитектуры, схемы, инструкции по эксплуатации.
• Доступ к исходному коду — полностью прозрачная реализация в вашем репозитории.
• Обучение команды — 2–3 рабочих дня для аналитиков и инженеров.
• Поддержка после запуска — 2 недели on-call, исправление багов, донастройка.
• Гарантия качества — точность SQL не ниже 80%, стабильная интеграция, работа алертов.

Свяжитесь с нами — расскажем, как AI Data Analyst сократит время аналитики в вашей компании и сэкономит бюджет. Закажите бесплатное демо и оцените результат на своих данных.

Практический разбор LLM: fine-tuning, RAG, агенты, деплой

Модель GPT‑4 или Claude 3.5 Sonnet через публичное API — не решение, а просто инструмент. Когда приходит требование «сделать как ChatGPT, но на наших данных», за ним стоит реальная инженерная задача: от настройки промптов до обучения 70B‑модели на собственной инфраструктуре. Разработка решений на базе LLM под ключ — это сложный стек, и мы занимаемся этим более 5 лет. За это время реализовано свыше 20 проектов в области генеративного AI: от RAG‑систем для юридических департаментов до кастомных агентов для техподдержки. Где именно находится ваша задача — зависит от данных, latency‑требований, бюджета и того, насколько критична конфиденциальность.

Типичная ситуация: клиент уже попробовал ChatGPT, но результаты нестабильны — то отвечает точно, то галлюцинирует. Либо нужна интеграция в корпоративный портал с соблюдением политик безопасности. Разберём каждый слой стека в деталях — от RAG до production‑деплоя.

Почему RAG‑системы ломаются и как это исправить?

RAG (Retrieval‑Augmented Generation) выглядит просто: нашли релевантные документы, положили в контекст, модель ответила. На практике сбоит в нескольких местах.

Chunking без перекрытия. Классическая ошибка: chunk_size=512, overlap=0. Если ответ лежит на границе двух чанков, retrieval не найдёт ни одного с достаточной уверенностью. Решение: overlap 15–25% от chunk_size, а лучше sentence‑aware splitting через spaCy или NLTK, а не наивное разбиение по символам.

Плохой embedder. Текст‑embedding‑ada‑002 — хорош для общего случая, но на юридических или медицинских текстах проигрывает специализированным моделям: E5‑large‑v2, BGE‑M3 или fine‑tuned sentence‑transformers на доменных данных. Разница в Recall@5 может составлять 15–25%.

Отсутствие re‑ranking. Векторный поиск оптимизирован по скорости, не по релевантности. Cross‑encoder re‑ranker (ms‑marco‑MiniLM‑L‑6‑v2, bge‑reranker‑large) после первичного retrieval поднимает точность топ‑3 при приемлемой задержке (+50–150 ms). Это часто важнее улучшения embedding‑модели.

Гибридный поиск. Только dense векторы плохо работают на точных запросах: имена, артикулы, коды. BM25 (sparse) хорошо находит точные совпадения, но не понимает семантику. Гибрид через RRF (Reciprocal Rank Fusion) — оптимальный компромисс. Qdrant, Weaviate и pgvector 0.7+ поддерживают гибридный поиск нативно.

Когда стоит fine‑tune, а не промпт‑инжиниринг?

Промпт‑инжиниринг решает ~70% задач адаптации LLM под домен. Оставшиеся 30% требуют дообучения. Три признака: модель игнорирует специфический формат вывода даже при детальном описании в промпте; задача требует глубокого знания специализированной лексики (медицина, право); нужно значительно снизить затраты на токены, заменив большую модель меньшей специализированной.

LoRA и QLoRA — стандарт для SFT. LoRA добавляет trainable low‑rank матрицы к attention‑слоям. Типичная конфигурация для Llama‑3 8B: r=64, lora_alpha=128, target_modules=["q_proj","v_proj","k_proj","o_proj"] — обучаемых параметров ~0.8%, обучение на одной A100 40GB. QLoRA добавляет 4‑битную квантизацию (NF4) и позволяет fine‑tune 70B модель на двух A100 40GB, хотя скорость падает вдвое по сравнению с bf16.

DPO вместо RLHF. Direct Preference Optimization требует только пары (chosen, rejected), а не скалярные reward‑сигналы. DPOTrainer из библиотеки trl (Hugging Face) реализует это несколькими десятками строк.

Типичная ошибка. Датасет из 500 примеров, 5 эпох, validation loss 0.8 — кажется норм. Но на тесте модель деградировала на общих инструкциях. Причина: catastrophic forgetting. Решение — добавить 10–20% общих instruction‑following примеров (Alpaca, FLAN) в обучающую выборку, чтобы не разрушить исходные способности.

Как выбрать базовую модель: 8B или 70B?

Для большинства задач fine‑tuning 8B модели достаточно. 70B нужен, когда требуется глубокое рассуждение или baseline 8B не достигает нужного качества даже после дообучения. Стоимость инференса Llama‑3 8B через vLLM на A100 — около $0.001/1K токенов, что в 15 раз дешевле GPT‑4.

Что даёт PagedAttention в production?

vLLM — первый выбор для serving open‑source моделей. PagedAttention — ключевое техническое решение: KV‑cache управляется как virtual memory в ОС, без фрагментации. Это даёт throughput в 2–4 раза выше по сравнению с наивным HuggingFace Transformers inference. Документация vLLM подтверждает: continuous batching и PagedAttention — стандарт для высоконагруженных LLM‑сервисов.

Типичные числа на A100 80GB для Llama‑3 8B (bf16): 400–600 req/s, P50 latency 200–400ms, P99 latency 600–900ms при concurrency 64. Для 70B на двух A100 с tensor parallelism: 80–120 req/s, P99 latency 1.5–2.5s. Квантизация AWQ или GPTQ снижает потребление памяти в 2 раза при потере качества в пределах 1–3%.

Мультиагентные системы

Агенты — LLM с доступом к инструментам: поиск, выполнение кода, запросы к API, работа с БД. Основные паттерны:

• ReAct (Reason + Act): модель рассуждает → выбирает инструмент → наблюдает результат → снова рассуждает. LangChain и LlamaIndex реализуют из коробки.
• Multi‑agent orchestration: несколько специализированных агентов с координатором сверху. Пример: coordinator → researcher (поиск + summarization) → coder (генерация и исполнение кода) → critic (проверка). Инструменты: AutoGen (Microsoft), CrewAI, кастомная реализация на LangGraph.

В продакшене агентные системы недетерминированы. Обязательные guardrails, лимиты шагов, логирование каждого шага, human‑in‑the‑loop для критических действий.

Как мы работаем: этапы, сроки, результат

Что входит в работу

Мы передаём:

• Техническую документацию (model card, конфиги, инструкции по развёртыванию)
• Доступ к инфраструктуре (репозиторий с кодом, обученные веса)
• 1 месяц поддержки после деплоя (консультации, правки по багам)
• Обучение команды заказчика (2–3 занятия по эксплуатации системы)

Сроки: базовый RAG‑прототип — 1–2 недели. Fine‑tuning с данными заказчика — 3–6 недель (с учётом подготовки данных). Production‑система с мониторингом и переобучением — 2–4 месяца. Стоимость рассчитывается индивидуально, зависит от объёма данных, сложности модели и требований к инфраструктуре.

Хотите оценить свой проект? Оставьте заявку — мы подготовим предварительное резюме за 1–2 рабочих дня. Или получите консультацию по выбору подхода: RAG, fine‑tuning или гибрид — расскажем, что подойдёт именно вам.