Human-in-the-Loop-Agents mit LangGraph

Veröffentlicht

3. Oktober 2026

Autor

Faizan Nadeem

Warum ein Pause-Knopf dauerhaft sein muss

Es lohnt sich, präzise darüber zu sein, was „Human in the Loop" tatsächlich erfordert, weil es leicht ist, etwas zu bauen, das in einer Demo richtig aussieht und in der Produktion zusammenbricht. Eine Demo-grade Pause ist: den Python-Prozess offen halten, bei input() blockieren und fortfahren, wenn jemand etwas tippt. Das funktioniert genau so lange, wie der Prozess am Leben bleibt und eine Person ein Terminal beobachtet.

Produktionsreifes Human-in-the-Loop braucht drei Dinge, die ein blockierender input()-Aufruf nicht geben kann: die Fähigkeit, über einen Prozessneustart hinaus zu pausieren (Deploys passieren; Server werden umgeplant), die Fähigkeit, dass das „Fortfahren"-Ereignis von einem völlig anderen Prozess kommt – einem Webhook von einem Slack-Genehmigungsknopf, einem Klick in einem internen Dashboard – und die Fähigkeit, viele pausierte Agents gleichzeitig zu haben, ohne für jeden einen Thread oder eine Verbindung offen zu halten. Das ist die eigentliche Anforderung, und das ist der Grund, warum das nicht mit einer Schleife und einem Flag lösbar ist – es braucht den State des Graphen, der irgendwo dauerhaft geschrieben wird, indiziert nach einer ID, die du später nachschlagen kannst. Wenn du die Entscheidung, ein graphenbasiertes Framework überhaupt zu verwenden, noch nicht getroffen hast, ist dieser Entscheidungspunkt – Chains versus Graphen – in einem eigenen Post behandelt; dieser hier setzt voraus, dass du dich bereits für LangGraph entschieden hast.

Die interrupt-Funktion

LangGraphs interrupt() ist eine Funktion, die du von innerhalb eines Nodes aufrufst. Wenn sie getroffen wird, stoppt der Graph die Ausführung an genau diesem Punkt, und was auch immer du an interrupt() übergeben hast, wird demjenigen angezeigt, der .invoke() ausführt.

from langgraph.types import interrupt

def request_approval(state: AgentState):
    decision = interrupt({
        "action": "send_refund",
        "amount": state["refund_amount"],
        "customer": state["customer_id"],
        "reason": state["reason"],
    })
    return {"approved": decision["approved"]}

Wenn request_approval ausgeführt wird und interrupt(...) trifft, stoppt die Ausführung des Graphen und die übergebene Payload wird dem Aufrufer als Teil des Aufruf-Ergebnisses zurückgegeben, markiert als Interrupt statt als normale Fertigstellung:

result = graph.invoke({"refund_amount": 42.50, "customer_id": "cust_9", "reason": "damaged item"}, config)

if "__interrupt__" in result:
    interrupt_data = result["__interrupt__"][0].value
    print(interrupt_data)
    # {'action': 'send_refund', 'amount': 42.5, 'customer': 'cust_9', 'reason': 'damaged item'}

Nichts nach dem interrupt()-Aufruf hat noch laufen – die Node-Funktion selbst hat nicht einmal zurückgegeben. Das ist das Detail, das Leute stolpern lässt: interrupt() gibt bei diesem Aufruf keinen Wert zurück, es hebt die Kontrolle vollständig aus dem Graphen heraus. Die Zeile decision = interrupt(...) bekommt tatsächlich erst beim nächsten Aufruf einen Wert zugewiesen – nach dem Fortsetzen – LangGraph replayed die Node-Funktion vom Anfang und diesmal gibt interrupt() sofort mit dem zurück, was du beim Fortsetzen übergeben hast.

Fortfahren mit Command

Um fortzufahren, rufst du .invoke() nicht mit einem neuen Input auf – du rufst es mit einem Command-Objekt auf, das den Wert einschließt, den der pausierte interrupt()-Aufruf empfangen soll.

from langgraph.types import Command

