Build a guardrails proxy with streaming-output tests

Problem

When you need this. Use this when streaming model tokens to a client and you need an output guardrail that holds back unchecked chunks and stop unsafe streams mid-flight.

The support assistant answers from a customer record that holds the customer’s email address and card number. Policy says an answer must not show either in full: the chat transcript is stored, forwarded to other agents and shown on shared screens. The model does not always follow that policy. Asked “What email and card do you have on file for me?”, it streamed this answer:

I have the following information on file for Jane Doe:

* **Email:** jane.doe@example.com
* **Card on file:** 4111 1111 1111 1111

Is there anything else I can help you with regarding your order or account?

The obvious fix is a filter on the model’s output. But the answer arrives as a stream: the model sends small chunks, often one token each, and the client shows them as they come. The email address arrived in 7 chunks, [" jane",".","doe","@","example",".","com"], and the card number in 19. A filter that checks each chunk on its own never sees either value whole, and lets both through:

I have the following information on file for Jane Doe:

* **Email:** jane.doe@example.com
* **Card on file:** 4111 1111 1111 1111

Is there anything else I can help you with regarding your order or account?

We will build a proxy that sits between the model server and the client. Its contract:

• Input: a chat request from the client, and the model’s stream of chunks.
• Output: a stream in the same newline-delimited JSON format in which no complete email address or card number appears, redacted with a marker or, in block mode, cut off before the first one; and a last line that reports what the guard found.
• Success check: both values are removed from the recorded answer, three answers without such values pass through unchanged, the output is the same for every way of splitting the text into chunks, and a block stops reading the model.

Concepts

Stream

An answer sent in pieces as the model generates it. Ollama’s chat API with "stream": true sends one JSON object per line, each with a piece of message.content, and a last object with "done": true and a done_reason.

Guardrail

A check outside the model that stops output the application must not send. This one finds two kinds of value with regular expressions: an email address, and a card number of 13 to 19 digits that may be grouped with spaces or hyphens. The card rule also requires the digits to pass the Luhn checksum, a mod-10 check that luhn_ok() computes. The record’s card number passes it and its order number does not, so the order number is left alone. The checksum separates digit runs; it does not prove that a number is a card.

Held text

The end of the text received so far that a later chunk could still turn into part of a value: the trailing run of characters an email address can contain, and the trailing run of digits, spaces and hyphens a card number can contain. Everything before it is final: no later chunk can change it.

Redact and block

Two actions on a value found in final text. Redact replaces it with a marker such as [email removed] and goes on. Block releases the text before the value, ends the stream with done_reason set to guardrail and stops reading the model.

The key property is that released text cannot be recalled. Once the client has jane.doe@exam, no later redaction removes it. A streaming guard can therefore only release text it has already checked and that no later chunk can extend. Holding back too little leaks; holding back everything until the end gives up streaming. The held text is one trailing run of candidate characters, usually a word or a group of digits. The implementation has no maximum run length: an uninterrupted sequence of candidate characters can retain the whole answer until the stream ends.

This is not a complete answer to data leaks. The guard knows two formats and nothing else; a name, an address or a card number written out in words passes through. The model in the recorded run also masked one card number on its own, as **1111**, which shows why output checks complement, and do not replace, giving the model less data: the record could simply leave out the full card number.

Architecture

Each chunk from the model enters StreamGuard, where pending_start() separates the held text from the final text. Value in final text? looks for complete values in the final part only. When there are none, the proxy can release it to the client, and the cobalt path returns the released text and reads the next chunk. When there is one, Action is block? decides. In redact mode, a marker replaces the value before release. In block mode, the proxy goes to End the stream: it sends the last line with done_reason guardrail and reads no further chunks.

Working code

Runtime: Python 3.11 or later. Dependencies: Python standard library only. Download the files of the example folder into one directory, keeping recorded/.

The guard

guard.py holds the rules and StreamGuard. Read the decisions in order:

• pending_start() finds where the held text begins: the earlier of the two trailing runs. Trailing spaces alone are not held.
• _release() also moves the release point back to the start of any complete value that would be cut by it. Without this, a card number followed by a full stop could be split, because the digits before the stop also look like the start of an email address.
• find() keeps a card match only when its digits pass luhn_ok().
• feed() returns what can be sent now; finish() returns the rest when the stream ends.

