🕸️ LangGraph Profundo

# 1. Estado: o coração do grafo
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
import operator

class AgentState(TypedDict):
    mensagem: str
    historico: Annotated[list, operator.add]  # acumula
    iteracoes: int
    aprovado: bool

# 2. Nós: funções que transformam o estado
def processar(state: AgentState) -> AgentState:
    return {"iteracoes": state["iteracoes"] + 1}

# 3. Grafo: conecta tudo
grafo = StateGraph(AgentState)
grafo.add_node("processar", processar)
grafo.add_edge(START, "processar")
grafo.add_edge("processar", END)

# Edge FIXA: sempre vai para o mesmo próximo nó
grafo.add_edge("analisar", "formatar")  # sempre

# Edge CONDICIONAL: função decide o próximo nó
def decidir_rota(state: AgentState) -> str:
    if state["aprovado"]:
        return "finalizar"
    elif state["iteracoes"] >= 3:
        return "escalar"
    else:
        return "revisar"

grafo.add_conditional_edges(
    "avaliar",          # nó de origem
    decidir_rota,       # função roteadora
    {                   # mapeamento string → nó
        "finalizar": "finalizar",
        "escalar": "escalar",
        "revisar": "revisar"
    }
)

# Múltiplos destinos possíveis do mesmo nó
grafo.add_node("avaliar", avaliar_resultado)
grafo.add_node("finalizar", gerar_output)
grafo.add_node("escalar", notificar_humano)
grafo.add_node("revisar", melhorar_resposta)

from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from typing import TypedDict

# 1. Define o estado
class AnaliseState(TypedDict):
    pergunta: str
    rascunho: str
    critica: str
    resposta_final: str
    tentativas: int

llm = ChatOpenAI(model="gpt-4o-mini")

# 2. Define os nós
def gerar_rascunho(state: AnaliseState):
    resposta = llm.invoke(f"Analise: {state['pergunta']}")
    return {"rascunho": resposta.content, "tentativas": state.get("tentativas", 0) + 1}

def criticar(state: AnaliseState):
    prompt = f"Critique esta análise: {state['rascunho']}"
    critica = llm.invoke(prompt)
    return {"critica": critica.content}

def refinar(state: AnaliseState):
    prompt = f"Melhore com base na crítica: {state['rascunho']}\nCrítica: {state['critica']}"
    melhorado = llm.invoke(prompt)
    return {"rascunho": melhorado.content}

def finalizar(state: AnaliseState):
    return {"resposta_final": state["rascunho"]}

# 3. Lógica de roteamento
def checar_qualidade(state: AnaliseState) -> str:
    if "APROVADO" in state["critica"] or state["tentativas"] >= 3:
        return "finalizar"
    return "refinar"

# 4. Monta o grafo
grafo = StateGraph(AnaliseState)
grafo.add_node("gerar", gerar_rascunho)
grafo.add_node("criticar", criticar)
grafo.add_node("refinar", refinar)
grafo.add_node("finalizar", finalizar)

grafo.add_edge(START, "gerar")
grafo.add_edge("gerar", "criticar")
grafo.add_conditional_edges("criticar", checar_qualidade, {
    "finalizar": "finalizar",
    "refinar": "refinar"
})
grafo.add_edge("refinar", "criticar")  # loop!
grafo.add_edge("finalizar", END)

# 5. Compila e executa
app = grafo.compile()
resultado = app.invoke({"pergunta": "Qual o impacto do LangGraph em produção?"})

# Desenvolvimento: memória (não persiste entre reinicializações)
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
app = grafo.compile(checkpointer=checkpointer)

# Produção: PostgreSQL (persiste entre restarts)
from langgraph.checkpoint.postgres import PostgresSaver
with PostgresSaver.from_conn_string(DATABASE_URL) as checkpointer:
    app = grafo.compile(checkpointer=checkpointer)

# thread_id identifica a sessão — SEMPRE use um
config = {"configurable": {"thread_id": "usuario_123_sessao_456"}}

# Cada invocação com o mesmo thread_id retoma do ponto anterior
resultado1 = app.invoke({"mensagem": "olá"}, config=config)
resultado2 = app.invoke({"mensagem": "continue"}, config=config)

# Inspeciona o estado salvo
estado_atual = app.get_state(config)
print(estado_atual.values)   # dados atuais
print(estado_atual.next)     # próximo nó a executar

# Histórico completo de todos os checkpoints
for snapshot in app.get_state_history(config):
    print(snapshot.config, snapshot.values)

# stream_mode="values": estado completo após cada nó
for estado in app.stream(input, stream_mode="values"):
    print(f"Estado: {estado}")

# stream_mode="updates": apenas o diff (mais eficiente)
for chunk in app.stream(input, stream_mode="updates"):
    no, atualizacao = list(chunk.items())[0]
    print(f"Nó '{no}' atualizou: {atualizacao}")

# stream_mode="debug": tudo, incluindo metadados internos
for evento in app.stream(input, stream_mode="debug"):
    print(evento)

# Streaming de tokens do LLM dentro dos nós
async for evento in app.astream_events(input, version="v2"):
    if evento["event"] == "on_chat_model_stream":
        token = evento["data"]["chunk"].content
        print(token, end="", flush=True)

# Integração com FastAPI SSE para UI em tempo real
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app_api = FastAPI()

@app_api.get("/executar")
async def executar_agente(pergunta: str):
    async def gerar():
        async for chunk in app.astream({"pergunta": pergunta}):
            yield f"data: {str(chunk)}\n\n"
    return StreamingResponse(gerar(), media_type="text/event-stream")