Як побудувати модульну систему агентів для LLM з динамічною маршрутизацією інструментів - Aitematic

Більшість LLM-агентів ламаються, щойно задача виходить за межі одного інструменту — модель або ігнорує доступні функції, або викликає їх у хибному порядку. Цей туторіал вирішує саме цю проблему: ми побудуємо модульну систему, де кожен агент відповідає за вузьку зону відповідальності, а центральний роутер динамічно вирішує, кого залучити до виконання завдання. На реалізацію піде приблизно 2–3 години, якщо вже маєш базовий досвід із Python і розумієш, що таке function calling у OpenAI або Anthropic API.

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

Python 3.11+ — основна мова реалізації; версія важлива через покращену підтримку asyncio та типізації
OpenAI API ключ (або Anthropic Claude API) — платний, але достатньо $5 на тести; використовуємо GPT-4o або Claude 3.5 Sonnet як мозок роутера
LangGraph 0.2+ — безкоштовна бібліотека для побудови графів агентів зі станом; встановлюється через pip
Pydantic v2 — для валідації схем інструментів і повідомлень між агентами; безкоштовно
Redis 7+ (локально або Redis Cloud free tier) — зберігання стану між викликами агентів; free tier достатньо для розробки
VS Code або PyCharm — зручний редактор із підтримкою дебагінгу async-коду

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

Крок 1: Ініціалізація проєкту та встановлення залежностей

Відкрий термінал і виконай команди по черзі: mkdir agent-router && cd agent-router, потім python -m venv .venv && source .venv/bin/activate (на Windows: .venv\Scripts\activate). Встанови залежності однією командою: pip install langgraph==0.2.* openai anthropic pydantic redis python-dotenv httpx. Створи файл .env у корені проєкту і додай до нього рядок OPENAI_API_KEY=sk-твій_ключ_тут — жодного разу не коміть цей файл у Git, одразу додай його до .gitignore. Перевір встановлення командою python -c "import langgraph; print(langgraph.__version__)" — має вивести версію без помилок.

Крок 2: Визначення схеми інструментів і базового агента

Створи файл agents/base.py — це фундамент усієї архітектури. Кожен агент у нашій системі — це клас, що успадковує BaseAgent і реалізує метод execute(state: AgentState) -> AgentState. Визнач Pydantic-модель стану прямо у цьому файлі:

from pydantic import BaseModel
from typing import Any, Optional

class ToolCall(BaseModel):
    tool_name: str
    arguments: dict[str, Any]
    result: Optional[Any] = None

class AgentState(BaseModel):
    user_input: str
    intent: Optional[str] = None
    tool_calls: list[ToolCall] = []
    final_answer: Optional[str] = None
    error: Optional[str] = None

Підводний камінь тут — не використовуй dict як тип стану у LangGraph напряму, бо втратиш валідацію. Завжди передавай Pydantic-модель і конвертуй через .model_dump() лише при передачі у граф.

Крок 3: Реалізація спеціалізованих агентів-виконавців

Тепер створи окремі файли для кожного агента у папці agents/. Почни з трьох: search_agent.py (пошук інформації через httpx + DuckDuckGo Instant Answer API), code_agent.py (генерація та виконання коду у пісочниці) і data_agent.py (обробка CSV/JSON даних). Кожен агент реєструє свої інструменти через декоратор — наприклад, у search_agent.py напиши функцію з анотаціями типів і декоратором @tool(description="Шукає актуальну інформацію в інтернеті") з LangGraph. Критично важливо: у полі description пиши максимально точно — саме цей текст роутер використовує для вибору агента. Уникай розмитих описів на кшталт “робить різні речі”.

Крок 4: Побудова динамічного роутера на базі LLM

Це серце всієї системи. Створи файл router/llm_router.py. Роутер отримує стан із намірами користувача і список зареєстрованих агентів із їхніми описами інструментів, а потім через function calling просить LLM вирішити, якого агента викликати і з якими аргументами. Реалізуй метод route(state: AgentState) -> str, який повертає ім’я наступного агента. У системному промпті чітко вкажи: “Ти роутер завдань. Обери ОДНОГО агента зі списку. Якщо завдання потребує кількох — обери першого в ланцюжку і встанови next_agent у стані.” Зареєструй граф у graph.py: виклич StateGraph(AgentState), додай вузли через graph.add_node("router", router.route) і graph.add_node("search", search_agent.execute), а ребра визнач через graph.add_conditional_edges("router", lambda s: s.intent, {"search": "search", "code": "code"}). Скомпілюй граф через app = graph.compile(checkpointer=RedisCheckpointer(redis_client)) — це активує збереження стану між запитами.

Крок 5: Тестування системи та підключення CLI-інтерфейсу

