Human in loop

1 - Fundamentos de Human in the Loop (HITL) e importância na orquestração de agentes: HITL é o padrão em que o grafo/agent pausa em pontos críticos, persiste o estado, expõe o que será feito e aguarda aprovação/edição humana antes de continuar. Em LangGraph isso é viabilizado por interrupts (pontos de pausa) + checkpointer/persistence (salvar checkpoints/snapshots para retomar depois), permitindo revisão humana, correções e retomada do fluxo sem perder contexto.

2 - Configurar ambiente com bibliotecas/padrões das aulas anteriores (código): nas suas aulas/repo aparecem pacotes como langgraph, langchain, langchain-openai, python-dotenv etc., além do padrão de instalar via pip/requirements.

# (opção A) instalar exatamente o que já estava no seu requirements.txt
pip install -r requirements.txt

# (opção B) instalar apenas o essencial do HITL (caso você queira mínimo)
pip install -U langgraph langchain-core langchain-openai python-dotenv

3 - Implementar a função reduceMessages para atualizar o estado substituindo ou anexando mensagens (código): o comportamento desejado é “append-only, exceto quando o ID coincide (substitui)”, que é justamente a ideia do reducer de mensagens (como add_messages).

from typing import List, Dict, Any

Message = Dict[str, Any]

def reduceMessages(left: List[Message], right: List[Message]) -> List[Message]:
    """
    Merge de mensagens:
    - se 'id' de uma mensagem nova já existir, substitui a antiga
    - senão, anexa ao final
    """
    if left is None:
        left = []
    if right is None:
        right = []

    index = {m.get("id"): i for i, m in enumerate(left) if m.get("id") is not None}
    merged = list(left)

    for m in right:
        mid = m.get("id")
        if mid is not None and mid in index:
            merged[index[mid]] = m
        else:
            merged.append(m)
            if mid is not None:
                index[mid] = len(merged) - 1
    return merged

4 - Usar UUID para garantir unicidade das mensagens (código):

from uuid import uuid4

def make_message(role: str, content: str) -> dict:
    return {
        "id": str(uuid4()),     # UUID garante unicidade
        "role": role,
        "content": content,
    }

5 - Adicionar “interrupt before action” para aprovação humana antes da execução (código): LangGraph suporta interrupções estáticas configuradas na compilação para pausar antes de um nó específico, viabilizando aprovação humana.

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()

# ... (builder = StateGraph(...); add_node etc.)

app = builder.compile(
    checkpointer=checkpointer,
    interrupt_before_nodes=["action"],  # pausa antes do nó "action"
)

6 - Estruturar o grafo com nós de ação, pausa e integração com ferramentas externas (código):

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()

# ... (builder = StateGraph(...); add_node etc.)

app = builder.compile(
    checkpointer=checkpointer,
    interrupt_before_nodes=["action"],  # pausa antes do nó "action"
)

7 - Capturar thread_id dinâmico para evitar conflito com históricos anteriores (código): em LangGraph, o thread_id é a “chave” da persistência: reutilizar retoma o histórico; usar novo começa “thread” limpa.

from uuid import uuid4

thread_id = str(uuid4())  # thread novo para não conflitar com históricos antigos

config = {
    "configurable": {
        "thread_id": thread_id
    }
}

8 - Configurar pausa e requerer intervenção humana em momentos críticos (código): o mecanismo base é interrupt + checkpointer; ao interromper, o estado é salvo e o fluxo aguarda retomada.

# (A) pausa estática: já configurada no compile com interrupt_before_nodes=["action"]
# Resultado: a execução para antes de rodar o nó "action".

# (B) pausa dinâmica: dentro de um nó, usando interrupt()
from langgraph.types import interrupt

def approval_node(state: AgentState):
    decision = interrupt({
        "question": "Aprova executar a ação X?",
        "draft_id": state.get("draft_id")
    })
    # decision será o payload passado no Command(resume=...) na retomada
    return {"approved": bool(decision.get("approved", False))}