Humano-en-el-bucle con Interrupciones

Implementa flujos de trabajo humano-en-el-bucle usando interrupciones, pausa la ejecución para entrada del usuario y reanuda la ejecución del grafo.

La función interrupt() de LangGraph pausa la ejecución del grafo y espera la entrada humana. Esto permite flujos de aprobación, solicitudes de aclaración y cualquier escenario en que una persona deba revisar o aportar antes de que el agente continúe.

¿Qué es una Interrupción?

Un interrupt pausa la ejecución del grafo en un nodo específico, muestra información del estado al usuario y espera a que el usuario proporcione la entrada antes de reanudar.

💡 Tip: Use Ctrl + Mouse Wheel to zoom, or click the fullscreen icon

La Función interrupt()

python

from langgraph.types import interrupt

def approval_node(state: State) -> dict:
    # Pause execution and ask for human input
    human_response = interrupt(
        # This value is presented to the human
        {
            "question": "Approve this action?",
            "action": state["proposed_action"],
            "risk_level": state["risk"]
        }
    )

    # human_response contains what the human provided
    if human_response.get("approved"):
        return {"approved": True}
    else:
        return {"approved": False, "feedback": human_response.get("feedback")}

📌Importante

interrupt() lanza una excepción especial que pausa el grafo. La función no sigue ejecutándose hasta que el grafo se reanude con invoke() o update_state().

Configurando Interrupciones

Los interrupts requieren un checkpointer (persistencia):

python

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt
from typing_extensions import TypedDict

class ApprovalState(TypedDict):
    query: str
    proposed_action: str
    approved: bool
    feedback: str

def analyze_node(state: ApprovalState) -> dict:
    # Agent proposes an action
    action = f"Search for: {state['query']}"
    return {"proposed_action": action}

def human_review(state: ApprovalState) -> dict:
    # INTERRUPT — pause and ask human
    response = interrupt({
        "query": state["query"],
        "proposed_action": state["proposed_action"],
        "prompt": "Approve this search?"
    })
    return {
        "approved": response.get("approved", False),
        "feedback": response.get("feedback", "")
    }

def execute_node(state: ApprovalState) -> dict:
    if state["approved"]:
        # Execute the approved action
        return {"result": f"Executed: {state['proposed_action']}"}
    return {"result": f"Skipped: {state.get('feedback', 'No reason given')}"}

# Build graph
builder = StateGraph(ApprovalState)
builder.add_node("analyze", analyze_node)
builder.add_node("human_review", human_review)
builder.add_node("execute", execute_node)
builder.add_edge(START, "analyze")
builder.add_edge("analyze", "human_review")
builder.add_edge("human_review", "execute")
builder.add_edge("execute", END)

# Compile with checkpointer
app = builder.compile(checkpointer=MemorySaver())

Ejecutando el Grafo con Interrupciones

python

# 1. Start the graph
config = {"configurable": {"thread_id": "approval-1"}}

# This will pause at the interrupt
try:
    result = app.invoke(
        {"query": "latest AI news", "proposed_action": "", "approved": False, "feedback": "", "result": ""},
        config
    )
except:  # interrupt() raises an exception
    pass

# 2. Check the paused state
state = app.get_state(config)
print(state.next)  # ('human_review',) — this node is pending
print(state.values)  # Current state at interruption point

# 3. Resume by providing human input
result = app.invoke(
    None,  # No new state — we're resuming
    {"configurable": {"thread_id": "approval-1"}},
    {"approved": True, "feedback": "Looks good!"}
)

print(result["result"])  # "Executed: Search for: latest AI news"

ℹ️Nota

Al reanudar, pasa None como entrada (no un estado nuevo) y proporciona la respuesta del interrupt mediante el config o a través de Command. El grafo continúa desde donde se pausó.

Usando Command para Reanudar

La clase Command ofrece una forma más limpia de reanudar:

python

from langgraph.types import Command

# Resume with human input using Command
result = app.invoke(
    Command(
        resume={"approved": True, "feedback": "Proceed"},
        update={"status": "human_reviewed"}  # Optional: update state too
    ),
    {"configurable": {"thread_id": "approval-1"}}
)

