Protect LlamaIndex RAG so retrieved PII is tokenized before the synthesis model and the answer is restored outside the query engine

The OGuardAI Python SDK ships an optional LlamaIndex integration for RAG. It gives you two complementary seams: a node postprocessor that tokenizes retrieved node text so the response-synthesis model only ever sees safe_text, and a query wrapper that transforms the query and restores the answer outside the query engine. Both chain a single sealed session, so a value that appears in both the query and a retrieved chunk gets the same token, and the final rehydrate resolves tokens from either source. Each seam is a thin layer over the SDK transform and rehydrate primitives.

Install

The integration is an optional extra, so LlamaIndex is only pulled in when you use it. The adapter imports llama_index lazily, so importing guardai_sdk never requires LlamaIndex.

pip install "oguardai-sdk[llamaindex]"

The two seams

The exported names are guarded_query, guarded_node_postprocessor, and NodeSessionSink.

guarded_query(query_engine, client, query_str, ...) is the boundary-safe path. It calls /v1/transform on the query and /v1/rehydrate on the answer outside the query engine. No LlamaIndex callback or tracer observes the raw query or the restored answer. Prefer this whenever a PII-exporting tracer or observability hook is attached to your engine.
guarded_node_postprocessor(client, ...) returns a BaseNodePostprocessor that replaces each retrieved node's text with tokenized safe_text before the response-synthesis model runs. It guarantees the synthesis model sees only tokens. It runs inside the query engine, after retrieval, so a tracer attached at or above the retrieval stage can still observe the raw retrieved chunk text before tokenization. To keep raw PII out of the vector store and its retrieval callbacks entirely, tokenize documents at ingestion (transform them before you index) so the store holds only safe_text in the first place.

Pair them through a shared NodeSessionSink so the query session and the node sessions are the same sealed session. That lets the final rehydrate resolve tokens produced by both the query and the retrieved nodes.

Python

from guardai_sdk import OGuardAIClient
from guardai_sdk.integrations.llamaindex import (
    NodeSessionSink,
    guarded_node_postprocessor,
    guarded_query,
)

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

# One sealed session shared by the query and every retrieved node.
sink = NodeSessionSink()

# Install the guarded postprocessor on your existing query engine so retrieved node
# text is tokenized before the synthesis model sees it. `index` is any LlamaIndex index.
query_engine = index.as_query_engine(
    node_postprocessors=[
        guarded_node_postprocessor(
            client,
            session_sink=sink,
            policy="german-support",
            language="de",
        ),
    ],
)

# Boundary-safe: the query is tokenized and the answer is restored OUTSIDE the engine.
answer = guarded_query(
    query_engine,
    client,
    "Was haben wir Erika Mustermann zur Kautionsrueckzahlung zugesagt?",
    session_sink=sink,
    policy="german-support",
    language="de",
    output_channel="user_output",
)
print(answer)

The synthesis model only ever sees text like Vorgang {{order:o_001:57e07d900ca4}} fuer {{person:p_001:ad4f97591c16}}, and the returned answer has the real values restored per the output channel.

Session chaining

NodeSessionSink is a small dataclass with a single field, session_state: str | None = None. When you pass the same sink to guarded_query and to the postprocessor installed on its engine:

guarded_query transforms the query and seeds the query session into the sink.
When the engine runs, the postprocessor reads the sink's session as its starting state, transforms each retrieved node onto it, and writes the final combined session back to the sink.
guarded_query reads the final combined session from the sink and rehydrates the answer with it, so tokens from both the query and the retrieved nodes resolve.

Because the session chains, the same raw value in the query and in a chunk yields the same token id. That token stability is what makes cross-document RAG answers restore correctly.

Standalone postprocessor

You can use guarded_node_postprocessor on its own, without guarded_query, when you drive retrieval and rehydration yourself. Pass a NodeSessionSink so you can read the sealed session after the engine runs, then rehydrate the answer with the SDK client:

from guardai_sdk import OGuardAIClient
from guardai_sdk.integrations.llamaindex import NodeSessionSink, guarded_node_postprocessor

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

query_engine = index.as_query_engine(
    node_postprocessors=[guarded_node_postprocessor(client, session_sink=sink)],
)

response = query_engine.query("...tokenized query...")
restored = client.rehydrate(
    str(response),
    session_state=sink.session_state,
    output_channel="user_output",
)
print(restored.restored_text)