"""Output guardrails for a streamed model answer. Python standard library only.

A model streams its answer in small chunks, often one token each. A value such as
an email address or a card number arrives spread over several chunks, so a check
that looks at one chunk at a time never sees it whole.

StreamGuard holds back the end of the text that could still become part of a value:
the trailing run of characters an email address can contain, and the trailing run of
digits, spaces and hyphens that a card number can contain. Everything before that
run is final, because no later chunk can change it. The guard looks for complete
values in the final text and releases it.

Two actions:
  redact  replace each value with a marker, such as [email removed]
  block   stop at the first value: release the text before it, then end the stream

Text already released cannot be recalled; that is why nothing that might still grow
into a value is released early.
"""
import re

RULES = {
    "email": re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9-]+(?:\.[A-Za-z0-9-]+)*\.[A-Za-z]{2,}"),
    "card": re.compile(r"(?<!\d)(?:\d[ -]?){12,18}\d(?!\d)"),
}
MARKERS = {"email": "[email removed]", "card": "[card removed]"}
EMAIL_CHARS = frozenset("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789._%+-@")
CARD_CHARS = frozenset("0123456789 -")


def luhn_ok(digits):
    """The Luhn mod-10 checksum. The card rule requires it, so most other long digit runs are left alone."""
    total, double = 0, False
    for char in reversed(digits):
        value = int(char) * (2 if double else 1)
        total += value - 9 if value > 9 else value
        double = not double
    return total % 10 == 0


def find(text):
    """Every complete value in text, as (start, end, rule), in order."""
    found = []
    for rule, pattern in RULES.items():
        for match in pattern.finditer(text):
            if rule == "card" and not luhn_ok(re.sub(r"\D", "", match.group())):
                continue  # a long order number is not a card number
            found.append((match.start(), match.end(), rule))
    return sorted(found)


def pending_start(text):
    """Where the trailing text that could still grow into a value begins."""
    email = len(text)
    while email and text[email - 1] in EMAIL_CHARS:
        email -= 1
    card = len(text)
    while card and text[card - 1] in CARD_CHARS:
        card -= 1
    digits = [i for i in range(card, len(text)) if text[i].isdigit()]
    card = digits[0] if digits else len(text)  # trailing spaces alone can be released
    return min(email, card)


class StreamGuard:
    def __init__(self, action="redact"):
        if action not in ("redact", "block"):
            raise ValueError("action must be redact or block")
        self.action, self.buffer, self.findings, self.blocked = action, "", [], None
        self.max_held = 0

    def _release(self, final):
        """Return the text that is final now, with values redacted; keep the rest."""
        text = self.buffer
        safe = len(text) if final else pending_start(text)
        values = find(text)
        for start, end, _ in reversed(values):
            if start < safe < end:
                safe = start  # never cut through a value that is already complete
        out, cursor = [], 0
        for start, end, rule in (v for v in values if v[1] <= safe):
            self.findings.append(rule)
            if self.action == "block":
                self.blocked, self.buffer = rule, ""
                return "".join(out) + text[cursor:start]
            out.append(text[cursor:start] + MARKERS[rule])
            cursor = end
        out.append(text[cursor:safe])
        self.buffer = text[safe:]
        return "".join(out)

    def feed(self, chunk):
        if self.blocked:
            return ""
        self.buffer += chunk
        self.max_held = max(self.max_held, len(self.buffer))  # Peak buffer before release, including this chunk.
        return self._release(final=False)

    def finish(self):
        return "" if self.blocked else self._release(final=True)


def per_chunk_filter(chunks):
    """The naive filter: redact values found inside each chunk on its own."""
    out = []
    for chunk in chunks:
        for start, end, rule in reversed(find(chunk)):
            chunk = chunk[:start] + MARKERS[rule] + chunk[end:]
        out.append(chunk)
    return "".join(out)


def guarded(chunks, action="redact"):
    """Run a whole stream through a StreamGuard; return (released pieces, guard)."""
    guard = StreamGuard(action)
    pieces = [guard.feed(chunk) for chunk in chunks]
    pieces.append(guard.finish())
    return pieces, guard

The proxy

