LangGraph Multi-Agent Systems: Nadzorcy, podgrafy i wysyłanie

Wymagania wstępne: przykłady zakładają, że potrafisz już tworzyć graf LangGraph z jednym agentem, wykorzystując odpowiednie narzędzia i pamięć, oraz że wiesz, jak wstrzymać i wznowić jego działanie za pomocą interrupt(). Wszystko, co jest poniżej, opiera się na tym założeniu, przechodząc od pytania, dlaczego chciałbyś mieć kilku agentów, do kompletnego asystenta badawczego wieloagentowego.

Dlaczego dzielić jednego agenta na wielu

Załóżmy, że asystent z jednym agentem działa dobrze. Wtedy użytkownicy zaczynają prosić go o sprawdzanie e-maili, badanie tematów, pisanie raportów oraz zarządzanie zadaniami, wszystko w ramach tej samej rozmowy. Zwykłą reakcją jest dodawanie kolejnych narzędzi oraz umieszczanie coraz większej ilości instrukcji w komendzie systemowej. Wkrótce pojawia się 15 narzędzi oraz komenda licząca 2000 słów, a jakość pracy asystenta stopniowo spada. Model zaczyna mieszać różne funkcje – używa narzędzia do e-maili podczas zadania badawczego i narzędzia badawczego podczas zadania związanej z e-mailem.

To jest przeładowanie poznawcze, a to główny powód istnienia projektów opartych na wielu agentach. Rozwiązaniem, na które od dziesięcioleci polegają inżynierowie oprogramowania, jest podział obowiązków. W LangGraph takie podejście rozwiązuje trzy odrębne problemy:

• Przeładowanie poznawcze. Jeden model musi zarządzać zbyt wieloma niespowiązanymi zadaniami. Specjalistyczni agenci, z których każdy jest dobry w jednej dziedzinie, redukują to obciążenie.
• Wąskie gardła sekwencyjne. Zadania, które mogłyby być wykonywane jednocześnie, są realizowane krok po kroku. Agenci równoległe skracają czas oczekiwania.
• Złożoność. Nikt nie jest w stanie zrozumieć grafu składającego się z 40 węzłów. Mniejsze podgrafy można analizować i testować pojedynczo.

Każdy z poniższych wzorców adresuje jeden z tych trzech problemów.

Pięć poziomów architektury

Traktuj tę progresję jak schody, na których każdy stopień dodaje jedną ideę.

Poziom 1: nadzorca. Jeden agent-nadzorca koordynuje pracowników zajmujących się konkretnymi zadaniami. Nadal jest to jedyna osoba podejmująca decyzje, ale deleguje zadania zamiast robić wszystko sam, co już odróżnia go od zwykłego chatbota używającego narzędzi.

Poziom 2: paralelizacja. Niezależne zadania są dzielone i wykonywane jednocześnie, a następnie łączone. Chodzi tu o rozgałęzianie i łączenie: kilka gałęzi działa równocześnie i dostarcza dane do kolejnego etapu. Jest to odpowiednie dla powtarzalnych zadań w wielu źródłach lub podzadaniach.

Poziom 3: podgrafy. Zamiast jednego ogromnego procesu pracy, każda gałąź staje się mniejszym grafem z własnymi wewnętrznymi krokami i schematem wyników. Skomplikowane elementy stają się łatwiejsze do budowania, testowania i zrozumienia.

Poziom 4: dynamiczny paralelizm. W przypadku wzorca Send, który opiera się na stylu map-reduce, liczba gałęzi nie jest znana z góry. Pracownicy są tworzeni w czasie wykonywania programu, po jednym na każdy element, temat lub analityka. Agenci nie są z góry zdefiniowani, wszyscy pracują jednocześnie, a każdy ma do wykonania konkretny zadanie.

Poziom 5: połączenie wszystkiego. Realistyczny system wykorzystuje wszystkie powyższe elementy razem, jak w przykładzie asystenta badawczego na końcu.

Krótki sposób na zapamiętanie: Poziom 1 polega na przekazywaniu zadań przez jednego nadzorcy, Poziom 2 uruchamia ustalony zbiór pracowników jednocześnie, Poziom 3 pozwala, by każda gałąź stanowiła pełny graf, a Poziom 4 tworzy pracowników na żądanie.

Poziom 1: wzorzec nadzorcy

To jest najprostszy wzorzec wielu agentów i odpowiedni punkt wyjścia. Supervisor otrzymuje prośbę użytkownika i decyduje, który specjalista powinien się nią zająć. Specjalista wykonuje pracę i przekazuje wyniki, a supervisor decyduje, co dalej robić.

Wyobraźmy sobie kancelarię prawniczą. Starszy partner rozmawia z klientem, rozumie jego potrzeby i przydziela zadania młodszym współpracownikom – jednemu do spraw umów, drugiemu do sporów sądowych, trzeciemu do kwestii zgodności z przepisami. Współpracownik wykonuje szczegółową pracę, a partner ją sprawdza i odpowiada klientowi.

Przepływ wygląda w ten sposób:

User Input
    ↓
[Supervisor Node]  ← decides who handles what
    ↓         ↓
[Specialist A] [Specialist B]   ← do the actual work
    ↓         ↓
[Supervisor Node]  ← reviews results, responds or routes again
    ↓
Response

Stan

Kluczową decyzją projektową jest pole next_agent. Nadzorca zapisuje w nim dane, a router je odczytuje.

# ── MODULE 2: STATE ─────────────────────────────────────────
from typing import TypedDict, Annotated, Literal
from langgraph.graph import MessagesState

