Як побудувати multi-agent AI pipeline для роботи з текстом та SQL запитами

Більшість команд витрачають години на ручну обробку даних: аналітик пише SQL, передає результати колезі, той формує звіт — і так по колу. Multi-agent pipeline автоматизує весь цей ланцюжок: один агент розуміє запит природною мовою, інший генерує SQL, третій форматує відповідь. Цей туторіал покаже, як зібрати таку систему з нуля за 2–3 години, використовуючи LangGraph та OpenAI GPT-4o. Для старту потрібен базовий досвід з Python та доступ до OpenAI API або сумісного провайдера.

🛠️ Що знадобиться

• Python 3.11+ — основне середовище виконання; рекомендую встановити через pyenv для чистого управління версіями
• LangGraph 0.2+ — фреймворк для побудови графів агентів; безкоштовний, open-source, встановлюється через pip
• OpenAI API (GPT-4o) — мозок агентів; платний, але для тестування вистачить $5 кредитів, достатньо ~200 запитів
• SQLite — вбудована база даних Python, жодних додаткових інсталяцій не потрібно; для продакшну можна замінити на PostgreSQL
• LangSmith — безкоштовний інструмент для трасування та дебагу агентів; реєструйся на smith.langchain.com
• VS Code або PyCharm — IDE для зручної роботи з кодом; обидва безкоштовні в базовій версії

📋 Покрокова інструкція

Крок 1: Налаштування середовища та встановлення залежностей

Відкрий термінал і створи новий проєкт: виконай mkdir multi-agent-pipeline && cd multi-agent-pipeline, потім python -m venv venv та активуй середовище командою source venv/bin/activate (на Windows — venv\Scripts\activate). Створи файл requirements.txt з таким вмістом: langgraph==0.2.55, langchain-openai==0.2.14, langchain-community==0.3.14, python-dotenv==1.0.1. Виконай pip install -r requirements.txt та створи файл .env, куди додай OPENAI_API_KEY=sk-твій-ключ і LANGCHAIN_API_KEY=ls-твій-ключ — ці змінні підхоплюються автоматично. Перевір установку командою python -c "import langgraph; print(langgraph.__version__)" — має вивести версію без помилок.

Крок 2: Створення тестової бази даних та схеми

Створи файл setup_db.py — він ініціалізує SQLite базу з тестовими даними. Вставляй код блоками: спочатку імпорт import sqlite3, потім підключення conn = sqlite3.connect("company.db"). Створи три таблиці: employees (id, name, department, salary, hire_date), projects (id, name, budget, status, lead_id) та tasks (id, project_id, assignee_id, title, status, due_date). Наповни кожну таблицю 10–15 тестовими рядками через conn.executemany() — це важливо, бо агент буде вчитися на реальній структурі. Виконай python setup_db.py і перевір результат: sqlite3 company.db ".tables" має показати всі три таблиці. Підводний камінь: не використовуй кирилицю в назвах колонок — деякі версії SQLite некоректно обробляють unicode в схемах.

Крок 3: Визначення агентів та їх ролей у графі

Створи файл agents.py — тут живе логіка кожного з чотирьох агентів. Перший агент — Router: приймає вхідний запит і вирішує, чи це питання до бази даних, чи задача на текстовий аналіз. Другий — Schema Inspector: підключається до SQLite, читає схему через SELECT name, sql FROM sqlite_master WHERE type="table" і передає її далі. Третій — SQL Generator: отримує схему плюс запит користувача і генерує валідний SQL через structured output GPT-4o з Pydantic-моделлю class SQLQuery(BaseModel): query: str; explanation: str. Четвертий — Response Formatter: бере сирі дані з бази та формує читабельну відповідь природною мовою. Кожен агент — це окрема функція, яка приймає і повертає словник стану AgentState.

Крок 4: Збірка графу в LangGraph та підключення умовних переходів

Створи файл pipeline.py та імпортуй from langgraph.graph import StateGraph, END. Визнач стан через TypedDict: class AgentState(TypedDict): user_input: str; schema: str; sql_query: str; raw_results: list; final_answer: str; route: str. Далі будуй граф: workflow = StateGraph(AgentState), потім додавай вузли командою workflow.add_node("router", router_agent) для кожного з чотирьох агентів. Додай умовні переходи через workflow.add_conditional_edges("router", lambda state: state["route"], {"sql": "schema_inspector", "text": "formatter", "end": END}) — це і є серце multi-agent логіки. Встанови точку входу workflow.set_entry_point("router") та скомпілюй граф app = workflow.compile(). Виклич app.get_graph().draw_mermaid_png(output_file_path="graph.png") — отримаєш візуалізацію графу, дуже корисно для дебагу.

Крок 5: Запуск pipeline та тестування наскрізного сценарію