If the engine returns a response object whose text is not a plain string or a .response attribute, pass response_extractor to guarded_query to pull the answer text out:

answer = guarded_query(
    query_engine,
    client,
    query_str,
    response_extractor=lambda r: r.response.message.content,
)

How it works

guarded_query calls /v1/transform on the query, replacing PII with semantic tokens, and seeds the sealed session into the shared sink.
query_engine.query runs on the tokenized query. The guarded postprocessor tokenizes each retrieved node's text, chaining the sealed session across nodes, so the synthesis model only ever sees safe_text.
guarded_query calls /v1/rehydrate on the engine's answer with the final combined session, restoring real values per output_channel and restore_mode.

output_channel defaults to user_output and restore_mode defaults to full. Channel rules may only tighten restoration, never widen it, so a value the policy caps (for example an IBAN masked under the german-support policy) stays capped even under restore_mode="full".

What the model and tracer see

The query engine receives only the tokenized query, never the raw query text.
Each retrieved node's text is replaced with tokenized safe_text before the response-synthesis model runs, so the model sees tokens and safe metadata, never a raw name, email, or IBAN.
The engine's answer still holds tokens. Rehydration happens after query_engine.query returns, outside the engine, so no LlamaIndex callback or tracer attached to the guarded_query boundary observes a raw or restored value.

The one place raw retrieved text still exists is between the retriever and the postprocessor, inside the engine. If a retrieval-stage tracer is a concern, tokenize documents at ingestion so the vector store never holds raw PII. Person and company detection needs the NER detector, and the example's german-support policy declares person detection as required, so a runtime without the NER sidecar rejects the transform (fail closed) rather than expose an un-tokenized name to the synthesis step. Run the full stack with the detector for name-level protection.

Fail closed

Both seams fail closed before any raw text is exposed. If a transform returns no session_state, guarded_query raises ValidationError and never runs the query, and guarded_node_postprocessor raises ValidationError and never exposes the node's text. A corrupted or expired session makes rehydrate raise a typed error (GUARDAI_SESSION_EXPIRED or GUARDAI_INVALID_INPUT) rather than return unresolved or raw text.

Runnable example: guarded support case-note RAG copilot

A support case-note RAG copilot for Musterfirma GmbH, a fictional Munich residential property manager: a guarded LlamaIndex query engine answers a support agent's question grounded in a knowledge base of tenant case notes, each dense with regulated PII (tenant name, email, phone, tenant number, case reference, refund IBAN). Retrieved node text becomes semantic tokens before the synthesis step, and the real values are restored per audience after the query engine returns.

What the example demonstrates:

Both LlamaIndex seams, one sealed session. guarded_node_postprocessor tokenizes each retrieved note before synthesis. guarded_query tokenizes the query and restores the answer outside the query engine. A shared NodeSessionSink chains one sealed session across the query and every note.
A grounded RAG answer the model never saw raw. The synthesis step reads only tokens and safe metadata, then answers on them. The answer is restored after query_engine.query returns.
Value-consistent tokenization. Because the session is shared, the case reference in the tokenized query gets the same token as the one in the tokenized note. The example asserts the two tokens are identical.
Channel-aware restore. The same answer is restored on two channels: the agent view (user_output) shows the case reference in full, the audit line (log_safe) is de-identified.
Fail closed. A corrupted session makes rehydrate raise a typed OGuardAI error instead of leaking raw text.

The policy is german-support: German formal restore, IBAN capped at masked, passport and SSN and health id redacted. The full source is in the repository at examples/llamaindex-python/guard.py.

There is no LLM and no embedding model. The synthesizer is a deterministic stand-in that reads the tokenized context and echoes tokens verbatim, so the example runs offline while the OGuardAI transform and rehydrate calls are real. Swap it for a real LlamaIndex query engine (index.as_query_engine(node_postprocessors=[...])) and the guard wiring is identical.

The flow

From the example README:

KB_NOTES (raw case notes, full of PII)
    |
    v
guarded_query(engine, client, question)     -> POST /v1/transform on the query (OUTSIDE the engine)
    |                                           seeds the shared sealed session into the sink
    v
engine.query(safe_query)                     the engine runs on the tokenized query
    |  guarded_node_postprocessor            -> POST /v1/transform on each retrieved note
    |                                           (INSIDE the engine) chaining the sealed session
    |  synthesize(tokens)                    the model answers on tokens only
    v
