Blog de Agentes IA
SISTEMA DE ACESSO COMPUTACIONAL

Máquinas de Estado e Agentes: Construindo Workflows Confiáveis com LangGraph

agentsarchitecturetutoriallanggraphautonomous-systems

A maioria dos tutoriais de agentes mostra um loop simples: perguntar ao Claude, analisar a resposta, chamar uma ferramenta, repetir. Isso funciona para demonstrações. Em produção, você precisa de determinismo, recuperação de erros, portões de aprovação humana e auditabilidade.

LangGraph traz máquinas de estado para os workflows de agentes. Em vez de um loop ad-hoc mantido por instruções if, você obtém um grafo explícito: nós nomeados (unidades de lógica), arestas tipadas (transições) e um esquema de estado compartilhado que flui por toda a execução.

Por Que Máquinas de Estado para Agentes?

O Problema do Loop Ad-Hoc

Um loop de agente típico se parece com isto:

messages = []
while True:
    response = claude.messages.create(model=..., messages=messages)
    if response.stop_reason == "end_turn":
        break
    for tool_call in get_tool_calls(response):
        result = execute_tool(tool_call)
        messages.append(tool_result(tool_call.id, result))

Isso é legível para duas ou três ferramentas. Adicione cinco ferramentas, caminhos condicionais, uma etapa de aprovação humana e lógica de reintento — e você terá centenas de linhas de fluxo de controle entrelaçado.

O problema mais profundo é o estado implícito. Em que estágio está o agente? Quais dados ele coletou? Tudo vive em messages — um blob sem tipo que cada nó lê e ao qual acrescenta, sem esquema imposto.

Máquinas de Estado como Solução

Uma máquina de estado torna o implícito explícito. Você define:

  • Nós — unidades de lógica discretas que recebem o estado atual, fazem uma coisa e retornam atualizações de estado.
  • Arestas — transições entre nós, incondicional (A → B sempre) ou condicional (se problemas: ir para revisão, senão: ir para resumo).
  • Estado — um dicionário tipado que flui pelo grafo completo. O esquema é validado a cada passo.

Quando NÃO Usar LangGraph

Para tarefas simples de uma única etapa, uma chamada direta à API é mais rápida e clara. Use LangGraph quando seu workflow tiver:

  • Múltiplos estágios distintos em sequência
  • Ramificação condicional baseada em resultados intermediários
  • Etapas com participação humana
  • Lógica de recuperação de erros ou reintento
  • Requisitos de auditabilidade

Fundamentos do LangGraph

Definindo o Estado

from typing import TypedDict


class ResearchState(TypedDict):
    query: str
    research_notes: str
    draft: str
    review_feedback: str
    is_approved: bool

Nós

import anthropic


client = anthropic.Anthropic()


def research_node(state: ResearchState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"Pesquise este tópico exaustivamente: {state['query']}"
        }]
    )
    return {"research_notes": response.content[0].text}


def draft_node(state: ResearchState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"Escreva um rascunho baseado nestas notas:\n{state['research_notes']}"
        }]
    )
    return {"draft": response.content[0].text}

Arestas e Roteamento Condicional

from typing import Literal


def route_after_check(state: ResearchState) -> Literal["human_review", "draft"]:
    if state.get("review_feedback"):
        return "human_review"
    return "draft"


workflow.add_conditional_edges(
    "research",
    route_after_check,
    {"human_review": "review_node", "draft": "draft_node"}
)

Construindo e Executando um Grafo

from langgraph.graph import StateGraph, END


workflow = StateGraph(ResearchState)
workflow.add_node("research", research_node)
workflow.add_node("draft", draft_node)
workflow.set_entry_point("research")
workflow.add_edge("research", "draft")
workflow.add_edge("draft", END)


graph = workflow.compile()
result = graph.invoke({"query": "O que é o protocolo MCP?"})
print(result["draft"])

Construindo um Workflow de Revisão de Documentos

Desenhando o Estado

class DocumentState(TypedDict):
    document: str
    key_terms: list[str]
    compliance_issues: list[str]
    summary: str
    human_feedback: str
    is_approved: bool

Implementando os Nós

import json


def extract_terms_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": (
                "Extraia exatamente 5 termos-chave deste documento. "
                "Retorne-os como um array JSON de strings.\n\n"
                f"Documento: {state['document']}"
            )
        }]
    )
    try:
        raw = response.content[0].text.strip()
        if raw.startswith("```"):
            raw = raw.split("```")[1]
            if raw.startswith("json"):
                raw = raw[4:]
        terms = json.loads(raw.strip())
    except (json.JSONDecodeError, IndexError):
        terms = [t.strip() for t in response.content[0].text.split("\n") if t.strip()][:5]
    return {"key_terms": terms}