proxy.py is a standard-library HTTP server on 127.0.0.1. For each request it reads the upstream stream chunk by chunk, feeds the guard, and writes one line for each piece of released text. It stops reading after a block and always ends with a line that carries the guard’s findings. ReplayUpstream serves streams recorded from gemma4:e2b, keyed by the SHA-256 of the exact request, so the tests need no model.

"""A guardrails proxy that streams a model's answer to the client through StreamGuard.

    python3 proxy.py --upstream replay   # serve recorded streams (default)
    python3 proxy.py --upstream ollama   # forward to a local Ollama server

POST /api/chat with an Ollama chat request and "stream": true. The proxy reads the
upstream stream chunk by chunk, sends the client only what the guard releases, as
newline-delimited JSON like Ollama's, and ends with one line that has "done": true,
the done_reason ("guardrail" when the guard blocked) and the guard's findings. When
the guard blocks, the proxy stops reading the upstream stream. Standard library only.
"""
import argparse
import hashlib
import json
import sys
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib import request as urlrequest

import guard

HERE = Path(__file__).resolve().parent
MODEL = json.loads((HERE / "model.json").read_text(encoding="utf-8"))
RECORDING = HERE / "recorded" / "streams.json"


def canonical(value):
    return json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(",", ":"))


def request_sha256(payload):
    return hashlib.sha256(canonical(payload).encode("utf-8")).hexdigest()


class OllamaUpstream:
    """Yields (content, final) for each line of a live Ollama stream, and records it."""

    def __init__(self, base_url="http://127.0.0.1:11434"):
        self.base_url, self.recorded = base_url, {}
        with urlrequest.urlopen(base_url + "/api/tags", timeout=30) as response:
            installed = {m["name"]: m["digest"] for m in json.load(response)["models"]}
        if installed.get(MODEL["ollama_tag"]) != MODEL["ollama_digest"]:
            raise RuntimeError(f"{MODEL['ollama_tag']} is missing or not the pinned digest")

    def stream(self, payload):
        req = urlrequest.Request(self.base_url + "/api/chat", data=json.dumps(payload).encode(),
                                 headers={"Content-Type": "application/json"}, method="POST")
        chunks = []
        with urlrequest.urlopen(req, timeout=600) as response:
            for line in response:
                event = json.loads(line)
                chunks.append(event["message"]["content"])
                if event.get("done"):
                    final = {k: event.get(k) for k in ("done_reason", "prompt_eval_count", "eval_count")}
                    self.recorded[request_sha256(payload)] = {"chunks": chunks, "final": final}
                    yield chunks[-1], final
                    return
                yield chunks[-1], None


class ReplayUpstream:
    """Yields the recorded chunks of the stream that answered this exact request."""

    def __init__(self, path=RECORDING):
        recording = json.loads(Path(path).read_text(encoding="utf-8"))
        if recording["model"] != MODEL:
            raise RuntimeError("recording was made with a different model pin")
        self.streams, self.read = recording["streams"], []

    def stream(self, payload):
        try:
            recorded = self.streams[request_sha256(payload)]
        except KeyError:
            raise LookupError("no recorded stream for this exact request") from None
        for i, chunk in enumerate(recorded["chunks"]):
            self.read.append(i)  # how far the proxy read: it stops early after a block
            yield chunk, recorded["final"] if i == len(recorded["chunks"]) - 1 else None


def make_handler(upstream, action):
    class Handler(BaseHTTPRequestHandler):
        def do_POST(self):
            payload = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
            self.send_response(200)
            self.send_header("Content-Type", "application/x-ndjson")
            self.end_headers()
            check, final = guard.StreamGuard(action), None
            for content, final in upstream.stream(payload):
                released = check.feed(content)
                if released:
                    self.send_line({"message": {"role": "assistant", "content": released}, "done": False})
                if check.blocked:
                    break  # stop reading the model's stream
            rest = check.finish()
            if rest:
                self.send_line({"message": {"role": "assistant", "content": rest}, "done": False})
            reason = "guardrail" if check.blocked else (final or {}).get("done_reason")
            self.send_line({"done": True, "done_reason": reason,
                            "guardrail": {"action": action, "findings": check.findings, "max_held": check.max_held}})

        def send_line(self, event):
            self.wfile.write((json.dumps(event, ensure_ascii=False) + "\n").encode("utf-8"))
            self.wfile.flush()

        def log_message(self, *args):
            pass

    return Handler


