LangGraph: construindo agentes com fluxo de estado e loops controlados

1. Introdução ao LangGraph e o Paradigma de Agentes com Estado

LangGraph é uma biblioteca desenvolvida pela LangChain que introduz um paradigma inovador para construção de agentes de IA: o uso de grafos de estado cíclicos. Diferente dos pipelines lineares tradicionais (DAGs), onde o fluxo de dados segue um caminho único e previsível, LangGraph permite que os agentes tomem decisões, executem ações e retornem a estados anteriores de forma controlada.

A principal inovação está na capacidade de manter estado explícito durante toda a execução do agente. Enquanto frameworks como AutoGPT ou implementações manuais do padrão ReAct tratam o estado como um artefato secundário, LangGraph coloca o estado como elemento central da arquitetura. Isso significa que cada nó do grafo pode ler e modificar um estado compartilhado, permitindo loops controlados, memória de longo prazo e tomada de decisão contextualizada.

Para agentes confiáveis, loops controlados são cruciais porque permitem que o sistema:
- Reavalie decisões com base em novas informações
- Execute múltiplas ferramentas em sequência
- Mantenha consistência através de múltiplas iterações
- Implemente limites de segurança contra loops infinitos

2. Arquitetura Fundamental: Nós, Arestas e Estado Compartilhado

A arquitetura do LangGraph é construída sobre três conceitos fundamentais:

StateGraph: O grafo principal que define o fluxo de execução. Cada grafo possui um esquema de estado tipado.

Nós (Nodes): Funções que processam o estado e retornam atualizações. Podem ser chamadas de LLM, ferramentas externas ou lógica condicional.

Arestas (Edges): Conexões entre nós que definem o fluxo. Existem dois tipos:
- Arestas fixas: fluxo determinístico entre nós
- Arestas condicionais: decisões baseadas no estado atual

O estado compartilhado é definido usando TypedDict ou modelos Pydantic:

from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END

class AgentState(TypedDict):
    messages: List[dict]
    next_action: str
    iteration_count: int
    tool_results: List[str]

3. Construindo o Primeiro Agente com Fluxo de Decisão

Vamos implementar um agente que decide entre pesquisar na web ou responder diretamente:

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_community.tools import DuckDuckGoSearchRun

# Definição do estado
class AgentState(TypedDict):
    messages: List[dict]
    needs_search: bool
    search_query: str

# Nó de decisão do LLM
def decide_action(state: AgentState) -> dict:
    llm = ChatOpenAI(model="gpt-4")
    last_message = state["messages"][-1]["content"]

    response = llm.invoke(
        f"Analyze: '{last_message}'. Reply 'SEARCH' if needs web search, else 'ANSWER'"
    )

    needs_search = "SEARCH" in response.content
    return {
        "needs_search": needs_search,
        "messages": state["messages"]
    }

# Nó de pesquisa
def search_web(state: AgentState) -> dict:
    search = DuckDuckGoSearchRun()
    query = state["messages"][-1]["content"]
    results = search.run(query)

    return {
        "tool_results": [results],
        "messages": state["messages"] + [{"role": "tool", "content": results}]
    }

# Nó de resposta final
def generate_answer(state: AgentState) -> dict:
    llm = ChatOpenAI(model="gpt-4")
    response = llm.invoke(state["messages"])

    return {
        "messages": state["messages"] + [{"role": "assistant", "content": response.content}]
    }

# Construção do grafo
graph = StateGraph(AgentState)
graph.add_node("decide", decide_action)
graph.add_node("search", search_web)
graph.add_node("answer", generate_answer)

graph.set_entry_point("decide")

# Arestas condicionais
graph.add_conditional_edges(
    "decide",
    lambda state: "search" if state["needs_search"] else "answer",
    {"search": "search", "answer": "answer"}
)

graph.add_edge("search", "answer")
graph.add_edge("answer", END)

app = graph.compile()

4. Loops Controlados: Execução Iterativa e Limites de Segurança

Loops controlados permitem que o agente execute múltiplas iterações até atingir um objetivo. O mecanismo principal é a aresta condicional que retorna ao mesmo nó:

class IterativeAgentState(TypedDict):
    messages: List[dict]
    iteration: int
    max_iterations: int
    task_complete: bool

def execute_iteration(state: IterativeAgentState) -> dict:
    llm = ChatOpenAI(model="gpt-4")
    response = llm.invoke(state["messages"])

    # Verifica se a tarefa foi concluída
    task_complete = "COMPLETE" in response.content

    return {
        "messages": state["messages"] + [{"role": "assistant", "content": response.content}],
        "iteration": state["iteration"] + 1,
        "task_complete": task_complete
    }

def check_completion(state: IterativeAgentState) -> str:
    if state["iteration"] >= state["max_iterations"]:
        return "end"
    if state["task_complete"]:
        return "end"
    return "continue"

# Configuração de segurança
graph = StateGraph(IterativeAgentState)
graph.add_node("execute", execute_iteration)
graph.set_entry_point("execute")