Створи файл main.py із простим циклом: зчитуй введення користувача, передавай у граф через result = await app.ainvoke({"user_input": query}, config={"thread_id": "session-1"}), виводь result["final_answer"]. Запусти Redis локально командою docker run -d -p 6379:6379 redis:7, потім стартуй систему: python main.py. Введи тестовий запит: “Знайди останні новини про квантові комп’ютери і склади короткий звіт у JSON”. Система має послідовно викликати search_agent для збору даних і data_agent для форматування — у консолі побачиш лог викликів інструментів і фінальну відповідь у структурованому форматі. Якщо все пройшло без помилок, ти маєш робочу модульну систему агентів із динамічним роутингом.

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

Роутер постійно викликає одного й того самого агента — перевір description у твоїх інструментах: вони мають чітко розмежовуватися семантично. Додай у системний промпт роутера приклади правильного розподілу завдань у форматі few-shot.
Стан агента не зберігається між викликами — майже завжди причина у тому, що Redis не запущений або thread_id різний між запитами. Зафікси thread_id для однієї сесії і перевір з’єднання командою redis-cli ping — має відповісти PONG.
Нескінченний цикл між агентами — встанови жорсткий ліміт ітерацій при компіляції графа: graph.compile(recursion_limit=10). Без цього два агенти можуть передавати стан один одному до вичерпання бюджету API.
Pydantic ValidationError при передачі стану — виникає, коли агент повертає поля, яких немає у AgentState. Розшир модель або використовуй model_config = ConfigDict(extra="allow") на час розробки, але перед продом приберись у схемі.

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

По-перше, логуй кожен виклик роутера у окремий файл із timestamp і вибраним агентом — через 50 тестових запитів одразу побачиш патерни помилкової маршрутизації і зможеш точечно підправити промпти. По-друге, для code_agent обов’язково використовуй пісочницю — запускай згенерований код через subprocess.run з таймаутом 5 секунд і без доступу до мережі (env={}), інакше ризикуєш виконати шкідливий код. По-третє, кешуй результати однакових tool_calls у Redis зі TTL 300 секунд — якщо роутер двічі поспіль хоче шукати одне й те саме, поверни кешований результат і зекономиш $0.01–0.05 на виклику. По-четверте, додай middleware для підрахунку токенів між кожним вузлом графа — це дозволить точно відстежувати вартість складних багатоетапних запитів і встановити hard limit перед деплоєм.

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

1. Чи можна використовувати локальні моделі замість OpenAI для роутера?
Так, підійде Ollama з моделлю qwen2.5:14b або llama3.3:70b — вони непогано справляються з function calling. Замість OpenAI client передай base_url="http://localhost:11434/v1", але очікуй на 20–40% гіршу точність маршрутизації порівняно з GPT-4o на складних мультиагентних сценаріях.

2. Як масштабувати систему на десятки агентів?
При більш ніж 10 агентах пряме перерахування всіх у промпті роутера стає неефективним. Перейди до ієрархічної маршрутизації: зроби мета-роутер, який спочатку визначає категорію (пошук, обчислення, комунікація), а потім вже спеціалізований роутер обирає конкретного агента всередині категорії.

3. Як тестувати агентів ізольовано, без виклику LLM?
Використовуй mock через unittest.mock.patch("openai.AsyncOpenAI") і підготуй фіксовані відповіді у форматі JSON для кожного тест-кейсу. LangGraph підтримує передачу стану напряму у вузол через agent.execute(mock_state) — тестуй логіку кожного агента окремо від роутера.

4. Що робити, якщо роутер не може визначити потрібний агент?
Додай fallback-вузол у граф — окремий агент clarification_agent, який запитує у користувача уточнення. У conditional_edges додай гілку "unknown": "clarification" і роутер автоматично туди потрапить, якщо впевненість у виборі нижче порогу (перевіряй через logprobs у відповіді OpenAI).

5. Чи підтримує LangGraph паралельне виконання кількох агентів?
Так, з версії 0.2 доступні паралельні гілки через graph.add_node("parallel_branch", RunnableParallel(...)). Це корисно, коли завдання можна розбити на незалежні підзадачі — наприклад, одночасно шукати інформацію і генерувати шаблон відповіді. Стани потім об’єднуються у merge-вузлі.

🏁 Підсумок

Ти побудував повноцінну модульну систему агентів із LLM-роутером, валідованим станом через Pydantic, персистентністю у Redis і чистим розподілом відповідальності між агентами — ця архітектура витримає реальне навантаження і легко розширюється новими агентами без переписування роутера.

Прямо зараз відкрий термінал, виконай перші три команди з Кроку 1 і напиши базовий AgentState — це займе 10 хвилин і дасть робочий скелет, на який ти наростиш решту системи за вихідні.