Módulo 9: Human-in-the-Loop

Breakpoints y Approvals

Descripción de la cápsula

En la cápsula anterior aprendiste interrupt() — una forma de pausar la ejecución dentro de un nodo, exactamente donde tú decidas. Pero hay otra forma de pausar: breakpoints. Los breakpoints pausan la ejecución entre nodos — antes o después de que un nodo específico se ejecute. No necesitas modificar el código del nodo. Los configuras al compilar el grafo.

¿Cuál es la diferencia práctica? Con interrupt() tú controlas el punto exacto dentro de la lógica de un nodo. Con breakpoints, tú dices "antes de que este nodo ejecute, pausa" o "después de que este nodo ejecute, pausa." Es una decisión a nivel de arquitectura, no de implementación.

Combinados con Command, los breakpoints te permiten construir flujos de aprobación completos: el agente llega a un punto crítico, se pausa, el humano inspecciona lo que el agente planea hacer, y decide si continuar, redirigir, o abortar. Todo sin tocar el código interno de los nodos.

Breakpoints vs interrupt(): dos herramientas, dos propósitos

Antes de ver código, la distinción:

Característica	`interrupt()`	Breakpoints
Dónde se define	Dentro del nodo (en el código)	Al compilar el grafo
Dónde pausa	En el punto exacto donde llamas `interrupt()`	Antes o después de un nodo
Modifica el nodo	Sí — agregas la llamada `interrupt()`	No — el nodo no cambia
Granularidad	Dentro de la lógica del nodo	Entre nodos
Caso de uso típico	Pedir input al usuario a mitad de proceso	Gate de aprobación antes de acciones peligrosas

Piénsalo así:

interrupt() → "Pausa aquí, en esta línea específica, y pregúntale algo al usuario"
interrupt_before → "Antes de que ejecute este nodo, pausa para aprobación"
interrupt_after → "Después de que ejecute este nodo, pausa para review"

interrupt_before: pausar ANTES de un nodo

El caso más común: tienes un nodo que ejecuta una acción peligrosa, costosa o irreversible. Quieres que el humano apruebe antes de que ese nodo se ejecute.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    task: str
    plan: str
    result: str
    log: Annotated[list[str], operator.add]

def planner(state: State) -> dict:
    plan = f"Plan para '{state['task']}': 1) Buscar datos, 2) Analizar, 3) Ejecutar acción"
    return {"plan": plan, "log": ["planner_done"]}

def dangerous_action(state: State) -> dict:
    result = f"Acción ejecutada según plan: {state['plan'][:50]}..."
    return {"result": result, "log": ["action_executed"]}

graph_builder = StateGraph(State)
graph_builder.add_node("planner", planner)
graph_builder.add_node("dangerous_action", dangerous_action)

graph_builder.add_edge(START, "planner")
graph_builder.add_edge("planner", "dangerous_action")
graph_builder.add_edge("dangerous_action", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["dangerous_action"],
)

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

print("=== Paso 1: Ejecutar hasta el breakpoint ===\n")
result = graph.invoke(
    {"task": "Enviar email masivo a 10,000 clientes", "plan": "", "result": "", "log": []},
    config,
)

state = graph.get_state(config)
print(f"Plan generado: {state.values['plan']}")
print(f"Siguiente nodo: {state.next}")
print(f"Resultado: '{state.values['result']}'")
print(f"Log: {state.values['log']}")

print("\n=== Paso 2: Humano aprueba → continuar ===\n")
result = graph.invoke(None, config)

state = graph.get_state(config)
print(f"Resultado: {state.values['result']}")
print(f"Log: {state.values['log']}")
print(f"Siguiente nodo: {state.next}")
# Output esperado:
# === Paso 1: Ejecutar hasta el breakpoint ===
#
# Plan generado: Plan para 'Enviar email masivo a 10,000 clientes': 1) Buscar datos, 2) Analizar, 3) Ejecutar acción
# Siguiente nodo: ('dangerous_action',)
# Resultado: ''
# Log: ['planner_done']
#
# === Paso 2: Humano aprueba → continuar ===
#
# Resultado: Acción ejecutada según plan: Plan para 'Enviar email masivo a 10,000 clien...
# Log: ['planner_done', 'action_executed']
# Siguiente nodo: ()

La mecánica:

interrupt_before=["dangerous_action"] — al compilar, defines que el grafo debe pausar antes de ejecutar dangerous_action
El primer invoke ejecuta planner y se detiene antes de dangerous_action
state.next muestra ('dangerous_action',) — el nodo está pendiente
El segundo invoke(None, config) reanuda: ejecuta dangerous_action y termina

El nodo dangerous_action no tiene ningún código de interrupción. No sabe que hay un breakpoint. Esa decisión vive en la compilación del grafo.

interrupt_after: pausar DESPUÉS de un nodo

A veces quieres que un nodo ejecute y luego pausar para que el humano revise el resultado antes de continuar:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    query: str
    analysis: str
    report: str
    log: Annotated[list[str], operator.add]

def analyze(state: State) -> dict:
    analysis = f"Análisis de '{state['query']}': 3 tendencias identificadas, 2 riesgos detectados"
    return {"analysis": analysis, "log": ["analysis_done"]}

def generate_report(state: State) -> dict:
    report = f"REPORTE FINAL: {state['analysis']}"
    return {"report": report, "log": ["report_done"]}

graph_builder = StateGraph(State)
graph_builder.add_node("analyze", analyze)
graph_builder.add_node("generate_report", generate_report)

graph_builder.add_edge(START, "analyze")
graph_builder.add_edge("analyze", "generate_report")
graph_builder.add_edge("generate_report", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["analyze"],
)

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

print("=== Paso 1: Ejecutar análisis y pausar para review ===\n")
graph.invoke(
    {"query": "Estado del mercado AI 2025", "analysis": "", "report": "", "log": []},
    config,
)

