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