27 · Se você já usa LangChain, LangGraph ou NeMo Guardrails¶

Este é o único capítulo escrito nos termos desses frameworks, para uma equipe que já tem um agente, um grafo ou um conjunto de rails. Nada na Parte I precisa dele: um agente da Parte I é um grafo LangGraph por baixo, e este capítulo é onde você o abre, ou leva o mesmo roteador e a mesma verificação de ferramenta a um código que já tem. Tudo aqui roda no seu ambiente, e cada decisão sobre a qual ele age é tomada pelo runtime licenciado na mesma máquina.

Idiomas

O seletor lê turnos escritos em inglês e em português do Brasil (pt-BR). O idioma é escolhido, nunca adivinhado a partir do turno: na página Settings do console (Language the selector reads), em language numa requisição de plano, ou no settings.language do seu arquivo de vocabulário; sem nenhum deles, inglês. Um turno em português pode ser digitado com os acentos ou sem eles. O português europeu não é uma gramática que o seletor tenha, e um turno em qualquer outra língua não é lido.

Do agente da Parte I ao seu grafo¶

agent.graph() é a porta: o grafo compilado que um agente da Parte I roda, verificado ao ser compilado.

desk = Agent.from_file("agent.yaml")
graph = desk.graph()                          # o grafo compilado que o agente da Parte I roda, provado ao compilar
print(sorted(graph.get_graph().nodes))

['__end__', '__start__', 'billing_model', 'billing_tools', 'compute', 'desk_model', 'desk_to_billing', 'desk_tools', 'rewind', 'rollback', 'route']
verified, findings: []
QG001 at quick_charge: 'charge_card' can run on a plain tool node, without the tool check
QG011 at quick_charge: the privileged 'charge_card' can run carrying the label untrusted:read_inbox with no confirmation or refusal
{"order_id": "4411", "total": "40.00", "card": "visa-0019", "customer": "c-208", "status": "unpaid"}
waiting for a person: charge_card [{'name': 'amount', 'value': 40, 'origin': 'typed'}, {'name': 'order_id', 'value': '4411', 'origin': 'typed'}]
the model was called 0 times
asked: charge_card amount=40.0 (model), order_id=4411 (model) | ['argument_origin: order_id', 'argument_origin: amount']
charged: {'charge_id': 'ch-4411', 'order_id': '4411', 'amount': '40.0'}

A primeira linha da saída é o grafo da loja: route lê cada turno, desk_model e desk_tools são o modelo da central e o seu nó de ferramentas protegido, desk_to_billing passa a conversa para a central de cobrança, com o seu próprio par, compute roda um cálculo, e rollback e rewind desfazem uma execução que falhou e depois a esquecem. Ele é um StateGraph como qualquer outro: rode-o com o seu checkpointer, faça streaming dele, acrescente nós (abaixo). agent.export() é o mesmo grafo como o verificador o lê, e stalwart_agents.export(path, out) o escreve num arquivo, que é o que stalwart verify --export-only faz.

`create_agent`, com o roteador e a verificação como middleware¶

Para um agente feito com o create_agent do LangChain, dois middlewares acrescentam o roteador e a verificação de ferramenta sem mudar mais nada (E13):

chat = ScriptedChatModel(script=[])           # nenhum turno abaixo precisa dele: cada um é lido ou verificado antes
agent = create_agent(chat, tools, checkpointer=InMemorySaver(), middleware=[
    StalwartRouterMiddleware(client, agent="desk", catalogue=catalogue, on_unread="refuse"),
    # Os fatos do rulebook, lidos das respostas das ferramentas como diz o `facts:` do agent.yaml.
    StalwartToolGuardMiddleware(client, agent="desk", catalogue=catalogue, context=Context(facts=desk.facts)),
])
thread = {"configurable": {"thread_id": "chat-17"}}
state = agent.invoke({"messages": [HumanMessage("mostre o pedido 4411")]}, thread)
print(state["messages"][-1].content)
state = agent.invoke({"messages": [HumanMessage("cobre o pedido 4411 em 40,00")]}, thread)
for waiting in state.get("__interrupt__", []):
    print("waiting for a person:", waiting.value["tool"], waiting.value["arguments"])
print("the model was called", chat.turns, "times")

