Protect Haystack pipelines with OGuardAI so PII is tokenized before the generator and restored after it

The OGuardAI Python SDK ships an optional Haystack integration: two pipeline components that wrap the framework-agnostic transform / rehydrate client. guarded_transform_component tokenizes the input text into safe_text (plus a sealed session_state) so any downstream component only ever sees tokens. guarded_rehydrate_component restores the model output using that same session_state. You wire the transform component before the generator and the rehydrate component after it.

Unlike a single wrapper function, the Haystack adapter is a pair of ordinary Haystack components. That gives you two ways to restore, with different trust properties. Read the trust boundary below before you pick one.

Install

The integration is an optional extra, so Haystack is only pulled in when you use it. The framework is imported lazily, so importing guardai_sdk never requires Haystack.

pip install "oguardai-sdk[haystack]"

Trust boundary

The core guarantee is the same in both usage patterns: the generator, and every component wired between the transform component and the restore point, only ever sees the tokenized safe_text. The model receives Beantworte diese Anfrage: {{person:p_001:ad4f97591c16}} ..., never the raw name, email, or IBAN. Raw PII stays in the trusted zone (the runtime and, when deployed, the Python NER sidecar), except for values an operator explicitly lists in a policy whitelist (the shipped strict policies have an empty whitelist).

Two honest caveats specific to Haystack:

The transform component receives the raw text at its own input socket, because something has to tokenize it. So a pipeline-wide tracer that instruments every component observes the raw input at the transform component's boundary.
The rehydrate component runs inside the pipeline and produces the restored PII at its output socket. A pipeline-wide tracer therefore observes the restored values at the rehydrate component's boundary.

The two patterns below differ only in where restoration happens relative to the pipeline.

Fail closed: guarded_transform_component refuses to emit text if the server returns no session_state. Its run raises ValidationError (GUARDAI_INVALID_INPUT) before any safe_text leaves the component, so a generator can never run on text that cannot be restored. On the restore side, an invalid or tampered session_state makes the underlying client raise rather than return raw text, so a broken session stops the pipeline instead of leaking.

Recommended: rehydrate outside the pipeline

Put guarded_transform_component and the generator in the pipeline, end the pipeline at the generator, and call client.rehydrate outside it. The restored PII is then never a pipeline value, so no Haystack tracer, logger, or intermediate component ever observes a restored name or IBAN. This is the path to prefer when a PII-exporting tracer is attached to the pipeline.

from haystack import Pipeline
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator

from guardai_sdk import OGuardAIClient
from guardai_sdk.integrations.haystack import guarded_transform_component

client = OGuardAIClient(base_url="http://localhost:3000")

pipe = Pipeline()
pipe.add_component("guard_in", guarded_transform_component(client, policy="german-support", language="de"))
pipe.add_component("prompt", PromptBuilder(template="Beantworte diese Mieteranfrage kurz und formal:\n{{ safe_text }}"))
pipe.add_component("llm", OpenAIGenerator(model="gpt-4o-mini"))

# Token-only text flows into the generator; the sealed session bypasses it.
pipe.connect("guard_in.safe_text", "prompt.safe_text")
pipe.connect("prompt.prompt", "llm.prompt")

result = pipe.run({"guard_in": {"text": "Bitte um Kautionsrueckzahlung, Erika Mustermann, erika.mustermann@example.de"}})

# The generator's reply still holds tokens. Restore it OUTSIDE the pipeline.
draft = result["llm"]["replies"][0]
session_state = result["guard_in"]["session_state"]
restored = client.rehydrate(draft, session_state=session_state, output_channel="customer_email", restore_mode="full")
print(restored.restored_text)

guarded_transform_component(client, *, policy=None, language=None) returns a component whose run takes text: str and emits safe_text: str and session_state: str. Because session_state is left unconnected here, Haystack surfaces it as a pipeline output alongside the generator's replies.

In-pipeline rehydration