result = graph.invoke(
    Command(resume={"approved": True}),
    config,  # same config, same thread_id, as the original call
)

Die config – speziell ihre thread_id – ist das, was dieses Fortsetzen mit der genauen pausierten Ausführung verknüpft. LangGraph schlägt den checkpointed State für diesen Thread nach, replayed den unterbrochenen Node, wobei interrupt() jetzt {"approved": True} zurückgibt, und setzt die Ausführung von dort durch den Rest des Graphen wie gewohnt fort.

config = {"configurable": {"thread_id": f"refund-{refund_id}"}}

# First call — pauses at request_approval
result = graph.invoke({"refund_amount": 42.50, "customer_id": "cust_9"}, config)

# ... hours or days pass, a human clicks "approve" in a dashboard ...

# Second call — resumes and completes
final = graph.invoke(Command(resume={"approved": True}), config)

Ein Node kann über die Lebensdauer eines Graphen auch mehr als einmal interrupt() aufrufen – eine Überprüfung, dann ein zweiter Bestätigungsschritt später – und jeder pausiert und setzt unabhängig mit demselben Command(resume=...)-Muster fort, indiziert nach derselben thread_id.

Checkpointer: Was wirklich gespeichert wird

Nichts davon funktioniert ohne einen Checkpointer, weil interrupt() allein nur ein Control-Flow-Signal ist – es ist der Checkpointer, der den State tatsächlich persistiert, damit ein anderer Prozess, möglicherweise Minuten oder Tage später, ihn aufgreifen kann. Jedes Mal, wenn ein Node die Ausführung beendet, schreibt der Checkpointer einen Snapshot des vollständigen Graphen-State, plus Metadaten darüber, welcher Node als nächstes kommt, indiziert nach thread_id.

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

MemorySaver ist für die lokale Entwicklung in Ordnung, aber unter der Haube ist es nur ein Python-Dict – beende den Prozess und jeder pausierte Thread ist weg. Für alles, das über die Lebensdauer eines Prozesses hinaus überleben muss, brauchst du einen durch echten Speicher gestützten Checkpointer.

Postgres-gestütztes Checkpointing in der Produktion

PostgresSaver (aus dem langgraph-checkpoint-postgres-Paket) persistiert denselben State in eine echte Postgres-Tabelle, was „pausieren" zu „unbegrenzt pausieren, über Neustarts hinweg, fortsetzbar aus jedem Prozess, der den Connection-String hat" macht.

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:password@localhost:5432/agent_db"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # creates the checkpoint tables on first run
    graph = builder.compile(checkpointer=checkpointer)
    result = graph.invoke(initial_state, config)

Für einen langlebigen Server-Prozess – eine FastAPI-App, die viele gleichzeitige Agents handhabt – willst du die Verbindung generell einmal beim Start öffnen statt pro Anfrage, und die async-Variante verwenden, damit das Pausieren eines Agents andere nicht blockiert:

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

async def build_graph():
    checkpointer = AsyncPostgresSaver.from_conn_string(DB_URI)
    await checkpointer.setup()
    return builder.compile(checkpointer=checkpointer)

# in a request handler
result = await graph.ainvoke(initial_state, config)

checkpointer.setup() ist idempotent und muss nur einmal pro Datenbank ausgeführt werden – in einem echten Deployment würdest du es typischerweise als Teil eines Migration-Schritts ausführen statt bei jedem App-Start, so wie du jede andere Schema-Migration behandelst.

Einen Graphen Tage später fortsetzen

Es lohnt sich, das explizit auszusprechen, weil es der eigentliche Punkt der Verwendung von Postgres statt MemorySaver ist: Der Prozess, der einen Agent fortsetzt, muss nicht derselbe sein, der ihn gestartet hat. Ein am Freitag pausierter Rückerstattungs-Genehmigungsagent kann von einer völlig anderen Server-Instanz am Montag fortgesetzt werden, solange sie die thread_id und eine Verbindung zur selben Postgres-Datenbank hat.

# On a completely different server, days later, triggered by a Slack button click:

async def handle_approval_webhook(refund_id: str, approved: bool):
    checkpointer = AsyncPostgresSaver.from_conn_string(DB_URI)
    graph = builder.compile(checkpointer=checkpointer)

    config = {"configurable": {"thread_id": f"refund-{refund_id}"}}
    result = await graph.ainvoke(Command(resume={"approved": approved}), config)
    return result

Das ist der praktische Unterschied zwischen einer Spielzeug-Pause und einer echten: Die thread_id ist effektiv ein dauerhafter Zeiger in Postgres, und jeder Prozess mit Datenbankzugang kann genau dort weitermachen, wo der Graph aufgehört hat – vollständige Message-History, alle Zwischenzustände, kein erneutes Ausführen von Nodes, die bereits abgeschlossen wurden. Du kannst jederzeit mit graph.get_state(config) prüfen, was pausiert ist – es gibt den aktuellen checkpointed State zurück und kritisch .next – das Tuple von Node-Namen, die ausgeführt werden, wenn du fortsetzt, sodass ein Dashboard „Wartet auf: request_approval" anzeigen kann, ohne zu raten.

Den Review-Schritt selbst gestalten

Die oben genannten Mechaniken sind notwendig, aber nicht hinreichend – ein Human-in-the-Loop-Schritt, der mechanisch korrekt, aber schlecht gestaltet ist, wird einfach ein langsamer, unwillkommener Engpass. Ein paar Dinge, die den Review-Schritt selbst tatsächlich funktionsfähig machen:

Übergib dem Reviewer alles, was er braucht, in der Interrupt-Payload, nicht einen Verweis, um nachzuschlagen. Wenn interrupt() {"action": "send_refund", "amount": 42.50, "reason": "damaged item"} zurückgibt, kann ein Reviewer allein davon entscheiden. Wenn es {"refund_id": "r_123"} zurückgibt, hast du gerade jeden Genehmigungsprozess zu einem zweistufigen gemacht.
Gestalte die Form der Resume-Payload so, dass sie dem entspricht, worauf der Node tatsächlich verzweigt. Setze nicht mit einem rohen Boolean fort, wenn du irgendwann auch einen Ablehnungsgrund brauchst – gestalte {"approved": bool, "note": str} von Anfang an, auch wenn du note noch nicht verwendest.
Begrenze zeitlich, was „pausiert" für deinen Anwendungsfall bedeutet. Eine Rückerstattungsgenehmigung, die nach fünf Tagen noch aussteht, ist wahrscheinlich ein Prozessfehler, kein geduldiger Reviewer – füge Ablauflogik hinzu (ein geplanter Job, der updated_at auf veralteten Threads prüft und eskaliert oder automatisch ablehnt), statt Threads für immer pausieren zu lassen.
Logge die Interrupt-Payload unabhängig vom Checkpoint. Postgres-Checkpoint-Tabellen sind nicht dafür gedacht, ein menschenlesbares Audit-Log zu sein – schreibe beim Auslösen von interrupt() eine separate Zeile in eine approvals-Tabelle, damit du eine abfragbare History hast, die nicht davon abhängt, LangGraphs internem State-Format zu deserialisieren.

Ein vollständiger Genehmigungsflow von Ende zu Ende

Es hilft, die oben genannten Teile in etwas zu sehen, das einem echten Service ähnelt – ein Agent, der eine Rückerstattung entwirft, für die Genehmigung pausiert und entweder sie ausstellt oder den Kunden über eine Ablehnung benachrichtigt, über eine kleine FastAPI-App exponiert.

from fastapi import FastAPI
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

class RefundState(TypedDict):
    customer_id: str
    amount: float
    reason: str
    approved: bool | None

def draft_refund(state: RefundState):
    # Any model calls or business logic that decide the refund amount happen here.
    return {"amount": round(state["amount"], 2)}

def request_approval(state: RefundState):
    decision = interrupt({
        "action": "approve_refund",
        "customer_id": state["customer_id"],
        "amount": state["amount"],
        "reason": state["reason"],
    })
    return {"approved": decision["approved"]}

