langgraph python agents workflows human-in-the-loop

LangGraph in Python: Zustände, parallele Knoten, Checkpoints und menschliche Überprüfung

Ein praktischer LangGraph-Überblick in Python: TypedDict-Zustände und Reduzierer, Kanten, Checkpoint-Funktionen, Unterbrechungen sowie ein E-Mail-Triage-Workflow mit menschlicher Freigabe.

Verlag: REACTAPP.TOPAuf dieser Website hinzugefügt 28. Okt. 202625 Min. Lesezeit5723 Wörter

Quellen: LangGraph Essentials in Python: Build AI Agent Workflows with State, Routing, and Human-in-the-Loop

Jeder, der über ein Spielzeug hinaus etwas auf der Basis eines LLM entwickelt hat, kennt drei wiederkehrende Probleme. Nacheinander durchgeführte Anrufe führen zu einer zunehmenden Latenz. Ein langer Workflow kann abstürzen und dabei alle bisherigen Fortschritte zerstören. Zudem liefert das Modell zum schlimmstmöglichen Zeitpunkt eine selbstsichere Antwort, die völlig falsch ist.

LangGraph behebt all diese Probleme. Es handelt sich um ein auf Graphen basierendes Orchestrierungsframework, das aus wenigen einfachen Komponenten besteht: Der Zustand wird als TypedDict gespeichert, Knoten sind gewöhnliche Funktionen und Kanten stellen die Routing-Logik dar. Diese Primitiven bilden zusammen Workflows, die Knoten parallel ausführen, den Zustand speichern, damit nach einem Absturz weitergearbeitet werden kann, und bei hohem Risiko eine Pause einlegen, damit ein Mensch die Situation überprüfen kann.

Die folgenden Abschnitte gehen Schritt für Schritt vor. Zunächst kommt ein einzelner Knoten, danach die parallele Ausführung, bedingte Routenbildung und persistente Speicherung, gefolgt von Unterbrechungen durch einen Menschen im Prozess. Am Ende kombiniert ein vollständiger Workflow zur Sortierung von E-Mails all diese Elemente. Jedes Beispiel soll unverändert ausgeführt werden. Das TypeScript SDK stellt nahezu dieselben APIs zur Verfügung, sodass derselbe Denkansatz dort ebenfalls gilt.

Einrichtung

Zuerst sind zwei Pakete erforderlich:

pip install langgraph langchain-openai

Dann die Umgebungsvariablen für OpenAI und, optional, das LangSmith-Tracking:

export OPENAI_API_KEY="your-key-here"
# Optional: enable LangSmith tracing for debugging
export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY="your-langsmith-key"

Die Beispiele wurden mit langgraph 1.1.3 (März 2026) und langchain-openai 1.1.10 getestet und erfordern Python 3.10 oder neuer. Ältere Anleitungen verwenden manchmal LANGCHAIN_TRACING_V2 und LANGCHAIN_API_KEY; beide funktionieren weiterhin, doch LANGSMITH_TRACING und LANGSMITH_API_KEY sind nun die bevorzugten Namen.

Alles in LangGraph basiert auf zwei Grundideen: Der Zustand speichert die Daten, und Knoten sind Funktionen, die diese Daten transformieren. Sobald man sich damit vertraut gemacht hat, fügt sich der Rest des Frameworks von selbst zusammen.

Zustand mit TypedDict definieren

Der Zustand in LangGraph ist ein einfacher Python TypedDict – es gibt keine speziellen Basisklassen oder framework-spezifischen Dekoratoren, nur typisierte Wörterbücher:

from typing import Annotated
from typing_extensions import TypedDict
import operator

class State(TypedDict):
    messages: Annotated[list[str], operator.add]
    summary: str

Die Klasse definiert die Struktur der durch das Graph fließenden Daten, und jeder Knoten liest von dieser gemeinsamen Struktur und schreibt darauf.

Warum ein TypedDict und nicht ein gewöhnliches Dictionary oder eine Dataclass? Es ermöglicht eine statische Typüberprüfung, da sowohl mypy als auch Pyright es verstehen, zusammen mit Typhinweisen, und es vermeidet den Laufzeit-Overhead einer Dataclass oder eines Pydantic-Modells. LangGraph kann außerdem die Anmerkungen prüfen, um das Schema zu ermitteln, was für strukturierte Ausgaben und Validierung wichtig ist.

Die Zeile, die man verstehen muss, ist Annotated[list[str], operator.add], mit der ein Reducer deklariert wird. Dies ist das wichtigste Konzept im Zustand von LangGraph und führt fast jeden beim ersten Mal in Verwirrung.

Ohne Reduktor ersetzt ein Knoten, der {"summary": "new value"} zurückgibt, einfach den gespeicherten Wert. Für skalare Felder wie summary ist das genau das gewünschte Verhalten. Probleme entstehen, wenn zwei parallele Knoten beide in messages schreiben: Nur derjenige, der zuletzt fertig wird, bleibt erhalten. Durch die Deklaration von operator.add weist man LangGraph an, hinzuzufügen, sodass jeder Beitrag zu messages sicher akkumuliert wird.

Eine komplexere Mergelogik erfordert einen benutzerdefinierten Reduktor:

def keep_latest_n(existing: list[str], new: list[str]) -> list[str]:
    """Keep only the last 10 messages."""
    combined = existing + new
    return combined[-10:]

class State(TypedDict):
    messages: Annotated[list[str], keep_latest_n]

Dieser Reducer fügt die vorhandene Liste mit den neuen Nachrichten zusammen und kürzt sie anschließend auf die zehn neuesten Einträge, was bei langen Konversationen hilft, bei denen die Tokenanzahl sonst in die Höhe schießen würde. operator.add dient zum einfachen Hinzufügen. Ein benutzerdefinierter Reducer ist die Wahl, wenn Duplikate beseitigt, die Listegröße begrenzt oder eine andere Art der Zusammenführung als das einfache Hinzufügen gewünscht ist.

Der klassische Fehler besteht darin, bei einem Feld, auf das parallele Knoten schreiben, einen Reducer zu vergessen. Ohne einen überlagern sich die Zweige stillschweigend, der letzte, der fertig wird, hat Vorrang, und es tritt niemals ein Fehler auf.

TypedDict versus Pydantic

