Fundamentos do Human in the Loop
O Human in the Loop é um mecanismo de governança em sistemas de IA que permite a intervenção humana em pontos críticos do fluxo do agente.
Objetivos:
Aumentar confiabilidade
Reduzir riscos
Garantir conformidade
Permitir controle humano antes de ações sensíveis
Melhorar qualidade das respostas
Muito utilizado em:
Envio automático de e-mails
Decisões financeiras
Publicação de conteúdo
Ações com impacto legal ou reputacional
Configuração do Ambiente
Instalação:
pip install langgraph langchain langchain-google-genai python-dotenv
Configuração Inicial
import os
import uuid
from dotenv import load_dotenv
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, END, add_messages
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.messages import HumanMessage, AIMessage
load_dotenv()
llm = ChatGoogleGenerativeAI(
model="gemini-1.5-flash",
temperature=0
)
AgentState com ThreadId
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
thread_id: str
Implementando reduceMessages
Permite atualizar o estado substituindo ou anexando mensagens.
def reduceMessages(existing_messages, new_messages):
# Estratégia simples: anexar
return existing_messages + new_messages
(Em cenário avançado, pode-se substituir mensagens pelo ID.)
UUID para Unicidade
def create_human_message(content):
return HumanMessage(
content=content,
additional_kwargs={"id": str(uuid.uuid4())}
)
Isso evita conflitos no histórico.
Nó do Modelo
def call_model(state: AgentState):
response = llm.invoke(state["messages"])
return {"messages": [response]}
⏸ Interrupt Before Action
Ponto crítico para aprovação humana.
def human_review(state: AgentState):
last_message = state["messages"][-1]
print("\n Resposta do agente:")
print(last_message.content)
decision = input("\nAprovar resposta? (sim/não): ").strip().lower()
if decision == "sim":
return {"messages": []}
else:
nova_resposta = input("Digite a resposta corrigida: ")
return {
"messages": [
AIMessage(
content=nova_resposta,
additional_kwargs={"id": str(uuid.uuid4())}
)
]
}
Estrutura do Grafo
builder = StateGraph(AgentState)
builder.add_node("model", call_model)
builder.add_node("human_review", human_review)
builder.set_entry_point("model")
builder.add_edge("model", "human_review")
builder.add_edge("human_review", END)
graph = builder.compile()
ThreadId Dinâmico
thread_id = str(uuid.uuid4())
Evita conflito entre sessões anteriores.
Snapshots
Registrar estado antes da intervenção:
snapshot = {
"messages": [],
"thread_id": thread_id
}
Snapshots permitem:
Restaurar estado
Auditoria
Rastreabilidade
Testando o Fluxo
initial_state = {
"messages": [create_human_message("Escreva um email cobrando pagamento atrasado.")],
"thread_id": thread_id
}
resultado = graph.invoke(initial_state)
print("\nEstado Final:")
for msg in resultado["messages"]:
print(msg.content)
