Graph Engineering para Agentes de IA: Do 'Prompt-and-Pray' a Máquinas de Estado Determinísticas

Em 2024, construir um agente de IA era escrever um loop while True, passar um prompt com ferramentas e torcer para o modelo não entrar em parafuso. Chamamos isso carinhosamente de Prompt-and-Pray. Funcionava para demos no Twitter e vídeos de 3 minutos no YouTube. Em produção, era um desastre anunciado: loops infinitos, chamadas repetidas de APIs pagas, decisões imprevisíveis e zero auditabilidade.

Em 2026, a engenharia de software retomou o controle. O divisor de águas entre brinquedos experimentais e sistemas corporativos de missão crítica tem nome: Graph Engineering (Engenharia de Grafos de Estado).

Em vez de dar liberdade irrestrita para o modelo decidir o que fazer a cada microssegundo, nós modelamos o comportamento do agente como uma Máquina de Estados Finita (FSM) e um Grafo Cíclico Dirigido (DCG). O LLM continua responsável pelo raciocínio e síntese dentro de cada nó, mas a topologia da execução, as regras de transição, os limites de retry e os portões de aprovação humana pertencem ao código.

Este guia é um manual prático e aprofundado para transformar agentes caóticos em sistemas determinísticos, resilientes e auditáveis usando LangGraph e Python.

O Colapso do Loop Aberto (Prompt-and-Pray)

Para entender por que Graph Engineering se tornou o padrão da indústria, precisamos dissecar a fragilidade do padrão ReAct puro (Reason + Act) quando executado em loop aberto:

# O anti-padrão de 2024: Loop aberto sem controle estrutural
while not task_completed:
    thought, action = llm.decide(history)
    if action == "FINISH":
        break
    observation = execute_tool(action)
    history.append(observation)

Esse padrão tem quatro falhas fatais em produção:

1. Incapacidade de impor regras de negócio estritas: Se a política da empresa exige que todo reembolso acima de R$ 500 passe por aprovação humana, um loop aberto confia que o prompt do LLM lembrará dessa regra em 100% das vezes. Na prática, uma pequena variação no vocabulário do usuário faz o agente pular a regra e chamar a tool de pagamento direto.
2. Ciclos de alucinação sem saída (Death Loops): Quando uma ferramenta externa retorna um erro 400 ou timeout, o LLM frequentemente tenta a mesma chamada repetidamente com parâmetros idênticos, queimando milhares de tokens até estourar a janela de contexto.
3. Falta de persistência atômica: Se o processo do agente reiniciar no 8º passo de uma tarefa complexa de 10 etapas, todo o progresso evapora. O agente precisa recomeçar do zero, reexecutando chamadas que já causaram efeitos colaterais no mundo real (como cobranças ou emails).
4. Impossibilidade de Time-Travel Debugging: Quando o agente erra, você não consegue inspecionar o snapshot exato do estado no nó 3 para reproduzir o bug isoladamente.

A Taxonomia da Orquestração: De Chains a Grafos

A evolução da orquestração de IA passou por três fases bem definidas:

Os 4 Pilares Fundamentais do Graph Engineering

Para construir agentes com Graph Engineering, você precisa dominar quatro conceitos arquiteturais:

1. State (Estado Central Compartilhado e Redutores)

O Estado é uma estrutura de dados tipada (geralmente baseada em TypedDict ou Pydantic) que representa a memória operacional do grafo. Cada nó recebe o estado atual, executa sua lógica isolada e retorna um dicionário com atualizações para o estado.

O ponto crítico aqui são os Redutores (Reducers). Em vez de simplesmente sobrescrever o estado, você pode definir regras de agregação. Por exemplo, mensagens de chat usam o redutor add_messages, que anexa novas mensagens ao histórico sem apagar as anteriores.

2. Nodes (Nós de Execução Pura)

Nós são funções Python síncronas ou assíncronas que realizam uma unidade discreta de trabalho:

• Um nó pode chamar um LLM para classificar um texto.
• Um nó pode executar uma consulta SQL.
• Um nó pode chamar uma ferramenta via MCP (Model Context Protocol).
• Um nó pode formatar o resultado final.

Regra de Ouro: Nós devem ser o mais puros e desacoplados possível. A lógica de decisão sobre para onde ir em seguida NÃO deve ficar dentro do nó, mas sim nas Arestas.

3. Edges & Conditional Routing (Arestas e Roteadores)

As arestas definem os caminhos permitidos no grafo:

