LangChain Human-in-the-Loop

In Produktionsumgebungen erfordern einige Vorgänge eine menschliche Genehmigung – wie das Senden von E-Mails, das Durchführen von Löschungen und die Verarbeitung von Zahlungen.

Human-in-the-Loop (HITL) ermöglicht es dem Agenten, an kritischen Momenten zu pausieren, auf eine menschliche Genehmigung zu warten und dann fortzufahren.

interrupt() — Pausieren der Ausführung in Tools

interrupt()Diese Funktion ermöglicht es einem Tool, mitten in der Ausführung zu pausieren, auf externe Eingaben zu warten und dann fortzufahren:

Beispiel

from langgraph.types import interrupt

# Im Tool mit interrupt() pausieren
def send_email(to: str, subject: str, body: str) -> str:
    """E-Mail senden (erfordert menschliche Genehmigung)"""
    # Ausführung pausieren und eine Genehmigungsanfrage nach außen senden
    approval = interrupt({
        "action": "send_email",
        "to": to,
        "subject": subject,
        "body": body,
        "message": "Bitte bestätigen Sie, ob diese E-Mail gesendet werden soll?"
    })

    # Warte auf externe Genehmigung, um fortzufahren
    if approval.get("approved"):
        return f"E-Mail gesendet an {to}"
    else:
        return f"E-Mail-Versand abgelehnt: {approval.get('reason', 'User cancelled')}"

Wie interrupt() funktioniert:

  1. Das Tool ruft interrupt() auf → der Agent pausiert die Ausführung
  2. Das externe System erhält die Interrupt-Informationen und zeigt sie dem Benutzer an
  3. Nachdem der Benutzer eine Entscheidung getroffen hat, wird die Ausführung über Command(resume=...) fortgesetzt
  4. interrupt() gibt den vom Benutzer übergebenen Wert zurück, und das Tool setzt die Ausführung fort

Vollständiges Beispiel — Genehmigungsprozess

Beispiel

from dotenv import load_dotenv
load_dotenv()

from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage


@tool
def delete_course(course_name: str) -> str:
    """Kurs löschen (Genehmigung erforderlich).

    Argumente:
course_name: der Name des zu löschenden Kurses
    """
    # Anhalten und auf Genehmigung warten
    approval = interrupt({
        "action": "delete_course",
        "course": course_name,
        "message": f"Kurs '{course_name}' löschen bestätigen? Diese Aktion kann nicht rückgängig gemacht werden."
    })

    if approval.get("confirmed"):
        return f"Kurs '{course_name}' gelöscht"
    else:
        return f"Löschung abgebrochen"


checkpointer = InMemorySaver()
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    tools=[delete_course],
    checkpointer=checkpointer,
    system_prompt="Du bist der Admin-Assistent des EXAMPLE-Tutorials.",
)

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

# Schritt 1: Löschanforderung initiieren (löst einen Interrupt aus)
print("=== Start der Ausführung ===")
result = agent.invoke(
    {"messages": [HumanMessage(content="Bitte lösche den Kurs 'Outdated Java Tutorial'.")]},
    config=config,
)

# Prüfen, ob der Agent pausiert wurde
state = .get_state(config)
print(f"Status: {state.next}")  # ('tools',) bedeutet pausiert am tools-Knoten
print(f"Interrupt-Info: {state.tasks)

# Schritt 2: Menschliche Genehmigung (simuliert, dass der Benutzer auf "Bestätigen" klickt)
print("\n=== Menschliche Genehmigung ===)
resume_value = {"confirmed": True, "operator": "Administrator Zhang San"}
result = agent.invoke(
    Command(resume=resume_value),
    config=config,
)
print(f"Endgültige Antwort: {result['messages'][-1].content}")

Ausgabe:

=== 开始执行 ===
Status: ('tools',)
中断信息: (Interrupt(value={'action': 'delete_course', ...}),)

=== 人工审批 ===
Final reply: The course "Outdated Java Tutorial" has been deleted.

interrupt_before / interrupt_after Parameter

Neben der Verwendung von interrupt() in Tools können Sie auch globale Interrupt-Punkte in create_agent() festlegen:

Beispiel

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[some_tool],
    checkpointer=checkpointer,

    # Pause vor Tool-Knoten (Genehmigung vor jedem Tool-Aufruf erforderlich)
    interrupt_before=["tools"],

    # Pause nach Modellknoten (nach jeder Modellantwort überprüfbar)
    # interrupt_after=["model"],
)
ParameterPausenzeitpunktAnwendungsfall
interrupt_before=["tools"]Vor jeder Tool-AusführungAlle Tool-Aufrufe erfordern eine Genehmigung
interrupt_before=["model"]Vor jedem ModellaufrufNachrichten vor der Modellverarbeitung manuell überprüfen
interrupt_after=["model"]Nach jeder ModellantwortModellausgabe überprüfen, bevor über die Fortsetzung entschieden wird
interrupt_after=["tools"]Nach jeder Tool-AusführungTool-Ergebnisse prüfen, bevor der nächste Schritt entschieden wird

Typische HITL-Architektur

In echten Webanwendungen wird HITL normalerweise so implementiert:

Beispiel

# Backend: Benutzernachricht empfangen, bis zum Interrupt-Punkt verarbeiten, Interrupt-Informationen zurückgeben
def handle_user_message(thread_id: str, message: str):
    config = {"configurable": {"thread_id": thread_id}}
    result = agent.invoke(
        {"messages": [HumanMessage(content=message)]},
        config=config,
    )
    state = .get_state(config)

    # Prüfen, ob auf Genehmigung gewartet wird
    if state.tasks and state.tasks[0].interrupts:
        return {
            "status": "pending_approval",
            "interrupt": state.tasks[0].interrupts[0].value,
            "thread_id": thread_id,
        }

    return {
        "status": "completed",
        "reply": result["messages"][-1].content,
    }


# Backend: Benutzergenehmigung verarbeiten
def handle_approval(thread_id: str, approved: bool, reason: str = ""):
    config = {"configurable": {"thread_id": thread_id}}
    result = agent.invoke(
        Command(resume={"confirmed": approved, "reason": reason}),
        config=config,
    )
    return {"status": "completed", "reply": result["messages"][-1].content}

HITL erfordert einen Checkpointer. Denn wenn der Agent bei interrupt() pausiert, muss sein Zustand persistiert werden, um die Ausführung korrekt fortzusetzen.

Andere Erweiterungen