Как создать Text-to-SQL агента для бизнеса: архитектура, guardrails и достижение 96% точности с Llama 3

Проблема: Бизнес-данные в тюрьме SQL

В каждом бизнесе есть золотая жила — базы данных с транзакциями, клиентами, логистикой. Но доступ к этой информации имеют только технические специалисты. Менеджеры, аналитики, руководители отделов вынуждены:

• Ждать дни или недели на выполнение SQL-запросов
• Формулировать потребности через посредников («мне нужно... как там...»)
• Получать не совсем то, что нужно, из-за потери контекста
• Терять возможность оперативного принятия решений

Решение: Архитектура с тройной защитой

Мы построили Text-to-SQL агента, который достигает 96% точности на реальных бизнес-запросах. Секрет не в магии, а в правильной архитектуре:

1. Strict Schema Binding — агент знает ТОЛЬКО вашу схему БД
2. Guardrails Layer — «вышибала» (Bouncer), который проверяет запросы до и после
3. Iterative Refinement — многоэтапная обработка с обратной связью

В статье «Production-ready AI-агенты: как превратить хайп в работающую систему для бизнеса» я подробно объяснял, почему архитектура важнее модели. Здесь мы применяем те же принципы.

Архитектура системы

1 Подготовка схемы базы данных

Первое и самое важное — строгая привязка к схеме. Мы не даём агенту «фантазировать» о несуществующих таблицах и полях.

import json
from sqlalchemy import create_engine, inspect

def extract_schema(db_uri):
    """Извлекаем полную схему БД в структурированном формате"""
    engine = create_engine(db_uri)
    inspector = inspect(engine)
    
    schema = {
        "tables": {},
        "relationships": [],
        "enums": {}
    }
    
    # Получаем все таблицы
    for table_name in inspector.get_table_names():
        columns = []
        for column in inspector.get_columns(table_name):
            col_info = {
                "name": column["name"],
                "type": str(column["type"]),
                "nullable": column["nullable"],
                "default": column.get("default")
            }
            columns.append(col_info)
        
        # Получаем первичные ключи
        pk_constraint = inspector.get_pk_constraint(table_name)
        primary_keys = pk_constraint.get("constrained_columns", [])
        
        # Получаем foreign keys
        foreign_keys = []
        for fk in inspector.get_foreign_keys(table_name):
            foreign_keys.append({
                "constrained_columns": fk["constrained_columns"],
                "referred_table": fk["referred_table"],
                "referred_columns": fk["referred_columns"]
            })
        
        schema["tables"][table_name] = {
            "columns": columns,
            "primary_keys": primary_keys,
            "foreign_keys": foreign_keys,
            "description": get_table_description(table_name)  # из метаданных
        }
    
    return json.dumps(schema, indent=2, ensure_ascii=False)

2 Создание агента с LangChain

Используем LangChain для создания агента с инструментами. В статье «Production-ready AI-агент с нуля: ReAct, Advanced RAG и работа с инструментами» я подробно разбирал основы.

from langchain.agents import AgentExecutor, create_react_agent
from langchain_groq import ChatGroq
from langchain.prompts import PromptTemplate
from langchain.tools import Tool
from langchain.memory import ConversationBufferMemory