def serve(upstream, action="redact"):
    """Start the proxy on a port the operating system chooses; return the server."""
    server = ThreadingHTTPServer(("127.0.0.1", 0), make_handler(upstream, action))
    return server


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--upstream", choices=["replay", "ollama"], default="replay")
    parser.add_argument("--action", choices=["redact", "block"], default="redact")
    options = parser.parse_args(argv)
    upstream = ReplayUpstream() if options.upstream == "replay" else OllamaUpstream()
    server = serve(upstream, options.action)
    print(f"listening on http://127.0.0.1:{server.server_address[1]}", flush=True)
    server.serve_forever()


if __name__ == "__main__":
    sys.exit(main())

Replay the recorded streams

The 4 streams were recorded from gemma4:e2b, the Gemma 4 E2B model under the Apache-2.0 license, through Ollama 0.32.14, with seed 1. run_guard.py sends each request to the proxy over HTTP, reads the stream back like a client, and does it once in redact mode and once in block mode.

python3 run_guard.py record   # live: Ollama with the pinned model
python3 run_guard.py          # offline replay; rewrites results.json
python3 run_guard.py --check  # offline; compares with results.json

The proxy removed both values from the first answer and changed nothing in the other three. The card-only answer was already masked by the model. The order number has 16 digits in groups of four, like a card number, and passes through because it fails the Luhn check. The client received fewer lines than the model sent chunks, because held text is released together with the chunk that ends it. At most 22 characters were buffered before a check in these answers. The historical max_held field counts the prior held suffix plus the newly received chunk, before safe text is released: the observed delay in these fixtures is about one word or one group of digits. This is not an upper bound for other streams.

In block mode, the first answer ends after "I have the following information on file for Jane Doe:\n\n* **Email:** " with done_reason guardrail. The proxy read 24 of the model’s 67 chunks and stopped.

{
  "model": "gemma4:e2b",
  "conversations": [
    {
      "id": "email-and-card",
      "chunks": 67,
      "answer": "I have the following information on file for Jane Doe:\n\n* **Email:** jane.doe@example.com\n* **Card on file:** 4111 1111 1111 1111\n\nIs there anything else I can help you with regarding your order or account?",
      "per_chunk_filter": "I have the following information on file for Jane Doe:\n\n* **Email:** jane.doe@example.com\n* **Card on file:** 4111 1111 1111 1111\n\nIs there anything else I can help you with regarding your order or account?",
      "redact": {
        "client_text": "I have the following information on file for Jane Doe:\n\n* **Email:** [email removed]\n* **Card on file:** [card removed]\n\nIs there anything else I can help you with regarding your order or account?",
        "client_lines": 38,
        "final": {
          "done": true,
          "done_reason": "stop",
          "guardrail": {
            "action": "redact",
            "findings": [
              "email",
              "card"
            ],
            "max_held": 21
          }
        },
        "upstream_chunks_read": 67
      },
      "block": {
        "client_text": "I have the following information on file for Jane Doe:\n\n* **Email:** ",
        "client_lines": 16,
        "final": {
          "done": true,
          "done_reason": "guardrail",
          "guardrail": {
            "action": "block",
            "findings": [
              "email"
            ],
            "max_held": 21
          }
        },
        "upstream_chunks_read": 24
      },
      "same_output_for_any_chunking": true
    },
    {
      "id": "card-only",
      "chunks": 18,
      "answer": "Your refund will be processed back to the card ending in **1111**.",
      "per_chunk_filter": "Your refund will be processed back to the card ending in **1111**.",
      "redact": {
        "client_text": "Your refund will be processed back to the card ending in **1111**.",
        "client_lines": 14,
        "final": {
          "done": true,
          "done_reason": "stop",
          "guardrail": {
            "action": "redact",
            "findings": [],
            "max_held": 14
          }
        },
        "upstream_chunks_read": 18
      },
      "block": {
        "client_text": "Your refund will be processed back to the card ending in **1111**.",
        "client_lines": 14,
        "final": {
          "done": true,
          "done_reason": "stop",
          "guardrail": {
            "action": "block",
            "findings": [],
            "max_held": 14
          }
        },
        "upstream_chunks_read": 18
      },
      "same_output_for_any_chunking": true
    },
    {
      "id": "order-number",
      "chunks": 26,
      "answer": "Your order number is **1234 5678 9012 3456**.",
      "per_chunk_filter": "Your order number is **1234 5678 9012 3456**.",
      "redact": {
        "client_text": "Your order number is **1234 5678 9012 3456**.",
        "client_lines": 7,
        "final": {
          "done": true,
          "done_reason": "stop",
          "guardrail": {
            "action": "redact",
            "findings": [],
            "max_held": 22
          }
        },
        "upstream_chunks_read": 26
      },
      "block": {
        "client_text": "Your order number is **1234 5678 9012 3456**.",
        "client_lines": 7,
        "final": {
          "done": true,
          "done_reason": "stop",
          "guardrail": {
            "action": "block",
            "findings": [],
            "max_held": 22
          }
        },
        "upstream_chunks_read": 26
      },
      "same_output_for_any_chunking": true
    },
    {
      "id": "no-personal-data",
      "chunks": 12,
      "answer": "Standard delivery takes 3 to 5 working days.",
      "per_chunk_filter": "Standard delivery takes 3 to 5 working days.",
      "redact": {
        "client_text": "Standard delivery takes 3 to 5 working days.",
        "client_lines": 9,
        "final": {
          "done": true,
          "done_reason": "stop",
          "guardrail": {
            "action": "redact",
            "findings": [],
            "max_held": 17
          }
        },
        "upstream_chunks_read": 12
      },
      "block": {
        "client_text": "Standard delivery takes 3 to 5 working days.",
        "client_lines": 9,
        "final": {
          "done": true,
          "done_reason": "stop",
          "guardrail": {
            "action": "block",
            "findings": [],
            "max_held": 17
          }
        },
        "upstream_chunks_read": 12
      },
      "same_output_for_any_chunking": true
    }
  ]
}