class SupervisorState(MessagesState):
    # The supervisor writes this field to indicate who should act next.
    # "researcher", "writer", "FINISH" are the possible values.
    next_agent: str

Narzędzia specjalistyczne

Każdy specjalista ma swoje własne, dedykowane narzędzia, które nie nakładają się na siebie.

# ── MODULE 3: TOOLS ─────────────────────────────────────────
from langchain_core.tools import tool

# Researcher's tools
@tool
def search_web(query: str) -> str:
    """Search the web for factual information, news, or research."""
    return f"[Web results for: {query}]"
@tool
def search_academic(query: str) -> str:
    """Search academic papers and journals for scholarly sources."""
    return f"[Academic results for: {query}]"
# Writer's tools
@tool
def format_as_report(content: str, title: str) -> str:
    """Format raw content into a structured report with headings."""
    return f"# {title}\n\n{content}"
@tool
def check_grammar(text: str) -> str:
    """Check and correct grammar in the provided text."""
    return f"[Grammar-checked]: {text}"
# Each specialist gets only its own tools
researcher_tools = [search_web, search_academic]
writer_tools = [format_as_report, check_grammar]
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", temperature=0)
researcher_llm = llm.bind_tools(researcher_tools)
writer_llm     = llm.bind_tools(writer_tools)

Węzły

Następnie przychodzą węzeł nadzorczy oraz węzły specjalistyczne. Odpowiedź nadzorcy w formie zwykłego tekstu stanowi sama decyzję o routingu.

# ── MODULE 4: NODES ─────────────────────────────────────────
from langchain_core.messages import SystemMessage, HumanMessage
from langgraph.prebuilt import ToolNode

# ── Supervisor ────────────────────────────────────────────
SUPERVISOR_PROMPT = """You are a supervisor coordinating a team of specialists.
Given the conversation, decide who should act next.
Your team:
- researcher: finds information, searches web and academic sources
- writer: formats content, writes reports, checks grammar
- FINISH: use this when the task is complete and you have a final answer
Respond with ONLY the name of the next agent: researcher, writer, or FINISH."""
def supervisor_node(state: SupervisorState) -> dict:
    """Reads the conversation history and routes to the right specialist."""
    messages = [SystemMessage(content=SUPERVISOR_PROMPT)] + state["messages"]
    response = llm.invoke(messages)
    # The supervisor's plain text response IS the routing decision
    return {"next_agent": response.content.strip()}

# ── Researcher Specialist ──────────────────────────────────
def researcher_node(state: SupervisorState) -> dict:
    """Specialist that handles all research and information-gathering tasks."""
    messages = [
        SystemMessage(content="You are a research specialist. Use your search tools to find accurate, thorough information.")
    ] + state["messages"]
    response = researcher_llm.invoke(messages)
    return {"messages": [response]}

# ── Writer Specialist ──────────────────────────────────────
def writer_node(state: SupervisorState) -> dict:
    """Specialist that handles all writing, formatting, and editing tasks."""
    messages = [
        SystemMessage(content="You are a writing specialist. Use your tools to produce well-structured, polished output.")
    ] + state["messages"]
    response = writer_llm.invoke(messages)
    return {"messages": [response]}

# Tool execution nodes for each specialist
researcher_tool_node = ToolNode(researcher_tools)
writer_tool_node     = ToolNode(writer_tools)

Ścieżki

# ── MODULE 5: ROUTING ────────────────────────────────────────
def route_after_supervisor(state: SupervisorState) -> Literal["researcher", "writer", "__end__"]:
    """Read the supervisor's decision and route accordingly."""
    decision = state["next_agent"]
    if decision == "FINISH":
        return "__end__"
    return decision  # "researcher" or "writer"

def route_researcher(state: SupervisorState) -> Literal["researcher_tools", "supervisor"]:
    """After researcher acts: did it call a tool, or produce a final answer?"""
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "researcher_tools"
    return "supervisor"  # Research done - report back to supervisor
def route_writer(state: SupervisorState) -> Literal["writer_tools", "supervisor"]:
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "writer_tools"
    return "supervisor"

Składanie grafu

# ── MODULE 6: GRAPH ASSEMBLY ────────────────────────────────
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

graph_builder = StateGraph(SupervisorState)
# Register all nodes
graph_builder.add_node("supervisor",        supervisor_node)
graph_builder.add_node("researcher",        researcher_node)
graph_builder.add_node("researcher_tools",  researcher_tool_node)
graph_builder.add_node("writer",            writer_node)
graph_builder.add_node("writer_tools",      writer_tool_node)
# Entry: always start at supervisor
graph_builder.add_edge(START, "supervisor")
# Supervisor routes to specialist or ends
graph_builder.add_conditional_edges(
    "supervisor",
    route_after_supervisor,
    {"researcher": "researcher", "writer": "writer", "__end__": END}
)
# Researcher loop: tools → researcher → supervisor
graph_builder.add_conditional_edges(
    "researcher", route_researcher,
    {"researcher_tools": "researcher_tools", "supervisor": "supervisor"}
)
graph_builder.add_edge("researcher_tools", "researcher")
# Writer loop: tools → writer → supervisor
graph_builder.add_conditional_edges(
    "writer", route_writer,
    {"writer_tools": "writer_tools", "supervisor": "supervisor"}
)
graph_builder.add_edge("writer_tools", "writer")
graph = graph_builder.compile(checkpointer=MemorySaver())