• Arestas Normais (Fixed Edges): Transição direta e garantida ($A \rightarrow B$).
• Arestas Condicionais (Conditional Edges): Funções de roteamento que inspecionam o Estado e decidem dinamicamente o próximo nó com base em regras determinísticas de código (e não em alucinações).

4. Checkpointers & Time-Travel (Persistência e Reversibilidade)

Um Checkpointer salva um snapshot atômico do Estado após a execução de cada nó em um banco de dados persistente (SQLite, PostgreSQL, Redis). Isso desbloqueia três superpoderes:

• Retomada transparente após falhas de infraestrutura.
• Human-in-the-loop assíncrono: o agente pausa antes de um nó perigoso, aguarda aprovação humana por horas ou dias e retoma de onde parou.
• Time-Travel: a capacidade de inspecionar estados passados, bifurcar execuções (branching) e editar o estado intermediário para reexecutar testes de regressão.

Anatomia de um StateGraph na Prática

Abaixo visualizamos como esses quatro pilares se conectam em uma arquitetura de suporte e execução automatizada:

Note a clareza do fluxo:

1. O input do usuário entra pelo nó START.
2. A Triagem categoriza o pedido.
3. A Validação aplica regras de negócio via schema Pydantic.
4. O Roteador Condicional avalia:
   • Se os dados são inválidos e o contador de retries for $< 3$, aciona o loop de autocorreção.
   • Se for uma ação destrutiva/crítica, aciona o Portão Humano (Human-in-the-loop).
   • Se for uma ação segura, envia direto para a Execução de Tool.
5. O Checkpointer grava o estado em cada nó, garantindo auditabilidade total.

Implementação Completa em Python com LangGraph

Vamos agora implementar um agente funcional de produção em Python. Este agente realiza triagem de solicitações técnicas, valida parâmetros estruturados, possui proteção contra loops infinitos e exige aprovação humana para comandos perigosos.

1. Instalação das Dependências

pip install langgraph langchain-core langchain-anthropic pydantic

2. Definição do Estado Central Tipado

from typing import Annotated, Literal, Optional
from typing_extensions import TypedDict
from pydantic import BaseModel, Field
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    """
    Estado compartilhado que viaja por todos os nós do grafo.
    """
    messages: Annotated[list[BaseMessage], add_messages]
    intent: str                          # "suporte_geral" | "execucao_sql" | "desconhecido"
    sql_query: Optional[str]             # Query extraída
    is_destructive: bool                 # True se for DROP, DELETE, UPDATE
    retry_count: int                     # Contador de autocorreção
    human_approved: Optional[bool]       # Decisão do operador humano
    final_response: Optional[str]        # Resposta consolidada ao usuário

3. Implementação dos Nós Especializados

from langchain_anthropic import ChatAnthropic

# Inicialização do modelo (ou utilize Claude 3.7 / Hermes local)
llm = ChatAnthropic(model="claude-3-5-sonnet-20241022", temperature=0)

def no_triagem(state: AgentState) -> dict:
    """
    Nó 1: Classifica a intenção do usuário e extrai dados iniciais.
    """
    last_message = state["messages"][-1].content
    
    prompt = """
    Analise a mensagem do usuario e identifique a intencao:
    Mensagem: \"\"\" + last_message + \"\"\"
    
    Responda apenas em formato JSON com chaves:
    intent: 'execucao_sql' ou 'suporte_geral'
    sql_query: 'query aqui ou null'
    """
    response = llm.invoke([HumanMessage(content=prompt)])
    
    # Simulação de parsing seguro
    content = response.content.strip()
    is_sql = "SELECT" in last_message.upper() or "DELETE" in last_message.upper() or "UPDATE" in last_message.upper()
    
    return {
        "intent": "execucao_sql" if is_sql else "suporte_geral",
        "sql_query": last_message if is_sql else None,
        "retry_count": state.get("retry_count", 0)
    }

def no_validacao_seguranca(state: AgentState) -> dict:
    """
    Nó 2: Aplica guardrails determinísticos de segurança sobre a query SQL.
    """
    query = state.get("sql_query") or ""
    upper_query = query.upper()
    
    # Regra de negócio estrita: DELETE, DROP ou UPDATE são destrutivos
    is_destructive = any(cmd in upper_query for cmd in ["DELETE", "DROP", "UPDATE", "TRUNCATE"])
    
    return {
        "is_destructive": is_destructive
    }