Sowohl werden in LangGraph verwendet, und die Trennung zwischen ihnen ist beabsichtigt. TypedDict dient dem Graphenzustand, dem Container, der sich durch den Graphen bewegt. LangGraph wurde um dieses Konzept herum entworfen – seine Felder unterstützen die Annotated-Reducer-Syntax, es serialisiert sauber zur Speicherung von Checkpoints und verursacht keinen Laufzeitaufwand.

Pydantic wird für Schemata verwendet, die mit dem LLM interagieren. Durch Aufruf von .with_structured_output(SomeModel) erhält das LLM ein JSON-Schema, an dem es sich orientieren kann, und es wird überprüft, ob die Antwort darin passt. Eine nicht passende Antwort löst sofort einen Fehler aus und verhindert, dass fehlerhafte Daten durch den Graphen weitergeleitet werden.

Der abschließende Workflow folgt genau dieser Aufteilung: EmailState ist ein TypedDict, das als Graphen-Container dient, während EmailClassification ein Pydantic-Modell ist, das die Ausgabe des LLMs beschreibt.

Erster Knoten schreiben

Jeder Knoten nimmt den vollständigen Zustand als Eingabe entgegen und gibt ein Dictionary mit einer teilweisen Aktualisierung zurück. LangGraph fügt diese Aktualisierung zum Zustand hinzu und wendet alle deklarierten Reduzierer an:

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END

class ChatState(TypedDict):
    messages: Annotated[list[str], operator.add]

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def chatbot(state: ChatState) -> dict:
    """Call the LLM with the conversation history."""
    response = llm.invoke(state["messages"])
    return {"messages": [response.content]}

# Build the graph
graph = StateGraph(ChatState)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)
app = graph.compile()
# Run it
result = app.invoke({"messages": ["What is LangGraph?"]})
print(result["messages"][-1])

Überblick über den Code:

StateGraph(ChatState) erstellt ein Graph, das ChatState als Schema verwendet, und LangGraph liest die Anmerkungen, um etwas über Reduzierer und Feldtypen zu erfahren.
graph.add_node("chatbot", chatbot) registriert die Funktion unter dem Namen "chatbot", auf den sich die Kantendefinitionen beziehen.
graph.add_edge(START, "chatbot") macht "chatbot" zum ersten zu ausführenden Knoten. START ist ein eingebauter Sentinel.
graph.add_edge("chatbot", END) beendet die Ausführung, nachdem chatbot abgeschlossen hat. END ist ein weiterer eingebauter Sentinel.
graph.compile() überprüft die Struktur, prüft auf hängende Kanten und unerreichbare Knoten und gibt einen ausführbaren CompiledGraph zurück.

app.invoke({"messages": [...]}) führt den Graphen synchron aus und gibt den endgültigen Zustand zurück.

Der Knoten umhüllt seinen String in eine Liste, da messages operator.add als Reducer verwendet. Das Zurückgeben von {"messages": [response.content]} fügt eine Liste mit einem Eintrag zur bestehenden Liste hinzu, während das Zurückgeben des einfachen Strings zur Laufzeit fehlschlagen würde, weil operator.add keinen String zu einer Liste hinzufügen kann.

Die Regel besteht darin, das, was ein Knoten zurückgibt, mit dem Typ des Feldes abzugleichen – einschließlich des Reducers. Ein Feld vom Typ list[str] mit operator.add erwartet ebenfalls einen Wert vom Typ list[str], während ein einfaches str-Feld ohne Reducer einen Wert vom Typ str erwartet.

Das ist eine vollständige, funktionierende LangGraph-Anwendung in etwa 20 Zeilen. Was innerhalb von invoke() geschieht, ist die nächste Frage.

Super-Schritte: Wie die Ausführung funktioniert

LangGraph führt Knoten in Super-Schritten aus. Jeder Knoten, der für denselben Schritt geplant ist, wird gleichzeitig ausgeführt, und alle müssen vor Beginn des nächsten Schritts abgeschlossen sein. Das Ein-Knoten-Beispiel umfasst einen Super-Schritt mit einem Knoten. Durch das Hinzufügen paralleler Kanten werden mehrere Knoten in denselben Super-Schritt gebracht.

Das ist aus zwei Gründen wichtig. Die parallele Ausführung ist von Anfang an in das Modell integriert und nicht später hinzugefügt. Zudem speichert LangGraph mit einem Checkpointer nach Abschluss jedes Super-Schritts ein vollständiges Zustandsbild. Dadurch wird eine Wiederherstellung nach einem Absturz möglich: Wenn der Prozess unterwegs abstirbt, setzt die Ausführung an dem letzten abgeschlossenen Super-Schritt fort.

Nun, da der Zustausströmung durch einen einzelnen Knoten verstanden wurde, stellt sich die nächste Frage: Wie kann man die Reihenfolge steuern oder Knoten gleichzeitig ausführen? Dafür sind Kanten zuständig.

Kanten: seriell, parallel und bedingt

Kanten steuern, wie die Ausführung zwischen Knoten erfolgt. LangGraph bietet drei Arten, die zusammen alle möglichen Routing-Muster abdecken. Die einfachste Form wurde bereits im Chatbot-Beispiel gezeigt.

Serialisierte Kanten

Im einfachsten Fall folgt ein Knoten auf den nächsten.

class DraftState(TypedDict):
    topic: str
    draft: str
    review: str

def generate_draft(state: DraftState) -> dict:
    response = llm.invoke(f"Write a short paragraph about: {state['topic']}")
    return {"draft": response.content}

def review_draft(state: DraftState) -> dict:
    response = llm.invoke(f"Review this draft and suggest improvements:\\n{state['draft']}")
    return {"review": response.content}

graph = StateGraph(DraftState)
graph.add_node("generate", generate_draft)
graph.add_node("review", review_draft)
graph.add_edge(START, "generate")
graph.add_edge("generate", "review")
graph.add_edge("review", END)

app = graph.compile()
result = app.invoke({"topic": "Why Python developers love type hints"})
print(result["review"])

generate_draft wird zuerst ausgeführt. Sobald es abgeschlossen ist und LangGraph das Feld draft in den Zustand integriert hat, wird review_draft ausgeführt, das den Entwurf aus state["draft"] lesen kann. Eine serielle Abfolge ist garantiert, da die Kante die Abhängigkeit vorschreibt und kein Sequenzierungscode erforderlich ist. Serielle Kanten kommen immer dann zum Tragen, wenn Knoten B von der Ausgabe von Knoten A abhängt: erst Erstellen, dann Überprüfen, erst Abrufen, dann Verarbeiten, erst Klassifizieren, dann Weiterleiten.

