AI Аналітик даних — цифровий працівник
AI Аналітик даних автономно відповідає на бізнес-питання через дані: генерує 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 (не неявні)
- Для часових рядів — 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": [
"Утримання когорт останні 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% (решта автоматично виправляються)
- Аналітики перезосередилися на: складному аналізі, експериментах, прогнозуванні
- Задоволеність команди: 4.3/5.0
Часова шкала
- Text-to-SQL з вашою схемою: 1–2 тижні
- Автоматичні звіти та візуалізації: 1–2 тижні
- Интеграция Slack/Teams для ad-hoc запитів: 1 тиждень
- Anomaly detection: 1 тиждень
- Разом: 4–6 тижнів