def check_compliance_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": (
                "Verifique este documento por problemas de conformidade. "
                "Procure: PII (nomes, CPF, e-mails, telefones), "
                "marcações confidenciais, afirmações não verificadas.\n\n"
                "Retorne um array JSON de descrições de problemas. "
                "Retorne um array vazio [] se nenhum problema for encontrado.\n\n"
                f"Documento: {state['document']}"
            )
        }]
    )
    try:
        raw = response.content[0].text.strip()
        if raw.startswith("```"):
            raw = raw.split("```")[1]
            if raw.startswith("json"):
                raw = raw[4:]
        issues = json.loads(raw.strip())
    except (json.JSONDecodeError, IndexError):
        issues = []
    return {"compliance_issues": issues}




def human_review_node(state: DocumentState) -> dict:
    print("\n=== REVISÃO HUMANA NECESSÁRIA ===")
    for issue in state["compliance_issues"]:
        print(f"  - {issue}")
    return {
        "human_feedback": "Revisado e aprovado com redação de PII",
        "is_approved": True
    }




def summarize_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=256,
        messages=[{
            "role": "user",
            "content": f"Resuma este documento em exatamente duas frases:\n\n{state['document']}"
        }]
    )
    return {"summary": response.content[0].text, "is_approved": True}

Montando o Grafo

def route_after_compliance(state: DocumentState) -> Literal["human_review", "summarize"]:
    if state.get("compliance_issues"):
        return "human_review"
    return "summarize"


workflow = StateGraph(DocumentState)
workflow.add_node("extract", extract_terms_node)
workflow.add_node("check", check_compliance_node)
workflow.add_node("review", human_review_node)
workflow.add_node("summarize", summarize_node)


workflow.set_entry_point("extract")
workflow.add_edge("extract", "check")
workflow.add_conditional_edges(
    "check", route_after_compliance,
    {"human_review": "review", "summarize": "summarize"}
)
workflow.add_edge("review", "summarize")
workflow.add_edge("summarize", END)


graph = workflow.compile()


result = graph.invoke({
    "document": "O paciente João Silva (CPF: 123.456.789-00) tem hipertensão.",
    "key_terms": [], "compliance_issues": [],
    "summary": "", "human_feedback": "", "is_approved": False,
})
print(result["summary"])

Pontos de Verificação Humanos e Interrupções

from langgraph.checkpoint.memory import MemorySaver


checkpointer = MemorySaver()
graph = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["review"]
)


thread_config = {"configurable": {"thread_id": "doc-review-001"}}
result = graph.invoke(initial_state, config=thread_config)


current_state = graph.get_state(thread_config)
print("Problemas:", current_state.values["compliance_issues"])


graph.update_state(
    thread_config,
    {"human_feedback": "Aprovado após redação manual de PII", "is_approved": True}
)


final_result = graph.invoke(None, config=thread_config)

Persistência e Implantação em Produção

from langgraph.checkpoint.postgres import PostgresSaver


with PostgresSaver.from_conn_string("postgresql://...") as checkpointer:
    graph = workflow.compile(checkpointer=checkpointer)


import asyncio


async def batch_process(documents: list[str]) -> list[dict]:
    return await asyncio.gather(*[
        graph.ainvoke({"document": doc, "key_terms": [], "compliance_issues": [],
                       "summary": "", "human_feedback": "", "is_approved": False})
        for doc in documents
    ])

Padrões Comuns e Armadilhas

  • Fan-Out / Fan-In: Execute vários nós em paralelo com Send e mescle os resultados.
  • Roteamento condicional excessivo: Se uma função de roteamento tem muitas ramificações, divida em subgrafos.
  • Sobrecarga de estado: Armazene referências (IDs de banco de dados, chaves S3) em vez de grandes artefatos.
  • Teste de nós em isolamento: Como os nós são funções simples, podem ser testados unitariamente diretamente.
def test_extract_terms_node():
    state = {
        "document": "A raposa marrom rápida pula sobre o cachorro preguiçoso.",
        "key_terms": [], "compliance_issues": [],
        "summary": "", "human_feedback": "", "is_approved": False,
    }
    result = extract_terms_node(state)
    assert "key_terms" in result
    assert isinstance(result["key_terms"], list)

LangGraph substitui loops ad-hoc por workflows explícitos, depuráveis e retomáveis. Comece com um grafo linear (A → B → C → FIM). Adicione uma aresta condicional quando precisar bifurcar. Adicione um ponto de verificação humano quando precisar de aprovação. A maioria dos workflows de agentes de produção precisa exatamente desses três padrões.

Artigos Relacionados