If no PII-exporting tracer is attached, you can keep the whole flow in one pipeline by adding guarded_rehydrate_component after the generator. This is the most compact wiring, at the cost of the restored PII being a pipeline value (see the tracer caveat above).

from haystack import Pipeline
from haystack.components.builders import PromptBuilder
from haystack.components.converters import OutputAdapter
from haystack.components.generators import OpenAIGenerator

from guardai_sdk import OGuardAIClient
from guardai_sdk.integrations.haystack import (
    guarded_rehydrate_component,
    guarded_transform_component,
)

client = OGuardAIClient(base_url="http://localhost:3000")

pipe = Pipeline()
pipe.add_component("guard_in", guarded_transform_component(client, policy="german-support", language="de"))
pipe.add_component("prompt", PromptBuilder(template="Beantworte diese Mieteranfrage kurz und formal:\n{{ safe_text }}"))
pipe.add_component("llm", OpenAIGenerator(model="gpt-4o-mini"))
# The generator emits replies: list[str]; the rehydrate component takes a single text: str.
pipe.add_component("pick", OutputAdapter(template="{{ replies[0] }}", output_type=str))
pipe.add_component("guard_out", guarded_rehydrate_component(client, output_channel="customer_email", restore_mode="full"))

pipe.connect("guard_in.safe_text", "prompt.safe_text")
pipe.connect("guard_in.session_state", "guard_out.session_state")
pipe.connect("prompt.prompt", "llm.prompt")
pipe.connect("llm.replies", "pick.replies")
pipe.connect("pick.output", "guard_out.text")

result = pipe.run({"guard_in": {"text": "Bitte um Kautionsrueckzahlung, Erika Mustermann, erika.mustermann@example.de"}})
print(result["guard_out"]["restored_text"])

guarded_rehydrate_component(client, *, output_channel="user_output", restore_mode="full") returns a component whose run takes text: str and session_state: str and emits restored_text: str. The output_channel decides what each audience is allowed to see, and restore_mode selects one of full, partial, masked, formatted, abstract, none. Channel rules may only tighten restoration, never widen it.

What the generator and tracer see

The transform component surfaces only safe_text and the sealed session_state. It does not surface entity metadata, so the generator sees the tokens and nothing else:

Beantworte diese Mieteranfrage kurz und formal:
Bitte um Kautionsrueckzahlung, {{person:p_001:ad4f97591c16}}, {{email:e_001:1bcaef1a4aff}}

The real name and email never enter the prompt, the generator, or any component between the two guards. Restoration happens either outside the pipeline (recommended pattern) or in the final component (in-pipeline pattern). The honorific and gender-correct form of a German name are resolved server-side from the sealed session at restore time, not by anything the generator is told.

How it works

guarded_transform_component.run(text) calls /v1/transform, replacing PII with semantic tokens, and emits safe_text plus the sealed session_state. If the server returns no session_state, it raises ValidationError instead of emitting text.
The generator, and any component wired on safe_text, run on tokens only.
Restoration calls /v1/rehydrate with the model output and the sealed session_state, restoring the real values per the output channel and restore mode. In the recommended pattern this is a client.rehydrate call outside the pipeline; in the compact pattern it is guarded_rehydrate_component as the last component.

If the transform fails, no safe_text is emitted and the generator never runs (fail closed).

Runnable example: guarded tenant-support reply copilot

A tenant-support reply copilot for Musterfirma GmbH, a fictional Munich property manager whose support team wants an LLM to draft formal German replies to tenant emails that are dense with regulated PII (name, email, phone, apartment address, tenant number, case reference, deposit IBAN, date of birth, and sometimes a passport number). The example leads with the boundary-safe wiring (transform plus generator in one pipeline, rehydrate outside it), then shows the compact in-pipeline variant, proves the generator only ever sees tokens, and shows a tampered session failing closed. The policy is german-support: German formal restore, IBAN capped at masked, passport, SSN, and health id hard-redacted.