StalwartRouterMiddleware lê o último turno do usuário num plano antes de o modelo ser chamado, e o plano decide:

O plano	O que o middleware faz
uma etapa de ferramenta que se resolve	devolve a chamada da ferramenta como resposta do modelo: nenhuma chamada ao modelo
uma pergunta	devolve a pergunta fixa como resposta: nenhuma chamada ao modelo
recusado	devolve a recusa nomeando a regra; o turno nunca é mostrado ao modelo
uma etapa de cálculo	o runtime monta o prompt, o modelo do seu agente escreve a função e o sandbox a roda (capítulo 23)
não lido	primeiro o pedido limitado para reformular, depois a sua configuração `on_unread` (capítulo 20)

Um plano de várias etapas roda em ordem, cada etapa de ferramenta pela própria execução de ferramentas do LangChain, então todo outro middleware ainda vê cada chamada; quando a última ferramenta responde, o resultado dela é a resposta.

StalwartToolGuardMiddleware envolve cada chamada de ferramenta, seja quem for que a escolheu -- o roteador, o modelo ou outro middleware -- e pergunta à verificação: rodá-la, recusá-la com o motivo numa frase, ou pausar para uma pessoa pelo interrupt do LangGraph, o mecanismo que o próprio middleware de human-in-the-loop do LangChain usa. Uma chamada só roda depois de uma aprovação exatamente dos argumentos mostrados.

Na saída, mostre o pedido 4411 é respondido com o pedido e cobre o pedido 4411 em 40,00 espera por uma pessoa, cada argumento mostrado com a sua origem; o modelo não foi chamado nenhuma vez. Ponha o middleware do NeMo Guardrails por fora, depois StalwartRouterMiddleware, depois o modelo, com StalwartToolGuardMiddleware em volta da execução de ferramentas. Nenhum dos nossos remove o gancho de outro middleware.

Um grafo seu¶

Para um grafo que você mesmo escreve, as partes são separadas. create_stalwart_agent(model, tools, catalogue=..., agent=...) é o grafo inteiro, gerado a partir do seu catálogo e compilado; build_stalwart_graph é o mesmo, sem compilar, para você acrescentar nós. validate_state(state) nomeia a primeira coisa que quebrou um estado, e validating(graph) roda um grafo verificando o estado depois de cada nó.

O estado¶

Use StalwartState como o seu estado, ou acrescente os nossos canais ao seu com with_stalwart_channels(YourState):

Canal	Contém	Quando duas escritas se encontram
`messages`	a conversa	`stalwart_messages` (abaixo)
`stalwart.session`	a `Session` do SDK, como JSON	a mais nova
`stalwart.history`	o histórico de verificações e o seu token	o mais longo, onde ele estende o outro; um histórico editado, reordenado ou bifurcado levanta `StateConflict`
`stalwart.labels`	rótulos de confiança, por mensagem, chamada e argumento	a união
`stalwart.saga`	a pilha de desfazer, as chamadas em andamento, uma falha	a `version` mais nova
`stalwart.pending`	interrupções abertas e os seus tokens	por id de interrupção; `None` fecha uma
`stalwart.plan`	o plano em execução, as origens dos argumentos de cada chamada, rotas recusadas	combinado por chave

Tudo o que o roteador e a verificação precisam entre turnos fica aqui, no seu checkpointer. Voltar um checkpoint volta isso junto, menos stalwart.history: as chamadas já verificadas ficam na cadeia.

As mensagens¶

stalwart_messages é o add_messages do LangGraph -- a mesma combinação por id, com RemoveMessage respeitado -- com quatro regras a mais: toda chamada de ferramenta recebe um resultado (uma chamada deixada sem resposta quando a conversa segue recebe um resultado de erro numa frase simples, para que a próxima requisição a um provedor seja bem formada); os resultados das chamadas de uma mensagem vêm depois dela na ordem das suas chamadas; cortar com trim_pairs nunca separa uma chamada do seu resultado; e nada nosso é acrescentado a uma mensagem, então nada nosso chega a um provedor.

Os nós¶

Cada nó pelo que lê, pelo que escreve e para onde vai:

Nó	Lê	Escreve	Vai para
`stalwart_route_node`	o último turno do usuário, `stalwart.session`	`stalwart.plan`, `stalwart.session`; uma resposta para uma pergunta, uma recusa	o nó de ferramentas protegido com a chamada escrita; o nó de cálculo; o seu nó de modelo, para um turno não lido que a sua configuração entrega; `END`
`StalwartToolNode(tools, client, agent=...)`	as chamadas da última mensagem, `stalwart.plan`, `stalwart.history`, `stalwart.labels`	o resultado de cada chamada, `stalwart.history`, `stalwart.saga`, `stalwart.pending`	ele mesmo, um passo por vez; de volta ao roteador ou ao modelo; o nó de rollback depois de uma falha terminal
`stalwart_compute_node`	a etapa de cálculo do plano e os resultados das suas fontes	o resultado verificado, com os rótulos das suas fontes	`END`
`stalwart_rollback_node`	`stalwart.saga`	o resultado de cada desfazer, verificado como o agente cuja escrita ele desfaz	o nó de rewind, quando a pilha esvazia
`stalwart_rewind_node`	as mensagens da execução	as mensagens da execução trocadas por um resumo: o que falhou, o que foi desfeito, o que não pôde ser	`END`

StalwartToolNode fica onde ficaria o ToolNode. Uma chamada recusada é zero invocações, e uma permitida exatamente uma; uma chamada que espera por uma pessoa pausa com interrupt() e, ao ser retomada, é verificada de novo com o token da aprovação. Dentro de uma ferramenta que ele roda, current_idempotency_key() é a chave da escrita em curso, para uma ferramenta que a repassa a uma API. stalwart_branches espalha as chamadas de uma mensagem em ramos Send, cada uma verificada no seu próprio ramo, revezando-se no único histórico.

As arestas¶

stalwart_edge(allowed={...}, choose=fn) é uma aresta condicional cuja função lê a saída do modelo; um destino fora do conjunto declarado não é seguido, e a aresta vai para um nó fixo que recusa e registra isso. add_stalwart_edge acrescenta esse nó onde o seu grafo não tem um. O conjunto declarado é o mapa de caminhos da aresta, então a exportação o conhece: uma aresta condicional sem mapa de caminhos, ou um nó Command sem a anotação Command[Literal[...]], pode chegar a qualquer nó, e o verificador relata isso.

Exportando um grafo para verificá-lo¶

O compile() de um grafo é onde ele é verificado: export_graph(compiled, catalogue) o escreve como o verificador o lê (qls-graph/1), o tipo de cada nó, as ferramentas que um nó de ferramentas liga e os destinos que cada aresta declara, e o caçador busca nele um caminho até uma ferramenta que contorne a verificação (E12):

client = Client()
catalogue = Agent.from_file("agent.yaml").catalogue()
tools = [get_order, charge_card, send_receipt, read_inbox]
model = ScriptedChatModel(script=[])

builder = build_stalwart_graph(model, tools, catalogue=catalogue, agent="desk", client=client)
print("verified, findings:", [f["rule"] for f in builder.compile().stalwart_findings if f["type"] == "vulnerability"])

# Um colega acrescenta um atalho que roda a cobrança com o executor de ferramentas comum.
builder = build_stalwart_graph(model, tools, catalogue=catalogue, agent="desk", client=client)
builder.add_node("quick_charge", ToolNode([charge_card]))
builder.add_conditional_edges("model", lambda state: "quick_charge", {"quick_charge": "quick_charge", END: END})
builder.add_edge("quick_charge", "model")
try:
    builder.compile()
except GraphNotVerified as refused:
    for finding in refused.findings:
        print(f"{finding['rule']} at {finding['node']}: {finding['message']}")

O grafo gerado é verificado sem nenhum achado. O atalho do colega não compila: GraphNotVerified nomeia cada caminho de passagem, aqui QG001, uma escrita num nó de ferramentas simples, e QG011, a mesma escrita alcançável depois de a caixa de entrada ser lida, sem perguntar a ninguém. Cada achado nomeia a sua regra, a sua ferramenta, o nó e o caminho testemunha; replay_witness roda esse caminho no seu próprio ambiente, com cada ferramenta substituída por um stub, para mostrar que ele é real (capítulo 31). Um veredito fica em cache pelo digest da exportação, então um processo que monta o seu grafo a cada requisição o verifica uma vez. service_verifier(client) verifica pelo modo graph hospedado; binary_verifier(path) roda o caçador instalado na sua máquina.