class TextToSQLAgent:
    def __init__(self, db_schema, llm_api_key):
        self.schema = db_schema
        self.llm = ChatGroq(
            temperature=0.1,  # НИЗКАЯ температура для детерминированности
            model_name="llama-3-70b-8192",
            groq_api_key=llm_api_key
        )
        
        # Создаём инструменты
        self.tools = self._create_tools()
        
        # Промпт с жёсткими инструкциями
        self.prompt = self._create_prompt()
        
        # Создаём агента
        self.agent = create_react_agent(
            llm=self.llm,
            tools=self.tools,
            prompt=self.prompt
        )
        
        self.agent_executor = AgentExecutor(
            agent=self.agent,
            tools=self.tools,
            memory=ConversationBufferMemory(
                memory_key="chat_history",
                return_messages=True
            ),
            verbose=True,
            handle_parsing_errors=True,
            max_iterations=3  # Защита от бесконечных циклов
        )
    
    def _create_prompt(self):
        """Создаём промпт со строгими правилами"""
        return PromptTemplate.from_template(
            """Ты — экспертный SQL-ассистент для бизнес-аналитики.

ДОСТУПНАЯ СХЕМА БАЗЫ ДАННЫХ:
{db_schema}

ПРАВИЛА, КОТОРЫЕ НЕЛЬЗЯ НАРУШАТЬ:
1. Используй ТОЛЬКО таблицы и колонки из схемы выше
2. Никогда не создавай новые таблицы или колонки
3. Все JOIN должны использовать существующие foreign keys
4. Для агрегаций используй только SUM, COUNT, AVG, MIN, MAX
5. Всегда добавляй LIMIT 100 если не указано иное
6. Никогда не выполняй DELETE, UPDATE, DROP, TRUNCATE
7. Если запрос слишком сложный — разбей его на подзапросы

ИСТОРИЯ ДИАЛОГА:
{chat_history}

ВОПРОС ПОЛЬЗОВАТЕЛЯ: {input}

Твой ответ должен содержать ТОЛЬКО валидный SQL-запрос. Никаких объяснений.
SQL: """
        )

3 Реализация Guardrails Layer («Bouncer»)

Это самая важная часть системы. «Вышибала» проверяет каждый запрос на:

• SQL-инъекции и вредоносные паттерны
• Соответствие схеме БД
• Сложность запроса (защита от runaway queries)
• Наличие LIMIT для больших выборок

class SQLBouncer:
    """Guardrails слой для проверки SQL запросов"""
    
    MALICIOUS_PATTERNS = [
        r"(?i)DELETE\\s+FROM",
        r"(?i)DROP\\s+TABLE",
        r"(?i)UPDATE\\s+.+\\s+SET",
        r"(?i)INSERT\\s+INTO",
        r"(?i)TRUNCATE",
        r"(?i);\\s*--",  # SQL injection attempt
        r"(?i)UNION\\s+SELECT",
        r"(?i)OR\\s+'1'='1'",
        r"(?i)EXEC\\s*(",
        r"(?i)XP_CMDSHELL",
    ]
    
    def __init__(self, allowed_tables, max_query_cost=1000):
        self.allowed_tables = set(allowed_tables)
        self.max_query_cost = max_query_cost
    
    def check_query(self, sql_query, user_context):
        """Проверяем запрос на безопасность и валидность"""
        
        # 1. Проверка на malicious patterns
        for pattern in self.MALICIOUS_PATTERNS:
            if re.search(pattern, sql_query):
                return {
                    "allowed": False,
                    "reason": f"Обнаружен вредоносный паттерн: {pattern}",
                    "suggestion": "Переформулируйте запрос"
                }
        
        # 2. Парсим SQL для анализа структуры
        try:
            parsed = sqlparse.parse(sql_query)[0]
        except:
            return {
                "allowed": False,
                "reason": "Невозможно распарсить SQL запрос",
                "suggestion": "Проверьте синтаксис"
            }
        
        # 3. Извлекаем используемые таблицы
        used_tables = self._extract_tables(parsed)
        
        # 4. Проверяем, что все таблицы разрешены
        unauthorized_tables = used_tables - self.allowed_tables
        if unauthorized_tables:
            return {
                "allowed": False,
                "reason": f"Использование неразрешенных таблиц: {unauthorized_tables}",
                "suggestion": f"Используйте только: {self.allowed_tables}"
            }
        
        # 5. Проверяем сложность запроса
        complexity_score = self._calculate_complexity(parsed)
        if complexity_score > self.max_query_cost:
            return {
                "allowed": False,
                "reason": f"Запрос слишком сложный (сложность: {complexity_score})",
                "suggestion": "Разбейте запрос на несколько простых"
            }
        
        # 6. Проверяем наличие LIMIT для SELECT без агрегаций
        if self._needs_limit(parsed):
            if not self._has_limit(parsed):
                return {
                    "allowed": True,  # Разрешаем, но добавляем LIMIT
                    "reason": "Добавлен LIMIT 100 для защиты производительности",
                    "modified_query": sql_query.rstrip(";") + " LIMIT 100;"
                }
        
        return {"allowed": True, "reason": "Запрос безопасен"}
    
    def _extract_tables(self, parsed_sql):
        """Извлекаем имена таблиц из SQL"""
        tables = set()
        for token in parsed_sql.tokens:
            if isinstance(token, sqlparse.sql.Identifier):
                tables.add(str(token).split()[0])
            elif token.is_group:
                tables.update(self._extract_tables(token))
        return tables