state = graph.get_state(config)
print(f"Análisis completado: {state.values['analysis']}")
print(f"Siguiente nodo: {state.next}")
print(f"Reporte: '{state.values['report']}'")

print("\n=== Paso 2: Humano revisa el análisis → aprueba ===\n")
result = graph.invoke(None, config)

state = graph.get_state(config)
print(f"Reporte generado: {state.values['report']}")
print(f"Log completo: {state.values['log']}")
# Output esperado:
# === Paso 1: Ejecutar análisis y pausar para review ===
#
# Análisis completado: Análisis de 'Estado del mercado AI 2025': 3 tendencias identificadas, 2 riesgos detectados
# Siguiente nodo: ('generate_report',)
# Reporte: ''
#
# === Paso 2: Humano revisa el análisis → aprueba ===
#
# Reporte generado: REPORTE FINAL: Análisis de 'Estado del mercado AI 2025': 3 tendencias identificadas, 2 riesgos detectados
# Log completo: ['analysis_done', 'report_done']

Con interrupt_after=["analyze"]:

El nodo analyze se ejecuta completamente
El grafo pausa después de analyze, antes de pasar a generate_report
El humano puede inspeccionar el análisis y decidir si proceder

El flujo de aprobación completo: aprobar, rechazar, redirigir

Pausar el grafo es solo el primer paso. Lo poderoso es lo que haces durante la pausa. Hay tres acciones fundamentales:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command

class State(TypedDict):
    task: str
    plan: str
    result: str
    status: str
    log: Annotated[list[str], operator.add]

def create_plan(state: State) -> dict:
    plan = f"Plan: ejecutar '{state['task']}' con 3 pasos automatizados"
    return {"plan": plan, "status": "plan_ready", "log": ["plan_created"]}

def execute_plan(state: State) -> dict:
    result = f"Ejecutado exitosamente: {state['plan']}"
    return {"result": result, "status": "executed", "log": ["executed"]}

def handle_rejection(state: State) -> dict:
    return {"result": "Tarea cancelada por el usuario", "status": "rejected", "log": ["rejected"]}

graph_builder = StateGraph(State)
graph_builder.add_node("create_plan", create_plan)
graph_builder.add_node("execute_plan", execute_plan)
graph_builder.add_node("handle_rejection", handle_rejection)

graph_builder.add_edge(START, "create_plan")
graph_builder.add_edge("create_plan", "execute_plan")
graph_builder.add_edge("execute_plan", END)
graph_builder.add_edge("handle_rejection", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["execute_plan"],
)

print("=== ESCENARIO 1: Aprobar ===\n")

config_approve = {"configurable": {"thread_id": "approve-flow"}}
graph.invoke(
    {"task": "Publicar artículo en blog", "plan": "", "result": "", "status": "", "log": []},
    config_approve,
)

state = graph.get_state(config_approve)
print(f"Plan: {state.values['plan']}")
print(f"Siguiente: {state.next}")

print("→ Humano aprueba: continuamos")
result = graph.invoke(None, config_approve)
print(f"Resultado: {graph.get_state(config_approve).values['result']}")

print("\n=== ESCENARIO 2: Rechazar (redirigir a otro nodo) ===\n")

config_reject = {"configurable": {"thread_id": "reject-flow"}}
graph.invoke(
    {"task": "Borrar base de datos de producción", "plan": "", "result": "", "status": "", "log": []},
    config_reject,
)

state = graph.get_state(config_reject)
print(f"Plan: {state.values['plan']}")
print(f"Siguiente: {state.next}")

print("→ Humano rechaza: redirigimos a handle_rejection")
graph.update_state(config_reject, None, as_node="execute_plan")
graph.update_state(
    config_reject,
    {"status": "redirected", "log": ["human_rejected"]},
    as_node="handle_rejection",
)

final_state = graph.get_state(config_reject)
print(f"Resultado: {final_state.values['result']}")
print(f"Status: {final_state.values['status']}")
print(f"Log: {final_state.values['log']}")
# Output esperado:
# === ESCENARIO 1: Aprobar ===
#
# Plan: Plan: ejecutar 'Publicar artículo en blog' con 3 pasos automatizados
# Siguiente: ('execute_plan',)
# → Humano aprueba: continuamos
# Resultado: Ejecutado exitosamente: Plan: ejecutar 'Publicar artículo en blog' con 3 pasos automatizados
#
# === ESCENARIO 2: Rechazar (redirigir a otro nodo) ===
#
# Plan: Plan: ejecutar 'Borrar base de datos de producción' con 3 pasos automatizados
# Siguiente: ('execute_plan',)
# → Humano rechaza: redirigimos a handle_rejection
# Resultado: Tarea cancelada por el usuario
# Status: redirected
# Log: ['plan_created', 'human_rejected']

Los tres caminos:

✅ Aprobar: graph.invoke(None, config) — el grafo continúa normalmente
❌ Rechazar y redirigir: graph.update_state() con as_node para simular que un nodo diferente respondió, cambiando el flujo
⚠️ Aprobar con modificaciones: editar el estado antes de continuar (lo verás en detalle en la cápsula 04)

Command: enviar instrucciones al reanudar

Cuando usas interrupt() dentro de un nodo, el valor de Command(resume=...) se devuelve directamente a la variable que recibe el interrupt(). Con breakpoints, el mecanismo es diferente: usas graph.invoke(None, config) para continuar, o graph.update_state() para modificar el estado antes de continuar.

Pero si combinas breakpoints con interrupt() dentro del nodo, obtienes lo mejor de ambos mundos:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class State(TypedDict):
    task: str
    plan: str
    human_feedback: str
    result: str
    log: Annotated[list[str], operator.add]

def create_plan(state: State) -> dict:
    plan = f"Plan automático para: '{state['task']}'"
    return {"plan": plan, "log": ["plan_created"]}