guarded_query rehydrate                      -> POST /v1/rehydrate on the answer (OUTSIDE the engine)
                                                user_output | log_safe

Prerequisites

Needs-NER. The tenant name needs 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,log_safe
docker compose -f deploy/docker/docker-compose.yml up -d

Builtin regex alone tokenizes structured PII (email, phone, IBAN, tenant number, case reference), it just cannot detect names. 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 synthesis step; run the full stack with the detector.

Install the Python dependencies (the SDK, editable from this repo, plus the LlamaIndex extra):

pip install -e ../../sdks/python[llamaindex]
# or, once published:
pip install -r requirements.txt

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

python guard.py

No LLM or provider key is needed. PII never reaches a model here, because there is no model call: the synthesizer is offline and only ever handles tokens.

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

# OGuardAI Python SDK with the LlamaIndex extra (pulls in llama-index-core).
oguardai-sdk[llamaindex]>=0.1
# For local development against this repo, install the SDK editable instead:
#   pip install -e ../../sdks/python[llamaindex]
llama-index-core>=0.11,<1

The knowledge base and the sentinels

The imports, config, the fictional case notes, the question, and the sentinel lists the proofs check against. From examples/llamaindex-python/guard.py:

from __future__ import annotations

import os
import re

from guardai_sdk import OGuardAIClient, OGuardAIError
from guardai_sdk.integrations.llamaindex import (
    NodeSessionSink,
    guarded_node_postprocessor,
    guarded_query,
)
from llama_index.core.postprocessor.types import BaseNodePostprocessor
from llama_index.core.query_engine import CustomQueryEngine
from llama_index.core.schema import NodeWithScore, TextNode

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

POLICY = "german-support"
LANG = "de"

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

# Fictional support case notes. All names, emails, IBANs, phones, and IDs are invented.
# Each note carries the structured PII a support answer would need to quote back.
KB_NOTES = [
    {
        "id": "note-4471",
        "text": (
            "Vorgang ORD-2026-4471, Mieterin Erika Mustermann, Kundennummer CUST-4892. "
            "Kaution nach Auszug protokolliert uebergeben. Rueckzahlung in voller Hoehe "
            "auf IBAN DE97 9900 0000 0123 4567 89 zugesagt, Frist 15.07.2026. "
            "Kontakt: erika.mustermann@example.de, Telefon 089 12345678."
        ),
    },
    {
        "id": "note-5522",
        "text": (
            "Vorgang ORD-2026-5522, Mieter Max Mustermann, Kundennummer CUST-5310. "
            "Adressaenderung zum 01.08.2026 hinterlegt. Neue Kontaktdaten bestaetigt an "
            "max.mustermann@example.de, Telefon 089 87654321."
        ),
    },
    {
        "id": "note-6690",
        "text": (
            "Vorgang ORD-2026-6690, Mieterin Erika Mustermann, Kundennummer CUST-6001. "
            "Nebenkostenabrechnung strittig, Pruefung laeuft. Rueckmeldung an "
            "erika.mustermann@example.de zugesagt bis 20.07.2026."
        ),
    },
]

QUESTION = (
    "Was haben wir Erika Mustermann im Vorgang ORD-2026-4471 zur Kautionsrueckzahlung "
    "zugesagt und wie erreichen wir sie?"
)

# Structured PII the builtin regex detectors tokenize, across the query and every note.
# The boundary proof asserts none of these literals survive into what synthesis saw.
STRUCTURED_MARKERS = [
    "ORD-2026-4471",
    "CUST-4892",
    "erika.mustermann@example.de",
    "DE97 9900 0000 0123 4567 89",
    "089 12345678",
    "ORD-2026-5522",
    "CUST-5310",
    "max.mustermann@example.de",
    "089 87654321",
    "ORD-2026-6690",
    "CUST-6001",
    "erika.mustermann@example.de",
]

# Tenant names live in KB_NOTES and the question. Only NER tokenizes them; the builtin
# regex leaves them raw. Whether a name crossing the boundary is a failure or a printed
# note depends on whether this runtime is configured to run NER.
NER_MARKERS = [
    "Erika Mustermann",
    "Max Mustermann",
    "Erika Mustermann",
]

The offline synthesizer and helpers

The token helpers and the deterministic synthesizer that stands in for a chat model. It builds a token-only German answer from the tokenized note, so it never sees a raw value and never calls a provider. From examples/llamaindex-python/guard.py:

def tokens_by_type(text: str) -> dict[str, str]:
    """Map each entity type to the EXACT token string the model must echo verbatim.
    Token body is 'type:id' or 'type:id:hash'; the type is the first segment."""
    out: dict[str, str] = {}
    for inner in TOKEN_RE.findall(text):
        etype = inner.split(":", 1)[0].lower()
        out.setdefault(etype, "{{" + inner + "}}")
    return out


def synthesize(_query_str: str, node: NodeWithScore) -> str:
    """Deterministic stand-in for a chat model. Builds a token-only German answer from
    the tokenized note, echoing tokens verbatim so rehydrate can resolve them. It never
    sees a raw value and never calls a provider."""
    content = _node_text(node)
    tok = tokens_by_type(content)
    order, person = tok.get("order"), tok.get("person")
    iban, email = tok.get("iban"), tok.get("email")

    subject = f"An {person}" if person else "An die Mieterin"
    ref = f" im Vorgang {order}" if order else ""
    pay = f" Die Kaution wird auf {iban} zurueckgezahlt." if iban else ""
    reach = f" Rueckfragen bitte an {email}." if email else ""
    return f"{subject}{ref}: Die Rueckzahlung der Kaution wurde zugesagt.{pay}{reach}"


def _node_text(item: NodeWithScore) -> str:
    node = getattr(item, "node", item)
    return node.get_content() if hasattr(node, "get_content") else node.text


def _most_relevant(query_str: str, nodes: list[NodeWithScore]) -> NodeWithScore:
    """Pick the note sharing the most tokens with the tokenized query. Because the sealed
    session is shared, a value common to the query and a note yields the SAME token, so
    token overlap is a real (if minimal) semantic match on the safe text."""
    q = set(TOKEN_RE.findall(query_str))
    return max(nodes, key=lambda n: len(q & set(TOKEN_RE.findall(_node_text(n)))), default=nodes[0])

The guarded query engine

A minimal CustomQueryEngine that retrieves the notes, runs the guarded node postprocessor so their text is tokenized before synthesis, then answers on tokens. transcript records every string synthesis saw and produced, which the boundary proof scans. guarded_answer wires the shared sink through guarded_query and the postprocessor. From examples/llamaindex-python/guard.py:

class SupportKBQueryEngine(CustomQueryEngine):
    """Minimal LlamaIndex query engine: retrieve the case notes, run the guarded node
    postprocessor so their text is tokenized before synthesis, then answer on tokens.
    `transcript` records every string synthesis saw and produced, for the boundary proof."""

    nodes: list[NodeWithScore]
    postprocessor: BaseNodePostprocessor
    transcript: list[str]

    def custom_query(self, query_str: str) -> str:
        self.transcript.append(query_str)
        processed = self.postprocessor.postprocess_nodes(list(self.nodes), query_str=query_str)
        for node in processed:
            self.transcript.append(_node_text(node))
        draft = synthesize(query_str, _most_relevant(query_str, processed))
        self.transcript.append(draft)
        return draft


def fresh_nodes() -> list[NodeWithScore]:
    # Fresh TextNodes per run: the postprocessor rewrites node content in place, so reused
    # nodes would be double-transformed. Raw source stays in KB_NOTES, untouched.
    return [
        NodeWithScore(node=TextNode(text=n["text"], id_=n["id"]), score=1.0)
        for n in KB_NOTES
    ]


def build_engine(client: OGuardAIClient, sink: NodeSessionSink) -> SupportKBQueryEngine:
    postproc = guarded_node_postprocessor(client, session_sink=sink, policy=POLICY, language=LANG)
    return SupportKBQueryEngine(nodes=fresh_nodes(), postprocessor=postproc, transcript=[])


def guarded_answer(
    client: OGuardAIClient, question: str, *, output_channel: str, restore_mode: str = "full"
) -> tuple[str, SupportKBQueryEngine]:
    sink = NodeSessionSink()
    engine = build_engine(client, sink)
    answer = guarded_query(
        engine,
        client,
        question,
        session_sink=sink,
        policy=POLICY,
        language=LANG,
        output_channel=output_channel,
        restore_mode=restore_mode,
    )
    return answer, engine

The four demo sections

The round trip, the channel contrast, the boundary proof (including the shared-session token equality check), the fail-closed proof, and the NER-active probe. From examples/llamaindex-python/guard.py:

# --- Sections -------------------------------------------------------------------------

def run_round_trip(client: OGuardAIClient) -> SupportKBQueryEngine:
    print("=" * 78)
    print("1. RAG ROUND TRIP -> tokenize query, retrieve + tokenize notes, restore answer")
    print("=" * 78)
    answer, engine = guarded_answer(client, QUESTION, output_channel="user_output")
    safe_query = engine.transcript[0]
    safe_context = " ".join(engine.transcript[1].split())
    print(f"   question (raw)        : {QUESTION}")
    print(f"   query the engine saw  : {' '.join(safe_query.split())}")
    print(f"   a note the engine saw : {safe_context[:200]}{'...' if len(safe_context) > 200 else ''}")
    print(f"   restored answer       : {answer}")
    print()
    # exit 0 is not proof the round trip restored anything. A revoked value comes back as
    # [DELETED] (revocation is value-digest-keyed and crosses sessions) and an unresolvable
    # token survives as a literal {{...}}; neither may appear in an answer shown to an agent.
    if "[DELETED]" in answer:
        raise SystemExit(f"unexpected [DELETED] in the restored answer: {answer}")
    if "{{" in answer:
        raise SystemExit(f"the restored answer still carries an unresolved token: {answer}")
    # When NER is active the note carried a person token, so the tenant name must restore on
    # user_output (a "[PERSON]" prefix a detector may add is tolerated: only the surname is
    # required). Without NER the name is never tokenized, so this tightening is skipped.
    if any("{{person" in seen for seen in engine.transcript) and "Mustermann" not in answer:
        raise SystemExit(f"expected the tenant name restored on user_output, got: {answer}")
    return engine


def run_boundary_proof(engines: list[SupportKBQueryEngine], ner_available: bool) -> list[str]:
    print("=" * 78)
    print("3. BOUNDARY PROOF -> nothing the synthesis step saw contains structured PII")
    print("=" * 78)
    crossings = [seen for engine in engines for seen in engine.transcript]
    leaked = sorted({m for m in STRUCTURED_MARKERS for seen in crossings if m in seen})
    if leaked:
        raise SystemExit(f"   FAIL: structured PII reached the engine: {leaked}")
    tokenized = sum(1 for seen in crossings if "{{" in seen)
    print(f"   engine handled {len(crossings)} strings during synthesis, "
          f"{tokenized} carried tokens, 0 carried structured PII.")

    name_leaks = sorted({m for m in NER_MARKERS for seen in crossings if m in seen})
    if name_leaks:
        if ner_available:
            raise SystemExit(
                "   FAIL: a tenant name reached the synthesis step while NER is active: "
                f"{name_leaks}"
            )
        print(f"   NOTE: {', '.join(name_leaks)} reached the synthesis step unmasked because "
              "this runtime runs without NER; structured PII was still tokenized. Start the "
              "full stack for name-level protection.")

    # Value-consistent tokenization: the case-reference token is identical in the query and
    # the note because the sealed session is shared across both.
    first = engines[0].transcript
    q_order = tokens_by_type(first[0]).get("order")
    note_order = next(
        (tokens_by_type(seen).get("order") for seen in first[1:]
         if tokens_by_type(seen).get("order")),
        None,
    )
    if q_order and q_order == note_order:
        print(f"   shared-session token: query and note both reference {q_order} (same sealed session).")
    else:
        print("   note: case-reference token not shared (check that a note matches the query's ref).")
    print()
    return name_leaks


def run_channel_contrast(client: OGuardAIClient) -> list[SupportKBQueryEngine]:
    print("=" * 78)
    print("2. CHANNEL CONTRAST -> one sealed session, restored for two audiences")
    print("=" * 78)
    agent_view, agent_engine = guarded_answer(client, QUESTION, output_channel="user_output")
    audit_view, audit_engine = guarded_answer(client, QUESTION, output_channel="log_safe")
    print(f"   -> agent view (user_output): {agent_view}")
    print(f"   -> audit line  (log_safe)  : {audit_view}")
    ref = "ORD-2026-4471"
    if ref in agent_view and ref not in audit_view:
        print(f"   channels tighten correctly: {ref} restored for the agent, de-identified in the audit line.")
    else:
        print("   note: check the policy channel rules if the case reference did not tighten as expected.")
    print()
    return [agent_engine, audit_engine]