def no_execucao_segura(state: AgentState) -> dict:
    """
    Nó 3: Executa consultas somente leitura com segurança.
    """
    query = state.get("sql_query")
    # Aqui entraria a conexão real com o banco de dados via MCP
    result_mock = f"✓ [SQL Executado com Sucesso]: 42 registros encontrados para '{query}'."
    
    return {
        "final_response": result_mock,
        "messages": [AIMessage(content=result_mock)]
    }

def no_execucao_critica(state: AgentState) -> dict:
    """
    Nó 4: Executa ações destrutivas somente após o portão humano.
    """
    if not state.get("human_approved"):
        msg = "✕ Ação cancelada: operação rejeitada pelo operador humano."
        return {"final_response": msg, "messages": [AIMessage(content=msg)]}
        
    query = state.get("sql_query")
    result_mock = f"⚠️ [OPERAÇÃO CRÍTICA EXECUTADA COM APROVAÇÃO HUMANA]: '{query}'."
    return {
        "final_response": result_mock,
        "messages": [AIMessage(content=result_mock)]
    }

def no_resposta_geral(state: AgentState) -> dict:
    """
    Nó 5: Responde a dúvidas gerais de suporte.
    """
    response = llm.invoke(state["messages"])
    return {
        "final_response": response.content,
        "messages": [response]
    }

4. Aresta Condicional de Roteamento (Decisões Determinísticas)

def roteador_principal(state: AgentState) -> Literal["execucao_segura", "portao_humano", "resposta_geral"]:
    """
    Roteador que inspeciona o estado e decide a próxima transição.
    """
    intent = state.get("intent")
    
    if intent == "execucao_sql":
        if state.get("is_destructive"):
            return "portao_humano"
        return "execucao_segura"
        
    return "resposta_geral"

5. Compilação do Grafo com Persistência (SqliteSaver)

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3

# 1. Instancia o grafo com o schema de estado
builder = StateGraph(AgentState)

# 2. Registra todos os nós
builder.add_node("triagem", no_triagem)
builder.add_node("validacao", no_validacao_seguranca)
builder.add_node("execucao_segura", no_execucao_segura)
builder.add_node("execucao_critica", no_execucao_critica)
builder.add_node("resposta_geral", no_resposta_geral)

# 3. Define as arestas fixas e condicionais
builder.add_edge(START, "triagem")
builder.add_edge("triagem", "validacao")

builder.add_conditional_edges(
    "validacao",
    roteador_principal,
    {
        "execucao_segura": "execucao_segura",
        "portao_humano": "execucao_critica",
        "resposta_geral": "resposta_geral"
    }
)

builder.add_edge("execucao_segura", END)
builder.add_edge("execucao_critica", END)
builder.add_edge("resposta_geral", END)

# 4. Configura banco de checkpoints SQLite para Time-Travel e HITL
conn = sqlite3.connect("checkpoints.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)

# 5. Compila o grafo com interrupção humana antes do nó crítico
app = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["execucao_critica"]  # O Portão Humano Obrigatório
)

Executando o Sistema: O Portão Humano em Ação

Vamos ver como o grafo se comporta diante de uma consulta inofensiva versus uma operação destrutiva:

Cenário A: Consulta Segura (Leitura)

thread_config = {"configurable": {"thread_id": "sessao_cliente_001"}}

input_data = {
    "messages": [HumanMessage(content="SELECT * FROM usuarios WHERE ativo = true;")]
}

for event in app.stream(input_data, config=thread_config):
    for node_name, state_update in event.items():
        print(f"--- Nó Executado: {node_name} ---")

# Saída no terminal:
# --- Nó Executado: triagem ---
# --- Nó Executado: validacao ---
# --- Nó Executado: execucao_segura ---
# ✓ [SQL Executado com Sucesso]: 42 registros encontrados...

Cenário B: Operação Destrutiva (Pausa no Portão Humano)

thread_config_critico = {"configurable": {"thread_id": "sessao_cliente_002"}}

input_critico = {
    "messages": [HumanMessage(content="DELETE FROM logs WHERE data < '2025-01-01';")]
}

# O grafo roda até o portão humano e CONGELA
for event in app.stream(input_critico, config=thread_config_critico):
    for node_name, state_update in event.items():
        print(f"--- Nó Executado: {node_name} ---")

# Saída no terminal:
# --- Nó Executado: triagem ---
# --- Nó Executado: validacao ---
# (A execução pausa automaticamente ANTES de 'execucao_critica')