W procesie montażu kluczowe jest kroku, który wraca do nadzorcy. Gdy specjalista kończy pracę, składa raport. Nadzorca analizuje całą rozmowę, a następnie wybiera jedną z trzech opcji: przekazuje zadanie innemu specjaliście, wysyła tego samego ponownie lub kończy pracę. W istocie to właśnie jest istota orkiestracji.

Poziom 2: paralelizacja z rozgałęzieniami i łączeniami

Wzorzec nadzorcy jest sekwencyjny – jednocześnie może pracować tylko jeden specjalista. Wiele rzeczywistych zadań zawiera niezależne podzadania, które mogą być wykonywane jednocześnie, a paralelizm rozwiązuje problem opóźnień.

Dobrym przykładem jest gotowanie. Nikt nie kończy przygotowywania sałatki, zanim nie zacznie gotować makaronu, a potem sosu. Przygotowujesz sałatkę, zagotowujesz wodę i jednocześnie zaczynasz robić sos, dzięki czemu kolacja jest gotowa szybciej.

W LangGraph węzeł z kilkoma wyjściowymi krawędziami powoduje równoległe podziałanie. LangGraph wykrywa tę strukturę rozgałęzienia i uruchamia wszystkie docelowe operacje jednocześnie, co nazywa superstepem. Wyniki są zbierane przed uruchomieniem jakiegokolwiek kolejnego węzła, co stanowi proces przyjmowania danych.

Istnieje jedna niezmienna zasada: gdy kilka węzłów w tym samym superstepie zapisuje dane do tego samego pola stanu, LangGraph musi umieć połączyć te zapisy. Bez funkcji redukcyjnej pojawia się błąd. Z użyciem operator.add wyniki są bezpiecznie łączone.

# ── MODULE 2: STATE FOR PARALLEL PATTERN ────────────────────
import operator
from typing import Annotated

class ResearchState(TypedDict):
    query: str
    # Without Annotated + operator.add, parallel writes to 'context' would CRASH.
    # With it, LangGraph concatenates results from all parallel nodes.
    context: Annotated[list[str], operator.add]
    final_answer: str

Przykład badań paralelnych wykorzystuje dwa źródła:

# ── MODULE 4: PARALLEL WORKER NODES ─────────────────────────

def web_search_node(state: ResearchState) -> dict:
    """Runs in parallel with academic_search_node."""
    result = f"[WEB] Results for: {state['query']}"
    return {"context": [result]}   # Returns a LIST - operator.add will concatenate
def academic_search_node(state: ResearchState) -> dict:
    """Runs in parallel with web_search_node."""
    result = f"[ACADEMIC] Papers about: {state['query']}"
    return {"context": [result]}   # Also a LIST
def synthesize_node(state: ResearchState) -> dict:
    """Runs AFTER both parallel nodes complete (fan-in point).
    state['context'] already contains results from BOTH parallel nodes."""
    combined = "\n".join(state["context"])
    prompt = [
        SystemMessage(content="Synthesize the following research into a clear answer."),
        HumanMessage(content=f"Research:\n{combined}\n\nQuery: {state['query']}")
    ]
    response = llm.invoke(prompt)
    return {"final_answer": response.content}

# ── MODULE 6: PARALLEL GRAPH ASSEMBLY ───────────────────────
from langgraph.graph import StateGraph, START, END
graph_builder = StateGraph(ResearchState)
graph_builder.add_node("web_search",      web_search_node)
graph_builder.add_node("academic_search", academic_search_node)
graph_builder.add_node("synthesize",      synthesize_node)
# Fan-out: START branches to BOTH search nodes simultaneously
graph_builder.add_edge(START,       "web_search")
graph_builder.add_edge(START,       "academic_search")
# Fan-in: BOTH search nodes must complete before synthesize runs
graph_builder.add_edge("web_search",      "synthesize")
graph_builder.add_edge("academic_search", "synthesize")
graph_builder.add_edge("synthesize", END)
parallel_graph = graph_builder.compile()

Rozgałęzienie polega po prostu na dwóch krawędziach wychodzących z tego samego źródła, START. LangGraph zauważa to i uruchamia oba docelowe węzły jednocześnie. Przyłączenie również odbywa się samoistnie, ponieważ funkcja synthesize jest wykonywana tylko raz, gdy obie przychodzące krawędzie zostaną spełnione, co oznacza, że oba węzły równoległe zostały zakończone.

Poziom 3: podgrafy do inkapsulacji

Płaskie węzły równoległe radzą sobie z prostymi przypadkami. Gdy każda gałąź jest złożona, składa się z kilku kroków, posiada własne narzędzia i logikę, lepszym rozwiązaniem są podgrafy. Podgraf to w pełni skompilowany StateGraph, który działa jako jeden węzeł w obrębie grafu nadrzędnego.

Pomyśl o działach w firmie. Dyrektor generalny, czyli graf nadrzędny, deleguje zadania do działu inżynieryjnego i marketingu. Każdy dział ma własne procesy i spotkania, ale dyrektor generalny jedynie przekazuje zadania i odbiera wyniki ich realizacji.

Rodzic i podgraf komunikują się za pośrednictwem wspólnych kluczy stanu. Każdy klucz obecny zarówno w stanie rodzica, jak i w stanie podgrafu jest przekazywany przy uruchamianiu podgrafu i zwracany po jego zakończeniu.

Parent State: {topic, cleaned_data, report_A, report_B, processed_ids}
                         ↓
Subgraph A State: {cleaned_data, internal_step_1, report_A, processed_ids}
                  ← reads 'cleaned_data' from parent
                  → writes 'report_A' and 'processed_ids' back to parent