Zufällige Zyklen sind eine häufige Falle. Standardmäßig erwartet LangGraph einen gerichteten azyklischen Graphen, weshalb das Hinzufügen einer Kante von review zurück zu generate zu einem Fehler bei der Kompilierung führt, es sei denn, der Graph ist ausdrücklich als zyklisch konfiguriert. Iterative Schleifen werden in der Regel mit bedingten Kanten erstellt, die zurück zu einem früheren Knoten geleitet werden können.

Parallelisierte Kanten

Wenn zwei Kanten von einem Knoten ausgehen, werden ihre Ziele gleichzeitig im selben Super-Schritt verarbeitet. Reduzierer werden hier unerlässlich:

class ResearchState(TypedDict):
    topic: str
    findings: Annotated[list[str], operator.add]
    outline: str
    final_summary: str

def research(state: ResearchState) -> dict:
    response = llm.invoke(f"List 3 key facts about: {state['topic']}")
    return {"findings": [f"Research: {response.content}"]}

def create_outline(state: ResearchState) -> dict:
    response = llm.invoke(f"Create a brief outline for an article about: {state['topic']}")
    return {"outline": response.content}

def combine(state: ResearchState) -> dict:
    research_text = "\\n".join(state["findings"])
    response = llm.invoke(
        f"Combine this research and outline into a summary:\\n"
        f"Research:\\n{research_text}\\n\\n"
        f"Outline:\\n{state['outline']}"
    )
    return {"final_summary": response.content}

graph = StateGraph(ResearchState)
graph.add_node("research", research)
graph.add_node("outline", create_outline)
graph.add_node("combine", combine)
# Fan-out: both start after START
graph.add_edge(START, "research")
graph.add_edge(START, "outline")
# Fan-in: both feed into combine
graph.add_edge("research", "combine")
graph.add_edge("outline", "combine")
graph.add_edge("combine", END)
app = graph.compile()
result = app.invoke({"topic": "LangGraph parallel execution"})
print(result["final_summary"])

Zwei Kanten von START planen research und outline in derselben Überschritt-Ebene, und LangGraph führt sie gleichzeitig aus. Beide haben außerdem eine Kante zu combine, wobei LangGraph combine erst dann ausführt, wenn alle zu ihm führenden Knoten abgeschlossen sind – dies fungiert als automatische Synchronisierungsbarriere.

findings verfügt über einen Reducer, obwohl hier nur research darauf schreibt, denn ein echtes System könnte vier parallele Forschungsagenten haben, die alle in dieselbe Liste schreiben; ohne Reducer würde nur ein Ergebnis übrig bleiben.

Die Berechnung der Latenz ist einfach. Wenn research 2 Sekunden und outline 3 Sekunden dauert, benötigt die serielle Version 5 Sekunden und die parallele Version 3 Sekunden – also die Länge der längeren Aufgabe. Bei unabhängigen LLM-Aufrufen führt die parallele Ausführung zu nahezu kostenlosen Latenzeinsparungen.

Kandidaten für die Parallelisierung sind alle Knotelpaare, deren Ausgaben voneinander unabhängig sind. Selbst das Verschieben zweier einsekündiger Aufrufe in denselben Super-Schritt hat bei Tausenden von Anfragen Auswirkungen. Achten Sie auf Knoten, die scheinbar unabhängig erscheinen, aber auf weniger offensichtliche Weise denselben Zustand teilen. Wenn zwei parallele Knoten ohne Reducer in ein Feld schreiben, entsteht eine Wettbedingung – der letzte, der fertig wird, bestimmt den Wert, und LangGraph gibt keine Warnung.

M dieses Fan-out- und Fan-in-Musters bildet die Grundlage des E-Mail-Projekts, bei dem vor der Erstellung einer Antwort eine Suchanfrage in einer Wissensdatenbank sowie eine Abfrage in einem Fehlerverfolgungssystem parallel ausgeführt werden.

Bedingte Kanten

Wenn der nächste Knoten von dem abhängt, was der aktuelle Knoten erzeugt hat, wird add_conditional_edges angewendet:

class ClassifyState(TypedDict):
    text: str
    sentiment: str
    result: str

def classify(state: ClassifyState) -> dict:
    response = llm.invoke(
        f"Classify the sentiment of this text as 'positive' or 'negative'. "
        f"Reply with just the word.\\n\\n{state['text']}"
    )
    return {"sentiment": response.content.strip().lower()}

def handle_positive(state: ClassifyState) -> dict:
    return {"result": f"Great feedback! Text was: {state['text']}"}

def handle_negative(state: ClassifyState) -> dict:
    return {"result": f"Escalating concern. Text was: {state['text']}"}

def route_by_sentiment(state: ClassifyState) -> str:
    """Routing function: inspect state, return a string key."""
    if state["sentiment"] == "positive":
        return "positive"
    return "negative"

graph = StateGraph(ClassifyState)
graph.add_node("classify", classify)
graph.add_node("handle_positive", handle_positive)
graph.add_node("handle_negative", handle_negative)
graph.add_edge(START, "classify")
graph.add_conditional_edges(
    "classify",
    route_by_sentiment,
    {"positive": "handle_positive", "negative": "handle_negative"}
)
graph.add_edge("handle_positive", END)
graph.add_edge("handle_negative", END)
app = graph.compile()
result = app.invoke({"text": "This product is absolutely fantastic!"})
print(result["result"])
# Great feedback! Text was: This product is absolutely fantastic!

Die Funktion route_by_sentiment prüft den Zustand und gibt einen String-Schlüssel zurück, wobei das Mapping-Dictionary, das an add_conditional_edges übergeben wird, diesen Schlüssel in einen Knotennamen umwandelt. Eine rein funktionierende Routenfunktion (Eingabe: Zustand, Ausgabe: String, keine Nebeneffekte) ist leicht zu testen, da sie ohne Einbeziehung einer KI unitgetestet werden kann.

add_conditional_edges nimmt drei Argumente entgegen: den Quellknoten, die Routenfunktion sowie ein Dictionary, das String-Schlüssel mit Zielknotennamen verknüpft. Jeder Schlüssel, den die Routenfunktion zurückgeben könnte, muss in diesem Dictionary vorhanden sein; andernfalls wirft LangGraph zur Laufzeit einen Fehler aus.

Die Alternative Command

Wenn ein Knoten den Zustand aktualisieren und gleichzeitig routen muss, kann er direkt einen Command zurückgeben und add_conditional_edges überspringen:

