AI Аналитик данных — цифровой сотрудник
AI Data Analyst автономно отвечает на бизнес-вопросы через данные: генерирует SQL, выполняет код, строит визуализации, интерпретирует результаты и готовит отчёты. В отличие от BI-инструментов с фиксированными дашбордами, AI-аналитик работает с произвольными ad-hoc запросами на естественном языке.
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
Практический кейс: e-commerce, 15 аналитических запросов/день
Ситуация: маркетинговая команда 5 человек, 15–20 ad-hoc аналитических вопросов в день к 2 аналитикам. Среднее время ответа — 4 часа.
AI Data Analyst:
- Доступ к PostgreSQL с 12 таблицами (заказы, клиенты, продукты, трафик)
- Генерация SQL + выполнение + визуализация (matplotlib/plotly)
- Slack-бот для ad-hoc запросов
Результаты:
- Среднее время ответа на вопрос: 4 часа → 2 минуты
- Правильность SQL с первой попытки: 81% (остальные исправляются автоматически)
- Аналитики перефокусировались на: комплексный анализ, эксперименты, prognosis
- Удовлетворённость команды: 4.3/5.0
Сроки
- Text-to-SQL с вашей схемой: 1–2 недели
- Автоматические отчёты и визуализации: 1–2 недели
- Slack/Teams интеграция для ad-hoc запросов: 1 неделя
- Anomaly detection: 1 неделя
- Итого: 4–6 недель