LangGraph-Multi-Agentensysteme: Überwacher, Untergraphen und Senden
LangGraph-Multi-Agentensysteme von einem überlasteten Agenten zu vielen: Supervisor-Routing, Fan-Out und Fan-In, Untergraphen, Senden von Map-Reduce sowie ein Forschungshilfskraft.
Quellen: LangGraph Multi-Agent Systems: From One Brain to Many
Voraussetzungen: Die Beispiele gehen davon aus, dass Sie bereits in der Lage sind, ein Graphenmodell mit einem einzigen Agenten sowie Tools und Speicher zu erstellen, und dass Sie wissen, wie man eine Ausführung mit
interrupt()pausiert und fortsetzt. Alles Weitere baut darauf auf und geht von der Frage aus, warum man mehrere Agenten benötigen könnte, bis hin zu einem vollständigen mehragentigen Forschungshilfsprogramm.
Warum einen Agenten in viele aufteilen?
Nehmen wir an, ein Assistent mit einem einzigen Agenten funktioniert gut. Dann beginnen die Nutzer, ihn zu bitten, E-Mails zu überprüfen, Themen zu recherchieren, Berichte zu schreiben und Aufgaben zu verwalten – alles im selben Gespräch. Die übliche Reaktion besteht darin, weitere Tools hinzuzufügen und mehr Anweisungen in den Systemprompt einzufügen. Bald gibt es 15 Tools sowie einen 2.000-Wörter-langen Prompt, und die Qualität sinkt allmählich. Das Modell beginnt, Dinge durcheinanderzubringen – es greift beispielsweise während einer Recherchetaufegabe zum E-Mail-Tool und während einer E-Mail-Aufgabe zum Recherche-Tool.
Das ist kognitive Überlastung, und sie ist der Hauptgrund für die Existenz von Mehr-Agenten-Designs. Die Lösung ist jene, auf die Softwareingenieure seit Jahrzehnten vertrauen: die Aufgaben aufteilen. In LangGraph behebt dies drei separate Probleme:
- Kognitive Überlastung. Ein Modell muss zu viele unzusammenhängende Aufgaben bewältigen. Spezialisierte Agenten, die jeweils in einer Sache gut sind, verringern den Druck.
- Sequenzielle Engpässe. Arbeiten, die gleichzeitig stattfinden könnten, werden nacheinander ausgeführt. Parallel arbeitende Agenten verringern die Latenz.
- Komplexität. Niemand kann einen Graphen mit 40 Knoten verstehen. Kleinere Untergraphen können einzeln verstanden und getestet werden.
Jedes der unten aufgeführten Muster bezieht sich auf eines dieser drei Probleme.
Fünf Ebenen der Architektur
Betrachten Sie den Fortschritt als eine Treppe, bei der jeder Schritt eine weitere Idee hinzufügt.
Level 1: Supervisor. Ein Supervisor-Agent koordiniert die arbeitenden Mitarbeiter. Es handelt sich weiterhin um einen einzigen Entscheider, doch dieser delegiert Aufgaben anstatt alles selbst zu erledigen, was ihn bereits über einen gewöhnlichen, tool-basierten Chatbot hinaushebt.
Level 2: Parallelisierung. Unabhängige Aufgaben werden aufgeteilt und gleichzeitig ausgeführt, bevor sie wieder zusammengeführt werden. Dabei geht es um Ausbreitung und Zusammenführung: Mehrere Zweige werden gleichzeitig ausgeführt und liefern Ergebnisse für einen späteren Schritt. Dies eignet sich für wiederholende Arbeiten an mehreren Quellen oder Unteraufgaben.
Level 3: Untergraphen. Anstelle eines einzigen, riesigen Arbeitsablaufs wird jeder Zweig zu einem kleineren Graphen mit eigenen internen Schritten und einem eigenen Ausgabeschema. So werden komplexe Teile einfacher zu erstellen, zu testen und zu verstehen.
Level 4: dynamischer Parallelismus. Mit dem Send-Muster, das dem Map-Reduce-Stil folgt, ist die Anzahl der Zweige im Voraus nicht bekannt. Worker werden zur Laufzeit erstellt, jeweils einer pro Element, Thema oder Analyst. Die Agenten sind nicht im Voraus definiert, sie arbeiten alle gleichzeitig, und jeder übernimmt eine spezifische Aufgabe.
Level 5: alles zusammen. Ein realistisches System nutzt alle oben genannten Ansätze gleichzeitig, wie im Fall des Forschungsassistenten am Ende beschrieben.
Eine kurze Erinnerungshilfe: Bei Level 1 laufen die Routen über einen einzigen Supervisor, bei Level 2 arbeitet eine feste Gruppe von Worker gemeinsam, bei Level 3 kann jeder Zweig ein vollständiger Graph sein und bei Level 4 werden Worker nach Bedarf erstellt.
Level 1: Das Supervisor-Muster
Das ist das einfachste Mehr-Agenten-Muster und der richtige Ausgangspunkt. Ein Supervisor erhält die Anfrage des Benutzers und entscheidet, welcher Spezialist sie bearbeiten soll. Der Spezialist erledigt die Arbeit und gibt die Ergebnisse zurück, wobei der Supervisor entscheidet, was als Nächstes geschieht.
Stellen Sie sich eine Anwaltskanzlei vor. Der Seniorpartner spricht mit dem Klienten, erfasst dessen Bedürfnisse und weist Aufgaben an Juniorpartner zu – einen für Verträge, einen für Rechtsstreitigkeiten, einen für Compliance. Der Juniorpartner erledigt die detaillierte Arbeit, der Partner prüft sie und antwortet anschließend dem Klienten.
Der Ablauf sieht so aus:
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
Zustand
Die entscheidende Gestaltungsentscheidung ist das Feld next_agent. Der Supervisor schreibt darin und der Router liest es aus.
# ── 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
Spezialwerkzeuge
Jeder Spezialist erhält seine eigenen, auf bestimmte Aufgaben ausgerichteten Werkzeuge, wobei es keine Überschneidungen zwischen ihnen gibt.
# ── 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)
Node
Danach kommen der Überwachungsnode und die Spezialistennode. Die Antwort des Überwachungsnodes in reinem Text ist selbst die Routing-Entscheidung.
# ── 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)
Routing
# ── 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"
Graphen-Assembly
# ── 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())
Wichtig bei der Zusammenführung ist die Verbindung, die zum Überwachungsprozess zurückkehrt. Wenn ein Spezialist fertig ist, meldet er sich zurück. Der Überwachungsprozess prüft das gesamte Gespräch und wählt anschließend eine von drei Optionen aus: Er weist die Aufgabe an einen anderen Spezialisten weiter, schickt denselben Spezialisten erneut los oder beendet die Aufgabe. Im Kern geht es dabei um Orchestrierung.
Level 2: Parallelisierung mit Fan-Out und Fan-In
Das Überwachungsprozess-Muster ist sequenziell, wobei jeweils nur ein Spezialist gleichzeitig aktiv ist. Viele reale Aufgaben enthalten unabhängige Teilaufgaben, die gleichzeitig ausgeführt werden können, und Parallelisierung löst das Problem der Verzögerung.
Kochen ist eine passende Analogie. Niemand fertigt den Salat vor dem Anbraten der Nudeln und anschließend der Soße ab. Man bereitet den Salat vor, setzt das Wasser auf und beginnt gleichzeitig mit der Soße – so ist das Abendessen schneller fertig.
In LangGraph führt ein Knoten mit mehreren ausgehenden Kanten zu einer parallelen Aufteilung. LangGraph erkennt diese Ausbreitung und führt alle Ziele gleichzeitig aus – was als Superstep bezeichnet wird. Die Ergebnisse werden gesammelt, bevor ein späterer Knoten startet, was die Einströmung darstellt.
Es gibt eine klare Regel: Wenn mehrere Knoten im selben Superstep in dasselbe Zustandsfeld schreiben, muss LangGraph wissen, wie diese Schreibvorgänge zusammengeführt werden. Ohne Reducer tritt ein Fehler auf. Mit operator.add werden die Ergebnisse sicher miteinander verbunden.
# ── 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
Ein Beispiel für parallele Forschung verwendet zwei Quellen:
# ── 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()
Die Fan-Out-Struktur besteht einfach aus zwei Kanten, die vom selben Ausgangspunkt START ausgehen. LangGraph erkennt dies und führt beide Ziele gleichzeitig aus. Auch das Fan-In findet automatisch statt, da synthesize erst dann ausgeführt wird, wenn beide einlaufenden Kanten erfüllt sind – das bedeutet, beide parallelen Knoten sind abgeschlossen.
Level 3: Subgraphen zur Kapselung
Einfache parallele Knoten bewältigen einfache Fälle. Wenn jeder Zweig komplex ist, mehrere Schritte umfasst, eigene Werkzeuge sowie eigene Logik benötigt, sind Subgraphen das bessere Werkzeug. Ein Subgraph ist ein vollständig kompiliertes StateGraph, das als einziger Knoten innerhalb eines Elterngraphen ausgeführt wird.
Stellen Sie sich Abteilungen in einem Unternehmen vor. Der CEO, der Elterngraph, delegiert Aufgaben an die Abteilungen für Ingenieurwesen und Marketing. Jede Abteilung hat interne Prozesse und Meetings, doch der CEO überträgt lediglich Aufgaben und erhält die Ergebnisse.
Elterngraph und Untergraph kommunizieren über gemeinsame Zustandschlüssel. Jeder Schlüssel, der sowohl im Zustand des Elterngraphs als auch im Zustand des Untergraphs vorhanden ist, wird beim Starten des Untergraphs übergeben und beim Beenden wieder zurückgesendet.
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
Schlüssel, die nur im Untergraphen vorhanden sind, wie internal_step_1 oben, bleiben privat, und der Elterngraphen sieht sie niemals.
Eine vollständige Implementierung folgt:
# ── 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()
Das Ausgabeschema, FailureAnalysisOutput und PerformanceAnalysisOutput in diesem Beispiel, ist der Teil, den viele Tutorials überspringen – aus diesem Grund treten dann Konflikte bezüglich der Zustandschlüssel auf. Es filtert genau heraus, welche Felder der Untergraph an seinen Elterngraph freigibt, und alles, was darin fehlt, gilt als privater interner Zustand.
Level 4: die Send API und Map-Reduce
Parallelverzweigungen und Untergraphen kommen dann zum Einsatz, wenn die Anzahl der Verzweigungen bereits zur Entwurfszeit bekannt ist. Doch was passiert, wenn ein Benutzer heute 5 Themen und morgen 20 untersuchen lassen möchte? Die Festlegung von 20 Verzweigungen im Code ist keine Option. Dafür kommt das Map-Reduce-Muster zum Einsatz, das durch Send angetrieben wird.
Stellen Sie sich einen Verlag vor, der Kapitel zuweist. Ein Lektor erhält das Manuskript, teilt es in Kapitel auf und gibt jedes gleichzeitig an einen anderen Korrektor weiter. Egal, ob es zwei oder zwanzig Kapitel sind – genauso viele Korrektoren werden eingestellt. Wenn alle fertig sind, werden die Ergebnisse zum endgültigen Buch zusammengefasst. Die Anzahl der parallel arbeitenden Personen wird zur Laufzeit festgelegt, nicht bereits bei der Planung.
Drei Schlüsselwörter sind wichtig:
Send(node_name, state), importiert auslanggraph.constants, erstellt eine neue Instanz eines Knotens mit eigenem Zustandsinhalt. Eine Routing-Funktion, die eine Liste dieser Knoten zurückgibt, lässt LangGraph alle gleichzeitig starten, wobei jeder seinen eigenen Zustand hat.operator.addim Sammelfeld ist der Reduzier-Schritt. Sobald die Arbeiter fertig sind, sammeln sich ihre Ergebnisse in einer immer länger werdenden Liste im Elternzustand an.
Hier ist die vollständige Implementierung von MapReduce:
# ── 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()
Die Magie liegt in continue_to_jokes. Es gibt eine Liste von Send-Objekten zurück anstelle einer Zeichenkette, und wenn LangGraph eine solche Liste von einer Routenfunktion erhält, startet es sofort alle davon. Drei Einträge in state["subjects"] bedeuten drei parallele generate_joke-Node, und zwanzig Einträge bedeuten zwanzig. Das Graph skaliert entsprechend dem, was die Laufzeit erzeugt.
Level 5: ein mehragentenbasierter Forschungsassistent
Nun wird alles kombiniert: supervisorbasierte Orchestrierung, menschliche Freigabe, parallele Untergraphen sowie die Send-API in einem systemnahen Produktionsumfeld. Es handelt sich um das Abschlussprojekt des LangGraph Academy-Kurses, das hier erläutert wird.
Das Verhalten: Ein Benutzer gibt ein Forschungsthema an. Das System erstellt ein Team aus KI-Analysten-Personas, die von einem Menschen genehmigt oder weiter verfeinert werden können. Jeder Analyst arbeitet anschließend als paralleler Interview-Sub-Agent, befragt einen KI‑„Experten“ mithilfe von Websuche und erstellt einen Abschnitt eines Berichts. Schließlich fasst ein Schreibknoten die Abschnitte zum fertigen Dokument zusammen.
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
Zustand auf drei Ebenen
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)
Erstellung von Analysten unter menschlicher Überprüfung
# ── 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"
Der Interview-Subgraph
Jeder Analyst läuft als eigenständiger Subgraph – ein kleiner ReAct-Agent, der einen „Experten“ (einen weiteren LLM, der diese Rolle übernimmt) befragt und im Internet sucht.
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()
Das äußere Orchestrierungsgraph
# ── 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
)
Das Ganze ausführen
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"])
Ein Muster auswählen
- Ein einzelner Agent, der bei Aufgaben in Verwirrung gerät, erfordert das Supervisor-Muster. Der Supervisor leitet die Aufgaben weiter und die Spezialisten führen sie aus.
- Unabhängige, langsame Aufgaben erfordern Fan-out und Fan-in. Führen Sie sie gleichzeitig aus und fügen Sie Reducer wie
operator.addfür gemeinsame Felder hinzu. - Parallel laufende, selbst komplizierte Zweige erfordern Subgraphen. Kompilieren Sie jeden Zweig als eigenständigen Graphen, integrieren Sie ihn als Knoten und verwenden Sie Ausgabeschemata, um zu steuern, was an den übergeordneten Knoten zurückgegeben wird.
- Die Anzahl der Arbeiter, die zur Laufzeit festgelegt wird, erfordert
Send. Ein Router erstellt eine Liste vonSend-Objekten, wobei jedes ein eigenständiges Arbeiterobjekt mit eigenem Zustand erzeugt.
Schlüsselwortreferenz
Supervisor-Struktur. supervisor_node ist der Orchestrierer. Er liest den Zustand, schreibt next_agent und übernimmt niemals spezialisierte Aufgaben. specialist_node kümmert sich mit eigenen Tools und Anweisungen um eine fokussierte Aufgabe und leitet anschließend die Ausführung zurück an den Supervisor. next_agent: str ist das übliche Zustandsfeld für die Weiterleitung: Der Supervisor schreibt einen Namen ein, und der Router liest ihn ab.
Parallelität. Fan-out bedeutet mehrere add_edge-Aufrufe von einem Quellknoten, die von LangGraph gleichzeitig ausgeführt werden. Fan-in bedeutet mehrere Kanten, die in einen Zielknoten führen, wobei dieser auf jeden Quellknoten wartet. Annotated[list, operator.add] ist auf jedem Feld erforderlich, auf das mehrere parallele Knoten schreiben, da ohne dieses Konstrukt parallele Schreibvorgänge zum Absturz führen, und operator.add ist der übliche Reduzierer, der Listen von gleichzeitig arbeitenden Knoten zusammenfügt.
Untergraphen. StateGraph(InternalState, output=OutputSchema) ist die zweiparametrige Form, bei der der zweite Parameter festlegt, welche Schlüssel an den Elterngraph zurückgegeben werden. Ein Ausgabeschema ist ein TypedDict, das nur die zu offenzlegenden Schlüssel enthält, während alles andere privat bleibt. Überschneidende Schlüssel fungieren als Kanal zwischen Elterngraph und Untergraph, da jeder Schlüssel, der in beiden Schemata vorhanden ist, automatisch geteilt wird. subgraph.compile() wandelt den Untergraph in eine aufrufbare Funktion um, die anschließend an parent.add_node("name", compiled_subgraph) übergeben wird. Durch Aufruf von graph.get_graph(xray=1).draw_mermaid() wird die interne Untergraph-Struktur im Diagramm dargestellt.
Map-reduce. Send(node_name, state_dict) steht für einen parallelen Aufruf eines Workers, wobei die übergebene Datenmenge nicht unbedingt mit dem Zustand des Elternteils übereinstimmen muss. Das Zurückgeben von [Send(...), Send(...)] durch einen Router startet alle Aufrufe gleichzeitig. Bei Verwendung von Send ist das dritte Argument von add_conditional_edges eine Liste – und nicht ein Dictionary – mit gültigen Zielnamen, die ausschließlich zur Überprüfung des Graphen verwendet werden. Schließlich zwingt with_structured_output(PydanticModel) nach einem Modellaufruf dazu, JSON im Einklang mit einem Pydantic-Schema zu erzeugen. Dies ist die übliche Methode, zuverlässige Entscheidungen des Supervisors sowie Analystenlisten zu erhalten.
Komposition ist die Fähigkeit
Der Wechsel von einem Agenten zu vielen ist eigentlich kein großer Sprung. Es handelt sich dabei um dieselben Bausteine, die wiederholt angewandt und miteinander verbunden werden. Jeder Agent in einem Mehr-Agenten-System ist ein Graph – ein Knoten oder ein Teilgraph – und die Primitiven ändern sich niemals. Entscheidend ist, zu wissen, wie man sie kombiniert.
Der hier gewählte Weg war bewusst: zunächst ein Supervisor, danach parallele Knoten, parallele Teilgraphen für komplexe Zweige, dynamische Arbeiter, die durch Send gesteuert werden, und schließlich alles zusammen. Wenn irgendein Schritt intuitiv verständlich war, dann hat das Design seine Aufgabe erfüllt, denn jede Ebene beruht sauber auf der vorherigen.
Der Forschungsassistent auf Stufe 5 kommt dem Inhalt einer produktiven Multi-Agenten-Codebasis nahe: menschliche Genehmigungsprozesse, parallele Spezialisten, strukturierte Modellausgaben, dynamisches Erstellen von Arbeitskraftinstanzen sowie ein abschließender Syntheseschritt. Zusammen mit einer kanonischen Projektstruktur, Speichermanagement und Sicherheitsmechanismen mit menschlicher Beteiligung umfassen diese Muster die Architektur hinter den meisten echten LangGraph-Anwendungen.