Input and Output Guardrail Chain

The distinction that matters: a system prompt asks the model to behave, a guardrail makes misbehaviour not reach anyone. Both belong in a production system, and only one of them is enforcement.

A check is a small, independent unit

Give every check the same interface so the chain is uniform and each check can be tested in isolation. Cheap deterministic checks — regex, denylists, length limits — go first because they are nearly free. Model-based checks go last and only run if the cheap ones passed.

class Action(StrEnum):
    ALLOW = "allow"
    REDACT = "redact"        # continue with modified content
    REGENERATE = "regenerate"  # output only: try again with feedback
    BLOCK = "block"


class CheckResult(BaseModel):
    action: Action
    reason: str
    modified: str | None = None


class Check(Protocol):
    name: str
    cost: Literal["free", "cheap", "model"]

    async def __call__(self, text: str, ctx: Context) -> CheckResult: ...


INPUT_CHECKS = [
    LengthLimit(max_chars=20_000),        # free
    PIIRedactor(),                        # cheap: regex + validators
    KnownJailbreakPatterns(),             # cheap
    PromptInjectionClassifier(),          # model
]

OUTPUT_CHECKS = [
    SecretScanner(),                      # free: never leak a key we generated
    PIIRedactor(),                        # cheap
    CompetitorMentionPolicy(),            # cheap
    GroundednessCheck(),                  # model: is this supported by retrieved context
]

Run the chain, short-circuit on block

Order by cost and run same-cost checks in parallel. Short-circuit as soon as something blocks — there is no value in running a model-based classifier after a regex has already decided to reject. Log every result, not just failures, because the pass rate of each check is what tells you whether it still does anything.

async def run_chain(text: str, checks: list[Check], ctx: Context) -> ChainOutcome:
    current, applied = text, []

    for tier in ("free", "cheap", "model"):
        tier_checks = [c for c in checks if c.cost == tier]
        if not tier_checks:
            continue

        results = await asyncio.gather(*(c(current, ctx) for c in tier_checks))

        for check, r in zip(tier_checks, results):
            audit.record(ctx.request_id, check.name, r.action, r.reason)
            if r.action is Action.BLOCK:
                return ChainOutcome(blocked=True, by=check.name, reason=r.reason)
            if r.action is Action.REGENERATE:
                return ChainOutcome(regenerate=True, by=check.name, feedback=r.reason)
            if r.action is Action.REDACT and r.modified is not None:
                current = r.modified
                applied.append(check.name)

    return ChainOutcome(text=current, redactions=applied)

Output checks and the regenerate path

Output guardrails have an option input guardrails do not: regeneration. When groundedness fails, feeding the failure back and generating once more often produces an acceptable answer — which is a better outcome than a blanket refusal. Bound it at one retry, and refuse cleanly if the second attempt also fails.

async def generate_guarded(prompt: str, ctx: Context) -> Reply:
    inbound = await run_chain(ctx.user_input, INPUT_CHECKS, ctx)
    if inbound.blocked:
        return Reply.refusal(user_message(inbound.by), audit_id=ctx.request_id)

    for attempt in range(2):
        draft = await model(prompt, user_input=inbound.text)
        outbound = await run_chain(draft, OUTPUT_CHECKS, ctx)

        if outbound.blocked:
            return Reply.refusal(user_message(outbound.by), audit_id=ctx.request_id)
        if outbound.regenerate and attempt == 0:
            prompt = prompt + f"\n\nYour previous attempt failed: {outbound.feedback}. Fix it."
            continue
        if outbound.regenerate:
            return Reply.refusal("I couldn't produce a well-grounded answer.", audit_id=ctx.request_id)

        return Reply.ok(outbound.text, redactions=outbound.redactions)

• Checks are small, uniform, independently testable units — never one tangled function
• Order by cost; parallelize within a tier; short-circuit on block
• Decide fail-open versus fail-closed per check, explicitly
• Prefer regenerate over refuse on output failures, bounded at one retry
• Log every decision — pass rates tell you which checks have stopped earning their latency