Skip to main content

Guard and AsyncGuard

A Guard is the one object your app talks to. You build it once from your policies and a decision model, then call it at each step of the agent.

Its job​

A guard owns three things:

  • the policies, the rules to apply;
  • the model, the decision model that answers judgments (Jev by default, or any other TypeSafe model);
  • the limits, how much work one check may do.

It runs a check and returns a Result. It does not call your LLM, run your tools, or remember the conversation. You call your model yourself, and a Redactions store carries the conversation. The guard holds no per-conversation state. One guard can serve every user and thread in your app.

Mental model​

Think of a checkpoint with five gates, one for each place text crosses into or out of your agent:

GateWhen to call it
check_inputA user message, before your model sees it.
check_untrustedRetrieved text, such as a web page or a file, before it enters the prompt.
check_tool_callA tool call your model wants to make, before it runs.
check_tool_resultWhat a tool returned, before your model reads it.
check_outputYour model's complete reply, before the user sees it.

Every policy declares which gates it runs at. pii runs at all of them. injection runs on input, untrusted text, and tool calls. You pass the full list once, and each gate picks the policies that apply.

Using it​

from jes import Guard
from jes.policies import injection, pii

guard = Guard(
    [pii(["EMAIL_ADDRESS"]), injection(threshold=0.8)],
    model="jev-latest",
)

def handle(user_text: str) -> str:
    incoming = guard.check_input(user_text)
    if not incoming.ok:
        return incoming.onward          # a short refusal

    reply = call_model(incoming.onward)  # the email is a placeholder here

    outgoing = guard.check_output(reply, prompt=incoming)
    return outgoing.onward              # the reply, email restored

Build the guard once, at startup, and keep it. Pass the input Result as prompt= on the later checks. That way judgments see the user's request, and the reply's placeholders are restored from the same store.

For asyncio apps, use AsyncGuard. It has the same constructor and the same five checks, as coroutines:

from jes import AsyncGuard

async with AsyncGuard(policies, model="jev-latest") as guard:
    incoming = await guard.check_input(user_text)

Good to know​

  • Close it when your app stops, or use with. Guard keeps a thread pool for its model requests. AsyncGuard holds no threads, but closes the same way.
  • A judgment needs a model. A guard with a judgment and no model= raises PolicyError when you build it. A guard of transforms alone runs offline.
  • Mistakes raise at startup, not mid-request. A bad threshold, two policies with the same name, or a judgment with no model raise PolicyError when you build the policies and the guard.
  • A check's model requests run in parallel. Adding judgments doesn't add their latencies up. deadline_s (30 seconds by default) bounds the whole check.
  • Backend failures raise by default. Set on_backend_error="block" to fail closed instead. See Errors.
  • Tool calls block instead of redacting. A tool call that carries an email or a secret is blocked, because its arguments can't be rewritten. The model still never sees the value.

Next​