После 10 лет внедрения BI-систем (Qlik Sense, Power BI, Data Lens) я понял одну вещь: дашборд — это не решение. Это данные для решения. А между данными и решением — пропасть, которую преодолевает человек.
В этой статье покажу, как построил RAG-систему с чат-интерфейсом для базы из 600 000 записей техники из Федресурса. Без философии — только архитектура, код и грабли.
Проблема: почему дашборды не работают
Типичный сценарий. Аналитик открывает дашборд с данными о технике в лизинге. Нужно найти топ-10 компаний с бензовозами в Московской области.
Что происходит:
Ищет нужный дашборд (их 15 штук)
Выбирает правильные фильтры
Понимает, что нужного среза нет
Идёт к разработчику
Ждёт доработку
Время: от часов до дней. И это для простого вопроса.
Ключевая проблема: чтобы найти ответ в BI, нужно знать, как задать вопрос системе. А пользователи хотят задавать вопросы на человеческом языке.
Архитектура решения
Компоненты
1. LLM Router — определяет тип запроса и маршрутизирует к нужному обработчику:
Точный запрос → SQL Generator
Семантический поиск → RAG
Агрегация → Aggregator
2. SQL Generator — генерирует SQL по естественному языку. Использую Claude API с few-shot примерами.
3. RAG Search — семантический поиск по векторной базе. Для случаев, когда пользователь не знает точных названий.
4. Aggregator — предрасчитанные агрегаты в Redis для быстрых ответов на частые вопросы.
Реализация SQL Generator
Самая интересная часть. Нужно научить LLM генерировать корректный SQL для конкретной схемы.
Схема данных (упрощённо)
CREATE TABLE vehicles ( id SERIAL PRIMARY KEY, vin VARCHAR(17), vehicle_type VARCHAR(100), -- "бензовоз", "автовоз", "экскаватор" brand VARCHAR(100), model VARCHAR(100), year_manufacture INTEGER, leasing_company VARCHAR(200), leasing_start DATE, leasing_end DATE, region VARCHAR(100), inn VARCHAR(12), company_name VARCHAR(300) ); -- Индексы для частых запросов CREATE INDEX idx_vehicle_type ON vehicles(vehicle_type); CREATE INDEX idx_region ON vehicles(region); CREATE INDEX idx_leasing_company ON vehicles(leasing_company);
Промпт для генерации SQL
SYSTEM_PROMPT = """Ты — SQL-генератор для базы техники в лизинге. Схема таблицы vehicles: - id: первичный ключ - vin: VIN-номер - vehicle_type: тип техники (бензовоз, автовоз, экскаватор, погрузчик...) - brand, model: марка и модель - year_manufacture: год выпуска - leasing_company: лизинговая компания - leasing_start, leasing_end: даты лизинга - region: регион РФ - inn, company_name: данные лизингополучателя Правила: 1. Генерируй ТОЛЬКО SELECT-запросы 2. Всегда добавляй LIMIT 100, если не указано иное 3. Для топ-N используй ORDER BY ... DESC LIMIT N 4. Регионы: "Москва", "Московская область", "Санкт-Петербург"... 5. Возвращай только SQL без объяснений Примеры: Вопрос: Сколько всего записей? SQL: SELECT COUNT(*) as total FROM vehicles; Вопрос: Топ-5 компаний по количеству экскаваторов SQL: SELECT company_name, COUNT(*) as cnt FROM vehicles WHERE vehicle_type = 'экскаватор' GROUP BY company_name ORDER BY cnt DESC LIMIT 5; Вопрос: Бензовозы в Московской области SQL: SELECT * FROM vehicles WHERE vehicle_type = 'бензовоз' AND region = 'Московская область' LIMIT 100; """
Код генератора
import anthropic from typing import Optional import re class SQLGenerator: def __init__(self): self.client = anthropic.Anthropic() self.model = "claude-sonnet-4-20250514" def generate(self, question: str) -> Optional[str]: response = self.client.messages.create( model=self.model, max_tokens=500, system=SYSTEM_PROMPT, messages=[{"role": "user", "content": question}] ) sql = response.content[0].text.strip() # Валидация: только SELECT if not sql.upper().startswith("SELECT"): return None # Защита от инъекций: проверяем на опасные конструкции dangerous = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "TRUNCATE", ";--"] if any(d in sql.upper() for d in dangerous): return None return sql def execute_safe(self, sql: str, conn) -> list: """Выполнение с таймаутом и лимитами""" # Принудительно добавляем LIMIT, если его нет if "LIMIT" not in sql.upper(): sql = sql.rstrip(";") + " LIMIT 100;" with conn.cursor() as cur: cur.execute("SET statement_timeout = '5s';") cur.execute(sql) return cur.fetchall()
Проблема #1: галлюцинации в SQL
LLM иногда генерирует несуществующие колонки или таблицы. Решение — валидация перед выполнением:
def validate_sql(sql: str, schema: dict) -> tuple[bool, str]: """Проверяем, что все упомянутые колонки существуют""" # Извлекаем имена колонок из SQL # Упрощённая версия, в проде используем sqlparse words = re.findall(r'\b([a-z_]+)\b', sql.lower()) valid_columns = set(schema.get("vehicles", [])) valid_columns.update(["select", "from", "where", "and", "or", "count", "group", "by", "order", "desc", "asc", "limit", "as"]) unknown = [w for w in words if w not in valid_columns and not w.isdigit()] if unknown: return False, f"Неизвестные колонки: {unknown}" return True, ""
RAG для семантического поиска
Пользователь пишет "грузовики для нефти". В базе это "бензовоз" или "автоцистерна". SQL-генератор не справится — нужен семантический поиск.
Индексация
from qdrant_client import QdrantClient from qdrant_client.models import Distance, VectorParams, PointStruct import openai client = QdrantClient(host="localhost", port=6333) openai_client = openai.OpenAI() def create_vehicle_index(): # Создаём коллекцию client.create_collection( collection_name="vehicle_types", vectors_config=VectorParams(size=1536, distance=Distance.COSINE) ) # Типы техники с синонимами vehicle_types = [ {"type": "бензовоз", "synonyms": "автоцистерна, топливозаправщик, грузовик для нефти, цистерна для топлива"}, {"type": "автовоз", "synonyms": "автомобилевоз, транспортировщик машин, перевозчик авто"}, {"type": "экскаватор", "synonyms": "землеройная машина, копатель, ковшовый экскаватор"}, # ... ещё 50+ типов ] points = [] for i, vt in enumerate(vehicle_types): text = f"{vt['type']}: {vt['synonyms']}" embedding = get_embedding(text) points.append(PointStruct( id=i, vector=embedding, payload={"vehicle_type": vt["type"]} )) client.upsert(collection_name="vehicle_types", points=points) def get_embedding(text: str) -> list[float]: response = openai_client.embeddings.create( model="text-embedding-3-small", input=text ) return response.data[0].embedding
Поиск
def find_vehicle_type(query: str) -> str: """Находим тип техники по семантике запроса""" query_vector = get_embedding(query) results = client.search( collection_name="vehicle_types", query_vector=query_vector, limit=1 ) if results and results[0].score > 0.7: return results[0].payload["vehicle_type"] return None
Интеграция в пайплайн
class QueryProcessor: def __init__(self): self.sql_gen = SQLGenerator() self.rag = RAGSearch() def process(self, question: str) -> str: # 1. Пробуем найти типы техники через RAG enriched_question = self.enrich_with_rag(question) # 2. Генерируем SQL sql = self.sql_gen.generate(enriched_question) if not sql: return "Не удалось понять запрос. Попробуйте переформулировать." # 3. Валидируем valid, error = validate_sql(sql, SCHEMA) if not valid: return f"Ошибка в запросе: {error}" # 4. Выполняем results = self.sql_gen.execute_safe(sql, self.conn) # 5. Форматируем ответ return self.format_response(results, question) def enrich_with_rag(self, question: str) -> str: """Заменяем синонимы на точные названия""" # Ищем упоминания техники vehicle_type = self.rag.find_vehicle_type(question) if vehicle_type: # Добавляем контекст для LLM return f"{question}\n[Тип техники в базе: {vehicle_type}]" return question
Проблема #2: медленные запросы
600K записей — не big data, но без оптимизации запросы выполняются секунды.
Решение: предрасчитанные агрегаты
# Ежедневный пересчёт популярных агрегатов AGGREGATES = { "top_companies_by_type": """ SELECT vehicle_type, company_name, COUNT(*) as cnt FROM vehicles GROUP BY vehicle_type, company_name """, "vehicles_by_region": """ SELECT region, vehicle_type, COUNT(*) as cnt FROM vehicles GROUP BY region, vehicle_type """, "leasing_companies_stats": """ SELECT leasing_company, COUNT(*) as total, COUNT(DISTINCT company_name) as clients FROM vehicles GROUP BY leasing_company """ } def precompute_aggregates(): """Запускается по cron раз в сутки""" redis = Redis() for name, sql in AGGREGATES.items(): with conn.cursor() as cur: cur.execute(sql) results = cur.fetchall() redis.set(f"agg:{name}", json.dumps(results), ex=86400)
Router для быстрых запросов
def route_query(question: str) -> str: """Определяем, можно ли ответить из кэша""" # Паттерны для быстрых ответов patterns = { r"топ.+компани.+по количеству": "top_companies_by_type", r"сколько.+в.+регион": "vehicles_by_region", r"лизинговые компании|рейтинг лизинг": "leasing_companies_stats" } for pattern, agg_name in patterns.items(): if re.search(pattern, question.lower()): cached = redis.get(f"agg:{agg_name}") if cached: return filter_cached_results(json.loads(cached), question) # Если нет в кэше — генерируем SQL return None
Форматирование ответа
LLM генерирует человекочитаемый ответ из табличных данных:
def format_response(results: list, question: str, sql: str) -> str: if not results: return "По вашему запросу ничего не найдено." # Для коротких результатов — текстовый ответ if len(results) <= 10: prompt = f"""Вопрос пользователя: {question} Данные из базы (SQL: {sql}): {json.dumps(results, ensure_ascii=False, indent=2)} Сформулируй краткий ответ на русском языке. Если это топ — пронумеруй. Укажи ключевые цифры.""" response = anthropic.Anthropic().messages.create( model="claude-sonnet-4-20250514", max_tokens=1000, messages=[{"role": "user", "content": prompt}] ) return response.content[0].text # Для длинных — предлагаем экспорт return f"Найдено {len(results)} записей. Показать первые 10 или экспортировать в Excel?"
Результаты
До (BI-дашборд):
Время на ответ: 5-30 минут
Требуется обучение: да, существенное
Доступность: только аналитики
После (AI-чат):
Время на ответ: 3-15 секунд
Требуется обучение: нет
Доступность: любой сотрудник с доступом в Telegram
Метрики за месяц
1,200 запросов обработано
94% успешно отвечены без участия человека
6% потребовали уточнения или ручной обработки
Типичные запросы
"Сколько экскаваторов в лизинге у Сбербанк Лизинг?" → 3 сек, точный ответ из SQL "Компании с грузовиками для перевозки нефтепродуктов в Татарстане" → 5 сек, RAG + SQL "Динамика лизинга спецтехники за 2024-2025" → 2 сек, из кэша агрегатов
Грабли и выводы
1. LLM галлюцинирует названия колонок. Решение: строгая валидация SQL перед выполнением.
2. Пользователи не знают точных терминов. Решение: RAG для семантического обогащения запроса.
3. Медленные аналитические запросы. Решение: предрасчитанные агрегаты + pattern matching для роутинга.
4. Сложно отлаживать. Решение: логирование всего пайплайна (вопрос → обогащённый вопрос → SQL → результат).
5. Пользователи ожидают магии. Решение: честные ограничения в интерфейсе. "Я могу искать технику по типу, региону, компании. Для сложной аналитики обратитесь к аналитику."
Стек
Backend: Python 3.12, FastAPI
LLM: Claude API (Claude Sonnet 4 для SQL), OpenAI Embeddings (text-embedding-3-small)
Векторная БД: Qdrant 1.12
Основная БД: PostgreSQL 17
Кэш: Redis 7
Frontend: Telegram Bot API (aiogram 3.x)
Код отдельных компонентов доступен в статье. Полная архитектура — под NDA клиента.
Если интересна тема text-to-SQL и RAG для бизнес-данных — пишите в комментариях, какие аспекты раскрыть подробнее.
