LangGraph: agentes como state machines (com cycles)

from typing import TypedDict, Annotated, Sequence
import operator
from langgraph.graph import StateGraph, START, END
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool

# 1. State schema
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]  # reducer: concat

# 2. Tools
@tool
def get_weather(city: str) -> str:
    """Retorna a temperatura atual da cidade."""
    return f"Em {city}: 22°C, ensolarado"

tools = [get_weather]
model = ChatAnthropic(model="claude-sonnet-4-7").bind_tools(tools)

# 3. Nodes
def call_model(state: AgentState) -> dict:
    response = model.invoke(state["messages"])
    return {"messages": [response]}

def call_tool(state: AgentState) -> dict:
    last_message = state["messages"][-1]
    tool_calls = last_message.tool_calls
    outputs = []
    for tc in tool_calls:
        tool_fn = {t.name: t for t in tools}[tc["name"]]
        result = tool_fn.invoke(tc["args"])
        outputs.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
    return {"messages": outputs}

# 4. Conditional router
def should_continue(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if not last_message.tool_calls:
        return END
    return "tools"

# 5. Build graph
graph = StateGraph(AgentState)
graph.add_node("agent", call_model)
graph.add_node("tools", call_tool)
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")  # ⬅️ CICLO — volta ao agente após tool

app = graph.compile()

result = app.invoke({"messages": [HumanMessage("Qual a temperatura em São Paulo?")]})
for m in result["messages"]:
    print(f"{m.type}: {m.content[:80]}")

from langgraph.checkpoint.postgres import PostgresSaver

# Checkpointer persiste estado a cada nó
checkpointer = PostgresSaver.from_conn_string("postgresql://localhost/langgraph")
checkpointer.setup()

app = graph.compile(checkpointer=checkpointer)

# thread_id identifica a conversa
config = {"configurable": {"thread_id": "user-42-conv-1"}}

# Primeira mensagem
app.invoke({"messages": [HumanMessage("Olá")]}, config=config)

# Segunda mensagem — engine carrega estado anterior
app.invoke({"messages": [HumanMessage("Continuando: temperatura em SP?")]}, config=config)

# Time travel — listar histórico
for snapshot in app.get_state_history(config):
    print(f"Step {snapshot.metadata['step']}: next={snapshot.next}")
    print(f"  Messages: {len(snapshot.values['messages'])}")

# Voltar a um estado específico e bifurcar
target = list(app.get_state_history(config))[3]   # snapshot de 3 steps atrás
new_config = app.update_state(target.config, {"messages": [HumanMessage("Pergunta diferente")]})
app.invoke(None, new_config)   # executa a partir do snapshot bifurcado

# Cenário: agente sugere ação destrutiva, humano precisa aprovar
from langgraph.graph import StateGraph
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    messages: Annotated[list, operator.add]
    pending_action: dict | None

def plan(state):
    # LLM propõe ação destrutiva (ex.: deletar arquivo)
    return {"pending_action": {"type": "delete", "target": "/tmp/important.db"}}

def execute(state):
    action = state["pending_action"]
    # Em produção: subprocess.run / API call etc.
    return {"messages": [AIMessage(f"Executado: {action}")]}

graph = StateGraph(State)
graph.add_node("plan", plan)
graph.add_node("execute", execute)
graph.add_edge(START, "plan")
graph.add_edge("plan", "execute")
graph.add_edge("execute", END)

# 🛑 PAUSAR antes de "execute" para esperar aprovação
app = graph.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["execute"],
)

config = {"configurable": {"thread_id": "req-1"}}
result = app.invoke({"messages": [], "pending_action": None}, config)
print("Pausado em:", app.get_state(config).next)
# > Pausado em: ('execute',)

# Backend retorna ao usuário: "Aprovar deletar /tmp/important.db?"
# Frontend mostra modal, usuário clica em "Aprovar"

# Resume — invoke com None continua do checkpoint
app.invoke(None, config)