Tutorials
Build a sensitive-data redaction workflow
Optimize entity identification with Seldon while keeping span validation, pseudonymization, and leakage checks in your application.
A document-preparation application needs to identify sensitive text before replacing it. Here, Seldon optimizes identification; your code owns the replacement policy and checks whether information was missed.
This walkthrough uses three short fictional documents. It reproduces the integration workflow, not the historical demo's benchmark. Paste the Python blocks below, in order, into pii_tutorial.py. They form one script with no private helper imports.
1. Set up the application
Complete the shared setup and dashboard walkthrough, including httpx, jsonschema, and the environment variables. The default script run is an offline wiring check. No Seldon requests happen unless you set RUN_PAID=1.
Hosted identification receives the original, unredacted text. Use fictional inputs while learning; review data handling before sending real documents. This example is not certified anonymization.
import hashlib
import hmac
import json
import os
import secrets
import time
import httpx
from jsonschema import validate
from jsonschema.exceptions import ValidationError
SYSTEM = (
"Identify every PERSON name and EMAIL address in the supplied text. "
"Return JSON with an entities array. Each entity has label, text, start, end. "
"Offsets are zero-based Python Unicode character offsets; end is exclusive. "
"Copy entity text exactly from the input. Return an empty array if none exist."
)
SCHEMA = {
"type": "object",
"properties": {
"entities": {
"type": "array",
"items": {
"type": "object",
"properties": {
"label": {"type": "string", "enum": ["PERSON", "EMAIL"]},
"text": {"type": "string", "minLength": 1},
"start": {"type": "integer", "minimum": 0},
"end": {"type": "integer", "minimum": 1},
},
"required": ["label", "text", "start", "end"],
"additionalProperties": False,
},
}
},
"required": ["entities"],
"additionalProperties": False,
}
def expected_entities(text, mentions):
return {"entities": [
{"label": label, "text": mention, "start": text.index(mention),
"end": text.index(mention) + len(mention)}
for label, mention in mentions
]}
FIXTURES = []
for text, mentions in [
("Ada Chen wrote to [email protected].",
[("PERSON", "Ada Chen"), ("EMAIL", "[email protected]")]),
("Ben Ito copied Ada Chen.", [("PERSON", "Ben Ito"), ("PERSON", "Ada Chen")]),
("The order arrived on Tuesday.", []),
]:
FIXTURES.append({"text": text, "expected": expected_entities(text, mentions)})These expected spans are application-side fixtures. They are not silently sent to the function or used as Seldon's evaluation dataset.
2. Define the existing request
The last user message is the variable document. Keep the system message fixed. This helper creates both the original paid request and the JSON you import into the dashboard.
def make_request(text, model):
return {
"model": model,
"messages": [
{"role": "system", "content": SYSTEM},
{"role": "user", "content": text},
],
"response_format": {
"type": "json_schema",
"json_schema": {"name": "sensitive_entities", "strict": True,
"schema": SCHEMA},
},
}
def invoke(text, model):
started = time.perf_counter()
result = {"output": None, "error": None, "usage": {}, "seldon": {}}
try:
# No automatic retries: a timeout may still have incurred usage.
with httpx.Client(timeout=httpx.Timeout(65.0, connect=10.0)) as client:
response = client.post(
os.environ["SELDON_BASE_URL"].rstrip("/") + "/chat/completions",
headers={"Authorization": "Bearer " + os.environ["SELDON_API_KEY"]},
json=make_request(text, model),
)
if response.status_code in {401, 402, 403, 429}:
raise RuntimeError(f"Batch stopped: authentication, credit, or rate limit error (HTTP {response.status_code})")
response.raise_for_status()
body = response.json()
if not isinstance(body, dict):
raise ValueError("Expected a chat-completion object")
result["usage"] = body.get("usage", {})
result["seldon"] = body.get("seldon", {})
output = json.loads(body["choices"][0]["message"]["content"])
validate(output, SCHEMA)
validated_spans(text, output)
result["output"] = output
except (httpx.HTTPError, ValueError, TypeError, KeyError, IndexError,
ValidationError) as exc:
result["error"] = type(exc).__name__
result["latency_seconds"] = time.perf_counter() - started
return resultErrors stay in the result set. The script deliberately does not print raw provider responses or document content on failure. The timeout is a client bound, not a latency SLO.
3. Optimize identification in the dashboard
Run the completed script once without RUN_PAID; it prints a credential-free request for the Import tab. Set SELDON_REFERENCE_MODEL to a currently routable model first. For an actual original-call baseline, set RUN_PAID=1 and run without SELDON_FUNCTION_MODEL.
# Assemble every Python block on this page before running.
RUN_PAID=0 python pii_tutorial.py
# Three paid original-model calls; explicitly ignore any existing function alias.
RUN_PAID=1 env -u SELDON_FUNCTION_MODEL python pii_tutorial.pyFollow the dashboard walkthrough. In Contract, verify that only the document text varies and the entity schema is unchanged. In Evaluation, check that the criteria cover completeness, allowed labels, and exact source spans—not just valid JSON. Test an omitted email and a fabricated span to see how the evaluator responds. Extra evaluation tests may incur usage within the compile cap.
Choose your priority and compilation cap, review the preview, then authorize optimization. The historical demo selected a GLiNER-based alternative, but your run may select something else or find no useful replacement.
Use Try on a fresh fictional document containing a name and email. Trial calls charge usage. Check the actual offsets and omissions before choosing Activate…. Unreliable or incomplete measurement is a reason to gather evidence, not to assume the candidate passed.
4. Keep replacement and safety checks in your application
Seldon returns entities. The following code validates offsets, deduplicates detections, merges overlapping replacements, and creates keyed, scope-specific pseudonyms. Adjacent entities remain separate. When overlapping entities merge, the pseudonym identifies the entire merged span rather than either entity alone.
def validated_spans(text, output):
validate(output, SCHEMA)
spans = set()
for entity in output["entities"]:
start, end = entity["start"], entity["end"]
if not (0 <= start < end <= len(text)) or text[start:end] != entity["text"]:
raise ValueError("Entity span does not match its source")
spans.add((start, end))
merged = []
for start, end in sorted(spans):
if merged and start < merged[-1][1]:
merged[-1] = (merged[-1][0], max(end, merged[-1][1]))
else:
merged.append((start, end))
return merged
def redact(text, spans, key, scope):
parts, cursor = [], 0
for start, end in spans:
digest = hmac.new(
key, (scope + "\0" + text[start:end]).encode(), hashlib.sha256
).hexdigest()[:16]
parts.extend([text[cursor:start], "[ENTITY_" + digest + "]"])
cursor = end
return "".join(parts) + text[cursor:]
def offline_records():
return [{"output": fixture["expected"], "error": None} for fixture in FIXTURES]
def score(records):
failures = covered = total = leaks = extra_chars = replaced_chars = 0
for index, fixture in enumerate(FIXTURES):
record = records[index] if index < len(records) else {}
try:
if record.get("error") or record.get("output") is None:
raise ValueError("Missing prediction")
predicted = validated_spans(fixture["text"], record["output"])
except (ValueError, TypeError, KeyError, ValidationError):
failures += 1
predicted = []
gold = validated_spans(fixture["text"], fixture["expected"])
predicted_chars = {i for start, end in predicted for i in range(start, end)}
gold_chars = {i for start, end in gold for i in range(start, end)}
covered += sum(set(range(start, end)) <= predicted_chars for start, end in gold)
total += len(gold)
leaks += bool(gold_chars - predicted_chars)
extra_chars += len(predicted_chars - gold_chars)
replaced_chars += len(predicted_chars)
return {
"documents": len(FIXTURES), "execution_failures": failures,
"entity_coverage": covered / total if total else None,
"documents_with_leaks_or_unprocessed_pii": leaks,
"unnecessary_replacement_character_rate": (
extra_chars / replaced_chars if replaced_chars else None
),
}Coverage here means covering the fixture's labeled character spans, not proving that all sensitive information was found. Missing or invalid outputs count as failures and leave labeled PII uncovered. For a failed document, block release; do not treat an unchanged document as successfully anonymized.
5. Call the activated function and check the result
In Use & monitor, copy the model alias into SELDON_FUNCTION_MODEL. The call below preserves the original messages and schema; only the model changes. These are imported single-call functions, not input-object workflows. Confirm the generated Use snippet before running. A pinned revision is useful for a controlled comparison.
def main():
reference = os.getenv("SELDON_REFERENCE_MODEL", "MODEL_FROM_MODELS_PAGE")
print("Import request (no credentials):")
print(json.dumps(make_request(FIXTURES[0]["text"], reference), indent=2))
if os.getenv("RUN_PAID") != "1":
print("Offline fixture wiring check; these are NOT model measurements.")
print(score(offline_records()))
return
if reference == "MODEL_FROM_MODELS_PAGE":
raise ValueError("Set SELDON_REFERENCE_MODEL before paid execution")
models = {"reference": reference}
if os.getenv("SELDON_FUNCTION_MODEL"):
models["candidate"] = os.environ["SELDON_FUNCTION_MODEL"]
# Random per-run key: stable within this run, not across sessions.
# Production applications must manage a secret key and scope lifecycle.
key = secrets.token_bytes(32)
for name, model in models.items():
records = [invoke(fixture["text"], model) for fixture in FIXTURES]
print(name, score(records))
for fixture, record in zip(FIXTURES, records):
if record["output"] is not None:
spans = validated_spans(fixture["text"], record["output"])
print(redact(fixture["text"], spans, key, "fictional-project"))
else:
print("BLOCKED: identification failed")
print({k: record[k] for k in ("error", "latency_seconds", "usage", "seldon")})
if __name__ == "__main__":
main()After activation, replace the placeholder with the model value from your function:
export SELDON_FUNCTION_MODEL="seldon/function/YOUR_FUNCTION_ID"
RUN_PAID=1 python pii_tutorial.pyRunning with both models makes six paid inference requests, in addition to any dashboard compilation or trial charges. Preserve your receipts and inspect fallback_used: a successful response may have used the original workflow after a detected failure. Fallback does not detect every semantic mistake.
Three documents cannot establish production quality or tail latency. Before release, use representative held-out labeled documents, include missed entity types and adversarial formatting, and assess leaks and over-redaction separately. Long documents also require an explicit chunking policy, offset translation, and cross-chunk deduplication; this short-document tutorial intentionally does not implement those steps.
Continue with trace analytics, or learn about managing revisions.