from langgraph.types import Command

def classify_and_route(state: ClassifyState) -> Command:
    response = llm.invoke(
        f"Classify sentiment as 'positive' or 'negative'.\\n\\n{state['text']}"
    )
    sentiment = response.content.strip().lower()
    if sentiment == "positive":
        return Command(update={"sentiment": sentiment}, goto="handle_positive")
    else:
        return Command(update={"sentiment": sentiment}, goto="handle_negative")

Ein Rückgabewert von Command enthält sowohl eine Zustandsaktualisierung (update=) als auch eine Routing-Entscheidung (goto=), sodass der Knoten in einem einzigen Schritt Daten schreibt und das nächste Ziel auswählt.

add_conditional_edges ist die richtige Wahl, wenn die Routing-Logik nur den Zustand liest und nichts Neues schreibt, wenn das Routing von der Knotenlogik getrennt bleiben soll oder wenn mehrere Quellknoten dieselbe Routing-Funktion teilen.

Command ist die richtige Wahl, wenn der Knoten neue Daten erzeugt, die sein eigenes Routing beeinflussen, wenn ein einmaliger Fall keine separate Routing-Funktion rechtfertigt oder wenn das Ziel von etwas abhängt, das innerhalb des Knotens berechnet wird und noch nicht im Zustand enthalten ist.

Für das menschliche Überprüfungsverfahren stützt sich die E-Mail-Verarbeitung auf add_conditional_edges, da die Routierung dort rein ist: Es wird nur die bereits im Zustand gespeicherte Priorität und Zuverlässigkeit überprüft.

Speicher und Checkpointer

Sobald Ordnung und Verzweigungen unter Kontrolle sind, stellt sich das nächste Problem der Persistenz: Was passiert, wenn der Prozess mitten in einem Gespräch neu gestartet wird?

Bis jetzt begann jede Aufrufung von Null aus. Gespräche und mehrschrittige Workflows benötigen einen Zustand, der zwischen den Aufrufen erhalten bleibt. LangGraph bewältigt dies mit Checkpointern, und die API ist so sauber strukturiert, dass der Wechsel von der Entwicklung zur Produktiv-Persistenz durch eine einzige Änderung erfolgt.

MemorySaver für die Entwicklung

MemorySaver speichert den Zustand im Speicher, was sich für Entwicklung und Testing eignet:

from langgraph.checkpoint.memory import MemorySaver

class ConversationState(TypedDict):
    messages: Annotated[list[str], operator.add]

def chatbot(state: ConversationState) -> dict:
    history = "\\n".join(state["messages"])
    response = llm.invoke(f"Conversation so far:\\n{history}\\n\\nRespond to the latest message.")
    return {"messages": [response.content]}

graph = StateGraph(ConversationState)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)

# Compile WITH a checkpointer
memory = MemorySaver()
app = graph.compile(checkpointer=memory)

# Thread 1: a conversation about Python
config_1 = {"configurable": {"thread_id": "user-alice"}}
app.invoke({"messages": ["Hi, I'm learning Python"]}, config=config_1)
result = app.invoke({"messages": ["What should I learn first?"]}, config=config_1)
print(result["messages"][-1])

# The response knows about the previous message because state persisted
# Thread 2: a completely independent conversation
config_2 = {"configurable": {"thread_id": "user-bob"}}
result = app.invoke({"messages": ["Tell me about Rust"]}, config=config_2)
print(result["messages"][-1])
# This conversation has no knowledge of Alice's conversation

Mit checkpointer=memory speichert das kompilierte Graph nach jedem Super-Step ein vollständiges Snapshot. Ein erneuter Aufruf von invoke() mit derselben thread_id lässt LangGraph zunächst den neuesten Checkpoint für diese Thread laden. Die thread_id funktioniert wie eine Session-ID: Wiederholen Sie deren Verwendung, um fortzusetzen, ändern Sie sie, um von vorne anzufangen.

Der zweite Aufruf kann frühere Nachrichten weglassen, da der Checkpointer diese bereits enthält. Wenn Alice mit {"messages": ["What should I learn first?"]} aufruft, lädt LangGraph den Checkpoint "user-alice", findet die gespeicherte Liste und fügt den neuen Eintrag über den Reducer operator.add hinzu. Sobald der chatbot läuft, enthält state["messages"] die gesamte Historie.

Die Wiederverwendung derselben thread_id für unzusammenhängende Gespräche ist ein Fehler. Wenn "user-alice" für ein anderes Thema wiederverwendet wird, bringt der Chatbot den gesamten vorherigen Gesprächsverlauf von Alice mit, weshalb jede neue Sitzung eine neue thread_id erhalten sollte. Während der Entwicklung zeigt app.get_state_history(config) den vollständigen Historieverlauf eines Threads an, und derselbe Mechanismus wird in LangSmith zur Fehlerbehebung durch Zeitreisen genutzt.

Belastbare Checkpointer für die Produktion

MemorySaver befindet sich im Prozessgedächtnis, sodass der Zustand mit dem Ende des Prozesses verloren geht. Für die Produktion ist ein belastbarer Backend erforderlich:

# Install: pip install langgraph-checkpoint-postgres
from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:pass@localhost:5432/mydb"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # Creates tables on first run
    app = graph.compile(checkpointer=checkpointer)
    # Same API, durable state
    result = app.invoke({"messages": ["Hello"]}, config=config_1)

checkpointer.setup() erstellt bei der ersten Ausführung die erforderlichen Datenbanktabellen und ist danach wirkungslos, weshalb es während der Initialisierung nur einmal aufgerufen werden muss. Die API ändert sich zwischen den Backend-Systemen nicht, da die Checkpointer hinter einer gemeinsamen Abstraktion liegen. MemorySaver, PostgresSaver und RedisSaver folgen demselben Protokoll, sodass der Graphen-Code nicht erkennen kann, mit welchem von ihnen er kommuniziert.

Wahlmöglichkeiten:

MemorySaver eignet sich für Entwicklung, Testing und Demos, bei denen es nicht darauf ankommt, nach einem Neustart weiterhin verfügbar zu sein.
PostgresSaver ist geeignet für Produktivumgebungen, die Zuverlässigkeit sowie die Möglichkeit zur Abfrage der Checkpoint-Geschichte benötigen.
RedisSaver aus langgraph-redis passt zu Hochleistungsanwendungen, die schnelle Lesevorgänge von Checkpoints benötigen und mit den Einschränkungen der Persistenz bei Redis leben können.