Промпт-инжиниринг для достижения 96% точности

Секрет высокой точности — не в сложных нейросетях, а в правильном промптинге. Мы используем несколько техник:

В статье «Agent Skills: как упаковать знания для LLM-агентов (не просто промпты)» я подробно разбираю эти техники.

Пример улучшенного промпта с Chain-of-Thought:

IMPROVED_PROMPT = """Ты — SQL эксперт. Сначала подумай, потом сгенерируй SQL.

СХЕМА:
{table_info}

ПРИМЕРЫ:
1. Вопрос: "Сколько заказов в прошлом месяце?"
   Мысли: Нужна таблица orders, поле created_at, фильтр по месяцу, COUNT
   SQL: SELECT COUNT(*) FROM orders WHERE created_at >= '2024-04-01' AND created_at < '2024-05-01';

2. Вопрос: "Топ-10 клиентов по сумме покупок"
   Мысли: Нужны таблицы orders и customers, JOIN по customer_id, GROUP BY, SUM(total), ORDER BY, LIMIT
   SQL: SELECT c.customer_name, SUM(o.total_amount) as total_spent FROM orders o JOIN customers c ON o.customer_id = c.id GROUP BY c.id, c.customer_name ORDER BY total_spent DESC LIMIT 10;

ТЕКУЩИЙ ВОПРОС: {question}

Сначала напиши "Мысли:" с пошаговым анализом, затем "SQL:" с запросом.
"""

Интеграция в бизнес-процессы

Text-to-SQL агент — не изолированная система. Его нужно интегрировать с:

• BI-панелями (Tableau, Power BI) через API
• Корпоративными чатами (Slack, Teams)
• Системами тикетов для отслеживания запросов
• Мониторингом производительности и использования

Как правильно организовать workflow, я рассказывал в статье «Как настроить агентный workflow: пример от мирового лидера целлюлозы Suzano».

Возможные ошибки и как их избежать

FAQ: Ответы на частые вопросы

Q: Почему именно Llama 3, а не GPT-4?

A: Llama-3-70B через Groq даёт сопоставимое качество с GPT-4 для SQL-генерации, но в 10 раз дешевле и без ограничений по rate limits. Для бизнес-приложений это критично.

Q: Как измеряли 96% точности?

A: На датасете из 500 реальных бизнес-вопросов от сотрудников. Критерии: SQL выполняется без ошибок + возвращает корректные данные + оптимален по производительности.

Q: Что делать с очень сложными запросами?

A: Используйте подход с суб-агентами, как описано в «Как правильно использовать суб-агентов в AI-разработке». Один агент разбивает вопрос, другой генерирует подзапросы.

Q: Как защититься от новых типов SQL-инъекций?

A: Регулярно обновляйте список MALICIOUS_PATTERNS на основе логов. Добавьте ML-классификатор для обнаружения аномалий.

Заключение

Text-to-SQL агент с guardrails — это не научная фантастика, а рабочий инструмент, который уже сегодня можно внедрить в бизнес. Ключевые элементы успеха:

1. Architecture-first подход: сначала guardrails, потом модель
2. Strict schema binding: агент знает только то, что ему разрешено
3. Iterative refinement: многоэтапная проверка и коррекция
4. Business context: понимание домена через few-shot examples

С такой архитектурой вы не просто даёте доступ к данным — вы создаёте безопасную, контролируемую и эффективную систему бизнес-аналитики. 96% точности — это не предел, а отправная точка для дальнейшей оптимизации.

Для тех, кто хочет глубже погрузиться в архитектуру AI-агентов, рекомендую статью «Как спроектировать современного AI-агента: от planner/executor до stateful memory».