LangChain Human-in-the-Loop

Dans les environnements de production, certaines opérations nécessitent une approbation humaine — comme l'envoi d'e-mails, l'exécution de suppressions et le traitement des paiements.

Human-in-the-Loop (HITL) permet à l'Agent de s'interrompre à des moments critiques, d'attendre l'approbation humaine, puis de continuer.

interrupt() — Mise en pause de l'exécution dans les outils

interrupt()Cette fonction permet à un outil de s'interrompre au milieu de son exécution, d'attendre une entrée externe, puis de continuer :

Exemple

from langgraph.types import interrupt

# Mise en pause dans l'outil avec interrupt()
def send_email(to: str, subject: str, body: str) -> str:
    """Envoyer un e-mail (nécessite une approbation humaine)"""
    # Suspendre l'exécution et envoyer une demande d'approbation vers l'extérieur
    approval = interrupt({
        "action": "send_email",
        "to": to,
        "subject": subject,
        "body": body,
        "message": "Veuillez confirmer si vous souhaitez envoyer cet e-mail ?"
    })

    # Attendre l'approbation externe pour continuer
    if approval.get("approved"):
        return f"E-mail envoyé à {to}"
    else:
        return f"Envoi de l'e-mail rejeté : {approval.get('reason', 'Annulé par l’utilisateur')}"

Fonctionnement de interrupt() :

  1. L'outil appelle interrupt() → l'Agent suspend l'exécution
  2. Le système externe reçoit l'information d'interruption et l'affiche à l'utilisateur
  3. Une fois que l'utilisateur a pris une décision, l'exécution reprend via Command(resume=...)
  4. interrupt() renvoie la valeur passée par l'utilisateur, et l'outil poursuit son exécution

Exemple complet — Processus d'approbation

Exemple

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:
    """Supprimer le cours (nécessite une approbation).

    Arguments :
course_name: le nom du cours à supprimer
    """
    # Mettre en pause et attendre l'approbation
    approval = interrupt({
        "action": "delete_course",
        "course": course_name,
        "message": f"Confirmer la suppression du cours '{course_name}' ? Cette action est irréversible."
    })

    if approval.get("confirmed"):
        return f"Le cours '{course_name}' a été supprimé"
    else:
        return f"Suppression annulée"


checkpointer = InMemorySaver()
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    tools=[delete_course],
    checkpointer=checkpointer,
    system_prompt="Vous êtes l'assistant administrateur du tutoriel EXAMPLE.",
)

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

# Étape 1 : Initier la demande de suppression (déclenchera une interruption)
print("=== Début de l'exécution ===")
result = agent.invoke(
    {"messages": [HumanMessage(content="Veuillez supprimer le cours 'Tutoriel Java obsolète'.")]},
    config=config,
)

# Vérifier si l'agent est en pause
state = agent.get_state(config)
print(f"Statut : {state.next}")  # ('tools',) signifie en pause au nœud tools
print(f"Infos d'interruption : {state.tasks)

# Étape 2 : validation humaine (simuler un clic de l'utilisateur sur "Confirmer")
print("\n=== Approbation humaine ===)
resume_value = {"confirmed": True, "operator": "Administrateur Zhang San"}
result = agent.invoke(
    Commande(resume=resume_value),
    config=config,
)
print(f"Réponse finale : {result['messages'][-1].content}")

Sortie :

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

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

Paramètres interrupt_before / interrupt_after

Outre l'utilisation de interrupt() dans tools, vous pouvez également définir des points d'interruption globaux dans create_agent() :

Exemple

from langgraph.checkpoint.memory import InMemorySaver

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

    # Pause avant les nœuds d'outils (approbation requise avant chaque appel d'outil)
    interrupt_before=["tools"],

    # Pause après les nœuds de modèle (possible après chaque réponse du modèle)
    # interrupt_after=["model"],
)
ParamètreMoment de la pauseScénario applicable
interrupt_before=["tools"]Avant chaque exécution d'outilTous les appels d'outils nécessitent une approbation
interrupt_before=["model"]Avant chaque appel de modèleExaminer manuellement les messages avant le traitement par le modèle
interrupt_after=["model"]Après chaque réponse du modèleExaminer la sortie du modèle avant de décider s'il faut continuer
interrupt_after=["tools"]Après chaque exécution d'outilVérifier les résultats des outils avant de décider de l'étape suivante

Architecture HITL typique

Dans les applications Web réelles, le HITL est généralement implémenté comme ceci :

Exemple

# Backend : recevoir le message utilisateur, traiter jusqu'au point d'interruption, renvoyer les informations d'interruption
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 = agent.get_state(config)

    # Vérifier si l'on attend une approbation
    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 : traiter l'approbation de l'utilisateur
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 nécessite un Checkpointer. Car lorsque l'Agent se met en pause au niveau de interrupt(), son état doit être persisté afin de reprendre correctement l'exécution.

Autres extensions