Was gespeichert wird

Bei jedem Super-Step wird ein vollständiger Snapshot gespeichert. Dadurch wird Zeitreisen ermöglicht: Der Zustand zu jedem Zeitpunkt in der Historie kann überprüft, von einem ausgewählten Checkpoint wiedergegeben oder zur Fehlerbehebung genutzt, um genau herauszufinden, wo etwas schiefgelaufen ist. Derselbe Mechanismus bildet die Grundlage für Unterbrechungen. Wenn ein Knoten interrupt() aufruft, benötigt LangGraph einen Ort, an dem der pausierte Zustand gespeichert werden kann, damit er später wieder aufgenommen werden kann. Ohne Checkpointer gibt es nichts, das ihn aufrechterhält, und das Unterbrechungsmuster bricht zusammen.

Mensch im Prozess mit Unterbrechungen

Einige Entscheidungen sind zu wichtig, um automatisiert zu werden. Bevor ein Agent eine E-Mail an einen Kunden sendet, eine Rückerstattung genehmigt oder Code bereitstellt, sollte eine Person dies überprüfen und genehmigen. LangGraph erledigt das mit der interrupt()-Funktion.

Die interrupt()-Funktion

Durch das Einfügen von interrupt() innerhalb eines Nodes wird die Ausführung gestoppt, der Zustand in den Checkpointer geschrieben und ein Payload an den Aufrufer zurückgegeben:

from langgraph.types import interrupt, Command

class ReviewState(TypedDict):
    messages: Annotated[list[str], operator.add]
    draft: str
    approved: bool

def generate(state: ReviewState) -> dict:
    response = llm.invoke("Write a short thank-you email to a customer.")
    return {"draft": response.content}

def review(state: ReviewState) -> dict:
    """Pause here for human review."""
    decision = interrupt(f"Please review this draft:\\n\\n{state['draft']}")
    if decision == "approve":
        return {"approved": True}
    return {"approved": False}

def send(state: ReviewState) -> dict:
    if state["approved"]:
        return {"messages": ["Email sent successfully."]}
    return {"messages": ["Email was rejected. Not sent."]}

Nach und nach führt der Aufruf vier Schritte aus:

Er stoppt die Ausführung genau an der Zeile, an der er aufgerufen wird.
Er speichert den vollständigen Zustand im Checkpointer – deshalb ist dieser erforderlich.
Er gibt den übergebenen Payload, hier den Entwurftext, an die Stelle zurück, die invoke() aufgerufen hat.
Er wartet und lässt das Graphen in einem suspendierten Zustand.

Die Variable decision erhält zunächst nichts, denn interrupt() gibt erst dann etwas zurück, wenn das Graphen mit Command(resume=value) wieder fortgesetzt wird. Zu diesem Zeitpunkt erhält decision den Inhalt, der mit resume= übergeben wurde.

Der Payload informiert den menschlichen Prüfer darüber, was er betrachtet, und kann daher alles enthalten, was benötigt wird: den Entwurf, eine Zusammenfassung oder ein strukturiertes Dictionary mit Kontextinformationen. Der Aufrufer erhält ihn als Teil der Unterbrechungsbenachrichtigung.

Fortsetzen mit Command

Das Fortsetzen nach einer Unterbrechung bedeutet, das Diagramm erneut mit Command(resume=value) aufzurufen:

memory = MemorySaver()

graph = StateGraph(ReviewState)
graph.add_node("generate", generate)
graph.add_node("review", review)
graph.add_node("send", send)
graph.add_edge(START, "generate")
graph.add_edge("generate", "review")
graph.add_edge("review", "send")
graph.add_edge("send", END)
app = graph.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "email-review-1"}}

# Step 1: Invoke the graph. It will run generate, then pause at review.
result = app.invoke({"messages": ["Start email workflow"]}, config=config)
# result contains the interrupt payload with the draft

# Step 2: The human reviews the draft and approves.
result = app.invoke(Command(resume="approve"), config=config)
print(result["messages"][-1])
# "Email sent successfully."

In dem ersten Schritt ruft LangGraph die Funktion generate auf, speichert den Entwurf im Zustand und gelangt zum Punkt interrupt() im Knoten review. Anschließend erstellt es einen Zustandscheckpunkt und gibt diesen zurück; der result aus dieser ersten invoke()-Aufruf enthält den Unterbrechungspayload sowie den Entwurf für den Prüfer. Im zweiten Schritt lässt Command(resume="approve") LangGraph den Zustandscheckpunkt für "email-review-1" laden, bei der Aufrufstelle interrupt() fortsetzen und "approve" als Rückgabewert übergeben. Der decision-Wert des Knotens review wird auf "approve" gesetzt, approved=True wird festgelegt und die Ausführung geht anschließend zum Schritt send.

Beide Aufrufe müssen denselben config verwenden, da LangGraph über thread_id den richtigen Checkpoint findet. Ein anderer thread_id beim zweiten Aufruf würde auf einen nicht existierenden Checkpoint verweisen, wodurch der Aufruf fehlschlägt.

Ein Checkpointer ist bei Unterbrechungen zwingend erforderlich. Ohne ihn hat der pausierte Zustand keinen Ort, an den er wechseln kann, weshalb app.compile(checkpointer=memory) hier nicht optional ist.

Seit LangGraph 1.1.0 akzeptiert invoke() den Parameter version="v2" und gibt einen GraphOutput zurück, der die Attribute .value und .interrupts enthält. Dadurch wird eine sicherere Handhabung von Unterbrechungen ermöglicht, die man einmal, nachdem man sich mit den Grundlagen vertraut gemacht hat, ausprobieren sollte. Ein Unterbrechungspayload kann jeder serialisierbare Wert sein – sei es eine Zeichenkette, ein Dictionary oder eine Liste. In Produktumgebungen werden in der Regel strukturierte Dictionaries übergeben, die alles enthalten, was ein Prüfer benötigt, wie zum Beispiel Entwurf, Klassifizierung, Absender sowie die Prüfungs-URL. Die Anwendungsschicht kann diese Informationen in einer Benutzeroberfläche anzeigen oder über Slack oder E-Mail weiterleiten.

Es gibt noch eine weitere Fallgrube. Die erneute Aufrufung von invoke() mit dem vollständigen Anfangszustand anstelle von Command(resume=value) ist ein Fehler: app.invoke({"messages": [...]}, config=config) auf derselben thread_id startet das Diagramm für diesen Thread erneut und ignoriert den pausierten Zustand. Ein unterbrochenes Diagramm sollte nur mit Command(resume=value) fortgesetzt werden.