def request_approval(state: State) -> dict:
    decision = interrupt({
        "question": "¿Apruebas este plan?",
        "plan": state["plan"],
        "options": ["aprobar", "aprobar_con_cambios", "rechazar"],
    })

    if isinstance(decision, dict):
        approved = decision.get("approved", False)
        feedback = decision.get("feedback", "")
    else:
        approved = decision == "aprobar"
        feedback = ""

    if approved:
        return {
            "human_feedback": feedback if feedback else "Aprobado sin cambios",
            "log": ["approved"],
        }
    else:
        return {
            "human_feedback": f"Rechazado: {feedback}",
            "result": "Cancelado por el usuario",
            "log": ["rejected"],
        }

def execute(state: State) -> dict:
    if "Rechazado" in state.get("human_feedback", ""):
        return {"result": "No ejecutado — rechazado", "log": ["skipped"]}
    result = f"Ejecutado: {state['plan']} | Feedback: {state['human_feedback']}"
    return {"result": result, "log": ["executed"]}

graph_builder = StateGraph(State)
graph_builder.add_node("create_plan", create_plan)
graph_builder.add_node("request_approval", request_approval)
graph_builder.add_node("execute", execute)

graph_builder.add_edge(START, "create_plan")
graph_builder.add_edge("create_plan", "request_approval")
graph_builder.add_edge("request_approval", "execute")
graph_builder.add_edge("execute", END)

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

config = {"configurable": {"thread_id": "command-demo"}}

print("=== Paso 1: Ejecutar hasta que pida aprobación ===\n")
graph.invoke(
    {"task": "Migrar base de datos", "plan": "", "human_feedback": "", "result": "", "log": []},
    config,
)

state = graph.get_state(config)
print(f"Plan: {state.values['plan']}")
print(f"Siguiente: {state.next}")

print("\n=== Paso 2: Humano aprueba con feedback ===\n")
result = graph.invoke(
    Command(resume={"approved": True, "feedback": "Procede pero haz backup primero"}),
    config,
)

final = graph.get_state(config)
print(f"Feedback: {final.values['human_feedback']}")
print(f"Resultado: {final.values['result']}")
print(f"Log: {final.values['log']}")
# Output esperado:
# === Paso 1: Ejecutar hasta que pida aprobación ===
#
# Plan: Plan automático para: 'Migrar base de datos'
# Siguiente: ('request_approval',)
#
# === Paso 2: Humano aprueba con feedback ===
#
# Feedback: Procede pero haz backup primero
# Resultado: Ejecutado: Plan automático para: 'Migrar base de datos' | Feedback: Procede pero haz backup primero
# Log: ['plan_created', 'approved', 'executed']

El patrón interrupt() + Command(resume=) te permite enviar datos estructurados de vuelta al nodo: no solo "sí" o "no", sino objetos completos con feedback, instrucciones, y contexto adicional.

Múltiples breakpoints: inspección en cada paso crítico

Puedes configurar breakpoints en múltiples nodos para crear un pipeline con checkpoints humanos:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    request: str
    research: str
    draft: str
    final: str
    log: Annotated[list[str], operator.add]

def research(state: State) -> dict:
    return {
        "research": f"Investigación sobre '{state['request']}': 5 fuentes encontradas",
        "log": ["research_done"],
    }

def draft(state: State) -> dict:
    return {
        "draft": f"Borrador basado en: {state['research'][:40]}...",
        "log": ["draft_done"],
    }

def finalize(state: State) -> dict:
    return {
        "final": f"DOCUMENTO FINAL: {state['draft']}",
        "log": ["finalized"],
    }

graph_builder = StateGraph(State)
graph_builder.add_node("research", research)
graph_builder.add_node("draft", draft)
graph_builder.add_node("finalize", finalize)

graph_builder.add_edge(START, "research")
graph_builder.add_edge("research", "draft")
graph_builder.add_edge("draft", "finalize")
graph_builder.add_edge("finalize", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["research", "draft"],
)

config = {"configurable": {"thread_id": "multi-bp"}}

print("=== Paso 1: Research ejecuta, pausa para review ===\n")
graph.invoke(
    {"request": "Análisis de competencia AI", "research": "", "draft": "", "final": "", "log": []},
    config,
)
state = graph.get_state(config)
print(f"Research: {state.values['research']}")
print(f"Siguiente: {state.next}")

print("\n=== Paso 2: Humano aprueba research → Draft ejecuta, pausa ===\n")
graph.invoke(None, config)
state = graph.get_state(config)
print(f"Draft: {state.values['draft']}")
print(f"Siguiente: {state.next}")

print("\n=== Paso 3: Humano aprueba draft → Finalize ejecuta ===\n")
graph.invoke(None, config)
state = graph.get_state(config)
print(f"Final: {state.values['final']}")
print(f"Siguiente: {state.next}")
print(f"Log: {state.values['log']}")
# Output esperado:
# === Paso 1: Research ejecuta, pausa para review ===
#
# Research: Investigación sobre 'Análisis de competencia AI': 5 fuentes encontradas
# Siguiente: ('draft',)
#
# === Paso 2: Humano aprueba research → Draft ejecuta, pausa ===
#
# Draft: Borrador basado en: Investigación sobre 'Análisis de compet...
# Siguiente: ('finalize',)
#
# === Paso 3: Humano aprueba draft → Finalize ejecuta ===
#
# Final: DOCUMENTO FINAL: Borrador basado en: Investigación sobre 'Análisis de compet...
# Siguiente: ()
# Log: ['research_done', 'draft_done', 'finalized']

El flujo tiene dos checkpoints humanos: uno después de research (¿las fuentes son buenas?) y otro después de draft (¿el borrador es correcto?). Solo después de dos aprobaciones se genera el documento final.

Cuándo usar breakpoints vs interrupt()

La regla general:

Situación	Usa
Necesitas pausar entre nodos sin modificar su código	`interrupt_before` / `interrupt_after`
Necesitas pausar en un punto específico dentro de un nodo	`interrupt()`
Gate de aprobación antes de una acción peligrosa	`interrupt_before`
Review de resultados parciales	`interrupt_after`
Recopilar input del usuario como parte de la lógica	`interrupt()` + `Command(resume=)`
Approval con datos estructurados de vuelta	`interrupt()` + `Command(resume=)`
Múltiples nodos necesitan aprobación, sin tocar su código	`interrupt_before` / `interrupt_after` en lista

En la práctica, los combinas. Un grafo de producción puede tener:

interrupt_before=["send_email", "execute_trade"] — gates automáticos para acciones irreversibles
interrupt() dentro de un nodo de planificación — para pedir feedback detallado al usuario

Workflow completo: Plan → Approve → Execute → Review → Deliver

Un flujo de aprobación realista de producción que combina múltiples técnicas:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class WorkflowState(TypedDict):
    objective: str
    plan: str
    execution_result: str
    review_notes: str
    deliverable: str
    log: Annotated[list[str], operator.add]

def plan_step(state: WorkflowState) -> dict:
    plan = f"Plan para '{state['objective']}':\n"
    plan += "  1. Recopilar datos de 3 fuentes\n"
    plan += "  2. Analizar tendencias\n"
    plan += "  3. Generar visualizaciones\n"
    plan += "  4. Compilar reporte ejecutivo"
    return {"plan": plan, "log": ["plan_created"]}

def approve_step(state: WorkflowState) -> dict:
    decision = interrupt({
        "message": "Revisa el plan antes de ejecutar",
        "plan": state["plan"],
        "action_required": "Responde con {approved: true/false, notes: '...'}",
    })

    if decision.get("approved"):
        notes = decision.get("notes", "Sin notas adicionales")
        return {"review_notes": f"Aprobado: {notes}", "log": ["plan_approved"]}
    else:
        return {
            "review_notes": f"Rechazado: {decision.get('notes', 'Sin razón')}",
            "deliverable": "Workflow cancelado",
            "log": ["plan_rejected"],
        }

def execute_step(state: WorkflowState) -> dict:
    if "Rechazado" in state["review_notes"]:
        return {"execution_result": "No ejecutado", "log": ["execution_skipped"]}
    result = f"Ejecutado exitosamente. Datos de 3 fuentes procesados. "
    result += f"Notas del reviewer: {state['review_notes']}"
    return {"execution_result": result, "log": ["executed"]}

def deliver_step(state: WorkflowState) -> dict:
    if state.get("deliverable") == "Workflow cancelado":
        return {"log": ["cancelled"]}
    deliverable = f"ENTREGABLE: {state['execution_result'][:60]}..."
    return {"deliverable": deliverable, "log": ["delivered"]}

graph_builder = StateGraph(WorkflowState)
graph_builder.add_node("plan", plan_step)
graph_builder.add_node("approve", approve_step)
graph_builder.add_node("execute", execute_step)
graph_builder.add_node("deliver", deliver_step)

graph_builder.add_edge(START, "plan")
graph_builder.add_edge("plan", "approve")
graph_builder.add_edge("approve", "execute")
graph_builder.add_edge("execute", "deliver")
graph_builder.add_edge("deliver", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["execute"],
)

config = {"configurable": {"thread_id": "full-workflow"}}

print("=== PASO 1: Plan se genera, approval se solicita ===\n")
graph.invoke(
    {
        "objective": "Reporte trimestral de ventas AI",
        "plan": "", "execution_result": "",
        "review_notes": "", "deliverable": "", "log": [],
    },
    config,
)

state = graph.get_state(config)
print(f"Estado: esperando aprobación en nodo '{state.next}'")

print("\n=== PASO 2: Humano aprueba con notas ===\n")
graph.invoke(
    Command(resume={"approved": True, "notes": "Incluir datos Q3 también"}),
    config,
)

state = graph.get_state(config)
print(f"Review: {state.values['review_notes']}")
print(f"Ejecución: {state.values['execution_result'][:80]}...")
print(f"Siguiente: {state.next}")

print("\n=== PASO 3: Humano revisa ejecución → continuar a delivery ===\n")
result = graph.invoke(None, config)

final = graph.get_state(config)
print(f"Entregable: {final.values['deliverable']}")
print(f"Log completo: {final.values['log']}")
# Output esperado:
# === PASO 1: Plan se genera, approval se solicita ===
#
# Estado: esperando aprobación en nodo '('approve',)'
#
# === PASO 2: Humano aprueba con notas ===
#
# Review: Aprobado: Incluir datos Q3 también
# Ejecución: Ejecutado exitosamente. Datos de 3 fuentes procesados. Notas del reviewer: A...
# Siguiente: ('deliver',)
#
# === PASO 3: Humano revisa ejecución → continuar a delivery ===
#
# Entregable: ENTREGABLE: Ejecutado exitosamente. Datos de 3 fuentes procesados....
# Log completo: ['plan_created', 'plan_approved', 'executed', 'delivered']

Este workflow tiene dos puntos de control humano:

Aprobación del plan: interrupt() dentro de approve — el humano envía datos estructurados con Command(resume=)
Review de la ejecución: interrupt_after=["execute"] — el humano revisa los resultados antes del delivery

Combinando interrupt_before e interrupt_after

Puedes usar ambos en el mismo grafo:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    data: str
    analyzed: str
    transformed: str
    saved: str
    log: Annotated[list[str], operator.add]

def analyze(state: State) -> dict:
    return {"analyzed": f"Análisis de: {state['data']}", "log": ["analyzed"]}

def transform(state: State) -> dict:
    return {"transformed": f"Transformado: {state['analyzed']}", "log": ["transformed"]}

def save_to_db(state: State) -> dict:
    return {"saved": f"Guardado en DB: {state['transformed'][:30]}...", "log": ["saved"]}

graph_builder = StateGraph(State)
graph_builder.add_node("analyze", analyze)
graph_builder.add_node("transform", transform)
graph_builder.add_node("save_to_db", save_to_db)

