Intervención humana en LangChain
En entornos de producción, algunas operaciones requieren confirmación humana, como enviar correos, ejecutar eliminaciones o procesar pagos.
La intervención humana (HITL, Human-in-the-Loop) permite que el agente se detenga en momentos críticos, espere la aprobación humana y luego continúe.
interrupt()——Pausar la ejecución en una herramienta
interrupt()La función permite que una herramienta se pause a mitad de ejecución, espere una entrada externa y luego continúe:
Ejemplo
from langgraph.types import interrupt
# Pausar con interrupt() dentro de una herramienta
def send_email(to: str, subject: str, body: str) -> str:
"""Enviar correo (requiere aprobación humana)"""
# Pausar la ejecución y enviar una solicitud de aprobación al exterior
approval = interrupt({
"action": "send_email",
"to": to,
"subject": subject,
"body": body,
"message": "¿Confirma el envío de este correo?"
})
# Esperar a que llegue la aprobación externa para continuar
if approval.get("approved"):
return f"Correo enviado a {to}"
else:
return f"El envío del correo fue rechazado: {approval.get('reason', 'Usuario canceló')}"
# Pausar con interrupt() dentro de una herramienta
def send_email(to: str, subject: str, body: str) -> str:
"""Enviar correo (requiere aprobación humana)"""
# Pausar la ejecución y enviar una solicitud de aprobación al exterior
approval = interrupt({
"action": "send_email",
"to": to,
"subject": subject,
"body": body,
"message": "¿Confirma el envío de este correo?"
})
# Esperar a que llegue la aprobación externa para continuar
if approval.get("approved"):
return f"Correo enviado a {to}"
else:
return f"El envío del correo fue rechazado: {approval.get('reason', 'Usuario canceló')}"
Flujo de trabajo de interrupt():
- La herramienta llama a interrupt() → el agente pausa la ejecución
- El sistema externo obtiene la información de interrupción y la muestra al usuario
- Cuando el usuario toma una decisión, se reanuda la ejecución mediante Command(resume=...)
- interrupt() devuelve el valor proporcionado por el usuario y la herramienta continúa su ejecución
Ejemplo completo——Flujo de aprobación
Ejemplo
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:
"""Eliminar curso (requiere aprobación).
Args:
course_name: nombre del curso a eliminar
"""
# Pausar y esperar aprobación
approval = interrupt({
"action": "delete_course",
"course": course_name,
"message": f"¿Confirmar la eliminación del curso «{course_name}»? Esta acción no se puede deshacer."
})
if approval.get("confirmed"):
return f"El curso «{course_name}» ha sido eliminado"
else:
return f"La eliminación ha sido cancelada"
checkpointer = InMemorySaver()
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[delete_course],
checkpointer=checkpointer,
system_prompt=Eres el asistente administrativo de Ejemplo (Tutorial de Ejemplo).,
)
config = {"configurable": {"thread_id": "admin-001"}}
# Primer paso: iniciar la solicitud de eliminación (activará la interrupción)
print("=== Iniciando ejecución ===")
result = agent.invoke(
{"messages": [HumanMessage(content="Por favor, elimina el curso «Tutorial de Java obsoleto»")]},
config=config,
)
# Verificar si el agente se ha pausado
state = agent.get_state(config)
print(f"Estado: {state.next}") # ('tools',) indica que se pausó en el nodo tools
print(f"Información de interrupción: {state.tasks)
# Segundo paso: aprobación humana (simulando que el usuario hace clic en 'Confirmar')
print("\n"=== Aprobación humana ===")
resume_value = {"confirmed": True, "operator": "Administrador Zhang San"}
result = agent.invoke(
Command(resume=resume_value),
config=config,
)
print(f"Respuesta final: {result['messages'][-1].content}")
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:
"""Eliminar curso (requiere aprobación).
Args:
course_name: nombre del curso a eliminar
"""
# Pausar y esperar aprobación
approval = interrupt({
"action": "delete_course",
"course": course_name,
"message": f"¿Confirmar la eliminación del curso «{course_name}»? Esta acción no se puede deshacer."
})
if approval.get("confirmed"):
return f"El curso «{course_name}» ha sido eliminado"
else:
return f"La eliminación ha sido cancelada"
checkpointer = InMemorySaver()
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[delete_course],
checkpointer=checkpointer,
system_prompt=Eres el asistente administrativo de Ejemplo (Tutorial de Ejemplo).,
)
config = {"configurable": {"thread_id": "admin-001"}}
# Primer paso: iniciar la solicitud de eliminación (activará la interrupción)
print("=== Iniciando ejecución ===")
result = agent.invoke(
{"messages": [HumanMessage(content="Por favor, elimina el curso «Tutorial de Java obsoleto»")]},
config=config,
)
# Verificar si el agente se ha pausado
state = agent.get_state(config)
print(f"Estado: {state.next}") # ('tools',) indica que se pausó en el nodo tools
print(f"Información de interrupción: {state.tasks)
# Segundo paso: aprobación humana (simulando que el usuario hace clic en 'Confirmar')
print("\n"=== Aprobación humana ===")
resume_value = {"confirmed": True, "operator": "Administrador Zhang San"}
result = agent.invoke(
Command(resume=resume_value),
config=config,
)
print(f"Respuesta final: {result['messages'][-1].content}")
Resultado de la ejecución:
=== 开始执行 ===
Estado: ('tools',)
中断信息: (Interrupt(value={'action': 'delete_course', ...}),)
=== 人工审批 ===
Respuesta final: El curso "Tutorial de Java desactualizado" se ha eliminado.
Parámetros interrupt_before / interrupt_after
Además de usar interrupt() en las herramientas, también puedes configurar puntos de interrupción globales en create_agent():
Ejemplo
from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()
agent = create_agent(
model="deepseek:deepseek-v4-flash",
tools=[some_tool],
checkpointer=checkpointer,
# Pausar antes del nodo de herramientas (cada llamada a una herramienta requiere aprobación)
interrupt_before=["tools"],
# Pausar después del nodo del modelo (se puede revisar cada respuesta del modelo)
# interrupt_after=["model"],
)
checkpointer = InMemorySaver()
agent = create_agent(
model="deepseek:deepseek-v4-flash",
tools=[some_tool],
checkpointer=checkpointer,
# Pausar antes del nodo de herramientas (cada llamada a una herramienta requiere aprobación)
interrupt_before=["tools"],
# Pausar después del nodo del modelo (se puede revisar cada respuesta del modelo)
# interrupt_after=["model"],
)
| Parámetro | Momento de pausa | Escenario de uso |
|---|---|---|
| interrupt_before=["tools"] | Antes de ejecutar cada herramienta | Todas las llamadas a herramientas requieren aprobación |
| interrupt_before=["model"] | Antes de cada llamada al modelo | Revisar manualmente los mensajes antes de que el modelo los procese |
| interrupt_after=["model"] | Después de cada respuesta del modelo | Revisar la salida del modelo antes de decidir si continuar |
| interrupt_after=["tools"] | Después de cada ejecución de herramienta | Revisar el resultado de la herramienta antes de decidir el siguiente paso |
Arquitectura típica de HITL
En aplicaciones web reales, HITL suele implementarse de la siguiente manera:
Ejemplo
# Backend: recibe el mensaje del usuario, lo procesa hasta el punto de interrupción y devuelve la información de interrupción
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)
# Verificar si está esperando aprobación
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: procesar la aprobación del usuario
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}
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)
# Verificar si está esperando aprobación
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: procesar la aprobación del usuario
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}
Otras extensionesHITL requiere el uso conjunto de Checkpointer. Esto se debe a que cuando el agente se detiene en interrupt(), su estado debe persistirse para poder continuar correctamente al reanudar.