Construindo um Grafo Simples
Construa um LangGraph simples de 3 etapas: processamento de entrada, transformação e geração de saída. Aprenda o fluxo completo de construir-compilar-invocar.
Construindo um Grafo Simples
Nesta lição, vamos construir um LangGraph completo de 3 etapas: entrada → processamento → saída. Esta é a fundação para cada agente que você construirá.
O Padrão de 3 Etapas
Toda aplicação LangGraph segue este padrão:
- Nó de entrada: Aceita e valida a entrada bruta
- Nó de processamento: Realiza a lógica principal (chamada LLM, computação, etc.)
- Nó de saída: Formata e retorna o resultado final
Passo 1: Definir o Estado
from typing_extensions import TypedDict
from typing import Optional
class SimpleState(TypedDict):
input_text: str # Entrada bruta do usuário
processed_text: str # Valor processado intermediário
output_text: str # Saída final
error: Optional[str] # Mensagem de erro (se houver)Inclua um campo error no seu estado desde o início. Isso torna o tratamento de erros muito mais limpo à medida que seu grafo cresce.
Passo 2: Definir os Nós
Nó de Entrada
def input_node(state: SimpleState) -> dict:
raw = state["input_text"].strip()
if not raw:
return {"error": "Input cannot be empty"}
return {"input_text": raw}Nó de Processamento
def process_node(state: SimpleState) -> dict:
if state.get("error"):
return {} # Pular processamento se houver erro
# Transformação de texto simples
processed = state["input_text"].upper()
word_count = len(state["input_text"].split())
return {
"processed_text": f"[{word_count} words] {processed}"
}Nó de Saída
def output_node(state: SimpleState) -> dict:
if state.get("error"):
return {"output_text": f"Error: {state['error']}"}
return {
"output_text": f"Result: {state['processed_text']}"
}O padrão de verificar state.get("error") em cada nó é uma forma básica de propagação de erro. Mais tarde substituiremos isso por arestas condicionais para roteamento mais limpo.
Passo 3: Construir o Grafo
from langgraph.graph import StateGraph, START, END
builder = StateGraph(SimpleState)
# Adicionar nós
builder.add_node("input", input_node)
builder.add_node("process", process_node)
builder.add_node("output", output_node)
# Adicionar arestas
builder.add_edge(START, "input")
builder.add_edge("input", "process")
builder.add_edge("process", "output")
builder.add_edge("output", END)
# Compilar
app = builder.compile()Passo 4: Invocar o Grafo
# Execução bem-sucedida
result = app.invoke({
"input_text": "hello world",
"processed_text": "",
"output_text": "",
"error": None
})
print(result["output_text"])
# Result: [2 words] HELLO WORLD
# Caso de erro
result = app.invoke({
"input_text": " ",
"processed_text": "",
"output_text": "",
"error": None
})
print(result["output_text"])
# Error: Input cannot be emptyPasso 5: Adicionar Streaming
Streaming permite observar a saída de cada nó enquanto executa:
for event in app.stream({
"input_text": "langgraph is awesome",
"processed_text": "",
"output_text": "",
"error": None
}):
for node_name, state_update in event.items():
if node_name == "__end__":
continue
print(f"---[{node_name}]---")
for key, value in state_update.items():
if value:
print(f" {key}: {value}")Saída:
---[input]---
input_text: langgraph is awesome
---[process]---
processed_text: [3 words] LANGGRAPH IS AWESOME
---[output]---
output_text: Result: [3 words] LANGGRAPH IS AWESOME
Streaming oferece visibilidade em tempo real da execução do seu grafo. Use durante o desenvolvimento para verificar o comportamento de cada nó.
Exemplo Completo
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
from typing import Optional
# 1. Estado
class SimpleState(TypedDict):
input_text: str
processed_text: str
output_text: str
error: Optional[str]
# 2. Nós
def input_node(state: SimpleState) -> dict:
raw = state["input_text"].strip()
if not raw:
return {"error": "Input cannot be empty"}
return {"input_text": raw}
def process_node(state: SimpleState) -> dict:
if state.get("error"):
return {}
processed = state["input_text"].upper()
word_count = len(state["input_text"].split())
return {"processed_text": f"[{word_count} words] {processed}"}
def output_node(state: SimpleState) -> dict:
if state.get("error"):
return {"output_text": f"Error: {state['error']}"}
return {"output_text": f"Result: {state['processed_text']}"}
# 3. Grafo
builder = StateGraph(SimpleState)
builder.add_node("input", input_node)
builder.add_node("process", process_node)
builder.add_node("output", output_node)
builder.add_edge(START, "input")
builder.add_edge("input", "process")
builder.add_edge("process", "output")
builder.add_edge("output", END)
app = builder.compile()
# 4. Executar
result = app.invoke({
"input_text": "hello langgraph",
"processed_text": "",
"output_text": "",
"error": None
})
print(result["output_text"])
# Result: [2 words] HELLO LANGGRAPHAdicionando um LLM ao Nó de Processamento
Vamos atualizar o nó de processamento para usar um LLM:
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
def process_with_llm(state: SimpleState) -> dict:
if state.get("error"):
return {}
prompt = ChatPromptTemplate.from_messages([
("system", "You are a text analyzer. Analyze the given text and provide:\n"
"1. A summary (1 sentence)\n"
"2. Sentiment (positive/negative/neutral)\n"
"3. Key topics"),
("human", "{text}")
])
chain = prompt | llm | StrOutputParser()
analysis = chain.invoke({"text": state["input_text"]})
return {"processed_text": analysis}Agora, em vez de transformação simples de string, seu nó de processamento realiza análise com IA.
Trocar um nó determinístico por um nó alimentado por LLM não requer alterações na estrutura do grafo. Apenas a função do nó muda. Este é o poder da abstração do grafo.
Adicionando um Loop (Prévia)
Mesmo em um grafo simples, você pode adicionar um loop. Vamos fazer o nó de processamento repetir até que o texto esteja limpo:
from langgraph.graph import START, END, StateGraph
from typing_extensions import TypedDict
class CleanState(TypedDict):
text: str
cleaned: bool
attempts: int
def clean_text(state: CleanState) -> dict:
original = state["text"]
cleaned = original.strip().lower()
is_clean = cleaned == original
return {
"text": cleaned,
"cleaned": is_clean,
"attempts": state["attempts"] + 1
}
def should_continue(state: CleanState) -> str:
if state["cleaned"] or state["attempts"] >= 3:
return "end"
return "continue"
builder = StateGraph(CleanState)
builder.add_node("clean", clean_text)
builder.add_edge(START, "clean")
builder.add_conditional_edges(
"clean",
should_continue,
{
"continue": "clean", # Loop de volta
"end": END
}
)
app = builder.compile()
result = app.invoke({"text": " HELLO WORLD ", "cleaned": False, "attempts": 0})
print(result["text"]) # hello world
print(result["attempts"]) # 2 (primeira passagem limpa, segunda confirma)Sempre tenha uma condição de terminação em loops. Sem a verificação attempts >= 3, um bug poderia causar um loop infinito. Sempre defina recursion_limit na configuração de invocação.
Testando Seu Grafo
# Teste 1: Entrada normal
result = app.invoke({"input_text": "Test", "processed_text": "", "output_text": "", "error": None})
assert "Error" not in result["output_text"]
# Teste 2: Entrada vazia
result = app.invoke({"input_text": "", "processed_text": "", "output_text": "", "error": None})
assert "Error" in result["output_text"]
# Teste 3: Entrada com espaços
result = app.invoke({"input_text": " ", "processed_text": "", "output_text": "", "error": None})
assert "Error" in result["output_text"]Escreva testes para cada nó individualmente (testes de função pura) e para o grafo completo (testes de integração). Isso captura bugs tanto no nível do nó quanto problemas de topologia.
Erros Comuns
Erro 1: Esquecer de tratar o caso de erro
def process_node(state: State) -> dict:
# BUG: Se o estado tem um erro, isso ainda executa
return {"result": expensive_computation(state["input"])}
# CORREÇÃO: Verificar erros primeiro
def process_node(state: State) -> dict:
if state.get("error"):
return {}
return {"result": expensive_computation(state["input"])}Erro 2: Transformar o estado diretamente
def bad_node(state: State) -> dict:
state["value"] = "new" # BUG: Não transforme o estado!
return {"value": "new"} # CORRETO: Retornar atualizações
def good_node(state: State) -> dict:
return {"value": "new"} # CORRETOErro 3: Aresta faltando para END
builder.add_edge("process", "output")
# BUG: Sem aresta de output para END — o grafo nunca termina!
builder.add_edge("output", END) # CORREÇÃOPerguntas de Prática
Qual é o padrão padrão de 3 etapas para um LangGraph simples?
O que um nó deve retornar quando um erro ocorreu upstream?
Como você observa saídas intermediárias de nós durante a execução do grafo?
O que acontece se você esquecer de adicionar uma aresta do último nó para END?
Qual é a melhor prática para lidar com erros em um grafo simples?
Você pode substituir um nó determinístico por um nó alimentado por LLM sem alterar a estrutura do grafo?
O que todo loop em LangGraph deve ter?
Qual é a maneira correta de atualizar o estado em um nó?
Qual é um bom uso para um nó que retorna None?
O que o método stream() retorna para cada evento?
Principais Conclusões
- O padrão Entrada → Processamento → Saída é a fundação de todas as apps LangGraph
- Inclua um campo
errorno seu estado para tratamento de erro limpo - Use
stream()durante o desenvolvimento para observar a execução dos nós - Sempre adicione uma aresta do último nó para END
- Nós podem ser atualizados de determinísticos para alimentados por LLM sem alterações no grafo
- Loops precisam de condições de terminação
- Nunca transforme o estado diretamente — retorne um dict de atualizações
- Teste nós individualmente e o grafo completo como testes de integração