Розробка AI-агента з доступом до бази даних

Розвиток AI-агента з доступом до бази даних

AI-агент з доступом до БД може самостійно запитувати дані, формувати SQL-запити на основі запитань природною мовою й виконувати операції запису з необхідними перевірками. Text-to-SQL — ключова технологія, яка дозволяє нетехнічним користувачам взаємодіяти з даними через діалог.

Text-to-SQL: Перетворення запитань на SQL

from langchain_openai import ChatOpenAI
from langchain_community.utilities import SQLDatabase
from langchain_community.agent_toolkits import SQLDatabaseToolkit
from langchain.agents import create_sql_agent

# Підключення до PostgreSQL
db = SQLDatabase.from_uri(
    "postgresql://user:password@localhost:5432/company_db",
    include_tables=["orders", "customers", "products", "inventory"],
    sample_rows_in_table_info=3,  # Включаємо приклади даних у схему
)

llm = ChatOpenAI(model="gpt-4o", temperature=0)
toolkit = SQLDatabaseToolkit(db=db, llm=llm)

# SQL агент з автоматичним виправленням помилок
agent = create_sql_agent(
    llm=llm,
    toolkit=toolkit,
    verbose=True,
    handle_parsing_errors=True,
    max_iterations=10,
)

# Приклади запитів
result = agent.invoke({"input": "Які топ-5 клієнтів за дохідом за останні 3 місяці?"})
result = agent.invoke({"input": "Покажи товари зі залишком на складі менше 10 одиниць"})

Безпека: тільки READ-доступ

from langchain_community.utilities import SQLDatabase
from sqlalchemy import create_engine, text

# Read-only користувач PostgreSQL
READ_ONLY_USER_URI = "postgresql://readonly_user:pass@localhost:5432/db"

# Додаткова валідація: заборона DML-операцій
def validate_sql_query(query: str) -> bool:
    """Перевіряє, що запит є тільки SELECT"""
    forbidden_keywords = ["INSERT", "UPDATE", "DELETE", "DROP", "CREATE", "ALTER", "TRUNCATE"]
    query_upper = query.upper()
    for keyword in forbidden_keywords:
        if keyword in query_upper:
            return False
    return True

class SafeSQLTool:
    def __init__(self, db_uri: str):
        self.engine = create_engine(db_uri)

    def execute_query(self, query: str) -> str:
        if not validate_sql_query(query):
            return "ERROR: Only SELECT queries are allowed"

        # Обмеження кількості рядків
        if "LIMIT" not in query.upper():
            query = f"{query.rstrip(';')} LIMIT 100"

        with self.engine.connect() as conn:
            result = conn.execute(text(query))
            rows = result.fetchall()
            columns = result.keys()
            return str([dict(zip(columns, row)) for row in rows])

Схема бази з контекстом для LLM

Якість Text-to-SQL критично залежить від якості описання схеми:

SCHEMA_CONTEXT = """
Таблиці бази даних:

1. orders (замовлення)
   - id: PK, INTEGER
   - customer_id: FK → customers.id
   - status: VARCHAR (pending, confirmed, shipped, delivered, cancelled)
   - total_amount: DECIMAL(12,2) — сума замовлення
   - created_at: TIMESTAMP
   - shipped_at: TIMESTAMP (NULL якщо не відправлено)

2. customers (клієнти)
   - id: PK
   - name: VARCHAR — назва компанії або ПІБ
   - inn: VARCHAR(12) — реєстраційний номер
   - segment: VARCHAR (enterprise, mid, small) — сегмент клієнта
   - manager_id: FK → employees.id — менеджер рахунку

3. products (товари)
   - sku: VARCHAR — артикул
   - name: VARCHAR
   - category: VARCHAR
   - price_rub: DECIMAL
   - cost_rub: DECIMAL — собівартість

ВАЖЛИВО: Статуси замовлення: 'delivered' = успішно виконано. 'cancelled' = скасовано.
         Дохід = сума total_amount замовлень зі статусом 'delivered'.
"""

system_prompt = f"""Ти — аналітик даних. Переводи запитання на SQL-запити.
Використовуй наступну схему бази даних:

{SCHEMA_CONTEXT}

Правила:
- Тільки SELECT запити
- Завжди додавай LIMIT (не більше 1000)
- Використовуй прозорі alias для колон
- Додавай ORDER BY при агрегуванні"""

Практичний кейс: BI-агент для e-commerce

Завдання: аналітичний помічник для комерційного директора — аналіз продажів, ABC-аналіз асортименту, воронка замовлень, cohort retention.

БД: PostgreSQL, 15 таблиць, 3M замовлень.

Приклади діалогів:

Користувач: «Яка конверсія з pending у delivered за останні 30 днів по сегментах клієнтів?»

Агент генерує:

SELECT
    c.segment AS сегмент,
    COUNT(*) FILTER (WHERE o.status = 'pending') AS очікуючих,
    COUNT(*) FILTER (WHERE o.status = 'delivered') AS доставлено,
    ROUND(
        COUNT(*) FILTER (WHERE o.status = 'delivered')::decimal /
        NULLIF(COUNT(*), 0) * 100, 1
    ) AS конверсія_pct
FROM orders o
JOIN customers c ON o.customer_id = c.id
WHERE o.created_at >= NOW() - INTERVAL '30 days'
GROUP BY c.segment
ORDER BY конверсія_pct DESC
LIMIT 100;

Результати:

• Час отримання аналітики: 2 дні → 3 хвилини
• Точність SQL (запитання → коректний SQL): 87%
• Типові помилки: неправильні JOIN при складних запитах (вирішується через few-shot приклади у промпті)

Few-shot приклади для підвищення точності

FEW_SHOT_EXAMPLES = """
Приклади коректних запитів:

Запитання: Топ-10 товарів за маржею за останній квартал
SQL:
SELECT p.name, p.sku,
       SUM(oi.quantity * (p.price_rub - p.cost_rub)) AS маржа_руб
FROM order_items oi
JOIN products p ON oi.sku = p.sku
JOIN orders o ON oi.order_id = o.id
WHERE o.status = 'delivered'
  AND o.created_at >= DATE_TRUNC('quarter', CURRENT_DATE) - INTERVAL '3 months'
GROUP BY p.name, p.sku
ORDER BY маржа_руб DESC
LIMIT 10;

Запитання: Середній чек по місяцях за 2025 рік
SQL:
SELECT DATE_TRUNC('month', created_at) AS місяць,
       ROUND(AVG(total_amount), 0) AS середній_чек,
       COUNT(*) AS кіл_замовлень
FROM orders
WHERE status = 'delivered'
  AND created_at BETWEEN '2025-01-01' AND '2025-12-31'
GROUP BY 1
ORDER BY 1;
"""

Графік

• Розвиток Text-to-SQL агента: 2–3 тижні
• Налаштування схеми та few-shot прикладів: 1 тиждень
• Тестування на реальних запитах: 1–2 тижні
• Всього: 4–6 тижнів