graph.add_conditional_edges(
    "execute",
    check_completion,
    {
        "continue": "execute",  # Loop controlado
        "end": END
    }
)

# Limite máximo de iterações
app = graph.compile()
app.recursion_limit = 10  # Prevenção de loops infinitos

5. Gerenciamento de Estado Avançado e Memória de Longo Prazo

Para estados complexos, podemos implementar memória persistente com checkpoints:

from langgraph.checkpoint import MemorySaver
from langgraph.graph import StateGraph
from typing import TypedDict, List, Optional

class AdvancedState(TypedDict):
    messages: List[dict]
    conversation_id: str
    user_preferences: dict
    cached_tool_results: List[str]
    summary: Optional[str]

# Checkpoint com SQLite
checkpointer = MemorySaver()

def summarize_history(state: AdvancedState) -> dict:
    if len(state["messages"]) > 10:
        llm = ChatOpenAI(model="gpt-4")
        summary = llm.invoke(
            f"Summarize this conversation: {state['messages'][-10:]}"
        )
        return {
            "summary": summary.content,
            "messages": state["messages"][-5:]  # Poda de mensagens
        }
    return state

# Configuração com persistência
graph = StateGraph(AdvancedState)
graph.add_node("summarize", summarize_history)
# ... outros nós

app = graph.compile(checkpointer=checkpointer)

6. Integração com Ferramentas Externas e APIs

Conectando ferramentas customizadas com tratamento de erros:

from langchain.tools import tool
import time

@tool
def query_vector_database(query: str) -> str:
    """Consulta uma vector database e retorna resultados"""
    try:
        # Simulação de consulta
        time.sleep(0.5)
        return f"Resultados para '{query}': [doc1, doc2, doc3]"
    except Exception as e:
        return f"Erro na consulta: {str(e)}"

@tool
def refine_response(context: str, question: str) -> str:
    """Refina a resposta com base no contexto recuperado"""
    llm = ChatOpenAI(model="gpt-4")
    response = llm.invoke(
        f"Context: {context}\nQuestion: {question}\nProvide refined answer:"
    )
    return response.content

# Nó de ferramenta com retry
def execute_tool_with_retry(state: dict) -> dict:
    max_retries = 3
    for attempt in range(max_retries):
        try:
            result = query_vector_database(state["query"])
            if "Erro" not in result:
                return {"tool_result": result}
        except Exception:
            if attempt == max_retries - 1:
                return {"tool_result": "Falha após múltiplas tentativas"}
            time.sleep(1)

7. Padrões de Projeto e Boas Práticas para Agentes Robutos

Separação de responsabilidades em nós especializados:

class RobustAgentState(TypedDict):
    plan: List[str]
    executed_steps: List[str]
    current_step: int
    reflection: str
    error_count: int

def planning_node(state: RobustAgentState) -> dict:
    llm = ChatOpenAI(model="gpt-4")
    plan = llm.invoke(f"Create plan for: {state['messages'][-1]}")
    return {"plan": plan.content.split("\n")}

def execution_node(state: RobustAgentState) -> dict:
    step = state["plan"][state["current_step"]]
    try:
        result = execute_step(step)
        return {
            "executed_steps": state["executed_steps"] + [step],
            "current_step": state["current_step"] + 1
        }
    except:
        return {"error_count": state["error_count"] + 1}

def reflection_node(state: RobustAgentState) -> dict:
    llm = ChatOpenAI(model="gpt-4")
    reflection = llm.invoke(
        f"Reflect on: {state['executed_steps']}\nErrors: {state['error_count']}"
    )
    return {"reflection": reflection.content}

# Subgrafos para modularização
subgraph = StateGraph(RobustAgentState)
subgraph.add_node("plan", planning_node)
subgraph.add_node("execute", execution_node)
subgraph.add_node("reflect", reflection_node)
# ... configuração de arestas

8. Comparação com Alternativas e Cenários de Uso no Mundo Real

LangGraph se destaca em cenários que exigem:

Quando escolher LangGraph:
- Assistentes de suporte multi-etapas (diagnóstico → solução → verificação)
- Automação de workflows com decisões condicionais
- Agentes de código que precisam compilar, testar e iterar
- Sistemas que exigem memória de longo prazo e consistência

Quando alternatives são melhores:
- RAG simples: LangChain básico é suficiente
- Tarefas lineares: pipelines tradicionais
- Prototipagem rápida: frameworks como CrewAI

Casos de uso reais:

# Agente de suporte técnico
support_graph = StateGraph(SupportState)
# Nós: identificar problema → pesquisar solução → executar → verificar

# Agente de automação de marketing
marketing_graph = StateGraph(MarketingState)
# Nós: analisar dados → gerar conteúdo → agendar → monitorar resultados

LangGraph oferece um equilíbrio único entre flexibilidade e controle, permitindo construir agentes verdadeiramente inteligentes que podem raciocinar, agir e aprender com suas próprias ações.

Referências