graph_builder.add_edge(START, "analyze")
graph_builder.add_edge("analyze", "transform")
graph_builder.add_edge("transform", "save_to_db")
graph_builder.add_edge("save_to_db", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["analyze"],
    interrupt_before=["save_to_db"],
)

config = {"configurable": {"thread_id": "combo-bp"}}

print("=== Paso 1: Analyze ejecuta, pausa después ===")
graph.invoke({"data": "ventas Q4", "analyzed": "", "transformed": "", "saved": "", "log": []}, config)
state = graph.get_state(config)
print(f"  Analyzed: {state.values['analyzed']}")
print(f"  Next: {state.next}")

print("\n=== Paso 2: Aprobamos análisis → Transform ejecuta → Pausa antes de save_to_db ===")
graph.invoke(None, config)
state = graph.get_state(config)
print(f"  Transformed: {state.values['transformed']}")
print(f"  Next: {state.next}")

print("\n=== Paso 3: Aprobamos guardado → save_to_db ejecuta ===")
graph.invoke(None, config)
state = graph.get_state(config)
print(f"  Saved: {state.values['saved']}")
print(f"  Log: {state.values['log']}")
# Output esperado:
# === Paso 1: Analyze ejecuta, pausa después ===
#   Analyzed: Análisis de: ventas Q4
#   Next: ('transform',)
#
# === Paso 2: Aprobamos análisis → Transform ejecuta → Pausa antes de save_to_db ===
#   Transformed: Transformado: Análisis de: ventas Q4
#   Next: ('save_to_db',)
#
# === Paso 3: Aprobamos guardado → save_to_db ejecuta ===
#   Saved: Guardado en DB: Transformado: Análisis de:...
#   Log: ['analyzed', 'transformed', 'saved']

Dos breakpoints diferentes:

interrupt_after=["analyze"] — "déjame ver el análisis antes de transformar"
interrupt_before=["save_to_db"] — "déjame aprobar antes de escribir en la base de datos"

Troubleshooting

Problema 1: "El grafo no se detiene en el breakpoint"

Síntoma: Configuras interrupt_before=["mi_nodo"] pero el grafo ejecuta todo sin pausar.

Causa: El nombre del nodo no coincide con el que registraste en add_node(), o el grafo no tiene checkpointer.

Solución: Verifica que el nombre sea idéntico al que usaste en graph_builder.add_node("nombre_exacto", fn) y que compilaste con checkpointer:

graph_builder.add_node("dangerous_action", my_function)

graph = graph_builder.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["dangerous_action"],
)

Problema 2: "graph.invoke(None, config) no reanuda"

Síntoma: Después del breakpoint, invoke(None, config) no hace nada o lanza error.

Causa: El thread_id en el config no coincide con el de la ejecución original.

Solución: Usa exactamente el mismo config:

config = {"configurable": {"thread_id": "mi-thread"}}
graph.invoke(input, config)
# ... breakpoint ...
graph.invoke(None, config)  # mismo config

Problema 3: "interrupt_before y interrupt_after no funcionan juntos"

Síntoma: Solo uno de los dos breakpoints pausa.

Causa: Esto no es un problema — es el comportamiento esperado. Si tienes interrupt_after=["A"] y interrupt_before=["B"], y A → B es una transición directa, el grafo pausa una vez (después de A, que es lo mismo que antes de B).

Solución: No necesitas ambos para la misma transición. Usa interrupt_after=["A"] O interrupt_before=["B"], no ambos.

Problema 4: "Quiero cancelar la ejecución, no continuar"

Síntoma: El grafo está pausado y quieres abortar completamente.

Causa: No hay un "abort" explícito — pero puedes modificar el estado para redirigir.

Solución: Usa update_state para marcar la tarea como cancelada y deja que el grafo termine limpiamente:

graph.update_state(config, {"status": "cancelled"})
graph.invoke(None, config)

Ejercicios

Ejercicio 1: Breakpoint básico con interrupt_before (Fácil)

Crea un grafo de 3 nodos: prepare → send_email → confirm. Configura un breakpoint antes de send_email. Ejecuta el grafo, verifica que se pausa antes de enviar, inspecciona el estado, y luego continúa.

Ver solución

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    recipient: str
    subject: str
    body: str
    sent: bool
    log: Annotated[list[str], operator.add]

def prepare(state: State) -> dict:
    body = f"Estimado usuario, este es un email sobre: {state['subject']}"
    return {"body": body, "log": ["prepared"]}

def send_email(state: State) -> dict:
    return {"sent": True, "log": ["email_sent"]}

def confirm(state: State) -> dict:
    return {"log": [f"confirmed_to_{state['recipient']}"]}

graph_builder = StateGraph(State)
graph_builder.add_node("prepare", prepare)
graph_builder.add_node("send_email", send_email)
graph_builder.add_node("confirm", confirm)

graph_builder.add_edge(START, "prepare")
graph_builder.add_edge("prepare", "send_email")
graph_builder.add_edge("send_email", "confirm")
graph_builder.add_edge("confirm", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["send_email"],
)

config = {"configurable": {"thread_id": "email-bp"}}
graph.invoke(
    {"recipient": "cliente@empresa.com", "subject": "Oferta especial", "body": "", "sent": False, "log": []},
    config,
)

state = graph.get_state(config)
print(f"Body preparado: {state.values['body']}")
print(f"¿Ya se envió?: {state.values['sent']}")
print(f"Siguiente nodo: {state.next}")

print("\n→ Aprobamos el envío")
graph.invoke(None, config)

state = graph.get_state(config)
print(f"¿Enviado?: {state.values['sent']}")
print(f"Log: {state.values['log']}")
# Output esperado:
# Body preparado: Estimado usuario, este es un email sobre: Oferta especial
# ¿Ya se envió?: False
# Siguiente nodo: ('send_email',)
#
# → Aprobamos el envío
# ¿Enviado?: True
# Log: ['prepared', 'email_sent', 'confirmed_to_cliente@empresa.com']