Check your understanding

Try it: feed the guard the text Card 4111 1111 1111 1111. as one chunk, then end the stream. Predict what feed() returns and what finish() returns.

Failure cases

Each behavior below has a named test in test_guard.py.

The other tests replay the recorded streams through the proxy over HTTP: the first answer leaks through the per-chunk filter but not the proxy, the three answers without values pass unchanged, every recorded answer gives the same output for any chunking, and results.json reproduces byte for byte. We removed each guard from the source, one at a time; every removal made its named test fail.

python3 -m unittest -v test_guard.py

The run ends with Ran 13 tests and OK on Python 3.11 and 3.15.

Deployment

No service was deployed. To put the proxy in front of a model server, run proxy.py --upstream ollama on the machine that can reach the server, point clients at the proxy instead of the server, and keep the model server unreachable from clients, or the proxy can be bypassed. Log each guardrail line with the request’s trace ID from the tracing build, and count findings per window as the monitoring build counts errors.

Scaling

No throughput was measured. The guard runs both regular expressions over the held and final text on every chunk, so its work per chunk grows with the length of the held text, which here stayed under a few dozen characters. The proxy is a ThreadingHTTPServer with one thread per request; for many concurrent streams, an asynchronous server avoids one thread per open stream. Measure the added time to first byte and the added delay per chunk with the real model before choosing.

Key takeaways

• A streamed value arrives in pieces. A check on each chunk on its own misses it.
• Release only text that no later chunk can change, and hold back the trailing run that could still become a value. In these fixtures that delays about one word or digit group; an arbitrarily long candidate run can delay the whole answer.
• Released text cannot be recalled. Block before the value, and stop reading the model after a block.
• Test a streaming guard with recorded streams and with every chunk size; its output must not depend on where the model happened to split the text.

Best practices

• Keep values out of the model’s context when the answer does not need them; an output guard is the last check, not the first.
• Report the guard’s findings on the stream’s last line, so clients and logs can tell a redacted answer from an unchanged one.
• Pair format rules with checksums or validators where they exist, such as the Luhn check in the card rule.

References

Official documentation checked October 20, 2026:

• Ollama: chat API, streaming responses — newline-delimited JSON objects with message.content, done and done_reason.
• Python: re and http.server
• Gemma 4 E2B model card