Neste artigo, você aprenderá como implementar interrupções gerenciadas pelo estado no LangGraph para que o fluxo de trabalho de um agente possa fazer uma pausa para aprovação humana antes de retomar a execução.
Os tópicos que cobriremos incluem:
- O que são interrupções gerenciadas pelo estado e por que são importantes em sistemas de IA de agente.
- Como definir um fluxo de trabalho LangGraph simples com um estado de agente compartilhado e nós executáveis.
- Como pausar a execução, atualizar o estado salvo com aprovação humana e retomar o fluxo de trabalho.
Proceed lendo para todas as informações.
Construindo uma porta de aprovação ‘human-in-the-loop’ para agentes autônomos
Imagem por Editor
Introdução
Em sistemas de IA agentesquando o pipeline de execução de um agente é interrompido intencionalmente, temos o que é conhecido como interrupção gerenciada pelo estado. Assim como um videogame salvo, o “estado” de um agente pausado – suas variáveis ativas, contexto, memória e ações planejadas – é salvo persistentemente, com o agente colocado em estado de suspensão ou espera até que um gatilho externo retome sua execução.
A importância das interrupções geridas pelo Estado cresceu juntamente com o progresso em aplicações de IA altamente autónomas e baseadas em agentes, por diversas razões. Eles não apenas agem como barreiras de segurança eficazes para a recuperação de ações que de outra forma seriam irreversíveis em ambientes de alto risco, mas também permitem aprovação humana e correção. Um supervisor humano pode reconfigurar o estado de um agente pausado e evitar consequências indesejadas antes que ações sejam executadas com base em uma resposta incorreta.
LangGraphuma biblioteca de código aberto para a construção de aplicativos de modelo de linguagem grande (LLM) com estado, oferece suporte a fluxos de trabalho baseados em agentes com mecanismos humanos no circuito e interrupções gerenciadas pelo estado, melhorando assim a robustez contra erros.
Este artigo reúne todos esses elementos e mostra, passo a passo, como implementar interrupções gerenciadas por estado usando LangGraph em Python sob uma abordagem human-in-the-loop. Embora a maioria dos processos de exemplo definidos abaixo devam ser automatizados por um agente, também mostraremos como fazer o fluxo de trabalho parar em um ponto-chave onde a revisão humana é necessária antes que a execução seja retomada.
Guia passo a passo
Primeiro, nós pip set up langgraph e faça as importações necessárias para este exemplo prático:
de digitando importar DigitadoDict de langógrafo.gráfico importar StateGraph, FIM de langógrafo.posto de controle.memória importar Economizador de memória |
Observe que uma das courses importadas é chamada StateGraph. LangGraph usa gráficos de estado para modelar fluxos de trabalho cíclicos e complexos que envolvem agentes. Há estados representando a memória compartilhada do sistema (também conhecida como carga útil de dados) e nós representando ações que definem a lógica de execução usada para atualizar esse estado. Tanto os estados quanto os nós precisam ser explicitamente definidos e verificados. Vamos fazer isso agora.
aula Estado do Agente(DigitadoDict): rascunho: str aprovado: bool enviado: bool |
O estado do agente é estruturado de forma semelhante a um dicionário Python porque herda de TypedDict. O estado atua como nosso “arquivo salvo” à medida que é passado entre os nós.
Em relação aos nós, definiremos dois deles, cada um representando uma ação: redigir um e-mail e enviá-lo.
definição rascunho_node(estado: Estado do Agente): imprimir(“(Agente): Redigindo o e-mail…”) # O agente cria um rascunho e atualiza o estado retornar {“rascunho”: “Olá! A atualização do seu servidor está pronta para ser implantada.”, “aprovado”: Falso, “enviado”: Falso} definição send_node(estado: Estado do Agente): imprimir(f“(Agente): Acordando! Verificando o standing de aprovação…”) se estado.pegar(“aprovado”): imprimir(“(Sistema): ENVIANDO E-MAIL ->”, estado(“rascunho”)) retornar {“enviado”: Verdadeiro} outro: imprimir(“(Sistema): Rascunho foi rejeitado. E-mail anulado.”) retornar {“enviado”: Falso} |
O draft_node() A função simula uma ação do agente que redige um e-mail. Para fazer o agente executar uma ação actual, você substituiria o print() instruções que simulam o comportamento com instruções reais que o executam. O detalhe principal a ser observado aqui é o objeto retornado pela função: um dicionário cujos campos correspondem aos da classe de estado do agente que definimos anteriormente.
Enquanto isso, o send_node() função simula a ação de envio do e-mail. Mas há um problema: a lógica central do mecanismo humano no circuito reside aqui, especificamente na verificação do authorized standing. Somente se o authorized campo foi definido como True – por um ser humano, como veremos, ou por uma intervenção humana simulada – o e-mail é realmente enviado. Mais uma vez, as ações são simuladas através de simples print() declarações por uma questão de simplicidade, mantendo o foco no mecanismo de interrupção gerenciado pelo estado.
O que mais precisamos? Um fluxo de trabalho de agente é descrito por um gráfico com vários estados conectados. Vamos definir uma sequência simples e linear de ações da seguinte forma:
fluxo de trabalho = StateGraph(Estado do Agente) # Adicionando nós de ação fluxo de trabalho.add_node(“rascunho_mensagem”, rascunho_node) fluxo de trabalho.add_node(“enviar_mensagem”, send_node) # Conectando nós através de bordas: Iniciar -> Rascunho -> Enviar -> Fim fluxo de trabalho.set_entry_point(“rascunho_mensagem”) fluxo de trabalho.add_edge(“rascunho_mensagem”, “enviar_mensagem”) fluxo de trabalho.add_edge(“enviar_mensagem”, FIM) |
Para implementar o mecanismo semelhante a um banco de dados que salva o estado do agente e para introduzir a interrupção gerenciada pelo estado quando o agente está prestes a enviar uma mensagem, usamos este código:
# MemorySaver é como nosso “banco de dados” para salvar estados memória = Economizador de memória() # ESTA É UMA PARTE CHAVE DO NOSSO PROGRAMA: dizer ao agente para fazer uma pausa antes de enviar aplicativo = fluxo de trabalho.compilar( ponto de verificação=memória, interromper_antes=(“enviar_mensagem”) ) |
Agora vem a verdadeira ação. Vamos execute o gráfico de ação definido há poucos momentos. Observe abaixo que um ID de thread é usado para que a memória possa acompanhar o estado do fluxo de trabalho entre as execuções.
configuração = {“configurável”: {“thread_id”: “demo-thread-1”}} estado_inicial = {“rascunho”: “”, “aprovado”: Falso, “enviado”: Falso} imprimir(“n— EXECUÇÃO DO GRÁFICO INICIAL —“) # O gráfico executará ‘draft_node’, atingirá o ponto de interrupção e fará uma pausa. para evento em aplicativo.fluxo(estado_inicial, configuração): passar |
Em seguida vem o momento humano no circuito, onde o fluxo é pausado e a aprovação humana é simulada pela configuração authorized para True:
imprimir(“n— GRÁFICO PAUSADO —“) estado_atual = aplicativo.get_state(configuração) imprimir(f“Próximo nó a ser executado: {current_state.subsequent}”) # Deve mostrar ‘send_message’ imprimir(f“Rascunho atual: ‘{current_state.values(‘draft’)}'”) # Simulando uma revisão humana e aprovação do rascunho do e-mail imprimir(“n (Humano): Revendo o rascunho… Parece bom. Aprovando!”) # IMPORTANTE: o estado está atualizado com a decisão do humano aplicativo.estado_atualização(configuração, {“aprovado”: Verdadeiro}) |
Isso retoma o gráfico e conclui a execução.
imprimir(“n— RETOMANDO GRÁFICO —“) # Passamos ‘None’, pois a entrada diz ao gráfico para continuar de onde parou para evento em aplicativo.fluxo(Nenhum, configuração): passar imprimir(“n— ESTADO FINAL —“) imprimir(aplicativo.get_state(configuração).valores) |
O geral saída impresso por este fluxo de trabalho simulado deve ficar assim:
—– CORRENDO INICIAL GRÁFICO —– (Agente): Rascunho o e-mail... —– GRÁFICO PAUSADO —– Próximo nó para executar: (‘enviar_mensagem’,) Atual Rascunho: ‘Olá! A atualização do seu servidor está pronta para ser implantada.’ (Humano): Revendo rascunho... Visible bom. Aprovando! —– RETOMANDO GRÁFICO —– (Agente): Acordando voltar acima! Verificando aprovação standing... (Sistema): ENVIANDO E-MAIL -> Olá! Seu servidor atualizar é preparar para ser implantado. —– FINAL ESTADO —– {‘rascunho’: ‘Olá! A atualização do seu servidor está pronta para ser implantada.’, ‘aprovado’: Verdadeiro, ‘enviado’: Verdadeiro} |
Concluindo
Este artigo ilustrou como implementar interrupções gerenciadas pelo estado em fluxos de trabalho baseados em agentes, introduzindo mecanismos humanos no circuito — um recurso importante em cenários críticos e de alto risco, onde a autonomia complete pode não ser desejável. Usamos LangGraph, uma biblioteca poderosa para construir aplicativos LLM orientados por agentes, para simular um fluxo de trabalho regido por essas regras.