Ejercicio 2: Review con interrupt_after (Fácil)

Crea un grafo donde un nodo generate_summary genera un resumen y luego se pausa para que el humano lo revise. Si el humano aprueba, el nodo publish lo publica. Usa interrupt_after en generate_summary.

Ver solución

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    topic: str
    summary: str
    published: bool
    log: Annotated[list[str], operator.add]

def generate_summary(state: State) -> dict:
    summary = f"Resumen sobre '{state['topic']}': 3 puntos clave identificados."
    return {"summary": summary, "log": ["summary_generated"]}

def publish(state: State) -> dict:
    return {"published": True, "log": ["published"]}

graph_builder = StateGraph(State)
graph_builder.add_node("generate_summary", generate_summary)
graph_builder.add_node("publish", publish)

graph_builder.add_edge(START, "generate_summary")
graph_builder.add_edge("generate_summary", "publish")
graph_builder.add_edge("publish", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["generate_summary"],
)

config = {"configurable": {"thread_id": "review-summary"}}
graph.invoke(
    {"topic": "LangGraph HITL", "summary": "", "published": False, "log": []},
    config,
)

state = graph.get_state(config)
print(f"Resumen generado: {state.values['summary']}")
print(f"¿Publicado?: {state.values['published']}")
print(f"Siguiente: {state.next}")

print("\n→ Humano revisa y aprueba")
graph.invoke(None, config)

state = graph.get_state(config)
print(f"¿Publicado?: {state.values['published']}")
print(f"Log: {state.values['log']}")
# Output esperado:
# Resumen generado: Resumen sobre 'LangGraph HITL': 3 puntos clave identificados.
# ¿Publicado?: False
# Siguiente: ('publish',)
#
# → Humano revisa y aprueba
# ¿Publicado?: True
# Log: ['summary_generated', 'published']

Ejercicio 3: Approval con Command(resume) (Medio)

Crea un workflow donde un nodo propose_action genera una propuesta y usa interrupt() para pedir aprobación. El humano responde con Command(resume={"approved": True, "priority": "high"}). El nodo siguiente (execute_action) usa la prioridad para decidir cómo ejecutar.

Ver solución

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class State(TypedDict):
    task: str
    proposal: str
    priority: str
    result: str
    log: Annotated[list[str], operator.add]

def propose_action(state: State) -> dict:
    proposal = f"Propuesta: automatizar '{state['task']}'"

    decision = interrupt({
        "message": "Revisa esta propuesta",
        "proposal": proposal,
        "respond_with": "{approved: bool, priority: 'low'|'medium'|'high'}",
    })

    return {
        "proposal": proposal,
        "priority": decision.get("priority", "medium"),
        "log": [f"proposal_{'approved' if decision.get('approved') else 'rejected'}"],
    }

def execute_action(state: State) -> dict:
    if state["priority"] == "high":
        method = "ejecución inmediata con todos los recursos"
    elif state["priority"] == "medium":
        method = "ejecución estándar"
    else:
        method = "ejecución en background, baja prioridad"

    result = f"{state['proposal']} → {method}"
    return {"result": result, "log": ["executed"]}

graph_builder = StateGraph(State)
graph_builder.add_node("propose_action", propose_action)
graph_builder.add_node("execute_action", execute_action)

graph_builder.add_edge(START, "propose_action")
graph_builder.add_edge("propose_action", "execute_action")
graph_builder.add_edge("execute_action", END)

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

config = {"configurable": {"thread_id": "priority-approval"}}

graph.invoke(
    {"task": "Deploy v2.0", "proposal": "", "priority": "", "result": "", "log": []},
    config,
)

state = graph.get_state(config)
print(f"Esperando en: {state.next}")

result = graph.invoke(
    Command(resume={"approved": True, "priority": "high"}),
    config,
)

final = graph.get_state(config)
print(f"Prioridad: {final.values['priority']}")
print(f"Resultado: {final.values['result']}")
print(f"Log: {final.values['log']}")
# Output esperado:
# Esperando en: ('propose_action',)
# Prioridad: high
# Resultado: Propuesta: automatizar 'Deploy v2.0' → ejecución inmediata con todos los recursos
# Log: ['proposal_approved', 'executed']

Ejercicio 4: Pipeline con múltiples breakpoints (Medio)

Crea un pipeline de 4 nodos: collect_data → clean_data → train_model → deploy_model. Configura breakpoints después de clean_data (para revisar la calidad de datos) y antes de deploy_model (para aprobar el deployment). Ejecuta todo el flujo con las 3 invocaciones necesarias.

Ver solución

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    dataset_name: str
    raw_rows: int
    clean_rows: int
    model_accuracy: float
    deployed: bool
    log: Annotated[list[str], operator.add]

def collect_data(state: State) -> dict:
    return {"raw_rows": 10000, "log": ["data_collected"]}

def clean_data(state: State) -> dict:
    clean = int(state["raw_rows"] * 0.85)
    return {"clean_rows": clean, "log": [f"cleaned_{clean}_rows"]}

def train_model(state: State) -> dict:
    accuracy = 0.92 if state["clean_rows"] > 5000 else 0.75
    return {"model_accuracy": accuracy, "log": [f"trained_accuracy_{accuracy}"]}

def deploy_model(state: State) -> dict:
    return {"deployed": True, "log": ["deployed"]}

graph_builder = StateGraph(State)
graph_builder.add_node("collect_data", collect_data)
graph_builder.add_node("clean_data", clean_data)
graph_builder.add_node("train_model", train_model)
graph_builder.add_node("deploy_model", deploy_model)

