Разработка 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 сократит время аналитики в вашей компании и сэкономит бюджет. Закажите бесплатное демо и оцените результат на своих данных.