def issue_refund(state: RefundState):
    process_payment_refund(state["customer_id"], state["amount"])
    return {}

def notify_rejection(state: RefundState):
    send_customer_email(state["customer_id"], "Your refund request was declined.")
    return {}

def route_on_approval(state: RefundState) -> str:
    return "issue_refund" if state["approved"] else "notify_rejection"

builder = StateGraph(RefundState)
builder.add_node("draft_refund", draft_refund)
builder.add_node("request_approval", request_approval)
builder.add_node("issue_refund", issue_refund)
builder.add_node("notify_rejection", notify_rejection)
builder.add_edge(START, "draft_refund")
builder.add_edge("draft_refund", "request_approval")
builder.add_conditional_edges("request_approval", route_on_approval)
builder.add_edge("issue_refund", END)
builder.add_edge("notify_rejection", END)

app = FastAPI()

@app.post("/refunds/{refund_id}/start")
async def start_refund(refund_id: str, customer_id: str, amount: float, reason: str):
    checkpointer = AsyncPostgresSaver.from_conn_string(DB_URI)
    graph = builder.compile(checkpointer=checkpointer)
    config = {"configurable": {"thread_id": f"refund-{refund_id}"}}
    result = await graph.ainvoke(
        {"customer_id": customer_id, "amount": amount, "reason": reason, "approved": None},
        config,
    )
    return {"status": "pending_review", "interrupt": result.get("__interrupt__")}

@app.post("/refunds/{refund_id}/decide")
async def decide_refund(refund_id: str, approved: bool):
    checkpointer = AsyncPostgresSaver.from_conn_string(DB_URI)
    graph = builder.compile(checkpointer=checkpointer)
    config = {"configurable": {"thread_id": f"refund-{refund_id}"}}
    result = await graph.ainvoke(Command(resume={"approved": approved}), config)
    return {"status": "completed"}

Beachte, dass draft_refund – alles, was ein Modell oder einen externen Preisservice aufrufen könnte – vor dem Interrupt läuft, und issue_refund – der eigentliche Side-Effect – erst nach einem Fortsetzen läuft, gesteuert durch route_on_approval. Diese Reihenfolge ist absichtlich: es ist dieselbe Regel „keine Side-Effects vor einem Interrupt, den du möglicherweise wiederholst" aus dem Fehlermodus-Abschnitt unten, nur auf einen konkreten Flow angewendet statt abstrakt formuliert.

Pause und Fortsetzen lokal testen

Da der gesamte Punkt dieses Patterns darin besteht, eine Prozessgrenze zu überleben, lohnt es sich, diese Grenze explizit zu testen, statt darauf zu vertrauen, dass MemorySaver-Verhalten in der Entwicklung auf Postgres in der Produktion überträgt. Ein nützlicher lokaler Test: Starte eine echte Postgres-Instanz (Docker ist in Ordnung), starte einen Graphen, beende den Python-Prozess vollständig, starte einen frischen Prozess und setze fort, indem du nur die thread_id und den Connection-String verwendest – kein gemeinsamer Speicher, keine Imports aus dem ersten Prozess.

import subprocess, time

# Process 1: start the graph, let it pause at the interrupt, then exit.
subprocess.run(["python", "start_refund.py", "--refund-id", "test-1"])

# Simulate a deploy, a crash, whatever — nothing from process 1 is still running.
time.sleep(1)

# Process 2: a completely fresh interpreter, resuming purely from Postgres state.
subprocess.run(["python", "resume_refund.py", "--refund-id", "test-1", "--approved", "true"])

Wenn dieser Roundtrip mit zwei wirklich getrennten Prozessen funktioniert, hast du die Eigenschaft validiert, die tatsächlich wichtig ist – nicht „Der Code kompiliert", sondern „Der State überlebt eine Grenze, über die der Code null Sichtbarkeit hat." Das Testen von Pause/Fortsetzen innerhalb eines einzigen Prozesses (zweimaliges Aufrufen von .invoke() im selben Skript) besteht selbst bei einem Checkpointer, der heimlich State durch eine Module-Level-Variable durchsickern lässt, und fängt das nicht, bis es ein Produktionsvorfall ist.

