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 combinam interrupts com roteamento condicional para criar processos estruturados com humano no loop. Esta lição cobre interrupts antes do nó, interrupts após o nó e padrões dinâmicos de aprovação.

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

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

O agente apresenta uma proposta, o humano decide e a execução segue de acordo.

Interrupção Antes do Nó

Interrompa antes de um nó executar, o humano decide se o nó deve rodar:

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

Um interrupt antes do nó impede que uma ação aconteça até ser aprovada. Este é o padrão de aprovação mais comum: ele evita ações não autorizadas ou arriscadas.

Interrupção Após o Nó

Interrompa depois de um nó executar, o humano revisa o resultado:

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

Interrupts após o nó são úteis para revisão de conteúdo, validação de resultados e qualquer fluxo de trabalho em que o humano deva verificar a saída antes de ela ser usada.

Roteamento Condicional de Aprovação

Roteie com base no que o humano aprovou ou rejeitou:

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

Sempre limite os laços de revisão. Sem um máximo, um revisor insatisfeito poderia ficar em laço para sempre.

Fluxo de Escalonamento

Roteie para um revisor sênior se o júnior rejeitar ou se a solicitação for de alto risco:

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

Inclua os dados relevantes na solicitação de aprovação:

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

Trate o caso em que o humano nunca responde:

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

O LangGraph não tem timeout integrado para interrupts. Implemente a lógica de timeout na camada da aplicação verificando há quanto tempo um thread está pausado.

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