The full source is in the repository at examples/haystack-python/guard.py. It has four sections that run against the server: section 0 prints exactly what the generator receives, section 1 is the boundary-safe pipeline with three restore audiences, section 2 is the in-pipeline pipeline, and section 3 is the fail-closed check.

Prerequisites

Needs-NER for the full scenario. Person detection plus German name restoration, and the Ausweisnummer in this email, need the NER detector, so start the OGuardAI server with the detector sidecar:

# From the repo root. The bundled stack requires these four secrets, and its client key ships
# least-privilege (channel user_output only), so widen it to the channels this example uses.
export GUARDAI_SESSION_SECRET=$(openssl rand -base64 32)
export GUARDAI_API_KEY_1=$(openssl rand -hex 32)
export GUARDAI_DETECTOR_API_KEY=$(openssl rand -hex 32)
export GUARDAI_REDIS_PASSWORD=$(openssl rand -hex 32)
export GUARDAI_API_KEY_1_ALLOWED_OUTPUT_CHANNELS=user_output,customer_email,internal_summary,log_safe
docker compose -f deploy/docker/docker-compose.yml up -d

Builtin regex alone still masks the structured PII (email, phone, IBAN, IDs, address, date of birth), it just cannot detect the name or this Ausweisnummer. The example's german-support policy declares person detection as required, so a runtime without the NER sidecar rejects the transform (fail closed, HTTP 503) rather than let an un-tokenized name reach the model; run the full stack with the detector.

Install the Python dependencies:

pip install -r requirements.txt

OpenAIGenerator is used only when OPENAI_API_KEY is set. Without a key the example runs fully offline with a deterministic drafter, and no provider is ever contacted.

Run

export GUARDAI_BASE_URL="http://localhost:3000"   # or GUARDAI_URL / GUARDAI_BASE
export GUARDAI_API_KEY="$GUARDAI_API_KEY_1"        # the bundled stack runs api_key auth
export OPENAI_API_KEY="sk-..."                     # optional: swaps in OpenAIGenerator

python guard.py

Dependencies (`examples/haystack-python/requirements.txt`)

# The [haystack] extra also pulls in haystack-ai (>=2,<3); it is listed below explicitly
# so the dependency is visible.
oguardai-sdk[haystack]
haystack-ai>=2,<3
# OpenAIGenerator ships inside haystack-ai and is used only when OPENAI_API_KEY is set.
# Without a key the example runs fully offline with a deterministic drafter.

The fixture, policy, and sentinels

The imports, the german-support policy, the fictional tenant email, and the sentinels the boundary checks scan for. The structured secrets (email, phone, IBAN) must be absent from every string that crosses to the generator; the name and the Ausweisnummer need NER, so they are handled separately. From examples/haystack-python/guard.py:

from __future__ import annotations

import os
import re
import sys
from typing import Any

from haystack import Pipeline, component
from haystack.components.builders import PromptBuilder
from haystack.components.converters import OutputAdapter

from guardai_sdk import OGuardAIClient
from guardai_sdk.errors import OGuardAIError
from guardai_sdk.integrations.haystack import (
    guarded_rehydrate_component,
    guarded_transform_component,
)

BASE = (
    os.environ.get("GUARDAI_BASE_URL")
    or os.environ.get("GUARDAI_URL")
    or os.environ.get("GUARDAI_BASE")
    or "http://localhost:3000"
).rstrip("/")
API_KEY = os.environ.get("GUARDAI_API_KEY") or None

# One vertical policy governs the whole run: German formal restore, IBAN capped at
# masked, passport/SSN/health-id hard-redacted. LANG pins detection to German.
POLICY = "german-support"
LANG = "de"