Fehlermodi und wie man sich schützt

Fortsetzen mit der falschen thread_id. LangGraph gibt keinen lauten Fehler aus, wenn du eine thread_id für einen Thread übergibst, der nicht wirklich bei einem Interrupt pausiert ist – im schlimmsten Fall läuft es einfach aus welchem State auch immer vorhanden ist (oder nichts, für eine neue ID). Generiere thread_ids deterministisch aus einer echten Entitäts-ID (f"refund-{refund_id}", nicht eine zufällige UUID, die du irgendwo anders speichern musst), damit keine separate Lookup-Tabelle aus dem Sync geraten kann.

Node-Code, der nicht sicher wiedergegeben werden kann. Weil LangGraph die unterbrochene Node-Funktion beim Fortsetzen von oben wiedergibt, läuft jeder Code vor dem interrupt()-Aufruf innerhalb dieses Nodes beim Fortsetzen erneut. Wenn dieser Code eine E-Mail sendet oder in ein externes System schreibt, sendest du doppelt. Halte Side-Effects nach dem interrupt()-Aufruf, nicht davor, oder mache sie idempotent.

Schema-Drift zwischen Pause und Fortsetzen. Wenn du dein AgentState TypedDict änderst – ein erforderliches Feld hinzufügst – zwischen dem Zeitpunkt, als ein Thread pausiert wurde, und dem Fortsetzen, kann das Deserialisieren des alten Checkpoints brechen. Behandle State-Schema-Änderungen genauso wie du eine Datenbankmigration behandeln würdest: standardmäßig additiv, und fehlende Felder defensiv im Node-Code behandeln statt anzunehmen, dass jeder Checkpoint dem heutigen Schema entspricht.

Connection-Pool-Erschöpfung bei vielen pausierten Threads. Pausierte Threads halten keine Datenbankverbindung offen – sie sind nur Zeilen in Postgres – aber ein Spike gleichzeitiger Fortsetzungen kann deinen Pool dennoch erschöpfen, wenn AsyncPostgresSaver nicht mit einer vernünftigen Pool-Größe für deine erwartete Gleichzeitigkeit konfiguriert ist. Dimensioniere den Pool für deine Fortsetzungsrate, nicht deine Pausierungsrate.

get_state() als Ersatz für deinen eigenen Audit-Trail behandeln. graph.get_state(config) ist für das interaktive Debuggen eines einzelnen pausierten Threads wirklich nützlich, aber es ist nicht dafür gebaut, „Zeig mir jede Rückerstattung, die gerade auf Genehmigung wartet" zu beantworten – das ist eine Abfrage, für die LangGraphs Checkpoint-Schema nicht skalierbar konzipiert wurde. Wenn ein Dashboard ausständige Genehmigungen auflisten muss, pflege das als normale Anwendungsebene-Tabelle (approvals mit einer Status-Spalte), die neben dem Checkpoint geschrieben wird, statt zu versuchen, den Checkpoint-State dafür abzuscannen.

Fazit

Der Unterschied zwischen „Unser Agent kann pausieren" und „Unser Agent kann eine Woche pausieren und auf einem anderen Rechner fortfahren, ohne etwas zu verlieren" ist genau der Unterschied zwischen dem Halten von State in einer Python-Variable und dem Schreiben in Postgres bei jedem Schritt. interrupt() gibt dir das Control-Flow-Primitive – eine saubere Möglichkeit, einen Node mitten in der Ausführung zu stoppen und die Kontrolle an denjenigen zurückzugeben, der den Graphen steuert – und PostgresSaver gibt dir die Dauerhaftigkeit, die das „Tage später fortfahren" tatsächlich wahr statt zu einem Demo-Trick macht. Sobald beide Teile vorhanden sind, ist die menschliche Überprüfung kein Sonderfall mehr, um den deine Architektur herumarbeiten muss – sie ist einfach ein weiterer Node im Graphen, einer, der zufällig auf eine Person statt auf ein Modell wartet.