Subgraph B State: {cleaned_data, internal_step_2, report_B, processed_ids}
                  ← reads 'cleaned_data' from parent
                  → writes 'report_B' and 'processed_ids' back to parent

Klucze, które istnieją tylko w podgrafie, takie jak internal_step_1 powyżej, pozostają prywatne, a element nadrzędny nigdy ich nie widzi.

Poniżej znajduje się pełna implementacja:

# ── MODULE 2: STATE SCHEMAS (Three Levels) ──────────────────

# Parent state - what the orchestrating graph tracks
class ParentState(TypedDict):
    raw_input: str
    cleaned_data: str                               # passed INTO both subgraphs
    failure_summary: str                            # written back by subgraph A
    performance_report: str                         # written back by subgraph B
    processed_ids: Annotated[list[int], operator.add]  # BOTH subgraphs write this

# Subgraph A's internal state - includes its private fields + shared keys
class FailureAnalysisState(TypedDict):
    cleaned_data: str        # shared input from parent
    failures: list[str]      # private intermediate state - parent never sees this
    failure_summary: str     # shared output to parent
    processed_ids: Annotated[list[int], operator.add]  # shared output (accumulated)

# Subgraph A's output schema - filters which keys return to parent
# Without this, ALL keys return, which can cause conflicts
class FailureAnalysisOutput(TypedDict):
    failure_summary: str     # only these two keys go back to parent
    processed_ids: Annotated[list[int], operator.add]

# Subgraph B's internal state
class PerformanceAnalysisState(TypedDict):
    cleaned_data: str
    metrics: list[str]          # private
    performance_report: str     # shared output
    processed_ids: Annotated[list[int], operator.add]

class PerformanceAnalysisOutput(TypedDict):
    performance_report: str
    processed_ids: Annotated[list[int], operator.add]

# ── SUBGRAPH A: Failure Analysis ─────────────────────────────
def extract_failures(state: FailureAnalysisState) -> dict:
    failures = [f"Error in {state['cleaned_data']}: timeout"]
    return {"failures": failures, "processed_ids": [1]}
def summarize_failures(state: FailureAnalysisState) -> dict:
    summary = f"Found {len(state['failures'])} failure(s): {state['failures']}"
    return {"failure_summary": summary}
failure_builder = StateGraph(FailureAnalysisState, output=FailureAnalysisOutput)
failure_builder.add_node("extract_failures",  extract_failures)
failure_builder.add_node("summarize_failures", summarize_failures)
failure_builder.add_edge(START, "extract_failures")
failure_builder.add_edge("extract_failures", "summarize_failures")
failure_builder.add_edge("summarize_failures", END)
failure_subgraph = failure_builder.compile()  # compiled → can be used as a node

# ── SUBGRAPH B: Performance Analysis ────────────────────────
def collect_metrics(state: PerformanceAnalysisState) -> dict:
    metrics = [f"Latency: 120ms", "Uptime: 99.8%"]
    return {"metrics": metrics, "processed_ids": [2]}
def write_report(state: PerformanceAnalysisState) -> dict:
    report = f"Performance Report:\n" + "\n".join(state["metrics"])
    return {"performance_report": report}
performance_builder = StateGraph(PerformanceAnalysisState, output=PerformanceAnalysisOutput)
performance_builder.add_node("collect_metrics", collect_metrics)
performance_builder.add_node("write_report",    write_report)
performance_builder.add_edge(START, "collect_metrics")
performance_builder.add_edge("collect_metrics", "write_report")
performance_builder.add_edge("write_report", END)
performance_subgraph = performance_builder.compile()

# ── PARENT GRAPH ─────────────────────────────────────────────
def clean_data_node(state: ParentState) -> dict:
    return {"cleaned_data": state["raw_input"].strip().lower()}
def combine_reports_node(state: ParentState) -> dict:
    combined = f"{state['failure_summary']}\n\n{state['performance_report']}"
    print(f"Final combined report:\n{combined}")
    return {}
parent_builder = StateGraph(ParentState)
parent_builder.add_node("clean_data",           clean_data_node)
parent_builder.add_node("failure_analysis",     failure_subgraph)    # subgraph as node
parent_builder.add_node("performance_analysis", performance_subgraph) # subgraph as node
parent_builder.add_node("combine_reports",      combine_reports_node)
parent_builder.add_edge(START, "clean_data")
# Fan-out to both subgraphs (parallel execution)
parent_builder.add_edge("clean_data",           "failure_analysis")
parent_builder.add_edge("clean_data",           "performance_analysis")
# Fan-in: combine only after both subgraphs complete
parent_builder.add_edge("failure_analysis",     "combine_reports")
parent_builder.add_edge("performance_analysis", "combine_reports")
parent_builder.add_edge("combine_reports", END)
parent_graph = parent_builder.compile()

Schemat wyjściowy, FailureAnalysisOutput oraz PerformanceAnalysisOutput w tym przykładzie, to element, który pomija wiele tutoriali, i właśnie dlatego ludzie napotykają konflikty kluczy stanu. Określa on dokładnie, które pola podgraf wykazuje swojemu rodzicowi, a wszystko, czego brakuje, jest traktowane jako prywatny stan wewnętrzny.

Poziom 4: Send API i map-reduce

Galęzie równoległe i podgrafy obejmują przypadki, gdy liczba gałęzi jest znana już na etapie projektowania. Ale co, jeśli użytkownik chce, aby dziś zbadano 5 tematów, a jutro 20? Kodowanie 20 gałęzi bezpośrednio nie jest możliwe. Wtedy przydaje się wzorzec map-reduce, realizowany za pomocą Send.

