from typing import TypedDict, Literal, Optional
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
# Estado compartilhado entre os nós
class ModState(TypedDict, total=False):
comentario: str
diretrizes: str
violacao: bool
motivo: str
sugestao: Literal["aprovar", "remover", "editar"]
comentario_final: str
justificativa_final: str
decisao_final: Literal["aprovar", "remover", "editar"]
# Base simples de diretrizes
DIRETRIZES = """
Remover: insultos, assédio, discriminação, spam, ameaças.
Editar: linguagem agressiva leve ou palavreado inadequado.
Aprovar: comentário respeitoso e construtivo.
"""
def analisar_comentario(state: ModState):
txt = state["comentario"].lower()
termos_graves = ["idiota", "burro", "ódio", "ameaça", "spam"]
termos_leves = ["ridículo", "lixo"]
if any(t in txt for t in termos_graves):
return {"violacao": True, "motivo": "Possível ofensa/spam",
"sugestao": "remover"}
if any(t in txt for t in termos_leves):
return {"violacao": True, "motivo": "Linguagem inadequada",
"sugestao": "editar"}
return {"violacao": False, "motivo": "Sem violação aparente",
"sugestao": "aprovar"}
def pesquisar_diretrizes(state: ModState):
return {"diretrizes": DIRETRIZES}
def revisao_humana(state: ModState):
payload = {
"comentario_original": state["comentario"],
"violacao": state["violacao"],
"motivo_agente": state["motivo"],
"sugestao_agente": state["sugestao"],
"diretrizes": state["diretrizes"],
"instrucao": "Responda com {'decisao':'aprovar|remover|editar', "
"'comentario_editado':'...', 'justificativa':'...'}"
}
resposta = interrupt(payload)
return {
"decisao_final": resposta["decisao"],
"comentario_final": resposta.get("comentario_editado", state["comentario"]),
"justificativa_final": resposta.get("justificativa", "Sem justificativa")
}
def finalizar(state: ModState):
print("\n=== RESULTADO FINAL ===")
print("Decisão:", state["decisao_final"])
print("Comentário final:", state["comentario_final"])
print("Justificativa:", state["justificativa_final"])
return {}
builder = StateGraph(ModState)
builder.add_node("analisar", analisar_comentario)
builder.add_node("diretrizes", pesquisar_diretrizes)
builder.add_node("humano", revisao_humana)
builder.add_node("finalizar", finalizar)
builder.add_edge(START, "analisar")
builder.add_edge("analisar", "diretrizes")
builder.add_edge("diretrizes", "humano")
builder.add_edge("humano", "finalizar")
builder.add_edge("finalizar", END)
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "mod-001"}}
# 1) Executa até pausar no interrupt
res = graph.invoke(
{"comentario": "Esse curso é lixo, autor ridículo."},
config=config
)
print(res["__interrupt__"])
# 2) Moderador humano decide e pode editar
res2 = graph.invoke(
Command(resume={
"decisao": "editar",
"comentario_editado": "Não gostei do curso e sugiro melhorias.",
"justificativa": "Mantido o sentido, removendo ofensa."
}),
config=config
)