graph_builder.add_edge(START, "collect_data")
graph_builder.add_edge("collect_data", "clean_data")
graph_builder.add_edge("clean_data", "train_model")
graph_builder.add_edge("train_model", "deploy_model")
graph_builder.add_edge("deploy_model", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_after=["clean_data"],
    interrupt_before=["deploy_model"],
)

config = {"configurable": {"thread_id": "ml-pipeline"}}

print("=== Fase 1: Collect + Clean → Pausa para review de datos ===")
graph.invoke(
    {"dataset_name": "ventas_2025", "raw_rows": 0, "clean_rows": 0,
     "model_accuracy": 0.0, "deployed": False, "log": []},
    config,
)
state = graph.get_state(config)
print(f"  Raw: {state.values['raw_rows']} → Clean: {state.values['clean_rows']}")
print(f"  Next: {state.next}")

print("\n=== Fase 2: Train → Pausa antes de deploy ===")
graph.invoke(None, config)
state = graph.get_state(config)
print(f"  Accuracy: {state.values['model_accuracy']}")
print(f"  Next: {state.next}")

print("\n=== Fase 3: Deploy aprobado ===")
graph.invoke(None, config)
state = graph.get_state(config)
print(f"  Deployed: {state.values['deployed']}")
print(f"  Log: {state.values['log']}")
# Output esperado:
# === Fase 1: Collect + Clean → Pausa para review de datos ===
#   Raw: 10000 → Clean: 8500
#   Next: ('train_model',)
#
# === Fase 2: Train → Pausa antes de deploy ===
#   Accuracy: 0.92
#   Next: ('deploy_model',)
#
# === Fase 3: Deploy aprobado ===
#   Deployed: True
#   Log: ['data_collected', 'cleaned_8500_rows', 'trained_accuracy_0.92', 'deployed']

Ejercicio 5: Approval workflow con rechazo y redirección (Avanzado)

Crea un workflow con 4 nodos: draft_email → review_email → send_email → log_sent. El nodo review_email usa interrupt() para pedir aprobación. Si el humano rechaza, el flujo debe regresar a draft_email con feedback para rehacer el borrador. Implementa un ciclo: draft → review → (rechazo → draft de nuevo) → review → (aprobación) → send → log.

Ver solución

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class State(TypedDict):
    recipient: str
    draft: str
    feedback: str
    revision_count: int
    sent: bool
    log: Annotated[list[str], operator.add]

def draft_email(state: State) -> dict:
    revision = state.get("revision_count", 0)
    feedback = state.get("feedback", "")

    if revision == 0:
        draft = f"Borrador v1 para {state['recipient']}: Propuesta de colaboración"
    else:
        draft = f"Borrador v{revision + 1} para {state['recipient']}: "
        draft += f"(Incorporando feedback: {feedback[:40]}...)"

    return {
        "draft": draft,
        "revision_count": revision + 1,
        "log": [f"draft_v{revision + 1}"],
    }

def review_email(state: State) -> dict:
    decision = interrupt({
        "message": "Revisa este borrador de email",
        "draft": state["draft"],
        "options": "Responde {approved: true} o {approved: false, feedback: '...'}",
    })

    if decision.get("approved"):
        return {"log": ["review_approved"]}
    else:
        return {
            "feedback": decision.get("feedback", "Necesita mejoras"),
            "log": ["review_rejected"],
        }

def send_email(state: State) -> dict:
    return {"sent": True, "log": ["email_sent"]}

def log_sent(state: State) -> dict:
    return {"log": [f"logged_email_to_{state['recipient']}"]}

def route_after_review(state: State) -> str:
    if state["log"][-1] == "review_rejected":
        return "draft_email"
    return "send_email"

graph_builder = StateGraph(State)
graph_builder.add_node("draft_email", draft_email)
graph_builder.add_node("review_email", review_email)
graph_builder.add_node("send_email", send_email)
graph_builder.add_node("log_sent", log_sent)

graph_builder.add_edge(START, "draft_email")
graph_builder.add_edge("draft_email", "review_email")
graph_builder.add_conditional_edges("review_email", route_after_review)
graph_builder.add_edge("send_email", "log_sent")
graph_builder.add_edge("log_sent", END)

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

config = {"configurable": {"thread_id": "email-cycle"}}

print("=== Ronda 1: Draft → Review ===")
graph.invoke(
    {"recipient": "ceo@empresa.com", "draft": "", "feedback": "",
     "revision_count": 0, "sent": False, "log": []},
    config,
)
state = graph.get_state(config)
print(f"Draft: {state.values['draft']}")

print("\n=== Ronda 1: Humano rechaza ===")
graph.invoke(
    Command(resume={"approved": False, "feedback": "Tono muy informal, hazlo más profesional"}),
    config,
)
state = graph.get_state(config)
print(f"Feedback: {state.values['feedback']}")
print(f"Draft actual: {state.values['draft']}")
print(f"Revisiones: {state.values['revision_count']}")

print("\n=== Ronda 2: Humano aprueba ===")
graph.invoke(
    Command(resume={"approved": True}),
    config,
)
final = graph.get_state(config)
print(f"¿Enviado?: {final.values['sent']}")
print(f"Log completo: {final.values['log']}")
# Output esperado:
# === Ronda 1: Draft → Review ===
# Draft: Borrador v1 para ceo@empresa.com: Propuesta de colaboración
#
# === Ronda 1: Humano rechaza ===
# Feedback: Tono muy informal, hazlo más profesional
# Draft actual: Borrador v2 para ceo@empresa.com: (Incorporando feedback: Tono muy informal, hazlo más profesion...)
# Revisiones: 2
#
# === Ronda 2: Humano aprueba ===
# ¿Enviado?: True
# Log completo: ['draft_v1', 'review_rejected', 'draft_v2', 'review_approved', 'email_sent', 'logged_email_to_ceo@empresa.com']

Ejercicio 6: Comparar breakpoints vs interrupt() en el mismo grafo (Avanzado)