def run_fail_closed(client: OGuardAIClient) -> None:
    print("=" * 78)
    print("4. FAIL-CLOSED PROOF -> a corrupted session raises, it never leaks")
    print("=" * 78)
    try:
        # Same mechanism guarded_query relies on internally: an unresolvable session must
        # raise a typed error rather than return raw or half-restored text.
        client.rehydrate("Vorgang {{order:o_001}}", session_state="tampered-session-blob")
        raise SystemExit("fail-closed check failed: a tampered session did not fail closed")
    except OGuardAIError as exc:
        print(f"   rehydrate raised as required, no output produced: [{exc.code}] {exc.message}\n")


def fetch_ner_active(client: OGuardAIClient) -> bool:
    """The one leak-exemption signal: whether this runtime is configured to run NER, read
    from /v1/capabilities. If it cannot be determined, treat NER as available so a raw
    tenant name is a hard failure rather than a silently excused leak."""
    try:
        value = client.capabilities().ner_active
    except (OGuardAIError, ValueError):
        return True
    return value is not False

Wiring it together

main checks server health, confirms the NER detector is active (which this example's german-support policy requires, so a runtime without it fails the transform closed), runs the four sections, and prints the closing line. From examples/llamaindex-python/guard.py:

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}")
    if health.status == "unhealthy":
        raise SystemExit(f"OGuardAI at {BASE} reports unhealthy; start the server first.")

    ner_available = fetch_ner_active(client)

    print(f"Server: {BASE}  policy={POLICY}  synthesizer=offline (no provider call)\n")

    engine = run_round_trip(client)
    channel_engines = run_channel_contrast(client)
    name_leaks = run_boundary_proof([engine, *channel_engines], ner_available)
    run_fail_closed(client)
    client.close()

    if name_leaks:
        print("Done. Structured PII stayed inside the runtime; the tenant name(s) "
              f"{', '.join(name_leaks)} reached the synthesis step because this runtime runs "
              "without NER. Start the full stack for name-level protection.")
    else:
        print("Done. Raw PII stayed inside the trusted zone; the query engine and its synthesis "
              "step saw tokens only, and the answer was restored outside the engine.")


if __name__ == "__main__":
    main()

Expected output

The example README documents this run output:

1. RAG ROUND TRIP -> tokenize query, retrieve + tokenize notes, restore answer
   question (raw)        : Was haben wir Erika Mustermann im Vorgang ORD-2026-4471 ...
   query the engine saw  : Was haben wir {{person:p_001}} im Vorgang {{order:o_001}} ...
   a note the engine saw : Vorgang {{order:o_001}} ... auf {{iban:ib_001}} ... an {{email:e_001}} ...
   restored answer       : An Frau Erika Mustermann im Vorgang ORD-2026-4471: ...

2. BOUNDARY PROOF -> nothing the synthesis step saw contains structured PII
   engine handled 4 strings during synthesis, 4 carried tokens, 0 carried structured PII.
   shared-session token: query and note both reference {{order:o_001}} (same sealed session).

3. CHANNEL CONTRAST -> one sealed session, restored for two audiences
   -> agent view (user_output): ... im Vorgang ORD-2026-4471 ...
   -> audit line  (log_safe)  : ... im Vorgang [masked] ...
   channels tighten correctly: ORD-2026-4471 restored for the agent, de-identified in the audit line.

4. FAIL-CLOSED PROOF -> a corrupted session raises, it never leaks
   rehydrate raised as required, no output produced: [GUARDAI_INVALID_INPUT] ...

The exact tokens and masked forms depend on the server and policy version. The three proofs (no structured PII in what the engine saw, shared-session token equality, and the channel tightening) hold regardless.

What the example proves

The query engine only ever sees tokens. The boundary proof asserts that none of the structured PII literals (case reference, tenant number, email, IBAN, phone) appear in any string the synthesis step handled.
The answer is restored outside the engine. Rehydration happens after query_engine.query returns, so the restored answer never re-enters the engine or its callbacks.
Restoration is channel-scoped. The case reference is restored in full on user_output and de-identified on log_safe, from the same sealed session. Channel rules may only tighten restoration, never widen it.

The one place raw retrieved text still exists is between the retriever and the postprocessor, inside the engine. In this example the notes start raw in KB_NOTES and are tokenized by the postprocessor at query time. In production, tokenize documents at ingestion so the vector store never holds raw PII and no retrieval-stage callback can see it.

LlamaIndex

On this page