27 · If you already use LangChain, LangGraph or NeMo Guardrails¶

This is the one chapter written in those frameworks' own terms, for a team that already has an agent, a graph or a set of rails. Nothing in Part I needs it: a Part I agent is a LangGraph graph underneath, and this chapter is where you open it, or bring the same router and tool check to code you already have. Everything here runs in your environment, and every decision it acts on is made by the licensed runtime on the same machine.

Language support

The selector reads turns written in English and in Brazilian Portuguese (pt-BR). The language is chosen, never guessed from the turn: the console's Settings page (Language the selector reads), language on a plan request, or the settings.language of your vocabulary file; with none of them, English. A Portuguese turn may be typed with its accents or without them. European Portuguese is not a grammar the selector has, and a turn in any other language is unread.

From Part I's agent to its graph¶

agent.graph() is the door: the compiled graph a Part I agent runs, verified as it compiles.

desk = Agent.from_file("agent.yaml")
graph = desk.graph()                          # the compiled graph Part I's agent runs, proved as it compiles
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'}

The first line of the output is the shop's graph: route reads each turn, desk_model and desk_tools are the desk's model and its guarded tool node, desk_to_billing hands a conversation to the billing desk with its own pair, compute runs a calculation, and rollback and rewind undo a failed run and then forget it. It is a StateGraph like any other: run it with your checkpointer, stream it, add to it (below). agent.export() is the same graph as the verifier reads it, and stalwart_agents.export(path, out) writes it to a file, which is what stalwart verify --export-only does.

`create_agent`, with the router and the check as middleware¶

For an agent made with LangChain's create_agent, two middlewares add the router and the tool check without changing anything else (E13):

