Stwórz routera wielu agentów za pomocą LangGraph oraz agenta nadzorczego
Jak podzielić jednego przeciążonego agenta na specjalistów z przełożonym w LangGraph, wykorzystując gotowy pomocnik oraz ręcznie tworzony StateGraph, a także projektowanie stanów i pułapki związane z pętlami.
Źródła: Build a Multi-Agent Router with LangGraph in 30 Minutes
Umieść jednego agenta przed rzeczywistym produktem, a szybko zacznie robić za dużo. Poproś jednego agenta o obsługę klienta, śledzenie zamówień i zwroty pieniędzy – ostatecznie będzie przeciętny we wszystkich trzech zadaniach. Jego instrukcje systemowe przeradzają się w 2000-słowny labirynt zdań typu „jeśli użytkownik pyta o X, rób Y, ale jeśli chodzi też o Z, wtedy...“. Rozwiązaniem nie jest sprytniejszy agent. Chodzi o większą liczbę mniejszych agentów wraz z jednym nadzorcą, który kieruje ich działaniami. To bez wątpienia najczęściej stosowany wzorzec w sztucznej inteligencji opartej na agentach obecnie, a jego działająca wersja mieści się w krótkim programie LangGraph.
Problemy z jednym gigantycznym agentem
Załóżmy, że budujemy asystenta, który odpowiada na ogólne pytania, wykonuje obliczenia arytmetyczne i wyszukuje aktualne fakty w internecie. Kuszącym rozwiązaniem jest jeden agent, jeden długi prompt oraz wszystkie narzędzia dołączone do niego. Gdy narzędzia i instrukcje się mnożą, model traci pewność co do wyboru odpowiedniego narzędzia, a jego rozumowanie staje się mniej precyzyjne. Naprawianie błędów zamienia się w koszmar, ponieważ nie można ustalić, dlaczego wybrał niewłaściwe narzędzie, skoro wszystko znajduje się w jednym promptie.
Rozwiązaniem jest podział obowiązków: jeden agent do matematyki, drugi do badań, oraz nadzorca, który jest po prostu routerem decydującym, który specjalista zajmie się poszczególnymi częściami zapytania.
Wzorzec nadzorcy
Pomyśl o centrum obsługi klienta. Dzwonisz z pytaniem i nie jesteś od razu połączony bezpośrednio z ekspertem. Najpierw słucha recepcjonista, decyduje, czy potrzebujesz pomocy w kwestiach fakturowania, wsparcia technicznego czy sprzedaży, a następnie cię przekazuje. Recepcjonista jest kierownikiem, a trzy zespoły to specjalistyczni pracownicy. Kierownik nigdy nie odpowiada na samo pytanie – jego jedynym zadaniem jest wybór osoby, która ma to zrobić.
Wizualnie wygląda to tak:
┌─────────────┐
│ User │
│ Request │
└──────┬──────┘
│
▼
┌───────────────┐
│ Supervisor │
│ (routes task)│
└───────┬───────┘
┌──────┴──────┐
▼ ▼
┌────────────┐ ┌────────────┐
│ Research │ │ Math │
│ Agent │ │ Agent │
└──────┬─────┘ └──────┬─────┘
│ │
└──────┬───────┘
▼
┌───────────────┐
│ Supervisor │
│ (final answer)│
└───────────────┘
Kierownik przejmuje zadanie, przekazuje je odpowiedniemu specjaliście, otrzymuje wynik, a następnie decyduje, czy przekierować je ponownie, czy udzielić użytkownikowi ostatecznej odpowiedzi.
Zalecana metoda budowy, krok po kroku
Kwestia, którą warto znać przed pisaniem kodu: zespół LangChain zaleca teraz budowanie wzorca kierownika bezpośrednio poprzez wywoływanie narzędzi, a nie poleganie wyłącznie na starszej bibliotece pomocniczej langgraph-supervisor, ponieważ wywoływanie narzędzi daje większą kontrolę nad projektowaniem kontekstu. Poniżej najpierw znajdziesz szybką wersję gotową, a potem tę ręcznie tworzoną, aby było jasne, co dzieje się w tle.
Krok 1: zainstaluj pakietów
pip install langgraph langchain-openai langgraph-supervisor
Krok 2: tworzenie specjalistów
Zacznij od importów:
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
Pierwszy specjalista zajmuje się matematyką, a jej narzędziami są dwie małe funkcje. Najpierw przychodzi dodawanie:
# Specialist 1: handles math
def add(a: float, b: float) -> float:
"""Add two numbers together."""
return a + b
Następnie mnożenie:
def multiply(a: float, b: float) -> float:
"""Multiply two numbers together."""
return a * b
Otocz je agentem ReAct z nazwą oraz promptem, które ograniczają jego rolę:
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."
)
Drugi specjalista prowadzi badania. Jednym z jego narzędzi jest funkcja wyszukiwania, przedstawiona tutaj w uproszczonej formie:
# 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]"
Zawróć to również w agencie:
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."
)
Każdy agent ma dokładnie jedno zadanie oraz tylko te narzędzia, których to zadanie wymaga, dzięki czemu nie ma zamieszania ani nadmiernie rozbudowanych poleceń. Przykłady wykorzystują modele OpenAI, chociaż sprawdziłaby się również darmowa alternatywa, takia jak Groq lub OpenRouter.
Krok 3: stworzenie nadzorcy
Importuj pomocnika:
from langgraph_supervisor import create_supervisor
Następnie podaj dwoje specjalistów: model oraz instrukcję określającą, kiedy używać każdego z nich:
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."
)
)
Zkompiluj proces pracy w aplikację, którą można uruchomić:
app = workflow.compile()
Krok 4: uruchom to
Wyślij pytanie matematyczne:
result = app.invoke({
"messages": [
{"role": "user", "content": "What's 1,847 multiplied by 293?"}
]
})
I wydrukuj końcowe wiadomość:
print(result["messages"][-1].content)
Oto co dzieje się w tle:
- Nadzorca czyta prośbę.
- Widzi, że chodzi o obliczenie, a nie o pytanie badawcze.
- Przekazuje zadanie
math_expert. math_expertoblicza odpowiedź za pomocą narzędziamultiply.- Nadzorca ponownie przejmuje kontrolę i zwraca odpowiedź.
W przypadku pytania badawczego, np. dotyczącego aktualnej liczby pracowników w dużych firmach technologicznych, ten sam nadzorca kieruje je do research_expert. Kod jest identyczny, różni się tylko decyzja o kierowaniu zadania, a LLM podejmuje tę decyzję samodzielnie.
Czym naprawdę są narzędzia przekazywania zadań
Ta część eliminuje wrażenie magii. Nadzorca nie przekazuje zadań za pośrednictwem jakiegoś ukrytego mechanizmu LangGraph – robi to poprzez wywołania narzędzi, dokładnie tak jak każdy zwykły agent używający narzędzi. W tle funkcja create_supervisor tworzy narzędzia do przekazywania zadań dla każdego specjalisty, o nazwach takich jak transfer_to_research_expert i transfer_to_math_expert. Model nadzorcy traktuje je jako narzędzia, które można wywołać, bez różnicy od zwykłych zapytań o pogodę; dlatego gdy stwierdzi, że potrzebne jest badanie, po prostu wywołuje funkcję przekazywania zadań. Konsekwencja jest uspokajająca – nadzorca wcale nie jest nowym pojęciem. To zwykły cykl agenta, w którym niektóre „narzędzia” to w rzeczywistości inne agenty, a nie wywołania API.
Budowanie ręcznie
Jeśli chcesz mieć pełną kontrolę nad routowaniem, co jest zalecane w fazie przygotowań do produkcji, oto wersja ręczna opracowana na bazie surowego StateGraph. Zacznij od importów:
from typing import Annotated, Sequence, TypedDict
import operator
from langchain_core.messages import BaseMessage
from langgraph.graph import StateGraph, END
Zdefiniuj stan wspólny, który przechowuje listę wiadomości oraz nazwę następnego węzła:
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]
next: str
Węzeł nadzorczy sprawdza najnowsze wiadomość:
def supervisor_node(state: AgentState):
"""Decide which specialist should act next."""
last_message = state["messages"][-1].content
W środowisku produkcyjnym decyzja byłaby podejmowana na podstawie wywołania modelu LLM. Tutaj sprowadza się to do prostych sprawdzeń słów kluczowych, dzięki czemu logika routingu pozostaje widoczna:
# 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"}
Każdy specjalista otrzymuje węzeł, który uruchamia jego agenta, a następnie przekazuje kontrolę z powrotem nadzorcy. Najpierw matematyka:
def math_node(state: AgentState):
result = math_agent.invoke(state)
return {"messages": [result], "next": "supervisor"}
Następnie badania:
def research_node(state: AgentState):
result = research_agent.invoke(state)
return {"messages": [result], "next": "supervisor"}
Następnie połącz graf, zarejestrujając węzły:
# 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)
Ustaw punkt wejścia:
workflow.set_entry_point("supervisor")
Dodaj krawędzie warunkowe, które wynikają z decyzji nadzorcy:
workflow.add_conditional_edges(
"supervisor",
lambda state: state["next"],
{"math_agent": "math_agent", "research_agent": "research_agent", "end": END}
)
Po tym odesłaj każdego specjalistę z powrotem do przełożonego:
workflow.add_edge("math_agent", "supervisor")
workflow.add_edge("research_agent", "supervisor")
I skompiluj:
app = workflow.compile()
Jest bardziej rozbudowany, ale nic nie jest ukryte. Można zobaczyć, jak przepływa stan, jak podejmowane są decyzje routingu oraz gdzie dodać własną logikę, taką jak rejestracja zdarzeń, mechanizmy kontroli lub niestandardowe warunki zakończenia.
Wybór projektu stanu, który najbardziej się starzeje
Ostrzeżenie z doświadczeń z produkcji: schemat stanu to decyzja, której najtrudniej jest później zmienić. Kuszące jest umieszczenie we wczesnym etapie wszystkich możliwych pól w jednym prostym słowniku, obejmującym komunikaty, kontekst użytkownika oraz wyniki pośrednie. To w porządku przy dwóch agentach. Jednak gdy jest osiem lub dziesięć węzłów, które wszystkie czytają i zapisują ten sam wspólny stan, staje się to uciążliwe, ponieważ sprawdzanie typów w LangGraph jest ścisłe, a uporanie się z rozdętym stanem oznacza konieczność edycji każdego węzła. Projektuj schemat celowo od pierwszego dnia, nawet w przypadku systemu z dwoma agentami. Zdecyduj, co każdy agent faktycznie musi czytać i zapisywać, a wszystko inne trzymaj poza wspólnym stanem.
Powódzenie, z którym się spotkasz
Systemy nadzorujące, które trafiają do produkcji, często wykazują taki wzorzec zachowania. Wyobraź sobie, że nadzorca deleguje zadanie. Specjalista je wykonuje, ale nadzorca uznaje wynik za niewystarczający i odrzuca go. Specjalista dokonuje poprawek, a nowa wersja okazuje się jeszcze gorsza od pierwszej. Nadzorca nadal jest niezadowolony i odsyła ją ponownie – jeśli się temu nie przeciwstawić, może to trwać w nieskończoność.
Rozwiązaniem jest ustalenie limitu liczby takich cykli pomiędzy nadzorcą a specjalistą, co może być realizowane za pomocą prostego licznika przechowywanego w stanie systemu, w połączeniu z jasną, deterministyczną warunkiem zatrzymania. Opinia modelu językowego o tym, czy coś jest wystarczająco dobre, nie stanowi niezawodnej reguły zatrzymywania procesu.
Kiedy nie należy go używać
Bądź szczery co do złożoności. Jeśli zadanie polega jedynie na pobraniu informacji, a następnie ich wygenerowaniu, jak w zwykłym systemie Q&A typu RAG, nadmiarowy nadzorca jest niepotrzebny. Dodatkowe zarządzanie stanem i obciążenia związane z routowaniem nie przynoszą żadnych korzyści, gdy nie ma rzeczywistej decyzji dotyczącej kierunku działania. Zastosuj wzorzec nadzorcy dopiero wtedy, gdy masz naprawdę odrębnych specjalistów i rzeczywisty wybór co do kierunku działania, a nie przyjmuj go jako standardu.
Podsumowanie
- Jeden przeciążony agent zawodzi wraz ze wzrostem złożoności.
- Nadzorca to agent routujący, który deleguje zadania specjalistom.
create_supervisorto szybka, gotowa metoda routowania.- W rzeczywistości przenoszenie obowiązków polega na zwykłych wywołaniach narzędzi.
- Ręczny
StateGraphzapewnia pełną kontrolę nad routowaniem. - Dwie lekcje z praktyki: zaprojektuj schemat stanu wcześnie i ogranicz pętle rewizji.
Literatura pokrewna
- Zespół wielu agentów zdolny do samoleczenia w LangGraph: architekt, tester, debugger — Stwórz maszynę stanową LangGraph w Pythonie, w której architekt, programista, środowisko testowe pytest, debugger i recenzent współpracują i powtarzają proces, aż testy się udają.
- Triage zgłoszeń obsługi klienta za pomocą wielu agentów z LangGraph i Ollama — Stwórz graf triażu zgłoszeń z trzema agentami przy użyciu LangGraph: klasyfikator, osoba odpowiadająca i człowiek dokonujący ostatecznej oceny, działający na lokalnym Ollama lub przez API przy użyciu jednej zmienionej linii kodu.