Załóżmy wydawcę, który dzieli rękopis na rozdziały. Jeden redaktor otrzymuje rękopis, dzieli go na rozdziały i w tym samym momencie przekazuje każdy z nich innemu korektorowi. Niezależnie od tego, czy jest to dwa, czy dwadzieścia rozdziałów, tylu korektorów zostaje zatrudnionych. Gdy wszyscy skończą pracę, uzyskane wyniki są łączone w ostateczną książkę. Liczba równoległych pracowników jest określana w trakcie wykonywania, a nie podczas projektowania.

Trzy kluczowe słowa są istotne:

• Send(node_name, state), importowane z langgraph.constants, tworzy nową instancję węzła z własnym obciążeniem stanowym. Funkcja routingu, która zwraca listę takich instancji, sprawia, że LangGraph uruchamia je wszystkie jednocześnie, przy czym każdy ma niezależny stan.
• operator.add w polu zbierającym stanowi krok redukcji. Gdy pracownicy kończą pracę, ich wyniki gromadzą się w rosnącej liście w stanie nadrzędnym.

Stan pracownika może być dostosowanym słownikiem, który nie musi odpowiadać stanowi elementu nadrzędnego. Pracownik przenosi jedynie dane dotyczące jego własnego fragmentu pracy.

Oto pełna implementacja map-reduce:

# ── MODULE 2: MAP-REDUCE STATE ──────────────────────────────
from langgraph.constants import Send

class OverallState(TypedDict):
    topic: str
    subjects: list[str]                         # populated by the map step
    jokes: Annotated[list[str], operator.add]   # accumulated by the reduce step
    best_joke: str
class JokeState(TypedDict):
    """The private state each parallel worker gets.
    Notice: this does NOT need to match OverallState."""
    subject: str

# ── MODULE 4: MAP-REDUCE NODES ──────────────────────────────
def generate_subjects(state: OverallState) -> dict:
    """MAP step 1: Expands the topic into a list of subjects to process."""
    prompt = [
        SystemMessage(content="Generate a list of 3 sub-topics related to the given topic. Return as comma-separated values only."),
        HumanMessage(content=state["topic"])
    ]
    response = llm.invoke(prompt)
    subjects = [s.strip() for s in response.content.split(",")]
    return {"subjects": subjects}

def generate_joke(state: JokeState) -> dict:
    """MAP step 2: Worker node. Each instance handles exactly ONE subject.
    Receives its own isolated JokeState - not the full OverallState."""
    prompt = [HumanMessage(content=f"Write a short, funny joke about: {state['subject']}")]
    response = llm.invoke(prompt)
    # Returns a LIST - operator.add in OverallState will accumulate these
    return {"jokes": [response.content]}

def pick_best_joke(state: OverallState) -> dict:
    """REDUCE step: All jokes are now in state['jokes']. Pick the winner."""
    jokes_text = "\n".join([f"{i}. {j}" for i, j in enumerate(state["jokes"])])
    prompt = [
        SystemMessage(content="You are a comedy judge. Pick the funniest joke from the list below. Reply with ONLY the number."),
        HumanMessage(content=jokes_text)
    ]
    response = llm.invoke(prompt)
    winning_index = int(response.content.strip())
    return {"best_joke": state["jokes"][winning_index]}

# ── MODULE 5: THE SEND ROUTER ────────────────────────────────
def continue_to_jokes(state: OverallState):
    """The routing function that LAUNCHES parallel workers.

    Returning a LIST of Send objects (not a string) is what triggers map-reduce.
    Each Send creates an independent, parallel execution of 'generate_joke'
    with its own state payload. The number of parallel workers is determined
    at runtime by how many subjects exist in state.
    """
    return [
        Send("generate_joke", {"subject": subject})
        for subject in state["subjects"]
    ]

# ── MODULE 6: MAP-REDUCE GRAPH ASSEMBLY ─────────────────────
graph_builder = StateGraph(OverallState)
graph_builder.add_node("generate_subjects", generate_subjects)
graph_builder.add_node("generate_joke",     generate_joke)
graph_builder.add_node("pick_best_joke",    pick_best_joke)
graph_builder.add_edge(START, "generate_subjects")
# This conditional edge uses the Send router - it fans out dynamically
graph_builder.add_conditional_edges(
    "generate_subjects",
    continue_to_jokes,   # returns a LIST of Send objects
    ["generate_joke"]    # list of possible destination nodes (for graph validation)
)
# Fan-in: pick_best_joke runs after ALL parallel generate_joke instances complete
graph_builder.add_edge("generate_joke", "pick_best_joke")
graph_builder.add_edge("pick_best_joke", END)
map_reduce_graph = graph_builder.compile()

Cud tkwi w continue_to_jokes. Zwraca on listę obiektów Send zamiast ciągu znaków, a gdy LangGraph otrzymuje taką listę od funkcji routującej, natychmiast uruchamia każdy z nich. Trzy elementy w state["subjects"] oznaczają trzy równoległe węzły generate_joke, a dwadzieścia elementów – dwadzieścia takich węzłów. Graf skaluje się zgodnie z tym, co generuje czas wykonywania.

Poziom 5: asystent badawczy oparty na wielu agentach

Teraz wszystko się łączy: orkiestracja w stylu nadzorcy, zatwierdzenie przez człowieka, równoległe podgrafy oraz API Send – to wszystko w jednym systemie typu produkcyjnego. Jest to projekt końcowy kursu LangGraph Academy, który znajduje się tutaj z adnotacjami.

