Створіть мультиагентний роутер за допомогою LangGraph та агента-керівника
Як розділити одного перевантаженого агента на спеціалістів із керівником у LangGraph, використовуючи готовий допоміжний інструмент та ручно створений StateGraph, а також принципи проектування станів та поширені помилки під час використання циклів.
Джерела: Build a Multi-Agent Router with LangGraph in 30 Minutes
Якщо поставити одного агента перед справжнім продуктом, він швидко починає робити занадто багато. Якщо попросити одного агента обробляти підтримку клієнтів, відстеження замовлень та повернення грошей, він виявиться посереднім у всіх трьох сферах. Його інструкції перетворюються на 2 000-слівний лабіринт формулювань на кшталт „якщо користувач запитує про X, робити Y, але якщо це також стосується Z, тоді...“. Рішенням не є створення розумнішого агента — це більше менших агентів разом із одним керівником, який координує їхню роботу. Це, мабуть, найпоширеніший патерн використання в агентському ШІ наразі, і його функціональна версія поміщається у короткій програмі LangGraph.
Проблеми з одним величезним агентом
Уявімо, що ви створюєте асистента, який відповідає на загальні запитання, виконує арифметичні операції та шукає актуальну інформацію в Інтернеті. Привабливим підходом є використання одного агента, одного великого запиту та всіх необхідних інструментів, приєднаних до нього. Коли кількість інструментів та інструкцій зростає, модель починає сумніватися, який інструмент обрати, а її міркування стають менш чіткими. Виправлення помилок перетворюється на справжній кошмар, адже неможливо з’ясувати, чому вона обрала неправильний інструмент, коли все знаходиться в одному запиті.
Рішенням є розподіл обов’язків: один агент для математики, один — для досліджень, а ще один — керівник, який фактично є маршрутизатором та вирішує, який спеціаліст буде обробляти кожну частину запиту.
Паттерн керівника
Уявіть собі кол-центр. Ви телефонуєте з питанням, але вас не спрямовують безпосередньо до фахівця. Спочатку слухає рецепціоніст, вирішує, чи потрібні вам послуги з билінгу, технічної підтримки чи продажів, і передає вас далі. Рецепціоніст є керівником, а три команди — це спеціалізовані агенти. Керівник ніколи не відповідає на саме питання; його єдине завдання — вибрати того, хто має це зробити.
Візуально це виглядає так:
┌─────────────┐
│ User │
│ Request │
└──────┬──────┘
│
▼
┌───────────────┐
│ Supervisor │
│ (routes task)│
└───────┬───────┘
┌──────┴──────┐
▼ ▼
┌────────────┐ ┌────────────┐
│ Research │ │ Math │
│ Agent │ │ Agent │
└──────┬─────┘ └──────┬─────┘
│ │
└──────┬───────┘
▼
┌───────────────┐
│ Supervisor │
│ (final answer)│
└───────────────┘
Керівник бере завдання, передає його відповідному фахівцю, отримує результат та потім вирішує, чи переслати його знову, чи надати користувачеві остаточну відповідь.
Рекомендований підхід, крок за кроком
Що варто знати перед написанням коду: команда LangChain тепер рекомендує створювати модель керівника безпосередньо з виклику інструментів, а не лише за допомогою старої бібліотеки-допоміжника langgraph-supervisor, оскільки виклик інструментів забезпечує більший контроль над формуванням контексту. Спочатку ви побачите швидку готову версію, а потім ручну, щоб було зрозуміло, що відбувається під час її роботи.
Крок 1: встановіть пакети
pip install langgraph langchain-openai langgraph-supervisor
Крок 2: створення спеціалістів
Почніть з імпорту:
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
Перший спеціаліст займається математикою, а її інструментами є дві невеликі функції. Спочатку йде додавання:
# Specialist 1: handles math
def add(a: float, b: float) -> float:
"""Add two numbers together."""
return a + b
Потім множення:
def multiply(a: float, b: float) -> float:
"""Multiply two numbers together."""
return a * b
Обгорніть їх агентом ReAct із ім’ям та запитом, які обмежують його роль:
math_agent = create_react_agent(
model="openai:gpt-4o",
tools=[add, multiply],
name="math_expert",
prompt="You are a math specialist. Solve numeric problems precisely."
)
Другий фахівець здійснює дослідження. Одним із його інструментів є функція пошуку, представлена тут у спрощеному вигляді:
# Specialist 2: handles research
def web_search(query: str) -> str:
"""Search the web for current information."""
# In production, replace this with a real search API call
return f"Search results for '{query}': [your real search integration here]"
Обгорніть це також агентом:
research_agent = create_react_agent(
model="openai:gpt-4o",
tools=[web_search],
name="research_expert",
prompt="You are a research specialist. Find accurate, current information."
)
Кожен агент має саме одну роботу та лише інструменти, необхідні для цієї роботи, тому немає плутанини та надмірних вимог до запиту. У прикладах використовуються моделі OpenAI, хоча підійде й безкоштовна альтернатива, така як Groq чи OpenRouter.
Крок 3: створення керівника
Імпортуйте допоміжний модуль:
from langgraph_supervisor import create_supervisor
Потім надайте йому двох спеціалістів — модель та інструкцію, які вказують, коли використовувати кожного з них:
workflow = create_supervisor(
[research_agent, math_agent],
model=ChatOpenAI(model="gpt-4o"),
prompt=(
"You are a team supervisor managing a research expert and a math expert. "
"For questions requiring current information or web lookups, use research_expert. "
"For anything involving calculation, use math_expert. "
"Route each request to exactly the right specialist."
)
)
Зіберіть робочий процес у запускове додаток:
app = workflow.compile()
Крок 4: запустіть його
Надішліть математичне запитання:
result = app.invoke({
"messages": [
{"role": "user", "content": "What's 1,847 multiplied by 293?"}
]
})
І виведіть остаточне повідомлення:
print(result["messages"][-1].content)
Ось що відбувається насправді:
- Керівник читає запит.
- Він бачить розрахунок, а не дослідницьке запитання.
- Він передає завдання
math_expert. math_expertобчислює відповідь за допомогою свого інструментуmultiply.- Керівник знову бере контроль та повертає відповідь.
У разі дослідницького запитання, наприклад щодо поточної кількості співробітників у великих технологічних компаніях, той самий керівник направляє його до research_expert. Код залишається ідентичним, відрізняється лише рішення щодо напрямку обробки, і LLM приймає це рішення самостійно.
Що насправді є інструментами передачі завдань
Ця частина усуває відчуття магії. Супервізор не використовує якийсь прихований механізм LangGraph для передачі завдань — він робить це за допомогою викликів інструментів, точно так само, як будь-який звичайний агент, що використовує інструменти. На задньому плані функція create_supervisor створює інструменти для передачі завдань кожному спеціалісту з назвами на кшталт transfer_to_research_expert та transfer_to_math_expert. Модель супервізора сприймає їх як інструменти, які можна викликати, не відрізняючись від звичайних пошуків інформації; тому, коли вона вирішує, що потрібні дослідження, вона просто викликає відповідну функцію передачі. Це дає впевненість: супервізор зовсім не є новою концепцією — це звичайний цикл агента, де деякими інструментами випадково є інші агенти замість викликів API.
Створення вручну
Якщо ви хочете повний контроль над маршрутизацією, що є рекомендованим під час переходу до продакшну, ось ручна версія на основі сирого StateGraph. Почніть з імпортів:
from typing import Annotated, Sequence, TypedDict
import operator
from langchain_core.messages import BaseMessage
from langgraph.graph import StateGraph, END
Визначте спільний стан, який містить список повідомлень та назву наступного вузла:
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]
next: str
Вузол керівництва переглядає останнє повідомлення:
def supervisor_node(state: AgentState):
"""Decide which specialist should act next."""
last_message = state["messages"][-1].content
У продакшені рішення прийматиметься на основі виклику LLM. Тут це спрощено до простих перевірок ключових слів, щоб логіка маршрутизації залишалася зрозумілою:
# In production, this decision comes from an LLM call —
# simplified here to show the routing logic clearly
if any(word in last_message.lower() for word in ["calculate", "multiply", "add", "sum"]):
return {"next": "math_agent"}
else:
return {"next": "research_agent"}
Кожен спеціаліст отримує вузол, який запускає його агента, а потім повертає керування наглядачу. Спочатку математика:
def math_node(state: AgentState):
result = math_agent.invoke(state)
return {"messages": [result], "next": "supervisor"}
Потім дослідження:
def research_node(state: AgentState):
result = research_agent.invoke(state)
return {"messages": [result], "next": "supervisor"}
Далі з’єднайте граф, зареєструвавши вузли:
# Wire the graph together
workflow = StateGraph(AgentState)
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("math_agent", math_node)
workflow.add_node("research_agent", research_node)
Встановіть точку входу:
workflow.set_entry_point("supervisor")
Додайте умовні ребра, які залежать від рішення керівника:
workflow.add_conditional_edges(
"supervisor",
lambda state: state["next"],
{"math_agent": "math_agent", "research_agent": "research_agent", "end": END}
)
Після цього надішліть кожного спеціаліста назад до керівника:
workflow.add_edge("math_agent", "supervisor")
workflow.add_edge("research_agent", "supervisor")
І скомпілюйте:
app = workflow.compile()
Це більш деталізовано, але нічого не приховано. Ви можете бачити, як передається стан, як приймаються кожні рішення щодо маршрутизації, та де можна додати власну логіку — наприклад, функції обліку дій, захисні механізми чи користувацькі умови завершення.
Вибір структури стану, який найшвидше втрачає ефективність
Попередження з практики роботи в продакшні: схема стану — це найскладніший елемент, який пізніше важко змінити. Існує спокуса вже на початку помістити всі можливі поля у один плоский словник, який би охоплював повідомлення, контекст користувача та проміжні результати. Це підходить для систем з двома агентами. Але коли вже вісім чи десять вузлів одночасно читають та записують у той самий спільний стан, це стає проблематичним, адже перевірка типів у LangGraph є суворою, а для впорядкування перевантаженого стану доводиться впливати на кожен вузол. Проектуйте схему свідомо з першого дня, навіть для системи з двома агентами. Вирішіть, що саме потрібно читати та записувати кожному агенту, і не включайте все інше до спільного стану.
Проблема, з якою ви зіткнетеся
Системи керування, які потрапляють у промислове використання, зазвичай демонструють таку модель. Уявіть собі, що система керування делегує завдання. Спеціаліст його виконує, але система керування вважає результат недостатнім та повертає його назад. Спеціаліст вносить корективи, але нова версія виявляється ще гіршою за першу. Система керування все одно незадоволена та знову її повертає, і якщо не вжити заходів, цей процес може тривати нескінченно.
Рішенням є обмеження кількості таких циклів, що може бути здійснено за допомогою простого лічильника у стані системи, разом із чітко визначеною умовою зупинки. Думка ШІ щодо того, чи є щось достатньо гарним, не є надійною єдиною умовою для зупинки.
Коли не варто його використовувати
Будьте чесні щодо складності завдання. Якщо робота полягає лише у пошуку інформації та її генерації, як у звичайних запитаннях-відповідях у стилі RAG, наявність керівника є зайвою. Додаткове керування станом та обробка маршрутів не приносять жодної користі, коли насправді немає необхідності у маршрутизації. Використовуйте паттерн керівника лише тоді, коли у вас є справді різні спеціалісти та реальний вибір маршруту, а не за замовчуванням.
Підсумок
- Один перевантажений агент призводить до проблем зі зростанням складності.
- Керівник — це агент-маршрутизатор, який делегує завдання спеціалістам.
create_supervisor— це швидкий, готовий до використання маршрут.- Під час передачі завдань використовуються звичайні виклики інструментів.
- Ручний
StateGraphзабезпечує повний контроль над маршрутизацією. - Два уроки з практики: розробляйте схему стану заздалегідь та обмежуйте кількість циклів корекції.