Um agente é um laço, não um prompt esperto

# .env   segredos, fora do código e fora do git.
# Cada linha vira uma variável de ambiente do processo.
ANTHROPIC_API_KEY=sk-ant-api03-...
LANGSMITH_TRACING=true
LANGSMITH_API_KEY=lsv2_pt_...
LANGSMITH_PROJECT=agente-suporte

# Endereços internos também moram aqui: eles mudam por ambiente,
# e ambiente não é assunto de código-fonte.
ERP_BASE_URL=https://erp.interno.local

# Guarde um .env.example COM as chaves vazias e sem segredo nenhum:
# é a documentação de quais variáveis o projeto precisa.

# .gitignore   escrito ANTES do primeiro commit.
# Depois que o .env entra no histórico, apagar o arquivo não basta:
# o commit antigo continua carregando a chave.
.env
.env.*
!.env.example

.venv/
__pycache__/
.pytest_cache/
*.sqlite
.langgraph_api/

# src/agente/estado.py
# O contrato entre os nós: tudo que atravessa o grafo está aqui.
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages


class Estado(TypedDict):
    # Annotated[tipo, redutor]: o redutor decide o que fazer com o
    # pedaço devolvido pelo nó. add_messages ACUMULA e substitui pelo
    # id quando a mensagem já existe. Sem ele, cada nó SOBRESCREVE
    # a lista inteira   e o agente esquece tudo entre um passo e outro.
    messages: Annotated[list, add_messages]

    # Campos sem redutor são substituídos, que é o certo para um valor
    # único. Guarde aqui o que o agente precisa lembrar mas não deve
    # reenviar ao modelo em toda chamada.
    cliente: str
    passos: int

# Atalho equivalente, quando o estado é só a conversa:
#   from langgraph.graph import MessagesState
#   class Estado(MessagesState): cliente: str

# src/agente/ferramentas.py
# Uma ferramenta é uma função comum com três coisas a mais:
# tipos anotados, docstring que o modelo lê, e erro que não explode.
import os
import httpx
from langchain.tools import tool

BASE = os.environ["ERP_BASE_URL"]
LIMITE = 1500   # caracteres devolvidos ao modelo, no máximo


@tool
def consultar_pedido(numero: str, incluir_itens: bool = False) -> str:
    """Busca a situação de um pedido no ERP pelo número.

    Use quando o usuário citar um número de pedido e quiser saber
    status, data de envio ou código de rastreio. Não use para trocas
    ou reembolso, que têm ferramenta própria.

    numero: identificador impresso na nota, formato PED-000000.
    incluir_itens: True traz a lista de itens, que é longa.

    Devolve uma linha por campo, ou "não encontrado".
    """
    try:
        # timeout SEMPRE: sem ele, a API lenta pendura o nó do grafo,
        # o servidor segura a conexão e a fila cresce atrás.
        r = httpx.get(f"{BASE}/pedidos/{numero}", timeout=5.0)
        if r.status_code == 404:
            return "não encontrado"
        r.raise_for_status()
    except httpx.HTTPError as e:
        # Erro VOLTA COMO TEXTO, não como exceção: o modelo lê,
        # entende e tenta outra coisa. Exceção mataria o turno.
        return f"erro ao consultar o ERP: {e}"

    dados = r.json()
    linhas = [f"{k}: {v}" for k, v in dados.items() if k != "itens"]
    if incluir_itens:
        linhas += [f"item: {i['sku']} x{i['qtd']}" for i in dados["itens"]]

    # Corte declarado: resposta gigante entra no histórico e fica lá,
    # sendo repaginada em TODOS os turnos seguintes.
    saida = "\n".join(linhas)
    return saida[:LIMITE] + ("\n…cortado" if len(saida) > LIMITE else "")


FERRAMENTAS = [consultar_pedido]   # a lista que o grafo importa

# src/agente/grafo.py   o arquivo que não dá para não ter.
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import InMemorySaver

from .estado import Estado
from .ferramentas import FERRAMENTAS

SISTEMA = SystemMessage("""Você atende o suporte de uma loja.
Responda em português, em no máximo 4 linhas.
Use as ferramentas para qualquer dado de pedido: nunca invente número,
data ou código de rastreio. Se a ferramenta falhar, diga isso.""")

# bind_tools é o que faz o modelo ENXERGAR as ferramentas: ele converte
# cada função no schema JSON que vai junto com o prompt.
modelo = init_chat_model("anthropic:claude-sonnet-5").bind_tools(FERRAMENTAS)


def no_agente(estado: Estado) -> dict:
    """Um nó recebe o estado inteiro e devolve só o pedaço que mudou."""
    resposta = modelo.invoke([SISTEMA] + estado["messages"])
    return {"messages": [resposta], "passos": estado.get("passos", 0) + 1}


construtor = StateGraph(Estado)
construtor.add_node("agente", no_agente)
construtor.add_node("ferramentas", ToolNode(FERRAMENTAS))

construtor.add_edge(START, "agente")

# A ARESTA CONDICIONAL é o laço inteiro: tools_condition olha a última
# mensagem   se ela pede ferramenta, vai para "ferramentas"; se não,
# vai para END. É uma função Python comum, e dá para escrever a sua.
construtor.add_conditional_edges(
    "agente", tools_condition, {"tools": "ferramentas", END: END}
)
construtor.add_edge("ferramentas", "agente")   # e volta: eis o ciclo

# O checkpointer salva o estado a cada super-passo. É o que permite
# continuar uma conversa, rebobinar e inspecionar passo a passo.
grafo = construtor.compile(checkpointer=InMemorySaver())


