1️ Explicação da arquitetura
2️ Script ( mainLangGraphPy.py)
3️ Como funciona o fluxo
4️ Como testar
1️ arquitetura
Roteador → Decida se o usuário quer praia ou montanha
Consultor de Praia → Especialista em destinos litorâneos
Consultor de Montanha → Especialista em destinos de altitude
LangGraph → Orquestra do fluxo de decisão
Fluxo: Usuário → Roteador → (Praia ou Montanha) → Resposta Final
2️ Script ( mainLangGraphPy.py)
# mainLangGraphPy.py
import os
import asyncio
from typing import TypedDict, Literal
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langgraph.graph import StateGraph, START, END
# ==============================
# 1. CONFIGURAÇÃO
# ==============================
load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")
modelo = ChatOpenAI(
model="gpt-4o-mini",
temperature=0.5,
api_key=api_key
)
parser = StrOutputParser()
# ==============================
# 2. PROMPTS DOS ASSISTENTES
# ==============================
prompt_roteador = ChatPromptTemplate.from_messages([
("system", "Você é um roteador inteligente. "
"Analise a pergunta do usuário e responda apenas com: "
"'praia' ou 'montanha'."),
("human", "{pergunta}")
])
prompt_praia = ChatPromptTemplate.from_messages([
("system", "Você é um consultor especialista em destinos de PRAIA no Brasil."),
("human", "{pergunta}")
])
prompt_montanha = ChatPromptTemplate.from_messages([
("system", "Você é um consultor especialista em destinos de MONTANHA no Brasil."),
("human", "{pergunta}")
])
# Cadeias individuais
cadeia_roteador = prompt_roteador | modelo | parser
cadeia_praia = prompt_praia | modelo | parser
cadeia_montanha = prompt_montanha | modelo | parser
# ==============================
# 3. ESTADO DO GRAFO
# ==============================
class Estado(TypedDict):
pergunta: str
destino: str
resposta: str
# ==============================
# 4. NÓS ASSÍNCRONOS
# ==============================
async def no_roteador(estado: Estado):
destino = await cadeia_roteador.ainvoke({
"pergunta": estado["pergunta"]
})
return {"destino": destino.strip().lower()}
async def no_praia(estado: Estado):
resposta = await cadeia_praia.ainvoke({
"pergunta": estado["pergunta"]
})
return {"resposta": resposta}
async def no_montanha(estado: Estado):
resposta = await cadeia_montanha.ainvoke({
"pergunta": estado["pergunta"]
})
return {"resposta": resposta}
# Função de decisão
def escolher_destino(estado: Estado) -> Literal["praia", "montanha"]:
if "praia" in estado["destino"]:
return "praia"
return "montanha"
# ==============================
# 5. CONSTRUÇÃO DO GRAFO
# ==============================
grafo = StateGraph(Estado)
grafo.add_node("roteador", no_roteador)
grafo.add_node("praia", no_praia)
grafo.add_node("montanha", no_montanha)
grafo.add_edge(START, "roteador")
grafo.add_conditional_edges("roteador", escolher_destino)
grafo.add_edge("praia", END)
grafo.add_edge("montanha", END)
app = grafo.compile()
# ==============================
# 6. EXECUÇÃO
# ==============================
async def executar():
pergunta_usuario = input("Digite sua pergunta: ")
resultado = await app.ainvoke({
"pergunta": pergunta_usuario
})
print("\nResposta do Assistente:")
print(resultado["resposta"])
if __name__ == "__main__":
asyncio.run(executar())
3️ Como funciona o fluxo
a Usuário faz pergunta
Exemplo: Quero viajar para um lugar com praias bonitas.
2️ Rotador classifica cano: praia
3️ Direção LangGraph Vai para o nó praia.
4️ Assistente responde Sugira destinos como:
Florianópolis
Porto de Galinhas
Jericoacoara
4️ Como testar
1️ Criar ambiente virtual
python -m venv langgraph
2️ Instalar dependências
pip install langchain langgraph langchain-openai python-dotenv
3️ Criar arquivo.env
OPENAI_API_KEY="SUA_CHAVE_AQUI"
4️ Executor
python mainLangGraphPy.py
Resultado
Você terá um sistema com:
Rotação inteligente
Assistentes especializados
Arquitetura baseada em grafite
Execução assíncrona
Separação clara de responsabilidades