Створи файл main.py з таким кодом: імпортуй app з pipeline.py, визнач тестові запити — наприклад, "Покажи топ-5 співробітників за зарплатою" та "Які проєкти зараз в статусі active?". Запускай через result = app.invoke({"user_input": query}) і виводь result["final_answer"]. Виконай python main.py — за 5–10 секунд маєш отримати читабельну відповідь з реальними даними з бази. Фінальний результат: система самостійно зрозуміла запит → знайшла схему → згенерувала SELECT e.name, e.salary FROM employees e ORDER BY salary DESC LIMIT 5 → виконала запит → відформатувала відповідь як список з іменами та зарплатами. Якщо бачиш відповідь у LangSmith Dashboard — граф агентів працює повністю.

⚠️ Типові помилки та як їх уникнути

• SQL ін’єкція через agent output — ніколи не виконуй згенерований SQL напряму без валідації; додай whitelist дозволених операцій: перевіряй що запит починається з SELECT через if not query.strip().upper().startswith("SELECT"): raise ValueError
• Нескінченні цикли між агентами — LangGraph за замовчуванням дозволяє рекурсію; явно встанови ліміт workflow.compile(recursion_limit=10), інакше при помилці агент може зациклитись і спалити весь API бюджет
• Втрата стану між викликами — TypedDict стан не персистентний між сесіями; якщо потрібна пам’ять між запитами — підключай MemorySaver через from langgraph.checkpoint.memory import MemorySaver та передавай thread_id в кожен виклик
• Галюцинації у SQL — GPT-4o іноді вигадує неіснуючі колонки; завжди передавай повну схему в промпт і додавай інструкцію “використовуй ТІЛЬКИ колонки зі схеми, не вигадуй нові”

💡 Поради для кращого результату

По-перше, використовуй structured outputs замість звичайного текстового парсингу — замість response.content підключи Pydantic через llm.with_structured_output(SQLQuery): це гарантує що агент поверне валідний об’єкт, а не “ось твій SQL: sql SELECT…”. По-друге, додай few-shot приклади в системний промпт SQL агента — 3–4 пари “запит → SQL” різко підвищують точність на складних джойнах. По-третє, кешуй результат Schema Inspector в стані між запитами одної сесії — схема не змінюється щохвилини, а зайвий виклик коштує токени та час. По-четверте, підключи LangSmith tracing з першого дня: встанови LANGCHAIN_TRACING_V2=true в .env і отримуй повний лог кожного кроку агента — це в рази скорочує час дебагу порівняно з print-statements.

❓ Часті запитання (FAQ)

1. Чи можна замінити OpenAI на локальну модель?
Так, LangGraph не прив’язаний до конкретного провайдера. Замінюй ChatOpenAI на ChatOllama з моделлю llama3.1:70b або qwen2.5-coder:32b — остання особливо добре справляється з генерацією SQL. Якість буде дещо нижча ніж у GPT-4o, але для простих запитів цілком прийнятна.

2. Як масштабувати pipeline на велику базу з сотнями таблиць?
Не передавай всю схему в промпт — це перевищить контекстне вікно. Замість цього зроби агент-пошук: спочатку векторний пошук по назвах таблиць через ChromaDB або pgvector, потім передавай лише релевантні 3–5 таблиць. Це так звана техніка “schema linking”.

3. Як додати авторизацію — щоб різні юзери бачили різні дані?
Додай в AgentState поле user_role: str і в промпт SQL агента додавай умову: наприклад, для ролі “manager” дозволяй запити тільки по його відділу. Більш надійний варіант — додавати WHERE clause на рівні виконання запиту, не генерації.

4. Скільки коштує запуск такого pipeline в продакшні?
Типовий запит проходить через 3–4 виклики LLM по ~1000–2000 токенів кожен. З GPT-4o mini (~$0.15 за 1M вхідних токенів) один наскрізний запит коштує приблизно $0.001–0.003. При 1000 запитів на день — це $1–3 на добу, цілком розумно для бізнес-інструменту.

5. Чи підходить цей підхід для real-time застосунків?
Затримка одного запиту — 3–8 секунд залежно від складності. Для аналітичних дашбордів та чат-інтерфейсів це норма. Для real-time систем з вимогою відповіді до 500мс — ні, тут потрібен інший підхід: попередньо згенеровані запити або кеш відповідей.

🏁 Підсумок

Ти побудував повноцінний multi-agent pipeline, де чотири спеціалізовані агенти спільно перетворюють звичайне питання на SQL запит, виконують його та повертають зрозумілу відповідь — без жодного ручного кроку. Архітектура на LangGraph дає тобі гнучкість: додавай нових агентів, міняй моделі, підключай будь-яку базу даних.

Прямо зараз відкрий термінал і виконай перший крок — створи папку проєкту та файл requirements.txt. Перші 30 хвилин налаштування дадуть тобі робочий скелет, а далі — лише додавати функціональність під свої задачі.