Capstone: Ein Workflow zur E-Mail-Triage

Sobald alle Komponenten vorhanden sind, können sie zu einem in der Produktion eingesetzten System zur E-Mail-Triage zusammengeführt werden. Dieses System liest eine eingehende E-Mail, klassifiziert den Zweck mithilfe strukturierter Ausgaben, durchsucht gleichzeitig eine Wissensdatenbank und einen Fehlertracker, erstellt einen Entwurf für eine Antwort und kann vor dem Versand zur menschlichen Überprüfung angehalten werden.

Jedes der früheren Konzepte entspricht einem Teil dieses Workflows:

Der TypedDict-Zustand mit Reduzern erscheint als EmailState, wobei operator.add auf search_results angewendet wird.
Seriale Kanten verbinden read_email mit classify_intent.
Parallelle Kanten ermöglichen es classify_intent, sich in search_docs und bug_tracking aufzuteilen.
Fan-in-Synchronisierung bedeutet, dass beide parallelen Knoten write_response versorgen.
Die bedingte Routierung erfolgt über route_by_priority, welches zwischen menschlicher Überprüfung und automatischer Übermittlung wählt.
Die strukturierte Ausgabe von Pydantic ist EmailClassification, die die Antwort des LLM einschränkt.
Die Unterbrechung erfolgt über human_review, wodurch eine Pause zur Genehmigung entsteht.
Der Checkpointer ist MemorySaver, der den Zyklus aus Unterbrechung und Wiederaufnahme ermöglicht.

Zustand und Schemata

Zuerst kommen der State und das Pydantic-Schema für die strukturierte Klassifizierung:

from typing import Annotated
from typing_extensions import TypedDict
import operator
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import MemorySaver

# Pydantic model for structured classification
class EmailClassification(BaseModel):
    intent: str = Field(description="The intent: 'bug_report', 'question', 'feedback', or 'other'")
    priority: str = Field(description="Priority level: 'high', 'medium', or 'low'")
    confidence: float = Field(description="Confidence score between 0.0 and 1.0")

# Graph state
class EmailState(TypedDict):
    email_subject: str
    email_body: str
    email_sender: str
    classification: dict
    search_results: Annotated[list[str], operator.add]
    bug_info: str
    draft_response: str
    approved: bool
    status: str
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

EmailClassification ist ein Pydantic-Modell und gehört nicht zu EmailState, welches die Trennung zwischen TypedDict und Pydantic darstellt. Es handelt sich dabei um die Struktur, die dem LLM als JSON-Schema-Ziel übergeben wird, wobei Pydantic die Antwort validiert und sofort einen Fehler auslöst, falls sie nicht übereinstimmt. EmailState hingegen ist der Graph-Container, ein TypedDict, da LangGraph die Reducer-Syntax sowie eine effiziente Serialisierung benötigt.

classification wird in EmailState als dict gespeichert, weil Pydantic die Antwort validiert hat und .model_dump() sie vor der Speicherung im Zustand in ein einfaches Dict umwandelt. Dadurch bleibt der Zustand von jedem Checkpointer serialisierbar, ohne dass zur Wiederherstellung eine Pydantic-Abhängigkeit erforderlich ist.

Das wichtigste Feld ist search_results: Annotated[list[str], operator.add]. Da search_docs und bug_tracking parallel laufen und beide Ergebnisse schreiben können, sammelt der Reducer ihre Ausgaben zusammen, während ohne ihn ein Zweig den anderen stillschweigend überschreiben würde.

Erstellung der Knoten

Knoten bleiben klein und auf eine Aufgabe beschränkt. Jeder holt nur die benötigten Werte aus dem Zustand, erledigt seine Aufgabe und gibt eine teilweise Aktualisierung zurück:

def read_email(state: EmailState) -> dict:
    """Parse the incoming email into structured fields."""
    return {
        "email_subject": state["email_subject"],
        "email_body": state["email_body"],
        "email_sender": state["email_sender"],
    }


def classify_intent(state: EmailState) -> dict:
    """Use structured output to classify the email."""
    structured_llm = llm.with_structured_output(EmailClassification)
    classification = structured_llm.invoke(
        f"Classify this email.\\n\\n"
        f"Subject: {state['email_subject']}\\n"
        f"Body: {state['email_body']}"
    )
    return {"classification": classification.model_dump()}

def search_docs(state: EmailState) -> dict:
    """Search knowledge base for relevant documentation."""
    intent = state["classification"]["intent"]
    # In production, this would hit a real vector store or search API
    response = llm.invoke(
        f"Suggest 2 relevant documentation links for a {intent} "
        f"about: {state['email_subject']}"
    )
    return {"search_results": [f"[Docs] {response.content}"]}

def bug_tracking(state: EmailState) -> dict:
    """Look up related bugs in the tracking system."""
    # In production, this would query Jira, Linear, GitHub Issues, etc.
    response = llm.invoke(
        f"Summarize any known bugs related to: {state['email_subject']}"
    )
    return {"bug_info": response.content}