Python simples: `Guard`¶

Para código sem grafo, Guard toma as mesmas decisões como funções simples -- route lê um turno, check verifica uma chamada e run é uma saga como gerenciador de contexto -- e não precisa nem de LangChain nem de LangGraph:

from stalwart_agents.langchain import Guard


def ask_the_person(confirmation):
    """A sua tela de confirmação: aqui, uma pessoa que lê a chamada e a aprova."""
    shown = ", ".join(f"{a['name']}={a['value']} ({a['origin']})" for a in confirmation["arguments"])
    print("asked:", confirmation["tool"], shown, "|", confirmation["reasons"])
    return "approve"


guard = Guard(client, agent="desk", catalogue=catalogue, tools=tools, on_confirm=ask_the_person)
with guard.run(guard.session()) as run:
    order = run.call(get_order, order_id="4411")
    # O $card_on_file do rulebook, como o seu código sabe.
    charged = run.call_with(charge_card, {"order_id": "4411", "amount": float(order["total"])},
                            context={"card_on_file": bool(order["card"])})
print("charged:", charged)

As duas últimas linhas da saída. Todo argumento dado a run.call conta como do modelo, então a cobrança é apresentada a uma pessoa (argument_origin); run.call_with(tool, arguments, origins={"amount": "typed"}) diz de onde veio cada um. Sem on_confirm, a chamada levanta ConfirmationRequired e a execução pausa. Uma falha terminal dentro do with desfaz as escritas da execução, da última para a primeira.

NeMo Guardrails¶

pip install "stalwart-agents[nemo]" acrescenta o NeMo Guardrails, nas versões de Python que ele suporta. Duas ações chamam o mesmo roteador e a mesma verificação que o middleware: stalwart_route lê a mensagem do usuário num plano e define $stalwart_declined onde o plano a recusou, e stalwart_check_tool verifica uma chamada de ferramenta antes de um fluxo rodá-la e só define $stalwart_allowed onde a verificação a permitiu. register(rails, client, agent=...) registra as duas no seu LLMRails, e snippet("1.0") ou snippet("2.x") dá o rail de entrada e a guarda em qualquer um dos dialetos de Colang. Mantenha a forma da guarda: a ação da ferramenta só roda no ramo onde $stalwart_allowed vale, e todo outro caminho termina numa recusa -- a forma que o caçador de bugs confere. O capítulo 29 a roda de ponta a ponta, e o capítulo 13 é a escrita em Colang 2.

As suas ferramentas, o feedback e os testes¶

tool_catalogue(tools, agent=...) faz o catálogo das suas ferramentas LangChain na sua máquina (capítulo 21), o tipo de cada ferramenta pelas marcas dos seus metadados (read_only, compensation, ...); tool_definitions(tools) é o que ele lê: o nome, a descrição, o esquema dos argumentos e os metadados de cada ferramenta. Nada dele é enviado até que uma verificação o leve.
StalwartFeedbackCallback lembra o turno e a ferramenta de cada execução, e thumbs(run_id, up=...) envia o polegar de uma pessoa (capítulo 25).
Sem modelo, sem chave de provedor. ScriptedChatModel(script=[...]) responde a partir de um roteiro -- tool_call(name, arguments, call_id) é um turno roteirizado que chama uma ferramenta -- e conta os seus turns; RespondingChatModel responde com uma função do que lhe foi mostrado. StubTools(catalogue) troca cada ferramenta por um stub que registra as suas chamadas e devolve o fixture_value da saída declarada; arguments_for(tool) preenche os argumentos de uma chamada pelos seus tipos declarados. Uma reprodução ou um teste montado com eles nunca chega a um sistema real.

Os limites¶

Os limites do capítulo 20 valem aqui, e mais estes:

A verificação lê o que a exportação mostra. Um nó seu é lido como código opaco, então o que ele faz por dentro não é provado; marque um nó de ferramentas como protegido só onde ele é.
O middleware e os nós agem sobre as decisões do roteador e da verificação. Uma ferramenta chamada fora deles -- direto do seu código, ou de um nó que não é nosso -- não é verificada, e a exportação a mostra como um caminho que contorna a verificação.