chat = ScriptedChatModel(script=[])           # no turn below needs it: each is read or checked first
agent = create_agent(chat, tools, checkpointer=InMemorySaver(), middleware=[
    StalwartRouterMiddleware(client, agent="desk", catalogue=catalogue, on_unread="refuse"),
    # The rulebook's facts, read from the tools' answers as agent.yaml's `facts:` says.
    StalwartToolGuardMiddleware(client, agent="desk", catalogue=catalogue, context=Context(facts=desk.facts)),
])
thread = {"configurable": {"thread_id": "chat-17"}}
state = agent.invoke({"messages": [HumanMessage("show me order 4411")]}, thread)
print(state["messages"][-1].content)
state = agent.invoke({"messages": [HumanMessage("charge order 4411 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 reads the latest user turn into a plan before the model is called, and the plan decides:

The plan	What the middleware does
a tool step that resolves	returns the tool call as the model's response: no model call
an ask	returns the fixed question as the reply: no model call
declined	returns the refusal naming the rule; the model is never shown the turn
a compute step	the runtime builds the prompt, your agent's model writes the function, and the sandbox runs it (chapter 23)
unread	first the bounded request to rephrase, then your `on_unread` setting (chapter 20)

A plan of several steps runs in order, each tool step through LangChain's own tool execution, so every other middleware still sees every call; once the last tool has answered, its result is the reply.

StalwartToolGuardMiddleware wraps every tool call, whoever chose it -- the router, the model or another middleware -- and asks the check: run it, refuse it with the reason in one sentence, or pause for a person through LangGraph's interrupt, the mechanism LangChain's own human-in-the-loop middleware uses. A call runs only after an approval of exactly the arguments shown.

In the output, show me order 4411 is answered with the order and charge order 4411 40.00 waits for a person, each argument shown with where it came from; the model was called no times. Put NeMo Guardrails' middleware outermost, then StalwartRouterMiddleware, then the model, with StalwartToolGuardMiddleware around tool execution. Neither of ours removes another middleware's hook.

A graph of your own¶

For a graph you write yourself, the parts are separate. create_stalwart_agent(model, tools, catalogue=..., agent=...) is the whole graph, generated from your catalogue and compiled; build_stalwart_graph is the same, uncompiled, for you to add to. validate_state(state) names the first thing that broke a state, and validating(graph) runs a graph checking the state after every node.

The state¶

Use StalwartState as your state, or add our channels to yours with with_stalwart_channels(YourState):

Channel	Holds	When two writes meet
`messages`	the conversation	`stalwart_messages` (below)
`stalwart.session`	the SDK `Session`, as its JSON	the newer
`stalwart.history`	the check history and its token	the longer, where it extends the other; an edited, reordered or forked history raises `StateConflict`
`stalwart.labels`	trust labels, by message, call and argument	the union
`stalwart.saga`	the undo stack, calls in flight, a failure	the newer `version`
`stalwart.pending`	open interrupts and their tokens	by interrupt id; `None` closes one
`stalwart.plan`	the plan being run, each call's argument origins, refused routes	merged by key

Everything the router and the check need between turns lives here, in your checkpointer. Rewinding a checkpoint rewinds it too, except stalwart.history: the calls already checked stay in the chain.

The messages¶

stalwart_messages is LangGraph's add_messages -- the same merge by id, RemoveMessage honoured -- with four rules on top: every tool call gets a result (a call left unanswered when the conversation moves on gets an error result in one plain sentence, so the next request to a provider is well formed); the results of one message's calls follow it in the order of its calls; trimming with trim_pairs never separates a call from its result; and nothing of ours is added to a message, so nothing of ours reaches a provider.

The nodes¶

Each node by what it reads, what it writes and where it goes:

Node	Reads	Writes	Goes to
`stalwart_route_node`	the latest user turn, `stalwart.session`	`stalwart.plan`, `stalwart.session`; a reply for an ask, a decline or a refusal	the guarded tool node with the call written; the compute node; your model node for an unread turn your setting hands over; `END`
`StalwartToolNode(tools, client, agent=...)`	the latest message's calls, `stalwart.plan`, `stalwart.history`, `stalwart.labels`	each call's result, `stalwart.history`, `stalwart.saga`, `stalwart.pending`	itself, one step a pass; back to the router or the model; the rollback node after a terminal failure
`stalwart_compute_node`	the plan's compute step and its sources' results	the checked result, with its sources' labels	`END`
`stalwart_rollback_node`	`stalwart.saga`	each undo's result, checked as the agent whose write it undoes	the rewind node, once the stack is empty
`stalwart_rewind_node`	the run's messages	the run's messages replaced by one summary: what failed, what was undone, what could not be	`END`

StalwartToolNode stands where ToolNode would. A refused call is zero invocations and an allowed one exactly one; a call that waits for a person pauses with interrupt() and, on resume, is checked again with the approval's token. Inside a tool it runs, current_idempotency_key() is the key of the write being made, for a tool that passes it on to an API. stalwart_branches fans one message's calls out as Send branches, each checked in its own branch, taking turns on the one history.

The edges¶

stalwart_edge(allowed={...}, choose=fn) is a conditional edge whose function reads model output; a destination outside the declared set is not taken, and the edge goes to a fixed node that refuses and records it. add_stalwart_edge adds that node where your graph has none. The declared set is the edge's path map, so the export knows it: a conditional edge with no path map, or a Command node with no Command[Literal[...]] annotation, can reach any node, and the verifier reports it.

Exporting a graph to verify it¶

A graph's compile() is where it is verified: export_graph(compiled, catalogue) writes it as the verifier reads it (qls-graph/1), each node's kind, the tools a tool node binds and the destinations each edge declares, and the hunter searches it for a path to a tool around the check (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"])

# A teammate adds a quick path that runs the charge with the plain tool runner.
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']}")

The generated graph verifies with no finding. The teammate's quick path does not compile: GraphNotVerified names each way through, here QG001, a write on a plain tool node, and QG011, the same write reachable after the inbox is read, with no person asked. Each finding names its rule, its tool, the node and the witness path; replay_witness runs that path in your own environment, with every tool stubbed, to show it is real (chapter 31). A verdict is cached by the export's digest, so a process that builds its graph on every request verifies it once. service_verifier(client) verifies through the hosted graph mode; binary_verifier(path) runs the hunter installed on your machine.

Plain Python: `Guard`¶

For code with no graph, Guard makes the same decisions as plain functions -- route reads a turn, check checks a call, and run is a saga as a context manager -- and needs neither LangChain nor LangGraph:

from stalwart_agents.langchain import Guard


def ask_the_person(confirmation):
    """Your confirmation screen: here, a person who reads the call and approves it."""
    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")
    # The rulebook's $card_on_file, as your code knows it.
    charged = run.call_with(charge_card, {"order_id": "4411", "amount": float(order["total"])},
                            context={"card_on_file": bool(order["card"])})
print("charged:", charged)

The last two lines of the output. Every argument run.call is given counts as the model's, so the charge is put to a person (argument_origin); run.call_with(tool, arguments, origins={"amount": "typed"}) says where each came from. Without on_confirm, the call raises ConfirmationRequired and the run pauses. A terminal failure inside the with undoes the run's writes, last in first out.

NeMo Guardrails¶

pip install "stalwart-agents[nemo]" adds NeMo Guardrails, on the Pythons it supports. Two actions call the same router and check as the middleware: stalwart_route reads the user's message into a plan and sets $stalwart_declined where the plan declined it, and stalwart_check_tool checks a tool call before a flow runs it and sets $stalwart_allowed only where the check allowed it. register(rails, client, agent=...) registers both on your LLMRails, and snippet("1.0") or snippet("2.x") gives the input rail and the guard in either Colang dialect. Keep the guard's shape: the tool action runs only in the branch where $stalwart_allowed holds, and every other path ends in a refusal -- the shape the bug hunter checks. Chapter 29 runs it end to end, and chapter 13 is Colang 2 authoring.

Your tools, feedback and tests¶

tool_catalogue(tools, agent=...) makes the catalogue of your LangChain tools on your machine (chapter 21), each tool's kind from its metadata's marks (read_only, compensation, ...); tool_definitions(tools) is what it reads: each tool's name, description, argument schema and metadata. Nothing of it is sent until a check carries it.
StalwartFeedbackCallback remembers each run's turn and tool, and thumbs(run_id, up=...) posts a person's thumb (chapter 25).
No model, no provider key. ScriptedChatModel(script=[...]) answers from a script -- tool_call(name, arguments, call_id) is a scripted turn that calls a tool -- and counts its turns; RespondingChatModel answers with a function of what it was shown. StubTools(catalogue) replaces every tool with a stub that records its calls and returns fixture_value of its declared output; arguments_for(tool) fills a call's arguments by their declared kinds. A replay or a test built from these never reaches a real system.

The limits¶

Chapter 20's limits hold here, and these besides:

Verification reads what the export shows. A node of your own is read as opaque code, so what it does inside is not proved; mark a tool node guarded only where it is one.
The middleware and the nodes act on the router's and the check's decisions. A tool called outside them -- directly from your code, or from a node that is not ours -- is not checked, and the export shows it as a way around the check.