Artigo
· 4 hr atrás 6min de leitura

Desvendando o LangGraph

Como Construir Aplicações com LangGraph: Um Guia Passo a Passo

Tags: #LangGraph #LangChain #AI #Agents #Python #LLM #StateManagement #Workflows


Olá pessoal, quero trazer para vocês aqui um pouco sobre o LangGraph, uma ferramenta que estou estudando e desenvolvendo.

Basicamente aplicações tradicionais de IA frequentemente enfrentam desafios ao lidar com fluxos de trabalho complexos e estados dinâmicos. O LangGraph oferece uma solução robusta, permitindo criar agentes com estado que podem gerenciar conversas complexas, tomar decisões baseadas em contexto e executar workflows sofisticados.

Este artigo fornece um guia passo a passo para construir aplicações utilizando LangGraph, um framework para criação de agentes multi-etapas com grafos de estado.


Passos de Implementação:

  1. Configurar o Ambiente e Instalar Dependências
  2. Definir o Estado da Aplicação
  3. Criar os Nós do Grafo
  4. Configurar o Grafo de Estado
  5. Executar o Agente

1. Configurar o Ambiente e Instalar Dependências

O primeiro passo é configurar o ambiente Python e instalar as bibliotecas necessárias:

pip install langgraph langchain langchain-openai

Configure suas credenciais de API:

import os
from langchain_openai import ChatOpenAI

# Configure sua API Key
os.environ["OPENAI_API_KEY"] = "sua-api-key-aqui"

# Inicialize o modelo
llm = ChatOpenAI(model="gpt-4", temperature=0)

2. Definir o Estado da Aplicação

O LangGraph utiliza um TypedDict para definir o estado que será compartilhado entre os nós do grafo:

from typing import TypedDict, Annotated
from operator import add

class AgentState(TypedDict):
    """Estado compartilhado entre os nós do grafo"""
    messages: Annotated[list, add]
    user_input: str
    response: str
    next_step: str

Este estado armazena:

  • messages: Histórico de mensagens trocadas
  • user_input: Entrada atual do usuário
  • response: Resposta gerada pelo agente
  • next_step: Próxima ação a ser executada

Estado do Grafo


3. Criar os Nós do Grafo

3.1 - Nó de Processamento de Entrada

Este nó processa a entrada do usuário e prepara o contexto:

def process_input(state: AgentState) -> AgentState:
    """Processa a entrada do usuário"""
    user_message = state["user_input"]
    
    # Adiciona a mensagem ao histórico
    state["messages"].append({
        "role": "user",
        "content": user_message
    })
    
    # Define próximo passo
    state["next_step"] = "analyze"
    
    return state

3.2 - Nó de Análise e Decisão

Este nó utiliza o LLM para analisar a entrada e decidir a próxima ação:

from langchain.prompts import ChatPromptTemplate

def analyze_request(state: AgentState) -> AgentState:
    """Analisa a requisição e decide a próxima ação"""
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", "Você é um assistente inteligente. Analise a requisição do usuário e determine a melhor forma de responder."),
        ("user", "{input}")
    ])
    
    chain = prompt | llm
    
    result = chain.invoke({
        "input": state["user_input"]
    })
    
    state["response"] = result.content
    state["next_step"] = "respond"
    
    return state

3.3 - Nó de Resposta

Este nó formata e retorna a resposta final:

def generate_response(state: AgentState) -> AgentState:
    """Gera a resposta final"""
    
    # Adiciona resposta ao histórico
    state["messages"].append({
        "role": "assistant",
        "content": state["response"]
    })
    
    state["next_step"] = "END"
    
    return state

4. Configurar o Grafo de Estado

4.1 - Criar o Grafo

Agora vamos conectar todos os nós em um grafo de estado:

from langgraph.graph import StateGraph, END

# Criar o grafo
workflow = StateGraph(AgentState)

# Adicionar os nós
workflow.add_node("process_input", process_input)
workflow.add_node("analyze", analyze_request)
workflow.add_node("respond", generate_response)

# Definir o ponto de entrada
workflow.set_entry_point("process_input")

# Adicionar as transições (edges)
workflow.add_edge("process_input", "analyze")
workflow.add_edge("analyze", "respond")
workflow.add_edge("respond", END)

# Compilar o grafo
app = workflow.compile()

4.2 - Visualizar o Grafo

O LangGraph permite visualizar a estrutura do grafo:

from IPython.display import Image, display

try:
    display(Image(app.get_graph().draw_mermaid_png()))
except Exception:
    print("Visualização do grafo requer dependências adicionais")

Fluxo do Grafo


5. Executar o Agente

5.1 - Executar uma Consulta Simples

def run_agent(user_input: str):
    """Executa o agente com a entrada do usuário"""
    
    # Estado inicial
    initial_state = {
        "messages": [],
        "user_input": user_input,
        "response": "",
        "next_step": ""
    }
    
    # Executar o grafo
    result = app.invoke(initial_state)
    
    return result["response"]

# Testar o agente
response = run_agent("Qual é a capital da França?")
print(f"Resposta: {response}")

Saída esperada:

Resposta: A capital da França é Paris.

5.2 - Executar com Streaming

Para aplicações interativas, você pode usar streaming:

async def run_agent_stream(user_input: str):
    """Executa o agente com streaming"""
    
    initial_state = {
        "messages": [],
        "user_input": user_input,
        "response": "",
        "next_step": ""
    }
    
    async for event in app.astream(initial_state):
        for node_name, node_state in event.items():
            print(f"\n--- {node_name} ---")
            if "response" in node_state and node_state["response"]:
                print(f"Resposta parcial: {node_state['response']}")

Funcionalidades Avançadas

Checkpoints e Persistência

O LangGraph suporta checkpoints para salvar o estado:

from langgraph.checkpoint.memory import MemorySaver

# Adicionar memória ao grafo
memory = MemorySaver()
app_with_memory = workflow.compile(checkpointer=memory)

# Executar com persistência
config = {"configurable": {"thread_id": "user-123"}}
result = app_with_memory.invoke(initial_state, config)

Condições e Roteamento Dinâmico

Você pode adicionar lógica condicional para roteamento:

def router(state: AgentState) -> str:
    """Determina o próximo nó baseado no estado"""
    
    if "urgente" in state["user_input"].lower():
        return "priority_handler"
    else:
        return "normal_handler"

# Adicionar roteamento condicional
workflow.add_conditional_edges(
    "analyze",
    router,
    {
        "priority_handler": "priority_node",
        "normal_handler": "normal_node"
    }
)

Casos de Uso

O LangGraph é ideal para:

  1. Chatbots Complexos: Gerenciar conversas multi-turno com contexto
  2. Agentes Autônomos: Criar agentes que tomam decisões baseadas em estado
  3. Workflows de Processamento: Orquestrar pipelines de processamento de dados
  4. Sistemas Multi-Agente: Coordenar múltiplos agentes especializados

Veja em Ação

Para mais detalhes e exemplos práticos, visite:


Conclusão

O LangGraph oferece uma abordagem poderosa e flexível para construir aplicações de IA com estado. Ao combinar grafos de estado com LLMs, você pode criar sistemas sofisticados que gerenciam conversas complexas, tomam decisões contextuais e executam workflows dinâmicos.

A estrutura modular do LangGraph permite escalar desde simples chatbots até sistemas multi-agente complexos, mantendo o código organizado e fácil de manter.

Discussão (0)1
Entre ou crie uma conta para continuar