intermediate⏱35 minutesLección 6 de 10

Flujos de Aprobación

Implementa flujos de aprobación estructurados con interrupciones condicionales, interrupciones antes/después de nodos y enrutamiento dinámico humano-en-el-bucle.

Flujos de Aprobación

Approval workflows combine interrupts with conditional routing to create structured human-in-the-loop processes. This lesson covers before-node interrupts, after-node interrupts, and dynamic approval patterns.

Visión General del Patrón de Aprobación

Agent proposes action → Human reviews → Approve? → Yes → Execute
                                              ↓ No
                                         Reject → Provide feedback → Revise or Stop

The agent presents a proposal, the human decides, and execution follows accordingly.

Interrupción Antes del Nodo

Interrupt before a node executes — the human decides whether the node should run:

python

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

class ApprovalState(TypedDict):
    query: str
    search_plan: str
    approved: bool
    feedback: str
    result: str

def plan_node(state: ApprovalState) -> dict:
    plan = f"Search for: {state['query']}"
    return {"search_plan": plan}

def approve_search(state: ApprovalState) -> dict:
    # BEFORE interrupt — human decides if search should happen
    response = interrupt({
        "type": "approval",
        "agent_plan": state["search_plan"],
        "prompt": "Approve this search operation?"
    })
    return {"approved": response.get("approved", False),
            "feedback": response.get("feedback")}

def execute_search(state: ApprovalState) -> dict:
    if not state["approved"]:
        return {"result": f"Rejected. Feedback: {state.get('feedback')}"}
    return {"result": f"Executed: {state['search_plan']}"}

def after_search_review(state: ApprovalState) -> dict:
    # AFTER interrupt — human reviews results
    response = interrupt({
        "type": "review",
        "result": state["result"],
        "prompt": "Are these results acceptable?"
    })
    return {"approved": response.get("approved", False)}

builder = StateGraph(ApprovalState)
builder.add_node("plan", plan_node)
builder.add_node("approve_search", approve_search)
builder.add_node("execute", execute_search)

builder.add_edge(START, "plan")
builder.add_edge("plan", "approve_search")
builder.add_edge("approve_search", "execute")
builder.add_edge("execute", END)

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

[!NOTA] A before-node interrupt prevents an action from happening until approved. This is the most common approval pattern — it prevents unauthorized or risky actions.

Interrupción Después del Nodo

Interrupt after a node executes — the human reviews the result:

python

def draft_email(state: EmailState) -> dict:
    email = llm.invoke(f"Draft email about: {state['topic']}")
    return {"draft": email.content}

def review_draft(state: EmailState) -> dict:
    # AFTER interrupt — human reviews the draft
    response = interrupt({
        "type": "review",
        "draft": state["draft"],
        "prompt": "Edit or approve this email draft?"
    })
    return {
        "approved": response.get("approved", False),
        "revision": response.get("revision", "")
    }

def finalize_or_revise(state: EmailState) -> dict:
    if state["approved"]:
        return {"final": state["draft"]}
    return {"draft": state["revision"], "approved": False}

[!CONSEJO] After-node interrupts are useful for content review, result validation, and any workflow where the human should verify output before it's used.

Enrutamiento Condicional de Aprobación

Route based on whether the human approved or rejected:

python

def approval_router(state: ApprovalState) -> str:
    if state["approved"]:
        return "approved"
    return "rejected"

builder.add_conditional_edges(
    "human_review",
    approval_router,
    {
        "approved": "execute",
        "rejected": "revise_plan"  # Send back for revision
    }
)

Patrón de Bucle de Revisión

python

class RevisionState(TypedDict):
    draft: str
    approved: bool
    revision_count: int

def generate_draft(state: RevisionState) -> dict:
    draft = llm.invoke(f"Write content about: {state['topic']}")
    return {"draft": draft.content}

def human_review(state: RevisionState) -> dict:
    response = interrupt({
        "draft": state["draft"],
        "revision_count": state["revision_count"],
        "prompt": "Approve, or provide revision instructions."
    })
    return {
        "approved": response.get("approved", False),
        "feedback": response.get("feedback", ""),
        "revision_count": state["revision_count"] + 1
    }

def revise_draft(state: RevisionState) -> dict:
    revised = llm.invoke(
        f"Revise this draft based on feedback:\n"
        f"Draft: {state['draft']}\nFeedback: {state['feedback']}"
    )
    return {"draft": revised.content}

def revision_router(state: RevisionState) -> str:
    if state["approved"]:
        return "finalize"
    if state["revision_count"] >= 5:
        return "max_revisions"
    return "revise"

builder.add_conditional_edges(
    "human_review",
    revision_router,
    {
        "finalize": "finalize",
        "revise": "revise_draft",
        "max_revisions": "error_handler"
    }
)
builder.add_edge("revise_draft", "human_review")  # Loop back for another review

[!ADVERTENCIA] Always cap revision loops. Without a maximum, an unsatisfied reviewer could loop forever.

Flujo de Escalada

Route to a senior reviewer if junior rejects or if the request is high-risk:

python

class EscalationState(TypedDict):
    request: str
    risk_level: str
    junior_approved: bool
    senior_approved: bool