# All fictional. The raw email is what a tenant actually sends; none of it may reach the
# model. Structured secrets must be absent from every crossing string on every lane; a raw
# name is a failure only under NER, else a printed limitation (builtin regex cannot mask names).
RAW_EMAIL = (
    "Betreff: Kautionsrueckzahlung nach Auszug, Vorgang KND-2024-0788\n"
    "Von: Erika Mustermann <erika.mustermann@example.de>\n"
    "Telefon: +49 89 55512340\n\n"
    "Sehr geehrte Damen und Herren,\n\n"
    "ich bin am 31.03.2024 aus der Wohnung in der Lindenstrasse 14, 80331 Muenchen "
    "ausgezogen (Mieternummer 4471039). Bitte ueberweisen Sie meine Kaution auf "
    "IBAN DE97 9900 0000 0123 4567 89. Mein Geburtsdatum zur Verifikation: 12.06.1985. "
    "Zur Identitaet lege ich meine Ausweisnummer L01X00T47 bei.\n\n"
    "Mit freundlichen Gruessen\nErika Mustermann"
)
STRUCTURED_SECRETS = {
    "email": "erika.mustermann@example.de",
    "phone": "+49 89 55512340",
    "iban": "DE97 9900 0000 0123 4567 89",
}
NAME_SECRET = "Erika Mustermann"
# The Ausweisnummer interleaves letters and digits, so the builtin passport regex cannot match it and
# the policy pattern is context-gated on words this email lacks; it is outside the builtin floor and
# needs NER, so it is never asserted as a structured secret and never claimed protected.
AUSWEIS_ID = "L01X00T47"

SYSTEM = (
    "Du bist im Kundenservice der Musterfirma GmbH und beantwortest Mieteranfragen. "
    "Antworte kurz, formal und auf Deutsch. Verwende die semantischen Token EXAKT wie "
    "gegeben und erfinde niemals Namen oder Nummern."
)
TEMPLATE = SYSTEM + "\n\nAnfrage:\n{{ safe_text }}"

TOKEN_RE = re.compile(r"\{\{([^}]+)\}\}")


def token_type(inner: str) -> str:
    """Type key from a token body: 'person:p_001:hash' -> 'person'."""
    return inner.rsplit(":", 2)[0] if inner.count(":") >= 2 else inner.split(":")[0]


def token_map(text: str) -> dict[str, str]:
    """Map each entity type to the EXACT token string (hash included) the model must
    echo. Short forms like {{person:p_001}} do not resolve; the hash is load-bearing."""
    out: dict[str, str] = {}
    for inner in TOKEN_RE.findall(text):
        out.setdefault(token_type(inner), "{{" + inner + "}}")
    return out

The offline generator

The deterministic drafter that lets the example run against just the server. It reads the tokenized prompt and echoes the tokens into a formal German reply, so it only ever sees tokens. Its prompt in, replies out contract matches Haystack's OpenAIGenerator, so the pipeline wiring is identical when OPENAI_API_KEY swaps the real generator in. From examples/haystack-python/guard.py:

@component
class OfflineSupportGenerator:
    """Deterministic stand-in for a chat model so the example runs offline against just
    an OGuardAI server. It reads the tokenized prompt and echoes the tokens into a formal
    German reply. It only ever sees tokens, which is exactly the guarantee this proves.
    Its interface (prompt in, replies out) matches Haystack's OpenAIGenerator, so the
    wiring is identical when OPENAI_API_KEY swaps in the real generator."""

    @component.output_types(replies=list[str])
    def run(self, prompt: str) -> dict[str, list[str]]:
        tok = token_map(prompt)
        person = tok.get("person")
        email = tok.get("email")
        iban = tok.get("iban")
        hello = f"Guten Tag {person}," if person else "Sehr geehrte Damen und Herren,"
        pay = f" Die Rueckzahlung Ihrer Kaution veranlassen wir auf {iban}." if iban else ""
        confirm = f" Eine Eingangsbestaetigung senden wir an {email}." if email else ""
        body = f"vielen Dank fuer Ihre Nachricht. Wir haben Ihr Anliegen aufgenommen.{pay}{confirm}"
        reply = f"{hello}\n\n{body}\n\nMit freundlichen Gruessen\nIhr Serviceteam der Musterfirma GmbH"
        return {"replies": [reply]}