Zachowanie: użytkownik podaje temat badań. System tworzy zespół postaci analityków AI, a człowiek może je zatwierdzić lub udoskonalić. Każdy analityk działa następnie jako oddzielny agent prowadzący wywiad, zadając pytania „ekspertowi” AI przy użyciu wyszukiwarki internetowej i tworząc jedną sekcję raportu. Na koniec węzeł pisarza łączy te sekcje w ostateczny dokument.

User provides topic
        ↓
[create_analysts] → LLM generates N analyst personas
        ↓
[human_feedback]  ← INTERRUPT: human approves or sends feedback
        ↓         ↑
[should_continue] → if feedback, loop back to create_analysts
        ↓
[initiate_research] → Send API: spawns N parallel interview subgraphs
        ↓↓↓ (N parallel workers running simultaneously)
[interview_subgraph × N] → each analyst interviews an "expert" via web search
        ↓
[write_report]    → LLM compiles all sections into a final report

Stan na trzech poziomach

from pydantic import BaseModel, Field

# ── Analyst persona - a Pydantic model for structured LLM output ──
class Analyst(BaseModel):
    name: str = Field(description="The analyst's full name")
    role: str = Field(description="Their professional role (e.g. 'Financial Analyst')")
    focus: str = Field(description="Their specific analytical focus area")

    @property
    def persona(self) -> str:
        return f"Name: {self.name}\nRole: {self.role}\nFocus: {self.focus}"
class Perspectives(BaseModel):
    analysts: list[Analyst]

# ── Outer graph state ─────────────────────────────────────────
class ResearchGraphState(TypedDict):
    topic: str
    max_analysts: int
    human_analyst_feedback: str      # human writes here during HITL pause
    analysts: list[Analyst]
    sections: Annotated[list, operator.add]  # accumulated from all parallel interviews
    final_report: str

# ── Interview subgraph state ──────────────────────────────────
class InterviewState(MessagesState):
    # inherits: messages: Annotated[list[BaseMessage], add_messages]
    max_num_turns: int
    context: Annotated[list, operator.add]  # search results accumulate here
    analyst: Analyst
    interview: str       # formatted transcript
    sections: list       # completed section (sent back to outer graph)

Tworzenie analityków z ludzką weryfikacją

# ── ANALYST GENERATION NODE ───────────────────────────────────
ANALYST_PROMPT = """You are generating a team of {max_analysts} research analysts
for the topic: {topic}.
{feedback_section}
Each analyst should have a unique perspective and area of focus.
Generate analysts with diverse viewpoints - financial, technical, social, etc."""
def create_analysts(state: ResearchGraphState) -> dict:
    """Generates analyst personas using structured LLM output."""
    feedback = state.get("human_analyst_feedback", "")
    feedback_section = (
        f"User feedback to incorporate:\n{feedback}" if feedback
        else "No feedback yet - generate your best initial set."
    )

    prompt = ANALYST_PROMPT.format(
        max_analysts=state["max_analysts"],
        topic=state["topic"],
        feedback_section=feedback_section
    )

    # with_structured_output: forces LLM to return a Pydantic model
    # instead of free-form text - ensures clean, usable data
    structured_llm = llm.with_structured_output(Perspectives)
    result = structured_llm.invoke([SystemMessage(content=prompt)])
    return {"analysts": result.analysts}

def human_feedback_node(state: ResearchGraphState) -> dict:
    """No-op node. Its only purpose is to be a pause point.
    The graph is compiled with interrupt_before=['human_feedback'].
    See Part 3 for the full HITL pattern."""
    pass  # Does nothing - the pause happens BEFORE this node via interrupt_before

def should_continue_to_research(state: ResearchGraphState) -> Literal["create_analysts", "initiate_research"]:
    """After human feedback: loop back to refine, or proceed to interviews."""
    if state.get("human_analyst_feedback"):
        return "create_analysts"
    return "initiate_research"

Podgraf wywiadu

Każdy analityk działa jako odrębny podgraf, mały agent ReAct, który przeprowadza wywiad z „ekspertem” (innym modelu LLM pełniącym tę rolę) oraz przeszukuje internet.

from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode

@tool
def web_search_tool(query: str) -> str:
    """Search the web for information on a topic."""
    return f"[Search results for '{query}']"  # Replace with real search API
interview_tools = [web_search_tool]
interview_llm = llm.bind_tools(interview_tools)

def generate_question(state: InterviewState) -> dict:
    """The analyst node: generates the next interview question."""
    analyst = state["analyst"]
    messages = [
        SystemMessage(content=(
            f"You are {analyst.name}, a {analyst.role} focused on {analyst.focus}. "
            f"You are interviewing an expert about the research topic. "
            f"Ask insightful questions that align with your analytical focus. "
            f"When you have enough information, say 'Thank you, that is all I needed.'"
        ))
    ] + state["messages"]
    response = interview_llm.invoke(messages)
    return {"messages": [response]}

def generate_answer(state: InterviewState) -> dict:
    """The expert node: answers the analyst's question, using web search."""
    messages = [
        SystemMessage(content=(
            "You are an expert being interviewed. Answer thoroughly and factually. "
            "Use the web search tool when you need current data."
        ))
    ] + state["messages"]
    response = interview_llm.invoke(messages)
    return {"messages": [response]}

def save_interview(state: InterviewState) -> dict:
    """Formats the full Q&A exchange into a clean transcript string."""
    transcript = []
    for msg in state["messages"]:
        if hasattr(msg, "content"):
            role = "Analyst" if type(msg).__name__ == "HumanMessage" else "Expert"
            transcript.append(f"{role}: {msg.content}")
    return {"interview": "\n\n".join(transcript)}

