I’m trying to build a guardrail system that runs in parallel with a multi-agent system and immediately stops the agent’s execution in case the guardrail check fails.
The guardrail is a async model call that analyzes the user’s input and try to detect a prompt-injection attempt. Meanwhile, I have a react agent (supervisor) that runs in parallel and outputs the tokens (`message` stream mode) to the client.
What is the langgraph's way of immediatelly interrupting/cancelling/aborting the parallel branch as soon as the guardrail fails?
There is a native mechanism that does cancel a running sibling.
Raise a real exception in the guardrail node. The Pregel runner waits with asyncio.wait(…, return_when=FIRST_COMPLETED) and, the moment one task raises a non‑interrupt exception, it cancels every still‑inflight sibling (_runner.py → _should_stop_others / _panic_or_proceed: inflight.pop().cancel()). That asyncio cancellation propagates into the orchestration subgraph and stops the LLM stream. (interrupt() is excluded because GraphInterrupt is a GraphBubbleUp, which _should_stop_others explicitly skips.)
interrupt() won’t do this, and the gate join can’t either - but raising an exception in the guardrail node will.
A few things about how LangGraph executes your fan-out:
interrupt() is human-in-the-loop pause/resume, not cancellation. Internally GraphInterrupt is a GraphBubbleUp, and the runner explicitly excludes those from its “a task failed → cancel the others” logic. So it will never abort a sibling branch.
gate can’t stop the agent. It’s a join node - it only runs after both branches finish. By then the agent is already done.
Branches can’t communicate mid-step. guardrail and orchestration are in the same superstep; their state writes are merged only after the step completes (graph-api docs), so polling state from inside the agent won’t see the verdict in time.
The native way to abort the running sibling: make guardrail raise. The async Pregel runner waits with asyncio.wait(..., return_when=FIRST_COMPLETED) and, the moment a task raises a non-interrupt exception, it cancels every still-inflight sibling (_runner.py → _should_stop_others / _panic_or_proceed: inflight.pop().cancel()). That asyncio cancellation propagates into the orchestration subgraph and stops the LLM stream.
class GuardrailTripwire(Exception): ...
async def guardrail(state):
verdict = await classifier.ainvoke(state["messages"])
if not verdict.safe:
raise GuardrailTripwire(verdict.reason) # cancels the agent branch
return {"guardrail": "passed"}
# catch at the stream boundary to send a clean "blocked" message:
try:
async for chunk in graph.astream(payload, stream_mode="messages"):
await client.send(chunk)
except GuardrailTripwire as e:
await client.send({"type": "blocked", "reason": str(e)})
One critical caveat for a security guardrail: running the check in parallel with a token-streaming agent is racy - cancellation is only “immediate” once the classifier returns, so unsafe tokens may already have reached the user. If you can’t tolerate that, run the guardrail sequentially before the agent (conditional edge from guardrail), so nothing streams until it clears. The parallel + raise-to-cancel pattern is best when first-token latency matters more than the small leakage window.
If you need a graceful mid-stream stop instead of an exception, pass a shared asyncio.Event via config["configurable"], set it in guardrail, and check it between tokens in the LLM node - that works because both branches share the event loop (state wouldn’t). And if the guardrail lives outside the graph (LangGraph Platform), use client.runs.cancel(thread_id, run_id).
