O que langgraph-human-in-the-loop faz?
INVOQUE ESTA HABILIDADE ao implementar padrões de humano no loop, pausar para aprovação ou lidar com erros em LangGraph. Abrange interrupt(), Command(resume=...), fluxos de aprovação/validação e a estratégia de tratamento de erros em 4 níveis.
Visão Geral
Esta habilidade descreve como implementar padrões de humano no loop (human-in-the-loop) em LangGraph. Esses padrões permitem que um agente pause a execução, aguarde por entrada humana e continue com base nessa entrada. Isso é essencial para construir sistemas confiáveis onde ações críticas ou resultados inesperados requerem revisão humana.
Plataformas Suportadas
- LangGraph (Python e TypeScript)
- LangGraph Cloud (para implantação em produção com suporte a persistência)
Padrões Principais
1. Pausando com interrupt()
Use a função interrupt() para pausar a execução do grafo. O grafo salvará seu estado atual e retornará o controle ao chamador (por exemplo, uma interface de usuário ou sistema de coordenador).
from langgraph.types import interrupt
def human_review_node(state):
# Pausar e enviar dados para revisão
review_data = {
"summary": state["summary"],
"next_action": state["next_action"]
}
interrupt(review_data) # Pausa aqui
# Continua após o humano fornecer entrada (via Command(resume=...))
return state
2. Retomando com Command(resume=...)
Para retomar um grafo pausado, use Command com resume. Isso fornece a entrada do humano de volta para o nó onde ocorreu o interrupt().
from langgraph.types import Command
# Exemplo: chamada de retomada
command = Command(resume={"approved": True, "feedback": "Looks good"})
3. Fluxos de Aprovação / Validação
Combine interrupt() com verificações condicionais para criar fluxos de aprovação. Por exemplo, um nó pode verificar se uma ação requer aprovação e, em caso positivo, pausar.
def check_approval(state):
if state["requires_approval"]:
response = interrupt({"action": state["action"], "reason": state["reason"]})
if not response.get("approved"):
state["error"] = "Ação rejeitada pelo humano"
return state
return state
4. Estratégia de Tratamento de Erros em 4 Níveis
A estratégia de tratamento de erros em 4 níveis garante que o sistema possa se recuperar de falhas de forma graciosa:
- Nível 1 – Erros recuperáveis: são erros que podem ser corrigidos automaticamente (ex.: timeout de API, tentar novamente). O grafo pode tentar novamente sem intervenção.
- Nível 2 – Erros que requerem confirmação: ações arriscadas que precisam de aprovação humana (ex.: exclusão de dados, grandes transações). O grafo pausa e espera aprovação.
- Nível 3 – Erros que requerem correção: situações onde o humano deve fornecer dados corretivos (ex.: entrada inválida, dados ausentes). O grafo pausa e o humano edita.
- Nível 4 – Erros fatais: falhas que não podem ser resolvidas (ex.: bug interno, dados corrompidos). O grafo registra o erro e aborta, notificando o humano.
Exemplo de Fluxo Completo
from langgraph.graph import StateGraph
from langgraph.types import interrupt, Command
class GraphState(TypedDict):
data: str
requires_review: bool
approved: bool
def process_data(state):
# Processar dados
state["data"] = state["data"].upper()
if state["requires_review"]:
result = interrupt({"data": state["data"]})
state["approved"] = result.get("approved", False)
return state
builder = StateGraph(GraphState)
builder.add_node("process", process_data)
builder.set_entry_point("process")
builder.set_finish_point("process")
graph = builder.compile()
# Execução inicial
initial_state = {"data": "hello", "requires_review": True}
for event in graph.stream(initial_state):
print(event)
Considerações de Implantação
- Persistência: Certifique-se de que seu gráfico use um checkpoint (ex.:
MemorySaver) para reter o estado durante pausas. - Interface de Usuário: A interface que invoca o grafo deve ser capaz de lidar com interrupções e exibir prompts para o humano.
- Segurança: Valide todas as entradas humanas para evitar injeção ou manipulação.