def write_section(state: InterviewState) -> dict:
    """The final node: uses the interview transcript to write a report section."""
    analyst = state["analyst"]
    prompt = [
        SystemMessage(content=(
            f"Based on the following interview conducted by {analyst.name} ({analyst.role}), "
            f"write a structured report section focused on {analyst.focus}. "
            f"Be concise, factual, and cite specific points from the interview."
        )),
        HumanMessage(content=state["interview"])
    ]
    response = llm.invoke(prompt)
    # sections is the key that Send passes back to the outer graph
    return {"sections": [response.content]}

def route_interview(state: InterviewState) -> Literal["generate_answer", "save_interview"]:
    """Continue interviewing OR wrap up when analyst is satisfied or max turns hit."""
    last_message = state["messages"][-1]
    if (
        "thank you, that is all" in last_message.content.lower()
        or len(state["messages"]) >= state.get("max_num_turns", 6) * 2
    ):
        return "save_interview"
    # Check if analyst made a tool call (web search)
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "generate_answer"
    return "generate_answer"

# ── INTERVIEW SUBGRAPH ASSEMBLY ───────────────────────────────
interview_builder = StateGraph(InterviewState)
interview_tool_node = ToolNode(interview_tools)
interview_builder.add_node("generate_question", generate_question)
interview_builder.add_node("generate_answer",   generate_answer)
interview_builder.add_node("interview_tools",   interview_tool_node)
interview_builder.add_node("save_interview",    save_interview)
interview_builder.add_node("write_section",     write_section)
interview_builder.add_edge(START, "generate_question")
interview_builder.add_conditional_edges(
    "generate_question",
    route_interview,
    {"generate_answer": "generate_answer", "save_interview": "save_interview"}
)
interview_builder.add_conditional_edges(
    "generate_answer",
    lambda s: "interview_tools" if (hasattr(s["messages"][-1], "tool_calls") and s["messages"][-1].tool_calls) else "generate_question",
    {"interview_tools": "interview_tools", "generate_question": "generate_question"}
)
interview_builder.add_edge("interview_tools", "generate_answer")
interview_builder.add_edge("save_interview",  "write_section")
interview_builder.add_edge("write_section",   END)
interview_graph = interview_builder.compile()

Graf zewnętrznej orkiestracji

# ── SEND ROUTER: Spawn one interview per analyst ─────────────
def initiate_research(state: ResearchGraphState):
    """Launches N parallel interview subgraphs via Send — one per analyst.
    This is the map-reduce Send pattern from Level 4, applied to subgraphs."""
    return [
        Send("interview", {
            "analyst": analyst,
            "messages": [HumanMessage(content=f"Research topic: {state['topic']}")],
            "context": [],
            "max_num_turns": 3,
            "interview": "",
            "sections": []
        })
        for analyst in state["analysts"]
    ]
# ── REPORT COMPILATION NODE ───────────────────────────────────
def write_report(state: ResearchGraphState) -> dict:
    """Fan-in point: all sections have arrived from parallel interviews.
    The LLM compiles them into a final report."""
    sections_text = "\n\n---\n\n".join(state["sections"])
    prompt = [
        SystemMessage(content=(
            "You are a senior editor. Compile the following report sections written by "
            "different analysts into one cohesive, well-structured final report. "
            "Add a proper introduction and conclusion. Do not repeat content."
        )),
        HumanMessage(content=f"Topic: {state['topic']}\n\nSections:\n{sections_text}")
    ]
    response = llm.invoke(prompt)
    return {"final_report": response.content}

# ── OUTER GRAPH ASSEMBLY ─────────────────────────────────────
outer_builder = StateGraph(ResearchGraphState)
outer_builder.add_node("create_analysts",   create_analysts)
outer_builder.add_node("human_feedback",    human_feedback_node)
outer_builder.add_node("interview",         interview_graph)   # subgraph as node
outer_builder.add_node("write_report",      write_report)
outer_builder.add_edge(START, "create_analysts")
outer_builder.add_edge("create_analysts", "human_feedback")
outer_builder.add_conditional_edges(
    "human_feedback",
    should_continue_to_research,
    {"create_analysts": "create_analysts", "initiate_research": "initiate_research"}
)
# initiate_research is a ROUTING FUNCTION (returns Send objects), not a node
outer_builder.add_conditional_edges(
    "human_feedback",
    initiate_research,   # returns list of Send objects
    ["interview"]        # validation: lists possible destinations
)
outer_builder.add_edge("interview",    "write_report")
outer_builder.add_edge("write_report", END)
research_graph = outer_builder.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["human_feedback"]  # HITL pause for analyst approval
)

Uruchamianie całego procesu

if __name__ == "__main__":
    config = {"configurable": {"thread_id": "research-001"}}

    # Step 1: Generate analysts
    result = research_graph.invoke(
        {
            "topic": "The impact of AI on the future of software engineering",
            "max_analysts": 3,
            "human_analyst_feedback": "",
            "analysts": [],
            "sections": [],
            "final_report": ""
        },
        config=config
    )

    # Step 2: HITL — graph paused at human_feedback node
    snapshot = research_graph.get_state(config)
    analysts = snapshot.values["analysts"]

    print("Generated analysts:")
    for a in analysts:
        print(f"  - {a.name} ({a.role}): {a.focus}")

    feedback = input("\nProvide feedback to refine analysts (or press Enter to approve): ")

    if feedback.strip():
        # Human has feedback — inject it and let supervisor regenerate
        research_graph.update_state(
            config,
            {"human_analyst_feedback": feedback},
            as_node="human_feedback"
        )
    else:
        # Approved — clear feedback and proceed to interviews
        research_graph.update_state(
            config,
            {"human_analyst_feedback": None},
            as_node="human_feedback"
        )

    # Step 3: Resume — N parallel interviews launch via Send, then report is compiled
    print("\nRunning parallel interviews...")
    final_result = research_graph.invoke(None, config=config)

    print("\n" + "="*60)
    print("FINAL RESEARCH REPORT")
    print("="*60)
    print(final_result["final_report"])