def build_generator() -> Any:
    """OpenAIGenerator when a key is present, else the offline drafter. Both are Haystack
    components with the same prompt/replies contract, so the pipeline is identical."""
    if os.environ.get("OPENAI_API_KEY"):
        from haystack.components.generators import OpenAIGenerator  # lazy: offline runs need no provider

        return OpenAIGenerator(model=os.environ.get("OPENAI_MODEL", "gpt-4o-mini"))
    return OfflineSupportGenerator()


def one_line(text: str) -> str:
    return " ".join(text.split())

The boundary check and the transform probe

fetch_ner_active confirms the NER detector is active, which this example's german-support policy requires: a runtime without it fails the transform closed rather than let an un-tokenized name reach the model. check_boundary guards one crossing string. prove_tokenized (section 0) runs the transform component alone and prints exactly what a downstream generator receives. From examples/haystack-python/guard.py:

def fetch_ner_active(client: OGuardAIClient) -> bool:
    """The one signal that decides whether a raw name crossing is a failure or a noted
    limitation: a NER-configured runtime tokenizes names, so a raw name is a real leak; a
    builtin-only runtime cannot, so a raw name is an expected, printed degradation. If the
    signal cannot be read, treat NER as available so a name leak still fails."""
    try:
        return client.capabilities().ner_active
    except (OGuardAIError, ValueError):
        return True


def check_boundary(where: str, text: str, ner_active: bool) -> list[str]:
    """Guard one string that crosses to the generator. Any structured secret present is a
    leak that fails on every lane. A raw name fails only when the runtime runs NER; when it
    does not it is a printed note, never silent. Returns the name note recorded, if any."""
    leaked = [label for label, raw in STRUCTURED_SECRETS.items() if raw in text]
    if leaked:
        raise SystemExit(f"FAIL: structured PII crossed to the generator in {where}: {', '.join(leaked)}")
    if NAME_SECRET in text:
        if ner_active:
            raise SystemExit(
                f"FAIL: the name {NAME_SECRET} crossed to the generator in {where}; "
                "this runtime runs NER, so names must be tokenized"
            )
        print(f"   NOTE: the name {NAME_SECRET} crossed in {where}; this runtime runs without "
              "NER, so names are not tokenized. Start the full stack for German name masking.")
        return [NAME_SECRET]
    return []


def prove_tokenized(client: OGuardAIClient, ner_active: bool) -> list[str]:
    """Run the transform component alone to show EXACTLY what a downstream generator
    receives. This is the input boundary of every pipeline below."""
    print("=" * 78)
    print("0. WHAT THE GENERATOR RECEIVES (transform component output)")
    print("=" * 78)
    probe = guarded_transform_component(client, policy=POLICY, language=LANG).run(text=RAW_EMAIL)
    safe_text = probe["safe_text"]

    n_tokens = len(TOKEN_RE.findall(safe_text))
    print(f"   tokens in safe_text: {n_tokens}")
    print(f"   safe_text: {one_line(safe_text)[:260]}{'...' if len(one_line(safe_text)) > 260 else ''}\n")

    for label, raw in STRUCTURED_SECRETS.items():
        print(f"   raw {label:8}: {'LEAKED' if raw in safe_text else 'absent'}")
    print()

    if n_tokens == 0:
        raise SystemExit("FAIL: the transform component emitted no tokens")
    notes = check_boundary("the transform output", safe_text, ner_active)
    if AUSWEIS_ID in safe_text:
        print(f"   NOTE: the Ausweisnummer {AUSWEIS_ID} is outside the builtin regex floor; masking it "
              "needs NER, so the structured guarantee here covers email, phone, and IBAN only.")
    return notes

Section 1: boundary-safe wiring

The recommended path. The pipeline is guard_in -> prompt -> llm and ends at the generator; session_state is left unconnected so Haystack surfaces it as a pipeline output. Restoration is three plain client.rehydrate calls outside the pipeline, one per audience, so no restored value is ever a pipeline value. From examples/haystack-python/guard.py:

def section_boundary_safe(client: OGuardAIClient, ner_active: bool) -> list[str]:
    """Recommended path. The pipeline ends at the generator; restoration is a plain
    client call OUTSIDE the pipeline, so no restored PII is ever a pipeline value."""
    print("=" * 78)
    print("1. BOUNDARY-SAFE: pipeline runs on tokens, rehydrate OUTSIDE the pipeline")
    print("=" * 78)

    pipe = Pipeline()
    pipe.add_component("guard_in", guarded_transform_component(client, policy=POLICY, language=LANG))
    pipe.add_component("prompt", PromptBuilder(template=TEMPLATE))
    pipe.add_component("llm", build_generator())
    # safe_text feeds the generator; session_state is left unconnected and surfaced.
    pipe.connect("guard_in.safe_text", "prompt.safe_text")
    pipe.connect("prompt.prompt", "llm.prompt")

    result = pipe.run({"guard_in": {"text": RAW_EMAIL}}, include_outputs_from={"prompt"})
    session_state = result["guard_in"]["session_state"]
    model_input = result["prompt"]["prompt"]
    draft = result["llm"]["replies"][0]

    notes = check_boundary("the model input", model_input, ner_active)

    # The model's own output still holds tokens: proof it worked on tokens, not raw values.
    if "{{" not in draft:
        raise SystemExit("FAIL: the generator output should still be token-only before rehydrate")
    notes += check_boundary("the model draft", draft, ner_active)
    print(f"   token-only draft (model output): {one_line(draft)[:200]}\n")

    # One sealed session, three audiences. The channel decides what each is allowed to see.
    customer = client.rehydrate(draft, session_state=session_state, output_channel="customer_email", restore_mode="full")
    agent = client.rehydrate(draft, session_state=session_state, output_channel="internal_summary")
    audit = client.rehydrate(draft, session_state=session_state, output_channel="log_safe")

    print(f"   -> customer reply : {one_line(customer.restored_text)[:200]}")
    print(f"   -> agent note     : {one_line(agent.restored_text)[:200]}")
    print(f"   -> audit line     : {one_line(audit.restored_text)[:200]}\n")
    return notes

Section 2: in-pipeline wiring

The compact variant that keeps the whole flow in one Pipeline by adding guarded_rehydrate_component as the last component. OutputAdapter bridges the generator's replies: list[str] to the rehydrate component's text: str. The restored text is now a pipeline output (guard_out.restored_text), so keep PII-exporting tracers off this pipeline. From examples/haystack-python/guard.py:

def section_in_pipeline(client: OGuardAIClient, ner_active: bool) -> list[str]:
    """Compact variant using both adapter components. The restored PII is now a pipeline
    output (guard_out.restored_text), so a pipeline-wide tracer would observe it. Use this
    only when no PII-exporting tracer is attached."""
    print("=" * 78)
    print("2. IN-PIPELINE: transform -> generator -> rehydrate, all in one Pipeline")
    print("=" * 78)

    pipe = Pipeline()
    pipe.add_component("guard_in", guarded_transform_component(client, policy=POLICY, language=LANG))
    pipe.add_component("prompt", PromptBuilder(template=TEMPLATE))
    pipe.add_component("llm", build_generator())
    # The generator emits replies: list[str]; the rehydrate component takes text: str.
    pipe.add_component("pick", OutputAdapter(template="{{ replies[0] }}", output_type=str))
    pipe.add_component("guard_out", guarded_rehydrate_component(client, output_channel="customer_email", restore_mode="full"))

    pipe.connect("guard_in.safe_text", "prompt.safe_text")
    pipe.connect("guard_in.session_state", "guard_out.session_state")
    pipe.connect("prompt.prompt", "llm.prompt")
    pipe.connect("llm.replies", "pick.replies")
    pipe.connect("pick.output", "guard_out.text")

    result = pipe.run({"guard_in": {"text": RAW_EMAIL}}, include_outputs_from={"prompt", "llm"})
    model_input = result["prompt"]["prompt"]
    draft = result["llm"]["replies"][0]
    notes = check_boundary("the in-pipeline model input", model_input, ner_active)
    notes += check_boundary("the in-pipeline model draft", draft, ner_active)

    restored = result["guard_out"]["restored_text"]
    print(f"   restored (customer_email, full): {one_line(restored)[:200]}")
    print("   note: restored PII is a pipeline output here; keep PII-exporting tracers off this pipeline.\n")
    return notes

