Humano-no-circuito com Interrupções

Implemente fluxos de trabalho humano-no-circuito usando interrupções, pause a execução para entrada do usuário e retome a execução do grafo.

A função interrupt() do LangGraph pausa a execução do grafo e aguarda a entrada humana. Isso viabiliza fluxos de aprovação, solicitações de esclarecimento e qualquer cenário em que um humano precise revisar ou contribuir antes de o agente continuar.

O que é uma Interrupção?

Um interrupt pausa a execução do grafo em um nó específico, mostra informações de estado ao usuário e aguarda que o usuário forneça a entrada antes de retomar.

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

A Função 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() lança uma exceção especial que pausa o grafo. A função não continua executando até que o grafo seja retomado com invoke() ou update_state().

Configurando Interrupções

Interrupts exigem um checkpointer (persistência):

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

Executando o Grafo com Interrupções

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

Ao retomar, passe None como entrada (e não um novo estado) e forneça a resposta do interrupt através do config ou por meio de Command. O grafo continua de onde pausou.

Usando Command para Retomar

A classe Command oferece uma forma mais limpa de retomar:

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

💡Dica

Command permite fornecer a resposta do interrupt e atualizar o estado simultaneamente. Use update para adicionar metadados sobre a revisão humana.

Múltiplos Pontos de Interrupção

Um grafo pode ter múltiplos pontos 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 é retomado de forma independente. O grafo sabe em qual interrupt está pausado.

Mensagens de Interrupção Dinâmicas

O valor passado para interrupt() pode ser qualquer dado serializável em 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}

Interrupção com Validação

Valide a 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

⚠️Aviso

Tenha cuidado com laços de validação. Cada tentativa inválida cria um novo checkpoint. Considere limitar as tentativas de nova prova para evitar frustração.

Exemplo Completo: Moderação de Conteúdo

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 Pausados

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
    )