intermediate⏱35 minutesLição 6 de 10

Fluxos de Aprovação

Implemente fluxos de aprovação estruturados com interrupções condicionais, interrupções antes/depois de nós e roteamento dinâmico humano-no-circuito.

Fluxos de Aprovação

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.

Visão Geral do Padrão de Aprovação

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.

Interrupção Antes do Nó

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.

Interrupção Após o Nó

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}

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

Roteamento Condicional de Aprovação

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
    }
)

Padrão de Loop de Revisão

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

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

Fluxo de Escalonamento

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"
    }
)

Aprovação com Dados 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 Aprovações

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.

Exemplo Completo de Fluxo de Aprovação

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())

Perguntas Práticas

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?

[!SUCESSO]

Principais Conclusões

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

Progresso60%

Humano-no-circuito com Interrupções

Fundamentos de Multi-Agentes

Contéudo do Curso

1. Gerenciamento Avançado de Estado 2. Persistência e Checkpointing 3. Memória de Conversação 4. Memória de Resumo 5. Humano-no-circuito com Interrupções 6. Fluxos de Aprovação 7. Fundamentos de Multi-Agentes 8. Comunicação e Supervisão entre Agentes 9. Streaming de Tokens 10. Projeto Intermediário: Sistema de Pesquisa Multi-Agente