Wybór wzorca

• Jeden agent, który myli zadania wymaga wzorca nadzorcy. Nadzorca kieruje przepływem zadań, a specjaliści je wykonują.
• Niezależne, powolne zadania wymagają struktur typu fan-out i fan-in. Wykonywaj je jednocześnie i dodaj reduktory operator.add dla wspólnych pól.
• Węzły równoległe, które same są złożone wymagają podgrafów. Skompiluj każdy węzeł jako odrębny graf, włącz go jako węzeł i użyj schematów wyjściowych, aby kontrolować, co wraca do węzła nadrzędnego.
• Liczba pracowników określana w czasie wykonywania wymaga użycia Send. Router tworzy listę obiektów Send, a każdy z nich uruchamia niezależnego pracownika z własnym stanem.
• Prawdziwy system produkcyjny wykorzystuje wszystko powyższe. Asystent poziomu 5 łączy wszystkie wzorce, ponieważ prawdziwe zadanie wymaga ich wszystkich.

Słowniczek terminów

Struktura nadzorcy. supervisor_node pełni rolę orkiestratora. Czyta stan, zapisuje next_agent i nigdy nie wykonuje zadań specjalistycznych. specialist_node zajmuje się jednym konkretnym zadaniem przy użyciu własnych narzędzi i instrukcji, a następnie przekazuje informacje z powrotem do nadzorcy. next_agent: str to standardowe pole stanu służące do kierowania przepływem danych: nadzorca zapisuje nazwę, a router ją odczytuje.

Paralelizm. Fan-out oznacza kilka wywołań funkcji add_edge z jednego węzła źródłowego, które LangGraph wykonywa równocześnie. Fan-in oznacza kilka krawędzi prowadzących do jednego węzła docelowego, który czeka na każdy z węzłów źródłowych. Funkcja Annotated[list, operator.add] jest obowiązkowa w każdym polu, do którego zapisują się kilka węzłów równoległych, ponieważ bez niej równoległe zapisy powodują awarię, a operator.add jest standardowym reduktorem łączącym listy z różnych węzłów działających jednocześnie.

Podgrafy. StateGraph(InternalState, output=OutputSchema) to forma z dwoma argumentami, przy której drugi argument filtrował klucze, które powracają do grafu nadrzędnego. Schemat wyjściowy to TypedDict zawierający tylko te klucze, które mają być udostępnione, a wszystko inne pozostaje prywatne. Pokrywające się klucze pełnią rolę kanału pomiędzy grafem nadrzędnym a potomnym, ponieważ każdy klucz obecny w obu schematach jest automatycznie udostępniany. Funkcja subgraph.compile() zamienia podgraf na funkcję wywoływalną, która następnie jest przekazywana do parent.add_node("name", compiled_subgraph). Wywołanie graph.get_graph(xray=1).draw_mermaid() pokazuje w diagramie wewnętrzną strukturę podgrafu.

Map-reduce. Send(node_name, state_dict) oznacza jedną paralelną wywołanie pracownika, a jego treść nie musi odpowiadać stanowi nadrzędnemu. Zwrócenie [Send(...), Send(...)] z routera uruchamia je wszystkie. Przy użyciu Send trzeci argument do add_conditional_edges to lista, a nie słownik, zawierająca ważne nazwy destynacji, używana wyłącznie do weryfikacji grafu. Wreszcie with_structured_output(PydanticModel) jest stosowany po wywołaniu modelu, aby zmusić do wygenerowania JSON zgodnego ze schematem Pydantic. To standardowy sposób na uzyskanie wiarygodnych decyzji nadzorczych oraz list analityków.

Kompozycja to kluczowa umiejętność

Przejście od jednego agenta do wielu wcale nie jest wielkim krokiem. Są to te same elementy stosowane wielokrotnie i łączone ze sobą. Każdy agent w systemie wieloagentowym stanowi graf – węzeł lub podgraf – a same podstawowe elementy nigdy się nie zmieniają. Ważne jest umiejętność ich łączenia.

Wybrana ścieżka była celowa: najpierw jeden nadzorca, potem proste, równoległe węzły, równoległe podgrafy dla złożonych gałęzi, dynamiczne elementy sterowane poleceniem Send, a na końcu całość razem. Jeśli któryś z kroków wydawał się naturalny, oznaczało to, że projekt spełnił swój cel, ponieważ każdy poziom spoczywał czysto na poprzednim.

Asystent badawczy na poziomie 5 jest bliski temu, co zawiera kod źródłowy wieloagentowego systemu produkcyjnego: pętle zatwierdzania przez ludzi, specjaliści pracujący równolegle, ustrukturyzowane wyniki modeli, dynamiczne tworzenie pracowników oraz krok końcowej syntezy. W połączeniu z kanoniczną strukturą projektu, zarządzaniem pamięcią oraz zabezpieczeniami zapewniającymi udział człowieka w procesie, te wzorce obejmują architekturę większości rzeczywistych aplikacji LangGraph.