💡Consejo

Command te permite proporcionar la respuesta del interrupt y actualizar el estado al mismo tiempo. Usa update para añadir metadatos sobre la revisión humana.

Múltiples Puntos de Interrupción

Un grafo puede tener múltiples puntos de interrupt:

python

def first_approval(state: State) -> dict:
    response = interrupt({"stage": "plan", "plan": state["plan"]})
    return {"plan_approved": response.get("approved")}

def second_approval(state: State) -> dict:
    response = interrupt({"stage": "execute", "result": state["result"]})
    return {"execution_approved": response.get("approved")}

# Graph: START → plan → first_approval → execute → second_approval → END

Cada interrupt se reanuda de forma independiente. El grafo sabe en qué interrupt está pausado.

Mensajes de Interrupción Dinámicos

El valor que se pasa a interrupt() puede ser cualquier dato serializable en JSON:

python

def review_node(state: State) -> dict:
    context = {
        "type": "approval_request",
        "timestamp": datetime.now().isoformat(),
        "agent_note": state["analysis"],
        "options": [
            {"value": "approve", "label": "Approve and continue"},
            {"value": "modify", "label": "Modify and continue"},
            {"value": "reject", "label": "Reject and stop"}
        ],
        "current_state": {k: v for k, v in state.items() if k != "secrets"}
    }
    response = interrupt(context)
    return {"human_decision": response}

Interrupción con Validación

Valida la entrada humana antes de continuar:

python

def validated_review(state: State) -> dict:
    while True:
        response = interrupt({"question": "Enter a number between 1 and 10:"})

        try:
            value = int(response.get("value", -1))
            if 1 <= value <= 10:
                return {"user_number": value}
        except (ValueError, TypeError):
            pass

        # Invalid — loop back to interrupt
        # The human will be prompted again

⚠️Advertencia

Ten cuidado con los bucles de validación. Cada intento no válido crea un nuevo checkpoint. Considera limitar los reintentos para evitar frustración.

Ejemplo Completo: Moderación de Contenido

python

from langgraph.graph import StateGraph, START, END, add_messages
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
from langchain_openai import ChatOpenAI
from typing_extensions import TypedDict, Annotated
from typing import List, Any

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

class ModerationState(TypedDict):
    messages: Annotated[List[Any], add_messages]
    content: str
    toxicity_score: float
    needs_review: bool
    approved: bool
    moderator_note: str

def analyze_content(state: ModerationState) -> dict:
    response = llm.invoke(f"Rate toxicity (0-1) of: {state['content']}")
    score = float(response.content.strip())
    return {"toxicity_score": score, "needs_review": score > 0.5}

def human_moderation(state: ModerationState) -> dict:
    response = interrupt({
        "content": state["content"],
        "toxicity_score": state["toxicity_score"],
        "prompt": "Approve this content?"
    })
    return {
        "approved": response.get("approved", False),
        "moderator_note": response.get("note", "")
    }

def auto_approve(state: ModerationState) -> dict:
    return {"approved": True, "moderator_note": "Auto-approved (low toxicity)"}

def router(state: ModerationState) -> str:
    if state["needs_review"]:
        return "human"
    return "auto"

builder = StateGraph(ModerationState)
builder.add_node("analyze", analyze_content)
builder.add_node("human_moderation", human_moderation)
builder.add_node("auto_approve", auto_approve)
builder.add_edge(START, "analyze")
builder.add_conditional_edges("analyze", router, {
    "human": "human_moderation",
    "auto": "auto_approve"
})
builder.add_edge("human_moderation", END)
builder.add_edge("auto_approve", END)

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

Verificando Grafos en Pausa

python

# List all threads with their status
# (You need to track thread IDs externally)

# Check if a specific thread is paused
state = app.get_state(config)
is_paused = len(state.next) > 0  # Has pending nodes

if is_paused:
    print(f"Graph is paused at: {state.next}")
    print(f"Interrupt values: {state.tasks}")

    # Resume
    result = app.invoke(
        Command(resume={"approved": True}),
        config
    )