if __name__ == "__main__":
    # thread_id é o identificador da conversa. Mesmo id = mesmo histórico.
    config = {"configurable": {"thread_id": "cliente-a"}, "recursion_limit": 12}
    estado = grafo.invoke(
        {"messages": [{"role": "user", "content": "cadê o PED-000123?"}]},
        config,
    )
    print(estado["messages"][-1].content)

# tests/test_ferramentas.py
# Ferramenta é função comum: testa-se como função comum, sem modelo,
# sem rede e sem gastar um centavo. Roda em milissegundos.
import httpx
import pytest

from agente.ferramentas import consultar_pedido


def test_pedido_inexistente(monkeypatch):
    # monkeypatch troca a chamada de rede por uma resposta fixa
    monkeypatch.setattr(httpx, "get", lambda *a, **k: httpx.Response(404))
    # .invoke porque @tool embrulha a função; .func chama a original
    assert consultar_pedido.invoke({"numero": "PED-000001"}) == "não encontrado"


def test_erro_de_rede_volta_como_texto(monkeypatch):
    def explode(*a, **k):
        raise httpx.ConnectTimeout("tempo esgotado")

    monkeypatch.setattr(httpx, "get", explode)
    saida = consultar_pedido.invoke({"numero": "PED-000001"})

    # A regra que este teste protege: ferramenta NUNCA levanta exceção
    # para o grafo. Erro é texto, e o modelo decide o que fazer com ele.
    assert "erro" in saida


def test_resposta_longa_e_cortada(monkeypatch):
    grande = {"campo": "x" * 5000, "itens": []}
    monkeypatch.setattr(httpx, "get", lambda *a, **k: httpx.Response(200, json=grande))
    assert len(consultar_pedido.invoke({"numero": "PED-1"})) < 1600

// langgraph.json   diz ao servidor local qual grafo publicar.
// Sem ele, `langgraph dev` sobe e o Studio fica vazio.
{
  "dependencies": ["."],
  "graphs": {
    // "nome-da-rota": "caminho/do/arquivo.py:variável_do_grafo"
    "suporte": "./src/agente/grafo.py:grafo"
  },
  "env": ".env"
}

# src/agente/rastro.py   o rastro é configuração, não código.
# Com estas três variáveis no ambiente, TODA chamada de modelo e
# TODO passo do grafo aparecem em smith.langchain.com, sem import.
import os
from dotenv import load_dotenv

load_dotenv()

FALTANDO = [k for k in ("LANGSMITH_TRACING", "LANGSMITH_API_KEY") if k not in os.environ]
if FALTANDO:
    # Falha cedo e em voz alta: agente sem rastro em produção é uma
    # caixa que às vezes responde errado e você não sabe por quê.
    raise RuntimeError(f"rastro desligado, faltam: {FALTANDO}")

os.environ.setdefault("LANGSMITH_PROJECT", "agente-suporte")

# Para marcar uma função sua no mesmo rastro:
#   from langsmith import traceable
#   @traceable(name="normaliza_numero_pedido")
#   def normaliza(n: str) -> str: ...

# src/agente/persistencia.py
# InMemorySaver perde tudo quando o processo reinicia   ótimo para
# estudar, inaceitável quando existe um usuário do outro lado.
import os
from langgraph.checkpoint.postgres import PostgresSaver

# pip install langgraph-checkpoint-postgres psycopg[binary]
URL = os.environ["POSTGRES_URL"]


def abrir_checkpointer():
    """Context manager: use com `with` no ponto de entrada do processo."""
    cm = PostgresSaver.from_conn_string(URL)
    return cm


# Uma vez, na primeira subida, para criar as tabelas:
#   with abrir_checkpointer() as cp:
#       cp.setup()
#       grafo = construtor.compile(checkpointer=cp)
#
# O que passa a persistir: todo o estado, a cada super-passo. Isso
# inclui o conteúdo das mensagens   trate o banco do checkpoint com
# o mesmo cuidado que você trata a tabela de clientes.

# Dockerfile   a mesma coisa, em qualquer máquina.
FROM python:3.12-slim

# uv dentro da imagem: instala rápido e respeita o lock
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv

WORKDIR /app
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev

COPY src ./src
COPY langgraph.json ./

# A chave NÃO entra na imagem: ela entra como variável de ambiente
# na hora de rodar   docker run -e ANTHROPIC_API_KEY=... 
ENV PYTHONUNBUFFERED=1
EXPOSE 2024

CMD ["uv", "run", "langgraph", "dev", "--host", "0.0.0.0", "--port", "2024"]

# src/agente/supervisor.py
# Quando dividir: um agente só começa a errar a escolha por volta de
# 10 a 15 ferramentas no mesmo cinto. A saída não é prompt maior,
# é mais de um agente, cada um com o seu punhado de ferramentas.
from typing import Literal
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command

from .estado import Estado


def supervisor(estado: Estado) -> Command[Literal["pedidos", "financeiro", END]]:
    """Lê a conversa e decide QUEM continua. Não responde ao usuário."""
    ultima = estado["messages"][-1].content.lower()

    # Roteamento por regra quando dá: é grátis, instantâneo e testável.
    # Só chame o modelo para decidir quando a regra não resolver.
    if "reembolso" in ultima or "estorno" in ultima:
        return Command(goto="financeiro")
    if "pedido" in ultima or "entrega" in ultima:
        return Command(goto="pedidos")
    return Command(goto=END)


construtor = StateGraph(Estado)
construtor.add_node("supervisor", supervisor)
construtor.add_node("pedidos", grafo_pedidos)       # um grafo pode ser nó
construtor.add_node("financeiro", grafo_financeiro)
construtor.add_edge(START, "supervisor")
supervisao = construtor.compile()