Revisão Conceitual
LangGraph
Nós (Nodes) → Funções/agentes especializados.
Arestas (Edges) → Conexões entre nós.
Arestas Condicionais → Controlam fluxo dinâmico.
Busca agêntica → Agente pesquisa, analisa e decide.
Multiagentes
Divisão de responsabilidades:
Planejar
Pesquisar
Escrever
Criticar
Refinar
Isso melhora:
Modularidade
Escalabilidade
Qualidade do conteúdo
Instalação de Dependências
pip install langgraph langchain langchain-google-genai tavily-python selenium beautifulsoup4 python-dotenv
Configuração Inicial
import os
import uuid
from dotenv import load_dotenv
from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_google_genai import ChatGoogleGenerativeAI
from tavily import TavilyClient
load_dotenv()
llm = ChatGoogleGenerativeAI(model="gemini-1.5-flash", temperature=0)
tavily = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))
Agent State
class AgentState(TypedDict):
task: str
plan: str
draft: str
critic: str
content: str
revision_number: int
Prompts dos 5 Agentes
Planejador
def planner_node(state: AgentState):
prompt = f"Crie um plano de redação detalhado para a tarefa: {state['task']}"
plan = llm.invoke(prompt).content
return {"plan": plan}
Pesquisador
def research_node(state: AgentState):
results = tavily.search(query=state["task"], max_results=3)
return {"content": str(results)}
Escritor (Geração)
def writer_node(state: AgentState):
prompt = f"""
Usando o plano: {state['plan']}
E os dados: {state['content']}
Escreva um artigo em 5 parágrafos.
"""
draft = llm.invoke(prompt).content
return {"draft": draft}
Crítico (Reflexão)
def critic_node(state: AgentState):
prompt = f"Critique tecnicamente o texto:\n{state['draft']}"
critic = llm.invoke(prompt).content
return {"critic": critic}
Pesquisa da Crítica (Refinamento)
def research_critic_node(state: AgentState):
results = tavily.search(query=state["critic"], max_results=2)
updated_content = state["draft"] + "\n\nAtualizações baseadas em pesquisa:\n" + str(results)
return {
"content": updated_content,
"revision_number": state["revision_number"] + 1
}
Aresta Condicional
def should_continue(state: AgentState):
if state["revision_number"] >= 2:
return END
return "writer"
Construindo o Grafo
builder = StateGraph(AgentState)
builder.add_node("planner", planner_node)
builder.add_node("research", research_node)
builder.add_node("writer", writer_node)
builder.add_node("critic", critic_node)
builder.add_node("research_critic", research_critic_node)
builder.set_entry_point("planner")
builder.add_edge("planner", "research")
builder.add_edge("research", "writer")
builder.add_edge("writer", "critic")
builder.add_edge("critic", "research_critic")
builder.add_conditional_edges(
"research_critic",
should_continue,
{
"writer": "writer",
END: END
}
)
graph = builder.compile()
Testando o Fluxo
initial_state = {
"task": "Impactos da Inteligência Artificial na Educação",
"plan": "",
"draft": "",
"critic": "",
"content": "",
"revision_number": 0
}
result = graph.invoke(initial_state)
print("\nConteúdo Final:\n")
print(result["content"])
Persistência e Streaming
Persistência (SQLiteSaver)
from langgraph.checkpoint.sqlite import SqliteSaver
memory = SqliteSaver.from_conn_string("sqlite:///multiagent.db")
graph = builder.compile(checkpointer=memory)
Streaming
for step in graph.stream(initial_state):
print(step)
Human-in-the-Loop
Adicionar pausa antes da publicação:
def human_review(state: AgentState):
print("Conteúdo gerado:\n", state["content"])
decision = input("Aprovar publicação? (sim/não): ")
if decision == "sim":
return {}
else:
correction = input("Digite a correção: ")
return {"content": correction}
Inserir antes do END.
Interface com Gradle
Fluxo:
Backend Python roda o grafo.
API REST expõe endpoint.
Interface Java (Gradle) consome API.
Exibe:
Plano
Rascunho
Crítica
Versão final
Gradle (build.gradle):
plugins {
id 'java'
}
repositories {
mavenCentral()
}