def junior_review(state: EscalationState) -> dict:
    response = interrupt({"request": state["request"], "risk": state["risk_level"]})
    return {"junior_approved": response.get("approved", False)}

def escalate(state: EscalationState) -> dict:
    response = interrupt({"request": state["request"],
                          "risk": state["risk_level"],
                          "note": "Escalated from junior review"})
    return {"senior_approved": response.get("approved", False)}

def escalation_router(state: EscalationState) -> str:
    if state["junior_approved"]:
        return "approved"
    if state["risk_level"] == "high":
        return "escalate"
    return "rejected"

builder.add_conditional_edges(
    "junior_review",
    escalation_router,
    {
        "approved": "execute",
        "escalate": "senior_review",
        "rejected": "reject_node"
    }
)

Aprobación con Datos Dinámicos

Include relevant data in the approval request:

python

def approval_with_context(state: State) -> dict:
    # Gather context for the human reviewer
    context = {
        "request_id": state["request_id"],
        "requester": state["user_name"],
        "action": state["proposed_action"],
        "cost_estimate": state.get("estimated_cost", "Unknown"),
        "similar_approved_count": state.get("similar_count", 0),
        "policy_reference": "Policy #42: External API calls require approval",
        "timestamp": datetime.now().isoformat()
    }
    response = interrupt(context)
    return {"approval_response": response}

Timeout para Aprobaciones

Handle the case where the human never responds:

python

def approval_with_timeout(state: State) -> dict:
    # Set a timeout via configuration
    config = state.get("config", {})
    timeout = config.get("approval_timeout_minutes", 60)

    response = interrupt({
        "prompt": "Approve within {timeout} minutes or it will be auto-rejected.",
        "request": state["request"]
    })

    # If human doesn't respond within timeout,
    # calling invoke again with a timeout command handles it
    return {"human_response": response}

[!NOTA] LangGraph does not have built-in timeout for interrupts. Implement timeout logic at the application layer by checking how long a thread has been paused.

Ejemplo Completo de Flujo de Aprobación

python

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

class WorkflowState(TypedDict):
    request: str
    risk_score: float
    draft_response: str
    approved: bool
    feedback: str
    result: str
    step: str

def analyze_request(state: WorkflowState) -> dict:
    risk = len(state["request"]) * 0.01  # Simplified risk calc
    return {"risk_score": min(risk, 1.0), "step": "analyzed"}

def generate_draft(state: WorkflowState) -> dict:
    draft = f"Response to: {state['request']}"
    return {"draft_response": draft, "step": "drafted"}

def human_approval(state: WorkflowState) -> dict:
    response = interrupt({
        "request": state["request"],
        "draft": state["draft_response"],
        "risk": state["risk_score"],
        "timestamp": datetime.now().isoformat()
    })
    return {"approved": response.get("approved", False),
            "feedback": response.get("feedback", ""),
            "step": "reviewed"}

def execute(state: WorkflowState) -> dict:
    if state["approved"]:
        return {"result": state["draft_response"], "step": "completed"}
    return {"result": f"Rejected: {state.get('feedback')}", "step": "rejected"}

builder = StateGraph(WorkflowState)
builder.add_node("analyze", analyze_request)
builder.add_node("draft", generate_draft)
builder.add_node("approve", human_approval)
builder.add_node("execute", execute)
builder.add_edge(START, "analyze")
builder.add_edge("analyze", "draft")
builder.add_edge("draft", "approve")
builder.add_edge("approve", "execute")
builder.add_edge("execute", END)

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

Preguntas Prácticas

Practice Question

What is a before-node interrupt?

Practice Question

What is an after-node interrupt useful for?

Practice Question

How do you create a revision loop in an approval workflow?

Practice Question

Why should revision loops have a maximum count?

Practice Question

What is an escalation workflow?

Practice Question

How do you include context for the human reviewer in an interrupt?

Practice Question

What determines which path the graph takes after an approval interrupt?

Practice Question

What happens when a human rejects a proposal in an approval workflow?

Practice Question

Can an approval workflow have multiple sequential approval steps?

Practice Question

How does the approval workflow pattern enhance LangGraph agents?

[!ÉXITO]

Conclusiones Clave

Before-node interrupts prevent actions until human approval
After-node interrupts let humans review results before proceeding
Conditional edges route based on approval/rejection decisions
Revision loops cycle until approval or max revisions reached
Escalation workflows route to senior reviewers when needed
Pass rich context in interrupt() to help humans make informed decisions
Always cap revision and escalation loops with maximum counts
Multiple sequential approvals can be chained in a single graph

Progreso60%

Humano-en-el-bucle con Interrupciones

Fundamentos de Multi-Agente

Contenido del Curso

1. Gestión Avanzada de Estado 2. Persistencia y Puntos de Control 3. Memoria de Conversación 4. Memoria de Resumen 5. Humano-en-el-bucle con Interrupciones 6. Flujos de Aprobación 7. Fundamentos de Multi-Agente 8. Comunicación y Supervisión entre Agentes 9. Streaming de Tokens 10. Proyecto Intermedio: Sistema de Investigación Multi-Agente