Output Filtering and Safe Completions

By Michael Samuel Naeem · March 9, 2025 · 5 min read

The coding assistant streamed a complete AWS access key into the chat bubble before the moderation job running on the finished response could flag it. Output filtering that runs only on completion.done is too late for streaming UIs — users screenshot tokens in seconds. Safe completions require treating model output as untrusted data that passes through validation, moderation, and policy gates before display, tool execution, or storage — regardless of how confident the model sounded generating it.

Output filtering pipeline

Model tokens → Chunk buffer → Moderation → Schema validate → Policy engine → User / Tool
                    ↓ fail              ↓ fail           ↓ fail
                 Block/retry        Block/retry      Block/replace

Each stage can block (discard), retry (regenerate with stricter prompt), or replace (safe canned response).

Moderation classifiers

Run provider or self-hosted classifiers on output text:

async def moderate_output(text: str) -> ModerationResult:
    resp = await openai_client.moderations.create(input=text)
    flagged = resp.results[0].flagged
    categories = {
        k: v for k, v in resp.results[0].categories.model_dump().items() if v
    }
    return ModerationResult(flagged=flagged, categories=categories)

async def safe_complete(prompt: str, max_retries: int = 2) -> str:
    for attempt in range(max_retries + 1):
        raw = await llm.generate(prompt)
        mod = await moderate_output(raw)
        if not mod.flagged:
            return raw
        prompt = append_restriction(prompt, mod.categories)
    return FALLBACK_MESSAGE

Category-specific retries work better than generic "try again" — "Remove violent content and answer factually" beats blank regeneration.

Streaming interruption

For SSE streaming, accumulate windows and check incrementally:

BUFFER_WINDOW = 200  # characters
blocked = False

async for chunk in llm.stream(prompt):
    if blocked:
        continue
    buffer += chunk
    if len(buffer) >= BUFFER_WINDOW:
        if await moderate_output(buffer).flagged:
            blocked = True
            yield "\n\n[Response removed: policy violation]"
            audit.log("stream_blocked", buffer_preview=buffer[:100])
            break
        yield buffer
        buffer = ""
if not blocked and buffer:
    if not (await moderate_output(buffer)).flagged:
        yield buffer

Trade latency for safety — larger windows catch more context but delay first token moderation.

PII and secret detection

Regex and NER models catch emails, phone numbers, SSN patterns, credit cards, and API keys:

PATTERNS = [
    (r"sk-[a-zA-Z0-9]{20,}", "openai_key"),
    (r"AKIA[0-9A-Z]{16}", "aws_access_key"),
    (r"\b\d{3}-\d{2}-\d{4}\b", "ssn"),
]

def redact_secrets(text: str) -> tuple[str, list[str]]:
    findings = []
    for pattern, kind in PATTERNS:
        if re.search(pattern, text):
            findings.append(kind)
            text = re.sub(pattern, f"[REDACTED_{kind}]", text)
    return text, findings

Block and alert on secret detection — redaction in user-facing chat may still expose that a secret existed. For internal logs, redact before write.

Schema validation for structured output

from pydantic import BaseModel, Field

class ProductSummary(BaseModel):
    title: str = Field(max_length=100)
    price_usd: float = Field(ge=0, le=1_000_000)
    category: Literal["electronics", "books", "home"]

def parse_and_validate(raw: str) -> ProductSummary:
    data = json.loads(strip_markdown_fence(raw))
    return ProductSummary.model_validate(data)

On validation failure, retry with "Return valid JSON matching schema" — cap retries to avoid cost loops.

Policy engine for business rules

Beyond harm moderation, enforce product policy:

def policy_check(text: str, context: RequestContext) -> PolicyResult:
    if context.brand == "acme" and "competitor_x" in text.lower():
        return PolicyResult(block=True, reason="competitor_mention")
    if context.domain == "medical" and not text.startswith(MEDICAL_DISCLAIMER):
        return PolicyResult(block=True, reason="missing_disclaimer")
    return PolicyResult(block=False)

Centralize rules in config — legal should update disclaimers without redeploying model weights.

Safe completion patterns

Constrained decoding — use grammar or JSON mode so output space is limited.

Refusal templates — consistent, non-jailbreakable messages for blocked categories.

Human escalation — route edge cases to review queue instead of guessing.

Caching safe responses — FAQ answers pre-approved, model only for novel queries.

Tool-call output filtering

When models invoke tools, filter arguments before execution, not just user-visible text:

def validate_tool_call(name: str, args: dict, context: RequestContext) -> None:
    if name == "send_email":
        if args.get("to") not in context.allowed_recipients:
            raise ToolCallBlocked("recipient_not_allowed")
        if contains_html_script(args.get("body", "")):
            raise ToolCallBlocked("unsafe_html")
    if name == "run_sql" and not is_read_only(args.get("query", "")):
        raise ToolCallBlocked("write_sql_forbidden")

A jailbroken model that streams safe-looking text can still emit malicious tool payloads. Treat tool JSON like user input — schema validate, allowlist destinations, and cap string lengths.

For multi-step agent loops, re-run output filters after each tool result before feeding it back into context. Tool returns may contain injected instructions from third-party APIs.

Coordinating filters with human review

Not every block should be a dead end. Route medium-severity flags to a review queue with the redacted draft attached:

if mod.flagged and mod.max_severity == "medium":
    review_id = queue.enqueue(response=redacted, categories=mod.categories)
    return f"Your response is pending review (ref {review_id})."

Humans approve, edit, or reject within SLA. Sampling low-severity passes catches false negatives without reviewing every completion.

Metrics and regression testing

Track:

• output_filter.block_rate by category
• output_filter.retry_count before success
• stream.mid_block_rate
• User override / report rate post-filter

Maintain golden-set evaluations — known harmful prompts must produce blocked or safe outputs after model updates.

Common production mistakes

Teams get safety output filtering wrong in predictable ways:

• Skipping failure-mode rehearsal — run a game day or fault injection exercise before peak traffic, not after the first outage.
• Missing correlation context — every error path should carry request, trace, or tenant identifiers so incidents are debuggable.
• Optimizing for demo, not steady state — load tests, cache warm-up, and cold-start paths matter more than local dev latency.
• Undocumented trade-offs — if you chose speed over strict correctness (or vice versa), write that down for the next engineer.

LLM features around safety output filtering break in production when prompts assume deterministic output, context windows are sized for dev datasets, or token costs are never budgeted per user session. Always log prompt hash, model version, and latency—not raw prompts with PII.

Resources

• OpenAI Moderation API
• NeMo Guardrails
• Microsoft Azure Content Safety
• JSON Schema specification
• OWASP LLM Top 10 — Improper Output Handling