# Inspecionando o estado pausado:
estado_atual = app.get_state(thread_config_critico)
print("\nPróximo nó na fila:", estado_atual.next)
# Próximo nó na fila: ('execucao_critica',)
print("Operação detectada como destrutiva?", estado_atual.values["is_destructive"])
# True

Cenário C: Aprovação e Retomada da Execução

O operador humano recebe um alerta no Slack ou painel, analisa a solicitação e injeta a decisão no estado:

# Humano aprova a operação
app.update_state(
    thread_config_critico,
    {"human_approved": True},
    as_node="validacao"
)

# Retomando a execução de onde parou (sem reexecutar triagem ou validação!)
for event in app.stream(None, config=thread_config_critico):
    for node_name, state_update in event.items():
        print(f"--- Nó Executado: {node_name} ---")

# Saída no terminal:
# --- Nó Executado: execucao_critica ---
# ⚠️ [OPERAÇÃO CRÍTICA EXECUTADA COM APROVAÇÃO HUMANA]: 'DELETE FROM logs...'

Matriz de Decisão: Quando Usar Cada Framework

Não tente usar State Graphs para tudo. Escolha a arquitetura adequada ao problema:

Diretrizes de Escolha Rápida:

• Use LangGraph quando o processo tiver regras de negócio estritas, necessidade de rollback/time-travel, ciclos de autocorreção com limites rígidos ou aprovação humana mandatória.
• Use CrewAI quando precisar simular uma equipe com personas definidas (ex: Redator + Revisor de SEO) em pipelines predominantemente lineares de conteúdo.
• Use Filas Assíncronas (SQS/RabbitMQ) quando o volume for de milhões de mensagens/dia e as tarefas forem independentes e assíncronas (conforme detalhado no guia Quando NÃO Usar MCP ou A2A).

As 3 Armadilhas Frequentes de Graph Engineering em Produção

Mesmo com grafos bem desenhados, times cometem erros clássicos ao levar máquinas de estado para produção:

1. Inchaço de Estado (State Bloat) e Degradação de Contexto

Se você anexar payloads brutos de ferramentas (ex: JSONs de 500 KB retornados por uma API) diretamente na chave messages do estado, em 3 nós o seu histórico estará com milhões de tokens desnecessários.

• Solução: Mantenha no AgentState apenas resumos estruturados e referências (entity_id, status, summary). Salve blobs brutos em Object Storage (S3/GCS) e passe apenas os ponteiros.

2. Loops Zumbis por Falta de Condição de Parada Rígida

Um ciclo de autocorreção ($Nó_Validação \rightarrow Nó_Correção \rightarrow Nó_Validação$) precisa ter um contador de tentativas explícito (retry_count).

• Solução: Sempre imponha uma aresta de escape: if state["retry_count"] >= 3: return "falha_humana". Além disso, configure o recursion_limit nativo do LangGraph ({"recursion_limit": 25}).

3. Falta de Idempotência nos Nós de Ação

Se a rede falhar durante a transição do nó de pagamento para o nó de confirmação, o checkpointer tentará reexecutar o nó de pagamento ao reiniciar.

• Solução: Todo nó com efeito colateral externo deve aceitar uma chave de idempotência única (idempotency_key = f"{thread_id}-{node_step}"), garantindo que uma cobrança ou envio de email nunca aconteça em duplicidade.

Checklist de Produção para Graph Engineering

Antes de colocar seu StateGraph em ambiente produtivo, valide os 7 itens:

• [ ] O AgentState possui tipagem estrita com TypedDict ou Pydantic.
• [ ] Todo nó de execução é uma função pura e a lógica de transição está 100% nas arestas.
• [ ] Todos os ciclos de autocorreção possuem limite explícito de tentativas (retry_count < N).
• [ ] Operações destrutivas ou de alto custo passam por portões humanos com interrupt_before.
• [ ] O checkpointer está configurado em banco de dados durável (PostgreSQL / SQLite).
• [ ] Chamadas a ferramentas externas utilizam chaves de idempotência.
• [ ] O sistema de observabilidade (OpenTelemetry / Traces) está conectado para monitorar a latência de cada nó.

Conclusão: Engenharia sobre o Caos

Agentes de IA não precisam ser caixas-pretas incontroláveis. Ao adotar Graph Engineering, você substitui o otimismo ingênuo do Prompt-and-Pray pelo rigor da engenharia de software tradicional: estados determinísticos, transições controladas, persistência atômica e segurança real em produção.

O LLM é o motor de cognição; o Grafo de Estados é a transmissão e o volante. É assim que construímos sistemas que escalam e crescem com você com confiança e previsibilidade.