LangGraph: Zyklische Workflows und State Management für AI-Agenten erklärt

LangGraph Deep Dive: Zyklische Workflows und State Management für AI-Agenten

LangGraph ist Teil des LangChain-Ökosystems, aber stärker auf graphbasierte Steuerung und explizites State-Management ausgerichtet. Im Gegensatz zu LangChain, das Arbeitsabläufe als geradlinige Ketten (Directed Acyclic Graphs) abbildet, erlaubt LangGraph beliebige gerichtete Graphen mit Schleifen und gemeinsamem Zustand. Dadurch sind komplexe, interaktive Agentenprozesse möglich, bei denen Schritte wiederholt oder in unterschiedlicher Reihenfolge ausgeführt werden können. LangGraph wurde explizit für zustandsbehaftete Multi-Agenten-Anwendungen konzipiert.

Graph-basierte Agentensteuerung mit LangGraph

In LangGraph fließt ein zentrales State-Objekt durch den Graphen. Jeder Knoten ist eine Python-Funktion, die das aktuelle state (z.B. als TypedDict oder Pydantic-Modell) empfängt und manipuliert. Man definiert einen Workflow beispielsweise so:

from langgraph.graph import StateGraph, MessagesState, START, END

def mock_llm(state: MessagesState) -> MessagesState:
    # Beispiel: KI-Modell-Antwort hinzufügen
    state.messages.append({"role": "ai", "content": "Hello!"})
    return state

# Graph mit dem Zustandsmodell MessagesState anlegen
graph = StateGraph(MessagesState)
graph.add_node("mock_llm", mock_llm)     # Knoten hinzufügen
graph.add_edge(START, "mock_llm")        # Start -> mock_llm
graph.add_edge("mock_llm", END)          # mock_llm -> Ende

# Graph kompilieren und ausführen
app = graph.compile()
result = app.invoke({"messages": [{"role": "user", "content": "Hi!"}]})

In diesem einfachen Beispiel wird nach dem Startknoten mock_llm ausgeführt und dann das Ende erreicht. Alle Knoten greifen auf dasselbe state zu, das zwischen den Aufrufen weitergegeben wird. Das state-Objekt kann komplexe Datenstrukturen enthalten – Nachrichtenlisten, Variablen, Zwischenergebnisse – die von jedem Knoten gelesen und erweitert werden können.

Zyklische Workflows und Conditional Edges

Abstrakte Darstellung zyklischer Workflows und bedingter Verzweigungen

LangGraph unterstützt zyklische Abläufe und bedingte Verzweigungen über Conditional Edges. Mit add_conditional_edges lässt sich festlegen, welcher Knoten nach einem Knoten ausgeführt wird, je nach Zustand. Angenommen, das State enthält eine boolesche Variable flag. Eine Routing-Funktion leitet dann je nach Wert weiter zum Knoten „positive" oder „negative":

def decide(state):
    return "positive" if state.get("flag", False) else "negative"

graph.add_node("positive", handle_positive)
graph.add_node("negative", handle_negative)
graph.add_conditional_edges("start", decide, {True: "positive", False: "negative"})

decide(state) prüft das flag-Feld. Ist es True, geht es zum Knoten „positive", sonst zu „negative". Man kann beliebig viele Zielknoten und sogar den END-Knoten eintragen. Intern erzeugt LangGraph dadurch einen flexiblen Workflow-Zweig, der abhängig vom Laufzeit-Zustand verzweigt und auch Schleifen ermöglicht. Durch solche Bedingungen sind if/else-Logiken direkt im Graphen umsetzbar und erlauben robuste, adaptive Steuerung des Agentenablaufs.

Zustandsverwaltung und Checkpointing

Ein zentrales Merkmal von LangGraph ist das explizite State-Management. Der gemeinsame Zustand (state) bleibt über den gesamten Agentenlauf persistiert. Für langlaufende oder unterbrechbare Prozesse bietet LangGraph eine Checkpointing-Funktion. Mit create_checkpoint(state) lässt sich an beliebigen Stellen ein Schnappschuss des aktuellen Zustands erstellen:

from langgraph.checkpoint import create_checkpoint

def process_with_checkpoint(state):
    # ... (Arbeitslogik hier)
    checkpoint = create_checkpoint(state)      # Zustand speichern
    return {"checkpoint": checkpoint, "state": state}

graph.add_node("process", process_with_checkpoint)

