Routing, Fan-Out, ReAct, Kritik und Genehmigung: Fünf LangGraph-Muster
Erlernen Sie fünf agierende Workflow-Muster in LangGraph – von Routern und ReAct-Schleifen über Evaluationsgateways bis hin zur menschlichen Freigabe – samt den notwendigen Sicherheitsmaßnahmen für die Einsatz in der Produktion.
Längere Anfragen beheben selten ein unzuverlässiges AI-Feature. Wenn ein System suchen, Code schreiben, Konformitätsprüfungen durchführen oder textliche Inhalte für Kunden bearbeiten muss, ist ein einziger nicht-deterministischer Modellaufruf zu instabil. Eine strukturierte Vorgehensweise hilft: Das Modell trifft Entscheidungen dort, wo Urteilsvermögen erforderlich ist, während der Code die Routen, Schleifen und das Beenden steuert. Im Folgenden sind fünf solcher Muster aufgeführt, jeweils mit einem ausführbaren LangGraph-Beispiel in Python sowie den Hinweisen, die vor der Produktion behoben werden müssen.
Warum Graphen zu Agenten-Arbeitsabläufen passen
Herkömmliche Programme laufen in einer geraden Linie ab. Agenten benötigen Schleifen, bedingte Ausführungspfade sowie einen persistenten Zustand: Wenn der generierte Code eine Prüfung nicht besteht, muss das System den Fehler erfassen, zurückgehen und es erneut versuchen.
LangGraph modelliert dies als gerichteten Graphen:
- Knoten sind Python-Funktionen, die eine einzige Aufgabe ausführen, wie beispielsweise eine SQL-Abfrage oder ein Modellaufruf.
Für eine ausführlichere Einführung in diese Primitiven siehe LangGraph in der Praxis: Zustand, Knoten und Kanten.
Muster 1: Der Router
Ein Router ist ein Klassifizierer am Eingangspunkt. Anstatt alles an ein großes, teures Modell zu senden, leitet ein leichtgewichtiges Element jede Anfrage an ein spezialisiertes Modell, einen Untergraphen oder ein lokales Tool weiter.
┌───> [Specialized Coding Agent] ───> [END]
[START] ──> [Router]
└───> [General Knowledge Agent] ───> [END]
Verwenden Sie ihn, um Latenz und Kosten zu senken oder Intentionen an spezialisierte Tools anzupassen.
Ein kleines Modell bei Temperatur 0 kennzeichnet die Abfrage mit coding oder general, und add_conditional_edges weist die Kennzeichnung auf einen Handler-Node zu. route_decision greift bei unerwarteten Ergebnissen des Modells auf general zurück.
from typing import TypedDict, Literal
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
# 1. Define the shared state
class RouterState(TypedDict):
query: str
route: str
response: str
# Use a fast, cost-effective model for classification
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# 2. Define the Nodes
def classify_query(state: RouterState):
prompt = f"""Classify the following user query into one of two categories: 'coding' or 'general'.
Respond with exactly one word, either 'coding' or 'general'.
Query: {state['query']}"""
response = model.invoke([HumanMessage(content=prompt)])
classification = response.content.strip().lower()
return {"route": classification}
def handle_coding(state: RouterState):
return {"response": "Executing advanced syntax processing and code compilation logic..."}
def handle_general(state: RouterState):
return {"response": "Processing casual conversation or general knowledge search..."}
# 3. Define Conditional Routing Logic
def route_decision(state: RouterState) -> Literal["coding", "general"]:
return state["route"] if state["route"] in ["coding", "general"] else "general"
# 4. Construct the Graph
workflow = StateGraph(RouterState)
workflow.add_node("classifier", classify_query)
workflow.add_node("coding_agent", handle_coding)
workflow.add_node("general_agent", handle_general)
workflow.add_edge(START, "classifier")
workflow.add_conditional_edges("classifier", route_decision, {
"coding": "coding_agent",
"general": "general_agent"
})
workflow.add_edge("coding_agent", END)
workflow.add_edge("general_agent", END)
# Compile and Run
app = workflow.compile()
result = app.invoke({"query": "How do I implement a binary search tree in Python?"})
print(f"Route Taken: {result['route']}\nResponse: {result['response']}")
Die Modellnamen waren zum Zeitpunkt der Erstellung des Beispiels aktuell; ersetzen Sie sie durch die aktuellen Namen Ihres Anbieters. Eine strukturierte Ausgabe ist robuster als das Parsen eines einzigen Wortes.
Muster 2: Orchestrator und Worker
Für Aufgaben, die für eine einzige Anfrage zu umfangreich sind, teilt ein Orchestrator das Ziel in unabhängige Unteraufgaben auf, die Worker führen diese aus und ein Synthesizer fügt die Ergebnisse zusammen.
┌───> [Worker A: Section 1] ───┐
[START] ──> [Orchestrator] ├───> [Worker B: Section 2] ───┼───> [Synthesizer] ───> [END]
└───> [Worker C: Section 3] ───┘
Es eignet sich für langformatige Inhalte wie Berichte sowie für Forschungen mit mehreren Quellen.
Der Orchesterer bittet um eine JSON-Liste mit zwei Unterthemen, der workers-Node schreibt einen Absatz für jedes und der Synthesizer kombiniert sie.
import json
from typing import TypedDict, List
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
class OrchestratorState(TypedDict):
topic: str
tasks: List[str]
worker_outputs: List[str]
final_report: str
model = ChatOpenAI(model="gpt-4o", temperature=0.2)
def orchestrator_plan(state: OrchestratorState):
prompt = f"Create a JSON list of exactly two sub-topics needed to write a comprehensive guide about: {state['topic']}. Return ONLY a valid JSON list of strings."
response = model.invoke([HumanMessage(content=prompt)])
tasks = json.loads(response.content.strip())
return {"tasks": tasks, "worker_outputs": []}
def worker_execute(state: OrchestratorState):
outputs = []
for task in state["tasks"]:
prompt = f"Write a brief, highly technical paragraph explaining: {task}"
response = model.invoke([HumanMessage(content=prompt)])
outputs.append(response.content)
return {"worker_outputs": outputs}
def synthesize_report(state: OrchestratorState):
combined_context = "\n\n".join(state["worker_outputs"])
prompt = f"Combine the following sections into a cohesive newsletter update regarding {state['topic']}:\n\n{combined_context}"
response = model.invoke([HumanMessage(content=prompt)])
return {"final_report": response.content}
# Graph Construction
orchestrator_flow = StateGraph(OrchestratorState)
orchestrator_flow.add_node("orchestrator", orchestrator_plan)
orchestrator_flow.add_node("workers", worker_execute)
orchestrator_flow.add_node("synthesizer", synthesize_report)
orchestrator_flow.add_edge(START, "orchestrator")
orchestrator_flow.add_edge("orchestrator", "workers")
orchestrator_flow.add_edge("workers", "synthesizer")
orchestrator_flow.add_edge("synthesizer", END)
app = orchestrator_flow.compile()
output = app.invoke({"topic": "Quantum Computing Security Implications"})
print(output["final_report"])
Zwei Einschränkungen: Dieser Worker-Node durchläuft die Aufgaben sequenziell, sodass nichts parallel ausgeführt wird; LangGraphs Send-API kann pro Aufgabe einen Worker einsetzen und die Ergebnisse über einen State-Reducer sammeln. Zudem umschließen Modelle manchmal JSON in Markdown-Fences, weshalb der Plan mit strukturiertem Ausgabeformat überprüft werden sollte, anstatt json.loads auf Rohtext zu vertrauen.
Muster 3: ReAct – Denken und Handeln in einem Loop
ReAct wechselt zwischen Denken und Handeln ab: Das Modell bewertet die Situation, ruft ein Tool wie eine Suche oder eine Datenbankabfrage auf, beobachtet das Ergebnis und stoppt, sobald es eine Antwort geben kann.
┌────────────────────────┐
▼ │
[START] ──> [Reasoner (Thought)] ───> (Should Call Tool?) ───> [Tool Executor (Act)]
│
└─ (Has Final Answer) ──> [END]
Es eignet sich für Forschungs-, Support- und Debugging-Agenten, bei denen die benötigten Daten nicht vorhergesagt werden können.
Der Reasoner bittet um ACTION: call_stock_api oder FINAL: ..., einschließlich der letzten Beobachtung. Das Tool gibt einen simulierten Kurs an, und die Rückkehr zum reasoner schließt den Kreislauf. should_continue begrenzt die Anzahl der Iterationen auf drei.
from typing import TypedDict, Literal
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
class ReActState(TypedDict):
user_input: str
agent_thought: str
tool_output: str
final_answer: str
loop_count: int
model = ChatOpenAI(model="gpt-4o", temperature=0)
def reason(state: ReActState):
loop_count = state.get("loop_count", 0) + 1
tool_context = f"\nTool Observation: {state.get('tool_output', '')}" if loop_count > 1 else ""
prompt = f"""You are a ReAct agent. Your goal is to find the current stock price of AAPL.
Current Loop: {loop_count} {tool_context}
Decide your next step. You must respond in one of two ways:
1. If you need data, say: 'ACTION: call_stock_api'
2. If you have the data, provide the answer starting with: 'FINAL: [your answer]'
User Request: {state['user_input']}"""
response = model.invoke([HumanMessage(content=prompt)]).content.strip()
if "FINAL:" in response:
return {"final_answer": response.replace("FINAL:", "").strip(), "loop_count": loop_count, "agent_thought": "done"}
else:
return {"agent_thought": "call_tool", "loop_count": loop_count}
def call_tool(state: ReActState):
print("-> System: Executing external stock database API call...")
mock_api_result = "$185.40 USD (Up 1.2% today)"
return {"tool_output": mock_api_result}
def should_continue(state: ReActState) -> Literal["call_tool", "end"]:
# Hard loop-break guardrail to prevent infinite execution loops
if state["agent_thought"] == "call_tool" and state["loop_count"] < 3:
return "call_tool"
return "end"
react_flow = StateGraph(ReActState)
react_flow.add_node("reasoner", reason)
react_flow.add_node("tool_executor", call_tool)
react_flow.add_edge(START, "reasoner")
react_flow.add_conditional_edges("reasoner", should_continue, {
"call_tool": "tool_executor",
"end": END
})
react_flow.add_edge("tool_executor", "reasoner")
app = react_flow.compile()
result = app.invoke({"user_input": "What is the market status of Apple right now?", "loop_count": 0})
print(f"\nFinal Agent Resolution:\n{result['final_answer']}")
Falls die Obergrenze vor einer FINAL:-Antwort erreicht wird, wird final_answer niemals gesetzt und der letzte print löst einen KeyError aus – daher muss dieser Fall behandelt werden. Reale Systeme verwenden in der Regel direkten Aufruf von Tools anstelle von Zeichenkettenmarkern. Zum Entsprechenden in TypeScript siehe bounded agentic loops for LLM tool use.
Muster 4: Evaluator und Optimierer
Ein Modell, das sein eigenes Werk bewertet, neigt dazu, es unkritisch zu akzeptieren; daher erzeugt und überarbeitet ein Optimierer den Inhalt, während ein separater, strengerer Evaluator Kritik äußert.
┌───> [Optimizer (Generate/Refine)] ───> [Evaluator (Critique)]
│ │
└──────────────── (If Rejected) ────────────────┼───> [Approved] ───> [END
Es eignet sich für das Erstellen von Entwürfen, Codegenerierung, Übersetzung sowie bei strengen Qualitäts- oder Regulierungsanforderungen. Ein günstigeres Modell bei einer Temperatur von 0,7 erstellt einen Slogan und berücksichtigt dabei Rückmeldungen aus späteren Durchgängen. Der Evaluator bei einer Temperatur von 0 prüft, ob der Slogan die Wörter future oder smart enthält, und antwortet in einem festgelegten Format von ACCEPTED / FEEDBACK; routing_gate leitet abgelehnte Ergebnisse zurück.
from typing import TypedDict, Literal
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
class EvaluationState(TypedDict):
task: str
draft: str
feedback: str
accepted: bool
iterations: int
generator_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
evaluator_llm = ChatOpenAI(model="gpt-4o", temperature=0)
def generate_draft(state: EvaluationState):
iterations = state.get("iterations", 0) + 1
feedback_context = f"\nPrevious Feedback to incorporate: {state.get('feedback', '')}" if iterations > 1 else ""
prompt = f"""Write a catchy, 3-sentence marketing slogan for: '{state['task']}'.
{feedback_context}
Provide ONLY the slogan."""
response = generator_llm.invoke([HumanMessage(content=prompt)]).content.strip()
return {"draft": response, "iterations": iterations}
def evaluate_draft(state: EvaluationState):
prompt = f"""Review the following marketing slogan for the product '{state['task']}':
Slogan: "{state['draft']}"
CRITERIA: The slogan must include the exact word 'future' or 'smart'.
Respond in EXACTLY the following format:
ACCEPTED: True or False
FEEDBACK: [If rejected, explain what needs fixing. If accepted, leave blank.]"""
response = evaluator_llm.invoke([HumanMessage(content=prompt)]).content.strip()
accepted = "ACCEPTED: True" in response
feedback = response.split("FEEDBACK:")[-1].strip() if not accepted else ""
return {"accepted": accepted, "feedback": feedback}
def routing_gate(state: EvaluationState) -> Literal["refine", "approve"]:
if state["accepted"] or state["iterations"] >= 3:
return "approve"
return "refine"
eval_flow = StateGraph(EvaluationState)
eval_flow.add_node("generator", generate_draft)
eval_flow.add_node("evaluator", evaluate_draft)
eval_flow.add_edge(START, "generator")
eval_flow.add_edge("generator", "evaluator")
eval_flow.add_conditional_edges("evaluator", routing_gate, {
"refine": "generator",
"approve": END
})
app = eval_flow.compile()
result = app.invoke({"task": "Eco-friendly Electric Skateboards", "iterations": 0})
print(f"Final Slogan: {result['draft']}\nTotal Iterations: {result['iterations']}")
Auch nach drei Iterationen genehmigt das System den Ausgangsinhalt unabhängig davon, sodass dieser möglicherweise abgelehnt wird; das accepted-Flag sollte daher zusammen mit dem Ergebnis gespeichert werden. Eine solche Schlüsselwortregel ist günstiger und zuverlässiger, wenn sie im Code implementiert wird.
Muster 5: Mensch im Prozess
Für risikoreiche Operationen wie das Löschen von Tabellen, das Ausgeben von Geld oder das Versenden von E-Mails an Kunden ermöglicht das Checkpointing es dem Graphen, vor einem sensiblen Knoten anzuhalten, den Zustand zu speichern und auf Genehmigung zu warten.
[START] ──> [Stager] ──> ⛔ (State Saved to DB / Graph Pauses)
│
[DevOps Manager Clicks "Approve"]
│
▼
[Executor (Run Production Deploy)] ──> [END]
Verwenden Sie es für Migrationen, Bereitstellungen, Zahlungen oder Massen-E-Mails.
stager bereitet einen Befehl vor und executor führt ihn aus. Die Kompilierung mit einem Checkpointer und interrupt_before=["executor"] stoppt nach dem Staging. Die Ausführungen werden durch thread_id identifiziert, sodass get_state die gespeicherten Werte sowie den ausstehenden ('executor',)-Schritt anzeigt; update_state protokolliert die Genehmigung und invoke(None, config) setzt die Ausführung fort.
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
class DeploymentState(TypedDict):
command: str
approved: bool
execution_log: str
# 1. Initialize thread checkpoint memory saver
memory = MemorySaver()
def stage_deployment(state: DeploymentState):
print("-> System: Staging server deployment commands...")
return {"command": "sudo systemctl restart production_api"}
def execute_deployment(state: DeploymentState):
print("-> System: Execution approved. Running command on production servers...")
return {"execution_log": f"Successfully executed: {state['command']}"}
hitl_flow = StateGraph(DeploymentState)
hitl_flow.add_node("stager", stage_deployment)
hitl_flow.add_node("executor", execute_deployment)
hitl_flow.add_edge(START, "stager")
hitl_flow.add_edge("stager", "executor")
hitl_flow.add_edge("executor", END)
# CRITICAL: Define the interrupt checkpoint before the executor node runs
app = hitl_flow.compile(checkpointer=memory, interrupt_before=["executor"])
# --- SIMULATING THE ACTIVE DEPLOYMENT WORKFLOW ---
config = {"configurable": {"thread_id": "prod_deploy_001"}}
# 1. Kick off the graph execution
initial_state = app.invoke({"command": "", "approved": False}, config)
# Verify the graph successfully halted its progress
print(f"\n[Current Graph State]: {app.get_state(config).values}")
print(f"[Next Pending Steps]: {app.get_state(config).next}") # Next step will say: ('executor',)
print("\n--- Halting Execution. Waiting for DevOps Manager Review... ---\n")
# 2. Simulate Human Reviewing the State and Updating with Approval
app.update_state(config, {"approved": True}, as_node="stager")
# 3. Resume execution thread seamlessly from the exact checkpoint
final_output = app.invoke(None, config)
print(f"[Final System Output]: {final_output['execution_log']}")
Drei Warnhinweise. MemorySaver arbeitet ausschließlich im Arbeitsspeicher; Pausen, die auch nach einem Neustart bestehen bleiben, benötigen einen durch eine Datenbank unterstützten Checkpointer. executor überprüft niemals den Status approved, daher sollte eine Überprüfung hinzugefügt werden oder eine bedingte Struktur, die den Ablauf bei Ablehnung beendet. Neuere Versionen von LangGraph bieten außerdem eine interrupt()-Funktion – prüfen Sie daher die aktuellen Dokumente nach dem empfohlenen Vorgehen.
Wahl eines Musters
Die Zuverlässigkeit ergibt sich aus der Anpassung der Struktur an das Problem, nicht aus größeren Modellen oder längeren Eingaben:
- Router: Viele Anfragenarten mit unterschiedlichen Kosten oder Fähigkeitsanforderungen.
- Orchestrator und Worker: Eine große Aufgabe, die in unabhängige Teile aufgeteilt wird.
- ReAct: Die benötigten Informationen sind erst zur Laufzeit ermittelbar.
Muster lassen sich kombinieren: Ein Router kann eine ReAct-Agentur anweisen, wobei die endgültige Aktion einer Genehmigung bedarf. Halten Sie überall drei Sicherheitsmaßnahmen ein: Begrenzen Sie jeden Prozesszyklus, überprüfen Sie die vom Code abhängigen Modellausgaben und dokumentieren Sie, ob ein Ergebnis genehmigt wurde oder die Anzahl der Versuche erschöpft ist. Dadurch muss das Modell nicht beim ersten Mal richtig liegen, da der Workflow es ermöglicht, weiterzuleiten, zu testen, erneut zu versuchen und Aufgaben an Menschen abzugeben.
Verwandte Literatur
- Erstellung einer ReAct-Forschungsagentur in LangGraph: Brain, Hands, Router — Lernen Sie, wie man den ReAct-Reason-Act-Observe-Zyklus als Untergraphen in LangGraph implementiert, inklusive erzwungener Reflexion, Iterationsbudgets sowie paralleler Scatter-Gather-Forschung.
- Erstellung einer KI-Agentur von Grund auf: Muster, ReAct und LangGraph — Erfahren Sie die grundlegenden Konzepte hinter KI-Agenturen – Planung, Werkzeugnutzung, Reflexion sowie das ReAct-Muster – und wie LangChain und LangGraph bei der manuellen Erstellung einer solchen Agentur helfen.