from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command
class AgentState(TypedDict, total=False):
comentario_original: str
analise_do_agente: str
politicas_relevantes: str
status_da_moderacao: str
justificativa_final: str
DIRETRIZES = {
"spam": "Proibido spam, links promocionais e autopromoção.",
"ofensivo": "Proibidos insultos e ataques pessoais."
}
def analisador(state: AgentState):
t = state["comentario_original"].lower()
if any(x in t for x in ["http", "grátis", "compre agora"]):
return {"analise_do_agente": "potencialmente problemático: spam"}
if any(x in t for x in ["idiota", "burro", "lixo"]):
return {"analise_do_agente": "potencialmente problemático: ofensivo"}
return {"analise_do_agente": "positivo"}
def rota(state: AgentState):
return "pesquisador" if "potencialmente problemático" in state["analise_do_agente"] else "aprovar"
def pesquisador(state: AgentState):
a = state["analise_do_agente"]
if "spam" in a:
return {"politicas_relevantes": DIRETRIZES["spam"]}
return {"politicas_relevantes": DIRETRIZES["ofensivo"]}
def revisor(state: AgentState):
a = state["analise_do_agente"]
p = state.get("politicas_relevantes", "")
if "spam" in a:
s = "Remover por Spam"
elif "ofensivo" in a:
s = "Editar por linguagem inadequada"
else:
s = "Aprovado"
return {"status_da_moderacao": s, "justificativa_final": f"{a}. {p}"}
def aprovar(state: AgentState):
return {
"status_da_moderacao": "Aprovado",
"justificativa_final": "Comentário sem violação."
}
def executar_acao_final(state: AgentState):
print("\n=== AÇÃO FINAL REGISTRADA ===")
print("Status:", state["status_da_moderacao"])
print("Justificativa:", state["justificativa_final"])
return {}
builder = StateGraph(AgentState)
builder.add_node("analisador", analisador)
builder.add_node("pesquisador", pesquisador)
builder.add_node("revisor", revisor)
builder.add_node("aprovar", aprovar)
builder.add_node("executar_acao_final", executar_acao_final)
builder.add_edge(START, "analisador")
builder.add_conditional_edges("analisador", rota, {
"pesquisador": "pesquisador",
"aprovar": "aprovar"
})
builder.add_edge("pesquisador", "revisor")
builder.add_edge("revisor", "executar_acao_final")
builder.add_edge("aprovar", END)
builder.add_edge("executar_acao_final", END)
memory = MemorySaver()
graph = builder.compile(checkpointer=memory, interrupt_before=["executar_acao_final"])
config = {"configurable": {"thread_id": "mod-3"}}
entrada = {"comentario_original": "Esse curso é lixo, compre agora grátis!"}
# Executa até o ponto de pausa
for _ in graph.stream(entrada, config=config):
pass
# Snapshot do estado pausado
snapshot = graph.get_state(config)
estado = snapshot.values
print("\n=== SNAPSHOT DO ESTADO ===")
print("Análise:", estado.get("analise_do_agente"))
print("Recomendação:", estado.get("status_da_moderacao"))
print("Justificativa atual:", estado.get("justificativa_final"))
nova_just = input("Digite nova justificativa_final (ou Enter para manter): ").strip()
if nova_just:
novos_valores = {
"justificativa_final": nova_just,
"status_da_moderacao": estado["status_da_moderacao"]
}
graph.update_state(config, novos_valores)
# Continua a execução a partir do estado atualizado
for _ in graph.stream(Command(resume={}), config=config):
pass
Conte com o apoio da comunidade Alura na sua jornada. Abraços e bons estudos!