Section 3: fail-closed, and wiring it together

section_fail_closed feeds a tampered session to rehydrate and asserts a typed OGuardAIError is raised instead of any raw text being returned. main checks server health, reads whether NER is active, runs the four sections in order, and prints the closing line. From examples/haystack-python/guard.py:

def section_fail_closed(client: OGuardAIClient) -> None:
    """The guard raises rather than leak. A tampered session cannot be resolved, so
    rehydrate returns a typed error and no raw text is produced."""
    print("=" * 78)
    print("3. FAIL-CLOSED: a tampered session must not leak raw text")
    print("=" * 78)
    try:
        client.rehydrate("Guten Tag {{person:p_001}}", session_state="tampered-session-blob", restore_mode="full")
        raise SystemExit("fail-closed check failed: a tampered session did not fail closed")
    except OGuardAIError as exc:
        print(f"   guard raised as required, no output produced: [{exc.code}] {exc.message}\n")


def main() -> None:
    client = OGuardAIClient(base_url=BASE, api_key=API_KEY)
    try:
        health = client.health()
    except OGuardAIError as exc:
        raise SystemExit(f"Cannot reach OGuardAI at {BASE}: [{exc.code}] {exc.message}")

    ner_active = fetch_ner_active(client)
    if not ner_active:
        print("WARNING: NER detector not active; person tokens and German name restoration "
              "will be missing. Start the full stack via docker compose.\n", file=sys.stderr)

    model = "OpenAIGenerator" if os.environ.get("OPENAI_API_KEY") else "offline drafter (no provider call)"
    print(f"Server: {BASE}  status={health.status}  policy={POLICY}  generator={model}\n")

    name_notes = prove_tokenized(client, ner_active)
    name_notes += section_boundary_safe(client, ner_active)
    name_notes += section_in_pipeline(client, ner_active)
    section_fail_closed(client)

    tracked = ", ".join(STRUCTURED_SECRETS)
    if name_notes:
        print(f"Done. The detectable structured secrets ({tracked}) stayed inside the runtime; the name "
              f"{NAME_SECRET} reached the generator because this runtime runs without NER. Start the full "
              "stack for German name masking.")
    else:
        print(f"Done. The detectable structured secrets ({tracked}) stayed inside the trusted zone and the "
              "generator saw them only as tokens.")


if __name__ == "__main__":
    main()

Expected output

Offline (no OPENAI_API_KEY), section 0 prints the tokenized text the generator receives and confirms each raw secret is absent. The example README documents it as:

0. WHAT THE GENERATOR RECEIVES (transform component output)
   tokens in safe_text: ...
   safe_text: Betreff: ... {{person:p_001:...}} <{{email:e_001:...}}> ... {{iban:ib_001:...}} ...
   raw name    : absent
   raw email   : absent
   raw iban    : absent
   raw passport: absent

Section 1 runs the generator on those tokens and restores the one draft into three audiences from one sealed session. Section 2 runs the full in-pipeline variant and surfaces restored_text as a pipeline output. Section 3 tampers with the session and shows the guard raising GUARDAI_INVALID_INPUT (or GUARDAI_SESSION_EXPIRED) rather than returning raw text. All names, emails, IBANs, phone numbers, and IDs in the example are fictional.

Because this example's german-support policy requires NER for the person name, a runtime without the detector rejects the transform (fail closed, HTTP 503) rather than let an un-tokenized name reach the model; start the full stack for German name masking. A structured secret (email, phone, or IBAN) crossing to the generator is always a hard failure and exits nonzero.

Haystack

On this page