Crea un grafo con 4 nodos: research → validate → execute → report. Usa interrupt_before en execute (gate de aprobación a nivel de compilación) y interrupt() dentro de validate para pedir al humano que confirme los datos. Implementa ambos flujos de pausa y reanudación.

Ver solución

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

class State(TypedDict):
    topic: str
    findings: str
    validated: bool
    human_confirmation: str
    execution_result: str
    report: str
    log: Annotated[list[str], operator.add]

def research(state: State) -> dict:
    findings = f"3 hallazgos sobre '{state['topic']}': tendencia alcista, riesgo moderado, oportunidad en Q2"
    return {"findings": findings, "log": ["researched"]}

def validate(state: State) -> dict:
    confirmation = interrupt({
        "message": "Valida estos hallazgos antes de continuar",
        "findings": state["findings"],
        "action": "Responde {confirmed: true/false, notes: '...'}",
    })

    confirmed = confirmation.get("confirmed", False)
    notes = confirmation.get("notes", "")
    return {
        "validated": confirmed,
        "human_confirmation": notes if notes else ("Validado" if confirmed else "Rechazado"),
        "log": [f"validation_{'passed' if confirmed else 'failed'}"],
    }

def execute(state: State) -> dict:
    if not state["validated"]:
        return {"execution_result": "No ejecutado — validación falló", "log": ["skipped"]}
    return {
        "execution_result": f"Ejecutado con confirmación: {state['human_confirmation']}",
        "log": ["executed"],
    }

def report(state: State) -> dict:
    return {"report": f"Reporte: {state['execution_result']}", "log": ["reported"]}

graph_builder = StateGraph(State)
graph_builder.add_node("research", research)
graph_builder.add_node("validate", validate)
graph_builder.add_node("execute", execute)
graph_builder.add_node("report", report)

graph_builder.add_edge(START, "research")
graph_builder.add_edge("research", "validate")
graph_builder.add_edge("validate", "execute")
graph_builder.add_edge("execute", "report")
graph_builder.add_edge("report", END)

checkpointer = MemorySaver()
graph = graph_builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["execute"],
)

config = {"configurable": {"thread_id": "combo-exercise"}}

print("=== Paso 1: Research completa, validate pide input (interrupt) ===")
graph.invoke(
    {"topic": "AI market 2025", "findings": "", "validated": False,
     "human_confirmation": "", "execution_result": "", "report": "", "log": []},
    config,
)
state = graph.get_state(config)
print(f"  Esperando en: {state.next}")

print("\n=== Paso 2: Humano confirma hallazgos (Command resume) ===")
graph.invoke(
    Command(resume={"confirmed": True, "notes": "Datos verificados con fuente interna"}),
    config,
)
state = graph.get_state(config)
print(f"  Validado: {state.values['validated']}")
print(f"  Confirmación: {state.values['human_confirmation']}")
print(f"  Esperando en: {state.next}")

print("\n=== Paso 3: Humano aprueba ejecución (breakpoint) ===")
graph.invoke(None, config)
final = graph.get_state(config)
print(f"  Resultado: {final.values['execution_result']}")
print(f"  Reporte: {final.values['report']}")
print(f"  Log: {final.values['log']}")
# Output esperado:
# === Paso 1: Research completa, validate pide input (interrupt) ===
#   Esperando en: ('validate',)
#
# === Paso 2: Humano confirma hallazgos (Command resume) ===
#   Validado: True
#   Confirmación: Datos verificados con fuente interna
#   Esperando en: ('execute',)
#
# === Paso 3: Humano aprueba ejecución (breakpoint) ===
#   Resultado: Ejecutado con confirmación: Datos verificados con fuente interna
#   Reporte: Reporte: Ejecutado con confirmación: Datos verificados con fuente interna
#   Log: ['researched', 'validation_passed', 'executed', 'reported']

Resumen

En esta cápsula aprendiste:

Breakpoints pausan entre nodos, sin modificar el código del nodo. Se configuran al compilar: interrupt_before=["nodo"] pausa antes, interrupt_after=["nodo"] pausa después. El nodo no sabe que hay un breakpoint
interrupt() pausa dentro de un nodo, en el punto exacto que tú elijas. Es más granular que breakpoints y te permite recopilar datos del usuario con Command(resume=)
El flujo de aprobación tiene tres caminos: aprobar (graph.invoke(None, config)), rechazar y redirigir (graph.update_state() con as_node), o aprobar con modificaciones (editar estado antes de continuar)
Command(resume=) permite enviar datos estructurados de vuelta al nodo que llamó interrupt(). No solo "sí/no" — puedes enviar objetos con feedback, prioridad, instrucciones
Puedes combinar múltiples breakpoints en el mismo grafo para crear pipelines con checkpoints humanos en cada paso crítico
Breakpoints + interrupt() se complementan: usa breakpoints para gates de aprobación a nivel de arquitectura, usa interrupt() para interacciones detalladas dentro de un nodo
Todo requiere checkpointer. Sin checkpointer, el estado se pierde entre la pausa y la reanudación. Cada ejemplo de HITL debe compilar con checkpointer=MemorySaver() (desarrollo) o un checkpointer durable (producción)

Próxima cápsula: Editable State — no solo aprobar o rechazar, sino inspeccionar y modificar el estado del agente durante una pausa. Corregir datos incorrectos, agregar contexto, cambiar la dirección de la ejecución.

Recursos adicionales

LangGraph — Human-in-the-Loop — Conceptos oficiales de HITL en LangGraph
How to add breakpoints — Guía práctica para configurar interrupt_before e interrupt_after
How to wait for user input — Patrones para recopilar input humano durante la ejecución
LangGraph — Command — Documentación del objeto Command para enviar instrucciones al grafo
How to review tool calls — Patrón de aprobación antes de ejecutar tool calls

Módulo 9 — LangChain & LangGraph: From Chains to Agents

← AnteriorInterrupts: Pausar Ejecución Siguiente →Editable State: Modificar Estado Durante Pausa