def write_response(state: EmailState) -> dict:
    """Draft an email response using all gathered context."""
    context_parts = [
        f"Original email from {state['email_sender']}:",
        f"Subject: {state['email_subject']}",
        f"Body: {state['email_body']}",
        f"\\nClassification: {state['classification']}",
        f"\\nRelevant docs: {chr(10).join(state['search_results'])}",
        f"\\nBug info: {state['bug_info']}",
    ]
    context = "\\n".join(context_parts)
    response = llm.invoke(
        f"Draft a helpful, professional reply to this customer email.\\n\\n

Auf den ersten Blick scheint read_email sinnlos zu sein, da es nur Felder zurückgibt, die bereits im Zustand vorhanden sind, doch es dient als expliziter Parseschritt. Eine Produktionsversion würde einen rohen webhook-Payload entgegennehmen und die Felder extrahieren, wodurch die hier vorhandene Funktion diese Extraktion ersetzt.

classify_intent ist der Punkt, an dem Pydantic nützlich wird. In structured_llm = llm.with_structured_output(EmailClassification) wird der LLM so umhüllt, dass die Antworten dem Schema entsprechen müssen. Er erhält ein JSON-Schema, das aus EmailClassification abgeleitet wird, und muss entsprechendes, gültiges JSON zurückgeben. LangChain versucht automatisch erneut, bevor ein Fehler ausgelöst wird, falls dies nicht geschieht. .model_dump() wandelt das validierte Objekt anschließend in ein einfaches Dict für den Zustand um.

Die beiden parallelen Knoten schreiben in separate Felder, search_results und bug_info, sodass ihre LLM-Aufrufe gleichzeitig ausgeführt werden. Für search_results gibt search_docs eine Liste zurück ([f"[Docs] {response.content}"]), da operator.add Liste-Eingaben benötigt. bug_tracking gibt für bug_info einen einfachen String zurück; da es keinen Reducer gibt, reicht eine gewöhnliche Zuweisung aus.

Der Fan-In-Knoten ist write_response. Wenn er startet, sind beide parallelen Knoten bereits abgeschlossen und ihre Ergebnisse befinden sich im Zustand, wodurch er den Prompt aus allem zusammenstellt und die Antwort entwirft.

human_review übermittelt an interrupt() ein strukturiertes Dictionary und nicht einen einfachen String, was eine gute Produktionspraxis darstellt. Das Dictionary enthält den Entwurf, die Klassifizierung sowie eine Nachricht für den Prüfer, sodass das System auf der anderen Seite – egal ob eine Web-Oberfläche, ein Slack-Bot oder eine E-Mail-Benachrichtigung – den gesamten Kontext erhält.

send_reply verwendet state.get("approved", True) als schützenden Standardwert; falls der Wert approved in dem Zustandsobjekt aus irgendeinem Grund fehlt, wird die Antwort dennoch gesendet. Dieser Standardwert kann auf False geändert werden, wenn das Risikoprofil vorsichtiger ist.

In der Produktion würden search_docs und bug_tracking echte Systeme wie eine Vektordatenbank, Jira oder GitHub Issues aufrufen, und der LangChain @tool-Decorator macht es einfach, diese Aufrufe zu verpacken. Die Knotenstruktur bleibt unverändert, wobei ein Tool-Aufruf den LLM-Aufruf ersetzt. Noch eine Fallstrick: write_response liest state[„search_results“] ab, sodass ein Zugriff fehlschlagen kann, wenn dies weiterhin eine leere Liste ist. Dies wird vermieden, indem search_results: [] im initialen Aufruf-Payload initialisiert wird, damit der Reducer von dort an anfängt hinzuzufügen.

Verbindung der Kanten

Serielle, parallele und bedingte Kanten verbinden anschließend die Knoten:

def route_by_priority(state: EmailState) -> str:
    """Route based on priority and confidence."""
    classification = state["classification"]
    if classification["priority"] == "high" or classification["confidence"] < 0.8:
        return "needs_review"
    return "auto_send"


graph = StateGraph(EmailState)

# Add all nodes
graph.add_node("read_email", read_email)
graph.add_node("classify_intent", classify_intent)
graph.add_node("search_docs", search_docs)
graph.add_node("bug_tracking", bug_tracking)
graph.add_node("write_response", write_response)
graph.add_node("human_review", human_review)
graph.add_node("send_reply", send_reply)

# Serial edges
graph.add_edge(START, "read_email")
graph.add_edge("read_email", "classify_intent")

# Parallel fan-out from classify_intent
graph.add_edge("classify_intent", "search_docs")
graph.add_edge("classify_intent", "bug_tracking")

# Parallel fan-in to write_response
graph.add_edge("search_docs", "write_response")
graph.add_edge("bug_tracking", "write_response")

# Conditional routing after write_response
graph.add_conditional_edges(
    "write_response",
    route_by_priority,
    {"needs_review": "human_review", "auto_send": "send_reply"}
)

# After human review, send the reply
graph.add_edge("human_review", "send_reply")
graph.add_edge("send_reply", END)

# Compile with memory for interrupt support
memory = MemorySaver()
app = graph.compile(checkpointer=memory)

route_by_priority sendet E-Mails mit hoher Priorität zur manuellen Überprüfung, unabhängig vom Vertrauensgrad. Klassifizierungen mit einem Vertrauensgrad unter 0,8 gehen ebenfalls zur Überprüfung, denn wenn das Modell sich nicht sicher ist, worum es in einer E-Mail geht, sollte eine Person entscheiden. Alles Weitere wird automatisch gesendet, was ein vernünftiger Standard für die Produktion ist, der anhand der tatsächlichen Fehlerraten angepasst werden sollte.

Diese Routenbildung erfolgt über add_conditional_edges und nicht über Command, weil route_by_priority eine reine Funktion ist: Sie prüft lediglich den Zustand und gibt einen Schlüssel zurück, ohne neue Daten zu erzeugen – genau das ist auch der Zweck von add_conditional_edges.

Hier sind alle Verbindungen im Graphen:

Ein serieller Ablauf beginnt bei START, geht über read_email zu classify_intent.
Die Ausführung von classify_intent startet gleichzeitig sowohl search_docs als auch bug_tracking.
Die Rückgabe der Ergebnisse dieser beiden Knoten wird an write_response weitergeleitet.
Nach write_response wählt eine bedingte Verbindung entweder human_review oder send_reply aus.
Sobald die Überprüfung abgeschlossen ist, gibt human_review die Daten an send_reply weiter, welches schließlich bei END endet.

Ausführung des Workflows von Anfang bis Ende

Zuerst wird eine Beispiel-E-Mail eingegeben, und die erste Ausführung läuft bis zum Unterbruch weiter:

config = {"configurable": {"thread_id": "email-001"}}

# Invoke with a sample email
result = app.invoke({
    "email_subject": "Login page broken after update",
    "email_body": (
        "Hi, since yesterday's update I can't log into my account. "
        "I get a 500 error every time. This is blocking my team. "
        "Please fix ASAP."
    ),
    "email_sender": "frustrated-customer@example.com",
    "search_results": [],
    "bug_info": "",
    "draft_response": "",
    "approved": False,
    "status": "",
}, config=config)

# The graph will classify this as high-priority, so it pauses at human_review.
# The interrupt payload contains the draft for review.
print("Interrupt payload received. Draft response ready for review.")

Bis zu diesem Zeitpunkt hat das Graph bereits viel Arbeit geleistet:

read_email hat die Nachricht analysiert.
classify_intent hat sie als Fehlerbericht hoher Priorität mit strukturiertem Pydantic-Ausgabeformat gekennzeichnet.
Die Dokumentensuche sowie die Abfrage im Fehlerverfolgungssystem wurden parallel innerhalb eines einzigen Schritts ausgeführt.
write_response hat eine Antwort entworfen, die all diesen Kontext nutzte.
Der Knoten human_review hat interrupt() aufgerufen und die Ausführung pausiert.

Der gesamte Verarbeitungspfad, einschließlich beider paralleler LLM-Aufrufe, war bereits vor der Pause abgeschlossen. Wenn das Graph anhält, wird der Zustand vollständig gespeichert, sodass der Prüfer Minuten oder Stunden Zeit hat, zu antworten, ohne etwas zu verlieren.

Anschließend genehmigt der Prüfer den Bericht:

# Resume after human approval
final = app.invoke(Command(resume="approve"), config=config)
print(final["status"])
# "Reply sent to frustrated-customer@example.com"

LangGraph lud den Checkpoint für "email-001" hoch, setzte die human_review-Prozedur nach dem Aufruf von interrupt() fort, gab "approve" als Rückgabewert zurück, setzte approved=True im Zustand und ging anschließend zu send_reply über, welches den Bestätigungsstatus zurückgab.

Jede frühere Idee kommt im Endprodukt zum Tragen: Zustand mit reducer-basiertem TypedDict, serielle und parallele Kanten, bedingte Routen, strukturierte Ausgabe mit Pydantic, menschliche Überprüfung durch Unterbrechung sowie durch einen Checkpointer gespeicherte Daten.

Für E-Mails mit niedriger Priorität, bei denen die Zuverlässigkeit hoch ist, gibt route_by_priority "auto_send" zurück und human_review wird übersprungen. Der Ablauf erfolgt in einem durchgehenden Prozess: Lesen, Klassifizieren, Suchen parallel, Entwurf erstellen und Versenden – die Unterbrechung tritt nur auf dem Weg auf, der sie benötigt. Das ist der Vorteil der bedingten Routierung.

Eine abschließende Warnung betrifft den Anfangszustand. Im Gegensatz zu Pydantic legt TypedDict keine Standardwerte für fehlende Felder fest. Wenn write_response auf state["search_results"] zugreift und der Schlüssel nicht im Anfangszustand vorhanden ist, tritt ein KeyError auf. Jedes Feld muss im Anfangszustands-Dict enthalten sein – auch wenn es zunächst als leere Zeichenkette, leere Liste oder False vorhanden ist.

Was als Nächstes?

Der hier beschrittene Weg beginnt mit einem einzigen Knoten und einem TypedDict und endet mit einem für die Produktion geeigneten Workflow, der LLM-Aufrufe nebeneinander ausführen, strukturierte Ausgaben überprüfen, bedingungsweise abzweigen, durch Checkpoints Krisen überstehen und auf einen Überprüfer warten kann. Von dort aus gibt es mehrere Weiterentwicklungsmöglichkeiten:

Die LangGraph Platform kümmert sich um die Bereitstellung, das verwaltete Checkpointing sowie das horizontale Skalieren, wodurch die Last des Betriebs der Infrastruktur entfällt, wenn das Ziel der Umgang mit echtem Traffic ist.
LangSmith verfolgt und debuggt jede Graphenausführung und zeigt den genauen Zustand bei jedem Schritt an, wenn sich ein Knoten falsch verhält. Dies wird wichtig, sobald die Graphen fünf oder mehr Knoten mit bedingten Pfaden aufweisen.
Subgraphen kombinieren kleinere Graphen zu größeren Mehr-Agenten-Systemen. Jeder kümmert sich um seinen eigenen Zustand und kommuniziert über einen Elterngraphen – ein natürlicher Schritt nach Workflows mit einzelnen Graphen.
Das über LangChain mittels @tool und ToolNode aufgerufene Tool ermöglicht es Agenten, externe APIs, Datenbanken und Dienste direkt aus dem Graphen heraus zu nutzen, wobei die Schlüsselknoten search_docs und bug_tracking offensichtliche Orte sind, um dies auszuprobieren.
Die Persistenz in der Produktion erfordert die Installation von langgraph-checkpoint-postgres, um einen dauerhaften Zustand zu gewährleisten, der auch nach Neustarts erhalten bleibt, und der API-Austausch erfolgt in nur einer Zeile.

Kurzer Überblick

Der Zustand ist ein TypedDict. Jedes Feld, das als Annotated[list, operator.add] definiert ist, sammelt Werte über mehrere Schreibvorgänge hinweg, während ein Feld ohne Reduktor bei jedem Schreibvorgang ersetzt wird.
Ein Knoten ist lediglich eine Funktion: Er erhält den vollständigen Zustand und gibt ein Teil-Update-Dict zurück, wobei LangGraph die Kombination vornimmt.
Die Kanten bestimmen den Ablauf. Verwenden Sie serielle Kanten, wenn Reihenfolge wichtig ist, parallele Kanten für unabhängige Aufgaben, die gleichzeitig ausgeführt werden, sowie bedingte Kanten, wenn die Daten die richtige Abzweigung wählen sollen.
Verwenden Sie ein TypedDict für den Graphenzustand und Pydantic mit .with_structured_output(), um festzulegen, was das LLM zurückgibt.
Checkpointer sorgen für Persistenz und ermöglichen Unterbrechungen: MemorySaver während der Entwicklung, PostgresSaver in der Produktion. Ohne Checkpointer keine Unterbrechungen.
Eine Unterbrechung ist nichts anderes als ein gespeicherter, pausierter Zustand. interrupt(payload) stoppt den Ablauf und erstellt einen Checkpoint, während Command(resume=value) genau an dieser Stelle fortsetzt.
Parallelität ist bereits integriert: Zwei Kanten, die von einem Knoten ausgehen, führen ihre Ziele gleichzeitig aus, und jedes Feld, das diese Knoten schreiben, benötigt einen Reducer.

Fragen, über die man nachdenken sollte

MemorySaver eignet sich für die Entwicklung und PostgresSaver für die Produktion. Welche anderen Speicherdienste verdienen bei einem Hochdurchsatzsystem eine Betrachtung, und welche Kompromisse würden die Wahl bestimmen?
route_by_priority sendet E-Mails mit hoher Priorität zur Überprüfung und sendet den Rest automatisch. Wie könnte dieser Schwellenwert in einem echten System angepasst werden, und welche Metriken zeigen an, ob er richtig eingestellt ist?
Der Abschlussschritt erfolgt nach write_response mit add_conditional_edges. Gibt es im E-Mail-Workflow Szenarien, in denen das Command-Muster besser geeignet wäre als eine separate Routenfunktion?