Später kann man mit load_checkpoint(checkpoint) den gespeicherten Zustand wiederherstellen. Dies ermöglicht es, auf Eingriffe oder Systemunterbrechungen zu reagieren: Der Agent kann an einem Checkpoint anhalten, der Entwickler prüft den Zustand und fährt danach an derselben Stelle fort. Die Workflow-Ausführung ist dadurch resilient gegenüber Fehlern und Pausen.

Menschliche Kontrolle (Human-in-the-Loop)

LangGraph unterstützt auch Human-in-the-Loop. Man kann Knoten definieren, die den aktuellen Zustand visualisieren und manuelle Eingaben abwarten. Beispielsweise könnte ein human_input-Knoten wie folgt aussehen:

def human_input(state):
    print("Zustand:", state)
    feedback = input("Geben Sie hier Anweisungen ein: ")
    state["human_feedback"] = feedback
    return state

graph.add_node("human_input", human_input)

Außerdem lassen sich bedingte Kanten nutzen, um – etwa basierend auf einer Unsicherheitsmetrik – zu entscheiden, ob menschliches Eingreifen nötig ist. So kann der Ablauf bei niedriger Vertrauenswahrscheinlichkeit in einen „Mensch"-Knoten verzweigen, andernfalls automatisch fortgesetzt werden. Dadurch bleiben komplexe Abläufe steuerbar, weil der Mensch bei Bedarf eingreifen und das gemeinsame State-Objekt manuell korrigieren kann.

Multi-Agenten-Orchestrierung und Tool-Calling

Vernetzte Knoten zur Darstellung der Multi-Agenten-Orchestrierung

LangGraph eignet sich besonders für Multi-Agenten-Setups. Man kann in einem Graphen verschiedene Knoten unterschiedlichen Spezialisten oder Werkzeugen zuordnen. Übliche Muster sind etwa Pipeline-Workflows (Schritt-für-Schritt) oder Hub-and-Spoke (Koordinator-Knoten verteilt Aufgaben). Dank des geteilten Zustands können Agenten einfach kooperieren – jeder Knoten liest und erweitert das gemeinsame Kontext-Objekt.

Ein typisches Beispiel sind Tool-Calling-Agenten: Hier bindet man externe Werkzeuge (z.B. Kalkulator, Such-API) ein und orchestriert sie im Graphen:

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain.tools import tool

# Zustand mit Konversationen, internem Gedächtnis etc.
class AgentState(TypedDict):
    messages: list
    ...

# LLM und Tools initialisieren
llm = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools([calculator, wikipedia])
tool_node = ToolNode([calculator, wikipedia])

# Agenten-Knoten: LLM-Aufruf
async def call_model(state: AgentState):
    response = llm.invoke(state["messages"])
    state["messages"].append(response)
    return state

# Routing-Entscheidung: Weiterleitung ans Tool oder Ende
def should_continue(state: AgentState) -> str:
    last = state["messages"][-1]
    if last.tool_calls:
        return "tools"
    return "end"

# Graph aufbauen
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": END})
workflow.add_edge("tools", "agent")
app = workflow.compile()

Der Knoten "agent" repräsentiert den LLM-Abruf, "tools" den Ausführungsmechanismus für zurückgegebene Funktionsaufrufe. Über eine bedingte Kante wird nach jedem LLM-Call überprüft, ob ein Tool-Aufruf vorliegt – falls ja, wird der Tool-Executor ausgeführt, andernfalls gelangt man zum Ende.

Fazit

LangGraph erweitert die gewohnten LLM-Frameworks um explizites State-Management und flexible, graphbasierte Steuerung. Zyklische Abläufe, bedingte Verzweigungen, Wiederaufnahme durch Checkpointing und Human-in-the-Loop-Unterbrechungen lassen sich damit leicht realisieren. Dies macht LangGraph ideal für robuste, mehrstufige Agentensysteme, bei denen verschiedene Modelle oder Tools zusammenarbeiten und den Workflow je nach Situation dynamisch anpassen müssen.

In der Praxis erstellt man einen StateGraph, fügt Knoten für einzelne Verarbeitungsschritte hinzu und verknüpft diese über Kanten und Bedingungen. So entstehen aus Bausteinen völlig neue Orchestrierungs-Pattern, die weit über lineare Prompt-Ketten hinausgehen. Mit LangGraph baut man ausführbare, visuelle Agenten-Flows, die aufwändige Szenarien handhaben können – von Dialogsystemen bis zu automatisierten Analyse-Pipelines – und dabei stets den Überblick über den gemeinsamen Zustand behalten.