8. Project: Sanitization Pipeline

Project description

This project closes Module 4 with the implementation of a complete Sanitization Pipeline — a reusable FastAPI middleware that integrates all the sanitization, validation, filtering, and guardrail layers you built in lessons 02-07. It's not a partial exercise — it's the production-ready artifact that demonstrates that your AI system not only withstands attacks (Module 3) but also maintains complete data-flow hygiene.

In the previous lessons you built the individual pieces: InputSanitizer (02), OutputValidator (03), ContentFilter (04), GuardrailChain (05), the integrated pipeline (06), and the production patterns (07). Now everything consolidates into a single system that:

1. Sanitizes inputs: Unicode normalization, removal of dangerous characters, length limits, HTML stripping
2. Validates outputs: Schema enforcement with Pydantic, JSON repair, retry strategies, fallback chains
3. Filters content: Toxicity (local), basic PII detection, off-topic detection, policy enforcement
4. Applies guardrails: Topic boundaries, language consistency, confidence calibration, custom business rules
5. Records auditing: Logging of each stage with timings, flags, and metrics

By the end you'll have a directory with executable Python code, unit tests, and a FastAPI server that demonstrates the pipeline in action. This is the guide's fourth artifact and it integrates with the Injection Defense Pipeline (M3) in Module 8 (Secured AI System).

---

Project goal

Build a FastAPI input/output sanitization middleware that processes each request through 5 layers (input sanitizer → output validator → content filter → guardrails → audit logger), with per-stage error handling, fallback strategies, and performance metrics — all in a reusable package.

---

Technical specifications

Stack

Python >= 3.10
pydantic >= 2.0
fastapi >= 0.100
uvicorn >= 0.20
openai >= 1.0
httpx >= 0.24
bleach >= 6.0

Deliverable structure

sanitization-pipeline/
├── pipeline/
│   ├── __init__.py
│   ├── sanitizer.py          # InputSanitizer
│   ├── validator.py           # OutputValidator
│   ├── content_filter.py      # ContentFilter
│   ├── guardrails.py          # Guardrails + GuardrailChain
│   ├── pipeline.py            # SanitizationPipeline (integration)
│   └── config.py              # Centralized configuration
├── app.py                     # FastAPI application
├── tests/
│   ├── test_sanitizer.py
│   ├── test_validator.py
│   ├── test_content_filter.py
│   ├── test_guardrails.py
│   └── test_pipeline.py
├── requirements.txt
└── README.md

---

Required features

1. InputSanitizer

• ✅ Unicode normalization (NFKC)
• ✅ Removal of zero-width characters
• ✅ Removal of control characters
• ✅ HTML tag stripping
• ✅ Whitespace normalization
• ✅ Length limits configurable per endpoint
• ✅ Result with action (pass/cleaned/truncated/rejected) and issues

2. OutputValidator

• ✅ JSON extraction from free-form text (with/without markdown)
• ✅ Validation against a Pydantic schema
• ✅ Repair of partial JSON (truncated by max_tokens)
• ✅ Fallback response when validation fails
• ✅ Result with action (valid/coerced/repaired/fallback/failed)

3. ContentFilter

• ✅ Local toxicity detection (patterns)
• ✅ Basic PII detection (email, phone, SSN, credit card)
• ✅ Configurable off-topic detection
• ✅ Content policy engine with custom rules
• ✅ Result with verdict (clean/flagged/blocked/redacted)

4. GuardrailChain

• ✅ At least 3 guardrails implemented
• ✅ Chained execution with fail-fast
• ✅ Timing per guardrail
• ✅ Result with passed/blocked_by/warnings

5. SanitizationPipeline

• ✅ Integration of the 4 components in a sequential flow
• ✅ Per-stage error handling (reject vs fallback)
• ✅ Audit logging with request_id, stages, timings, flags
• ✅ Configurable fallback response
• ✅ PipelineContext with complete tracking

---

Implementation code

Step 1: Project setup

mkdir sanitization-pipeline && cd sanitization-pipeline
mkdir pipeline tests

Create requirements.txt:

pydantic>=2.0
fastapi>=0.100
uvicorn>=0.20
openai>=1.0
httpx>=0.24
bleach>=6.0
pytest>=7.0

pip install -r requirements.txt

Step 2: Centralized configuration

Create pipeline/config.py:

from pydantic import BaseModel, Field
from typing import Optional


class SanitizerConfig(BaseModel):
    max_length: int = 4000
    max_lines: int = 50
    normalize_unicode: bool = True
    strip_html: bool = True
    strip_markdown: bool = False
    remove_zero_width: bool = True
    normalize_whitespace: bool = True
    on_overlength: str = "truncate"


class ValidatorConfig(BaseModel):
    allow_repair: bool = True
    allow_coercion: bool = True
    fallback_response: dict = Field(
        default_factory=lambda: {
            "answer": "I couldn't process your request. Try rephrasing your question.",
            "confidence": 0.0,
        }
    )


class ContentFilterConfig(BaseModel):
    check_toxicity: bool = True
    check_pii: bool = True
    check_off_topic: bool = True
    forbidden_topics: list[str] = Field(default_factory=lambda: [
        r"\b(política|elecciones|gobierno)\b",
        r"\b(religión|iglesia)\b",
    ])
    policy_rules: list[dict] = Field(default_factory=list)


class GuardrailConfig(BaseModel):
    max_output_length: int = 2000
    blocked_topics: list[str] = Field(default_factory=list)
    check_language_consistency: bool = True
    low_confidence_threshold: float = 0.5
    disclaimer: str = (
        "\n\n⚠️ This response may not be fully accurate. "
        "Verify with official sources."
    )


class PipelineConfig(BaseModel):
    sanitizer: SanitizerConfig = Field(default_factory=SanitizerConfig)
    validator: ValidatorConfig = Field(default_factory=ValidatorConfig)
    content_filter: ContentFilterConfig = Field(default_factory=ContentFilterConfig)
    guardrails: GuardrailConfig = Field(default_factory=GuardrailConfig)
    max_output_retries: int = 2
    enable_audit_log: bool = True
    system_prompt: str = (
        "You are a customer service assistant for TechStore. "
        "Respond in JSON with fields: answer (string), confidence (float 0-1). "
        "You only answer questions about electronic products."
    )

Step 3: InputSanitizer

Create pipeline/sanitizer.py:

import re
import unicodedata
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional

try:
    import bleach
    HAS_BLEACH = True
except ImportError:
    HAS_BLEACH = False

from .config import SanitizerConfig


class SanitizationAction(Enum):
    PASS = "pass"
    CLEANED = "cleaned"
    TRUNCATED = "truncated"
    REJECTED = "rejected"


@dataclass
class SanitizationResult:
    original: str
    sanitized: Optional[str]
    action: SanitizationAction
    issues: list[str] = field(default_factory=list)
    metrics: dict = field(default_factory=dict)

    @property
    def passed(self) -> bool:
        return self.action != SanitizationAction.REJECTED


class InputSanitizer:
    ZERO_WIDTH_CHARS = set(
        "\u200b\u200c\u200d\u200e\u200f"
        "\u202a\u202b\u202c\u202d\u202e"
        "\u2060\u2061\u2062\u2063\u2064"
        "\ufeff"
    )
    CONTROL_CHAR_RE = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]")
    HTML_TAG_RE = re.compile(r"<[^>]+>")

    def __init__(self, config: Optional[SanitizerConfig] = None):
        self.config = config or SanitizerConfig()

    def sanitize(self, text: str) -> SanitizationResult:
        if not text or not text.strip():
            return SanitizationResult(
                original=text, sanitized=None,
                action=SanitizationAction.REJECTED,
                issues=["Empty or whitespace-only input"],
            )

        issues: list[str] = []
        cleaned = text
        metrics = {"original_length": len(text)}

        if self.config.normalize_unicode:
            normalized = unicodedata.normalize("NFKC", cleaned)
            if normalized != cleaned:
                issues.append("Unicode normalized (NFKC)")
                cleaned = normalized

        if self.config.remove_zero_width:
            zw_count = sum(1 for c in cleaned if c in self.ZERO_WIDTH_CHARS)
            if zw_count > 0:
                cleaned = "".join(c for c in cleaned if c not in self.ZERO_WIDTH_CHARS)
                issues.append(f"Removed {zw_count} zero-width characters")

        control_matches = self.CONTROL_CHAR_RE.findall(cleaned)
        if control_matches:
            cleaned = self.CONTROL_CHAR_RE.sub("", cleaned)
            issues.append(f"Removed {len(control_matches)} control characters")

        if self.config.strip_html:
            html_tags = self.HTML_TAG_RE.findall(cleaned)
            if html_tags:
                if HAS_BLEACH:
                    cleaned = bleach.clean(cleaned, tags=[], strip=True)
                else:
                    cleaned = self.HTML_TAG_RE.sub("", cleaned)
                issues.append(f"Stripped {len(html_tags)} HTML tags")

        if self.config.normalize_whitespace:
            before_len = len(cleaned)
            cleaned = re.sub(r" {2,}", " ", cleaned)
            cleaned = re.sub(r"\n{3,}", "\n\n", cleaned)
            cleaned = cleaned.strip()
            diff = before_len - len(cleaned)
            if diff > 0:
                issues.append(f"Normalized whitespace (saved {diff} chars)")

        metrics["cleaned_length"] = len(cleaned)
        metrics["estimated_tokens"] = len(cleaned) // 4

        if len(cleaned) > self.config.max_length:
            if self.config.on_overlength == "reject":
                return SanitizationResult(
                    original=text, sanitized=None,
                    action=SanitizationAction.REJECTED,
                    issues=[f"Exceeds max length: {len(cleaned)} > {self.config.max_length}"],
                    metrics=metrics,
                )
            cleaned = cleaned[:self.config.max_length]
            issues.append(f"Truncated to {self.config.max_length} chars")
            metrics["truncated"] = True

        action = SanitizationAction.PASS
        if metrics.get("truncated"):
            action = SanitizationAction.TRUNCATED
        elif issues:
            action = SanitizationAction.CLEANED

        return SanitizationResult(
            original=text, sanitized=cleaned,
            action=action, issues=issues, metrics=metrics,
        )

Step 4: OutputValidator

Create pipeline/validator.py:

import json
import re
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional, Any

from pydantic import BaseModel, ValidationError

from .config import ValidatorConfig


class ValidationAction(Enum):
    VALID = "valid"
    REPAIRED = "repaired"
    FALLBACK = "fallback"
    FAILED = "failed"


@dataclass
class ValidationResult:
    action: ValidationAction
    data: Optional[dict]
    raw_output: str
    issues: list[str] = field(default_factory=list)

    @property
    def success(self) -> bool:
        return self.action in (ValidationAction.VALID, ValidationAction.REPAIRED)


class OutputValidator:
    JSON_BLOCK_RE = re.compile(r"```(?:json)?\s*([\s\S]*?)```")
    JSON_OBJECT_RE = re.compile(r"\{[\s\S]*\}")

    def __init__(
        self,
        schema: type[BaseModel],
        config: Optional[ValidatorConfig] = None,
    ):
        self.schema = schema
        self.config = config or ValidatorConfig()

    def validate(self, raw_output: str) -> ValidationResult:
        issues: list[str] = []

        extracted = self._extract_json(raw_output)

        if extracted is None:
            repaired = self._repair_json(raw_output)
            if repaired is not None:
                extracted = repaired
                issues.append("JSON repaired from partial output")

        if extracted is None:
            return ValidationResult(
                action=ValidationAction.FALLBACK,
                data=self.config.fallback_response,
                raw_output=raw_output,
                issues=["No JSON found in output, using fallback"],
            )

        try:
            validated = self.schema(**extracted)
            action = ValidationAction.REPAIRED if issues else ValidationAction.VALID
            return ValidationResult(
                action=action,
                data=validated.model_dump(),
                raw_output=raw_output,
                issues=issues,
            )
        except ValidationError as e:
            return ValidationResult(
                action=ValidationAction.FALLBACK,
                data=self.config.fallback_response,
                raw_output=raw_output,
                issues=issues + [f"Validation error: {e.error_count()} errors"],
            )

    def _extract_json(self, text: str) -> Optional[dict]:
        match = self.JSON_BLOCK_RE.search(text)
        if match:
            try:
                return json.loads(match.group(1).strip())
            except json.JSONDecodeError:
                pass

        try:
            return json.loads(text)
        except json.JSONDecodeError:
            pass

        match = self.JSON_OBJECT_RE.search(text)
        if match:
            try:
                return json.loads(match.group())
            except json.JSONDecodeError:
                pass

        return None

    def _repair_json(self, text: str) -> Optional[dict]:
        cleaned = text.strip()
        open_braces = cleaned.count("{")
        close_braces = cleaned.count("}")

        if open_braces > close_braces:
            last_comma = cleaned.rfind(",")
            last_colon = cleaned.rfind(":")
            if last_comma > last_colon:
                cleaned = cleaned[:last_comma]
            cleaned += "}" * (open_braces - close_braces)
            try:
                return json.loads(cleaned)
            except json.JSONDecodeError:
                pass

        return None

Step 5: ContentFilter

Create pipeline/content_filter.py:

import re
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional

from .config import ContentFilterConfig


class FilterVerdict(Enum):
    CLEAN = "clean"
    FLAGGED = "flagged"
    BLOCKED = "blocked"


@dataclass
class FilterResult:
    verdict: FilterVerdict
    text: Optional[str]
    flags: list[dict] = field(default_factory=list)


class ContentFilter:
    TOXICITY_PATTERNS = [
        (re.compile(r"\b(idiota|estúpido|imbécil|inútil)\b", re.I), "insult", 0.7),
        (re.compile(r"\b(matar|destruir|explotar|asesinar)\b", re.I), "violence", 0.9),
        (re.compile(r"\b(odio|repugnante|asco)\b", re.I), "hate", 0.6),
    ]

    PII_PATTERNS = [
        (re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b"), "email"),
        (re.compile(r"\b(?:\+\d{1,3}\s?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b"), "phone"),
        (re.compile(r"\b\d{3}-\d{2}-\d{4}\b"), "ssn"),
        (re.compile(r"\b(?:\d{4}[-\s]?){3}\d{4}\b"), "credit_card"),
    ]

    def __init__(self, config: Optional[ContentFilterConfig] = None):
        self.config = config or ContentFilterConfig()
        self.forbidden_topic_patterns = [
            re.compile(p, re.IGNORECASE)
            for p in self.config.forbidden_topics
        ]

    def filter(self, text: str) -> FilterResult:
        flags = []
        verdict = FilterVerdict.CLEAN

        if self.config.check_toxicity:
            for pattern, category, severity in self.TOXICITY_PATTERNS:
                matches = pattern.findall(text)
                if matches:
                    flags.append({
                        "layer": "toxicity",
                        "category": category,
                        "severity": severity,
                        "matches": matches,
                    })
                    if severity >= 0.8:
                        verdict = FilterVerdict.BLOCKED

        if self.config.check_pii:
            for pattern, pii_type in self.PII_PATTERNS:
                matches = pattern.findall(text)
                if matches:
                    flags.append({
                        "layer": "pii",
                        "type": pii_type,
                        "count": len(matches),
                    })
                    if verdict != FilterVerdict.BLOCKED:
                        verdict = FilterVerdict.FLAGGED

        if self.config.check_off_topic:
            for pattern in self.forbidden_topic_patterns:
                matches = pattern.findall(text)
                if matches:
                    flags.append({
                        "layer": "off_topic",
                        "matches": matches,
                    })
                    if verdict != FilterVerdict.BLOCKED:
                        verdict = FilterVerdict.FLAGGED

        return FilterResult(
            verdict=verdict,
            text=text if verdict != FilterVerdict.BLOCKED else None,
            flags=flags,
        )

Step 6: Guardrails

Create pipeline/guardrails.py:

import re
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional

from .config import GuardrailConfig


class GuardrailAction(Enum):
    PASS = "pass"
    WARN = "warn"
    BLOCK = "block"
    MODIFY = "modify"


@dataclass
class GuardrailResult:
    name: str
    action: GuardrailAction
    message: str = ""
    modified_text: Optional[str] = None
    execution_time_ms: float = 0.0


class Guardrail(ABC):
    def __init__(self, name: str, enabled: bool = True):
        self.name = name
        self.enabled = enabled

    @abstractmethod
    def check(self, text: str, context: Optional[dict] = None) -> GuardrailResult:
        pass

    def execute(self, text: str, context: Optional[dict] = None) -> GuardrailResult:
        if not self.enabled:
            return GuardrailResult(name=self.name, action=GuardrailAction.PASS)
        start = time.perf_counter()
        result = self.check(text, context)
        result.execution_time_ms = (time.perf_counter() - start) * 1000
        return result


class LengthGuardrail(Guardrail):
    def __init__(self, max_length: int = 2000):
        super().__init__("length_check")
        self.max_length = max_length

    def check(self, text, context=None):
        if len(text) > self.max_length:
            return GuardrailResult(
                name=self.name, action=GuardrailAction.MODIFY,
                message=f"Truncated from {len(text)} to {self.max_length}",
                modified_text=text[:self.max_length] + "...",
            )
        return GuardrailResult(name=self.name, action=GuardrailAction.PASS)


class TopicGuardrail(Guardrail):
    def __init__(self, blocked_topics: list[str]):
        super().__init__("topic_boundary")
        self.patterns = [re.compile(t, re.IGNORECASE) for t in blocked_topics]

    def check(self, text, context=None):
        for pattern in self.patterns:
            if pattern.search(text):
                return GuardrailResult(
                    name=self.name, action=GuardrailAction.BLOCK,
                    message=f"Blocked topic: {pattern.pattern}",
                )
        return GuardrailResult(name=self.name, action=GuardrailAction.PASS)


class ConfidenceGuardrail(Guardrail):
    def __init__(self, threshold: float = 0.5, disclaimer: str = ""):
        super().__init__("confidence_calibration")
        self.threshold = threshold
        self.disclaimer = disclaimer or (
            "\n\n⚠️ This response may not be fully accurate."
        )

    def check(self, text, context=None):
        confidence = (context or {}).get("confidence", 1.0)
        if confidence < self.threshold:
            return GuardrailResult(
                name=self.name, action=GuardrailAction.MODIFY,
                message=f"Low confidence ({confidence}), adding disclaimer",
                modified_text=text + self.disclaimer,
            )
        return GuardrailResult(name=self.name, action=GuardrailAction.PASS)


@dataclass
class ChainResult:
    passed: bool
    final_text: Optional[str]
    results: list[GuardrailResult] = field(default_factory=list)
    blocked_by: Optional[str] = None
    total_time_ms: float = 0.0


class GuardrailChain:
    def __init__(self, guardrails: list[Guardrail]):
        self.guardrails = guardrails

    def run(self, text: str, context: Optional[dict] = None) -> ChainResult:
        results = []
        current = text
        start = time.perf_counter()

        for guardrail in self.guardrails:
            result = guardrail.execute(current, context)
            results.append(result)

            if result.action == GuardrailAction.BLOCK:
                return ChainResult(
                    passed=False, final_text=None, results=results,
                    blocked_by=guardrail.name,
                    total_time_ms=(time.perf_counter() - start) * 1000,
                )
            if result.action == GuardrailAction.MODIFY and result.modified_text:
                current = result.modified_text

        return ChainResult(
            passed=True, final_text=current, results=results,
            total_time_ms=(time.perf_counter() - start) * 1000,
        )


def create_guardrail_chain(config: GuardrailConfig) -> GuardrailChain:
    guardrails = [
        LengthGuardrail(max_length=config.max_output_length),
        ConfidenceGuardrail(
            threshold=config.low_confidence_threshold,
            disclaimer=config.disclaimer,
        ),
    ]
    if config.blocked_topics:
        guardrails.insert(1, TopicGuardrail(config.blocked_topics))
    return GuardrailChain(guardrails)

Step 7: Integrated pipeline

Create pipeline/pipeline.py:

import json
import time
import uuid
import logging
from dataclasses import dataclass, field
from typing import Optional, Callable

from pydantic import BaseModel

from .sanitizer import InputSanitizer, SanitizationAction
from .validator import OutputValidator, ValidationAction
from .content_filter import ContentFilter, FilterVerdict
from .guardrails import GuardrailChain, GuardrailAction
from .config import PipelineConfig

logger = logging.getLogger("sanitization_pipeline")


@dataclass
class PipelineContext:
    request_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
    user_id: str = "anonymous"
    endpoint: str = ""
    stages_passed: list[str] = field(default_factory=list)
    stages_failed: list[str] = field(default_factory=list)
    flags: list[dict] = field(default_factory=list)
    timings: dict[str, float] = field(default_factory=dict)
    total_time_ms: float = 0.0


class PipelineError(Exception):
    def __init__(self, stage: str, status_code: int, user_message: str):
        self.stage = stage
        self.status_code = status_code
        self.user_message = user_message


class SanitizationPipeline:
    def __init__(
        self,
        input_sanitizer: InputSanitizer,
        output_validator: OutputValidator,
        content_filter: ContentFilter,
        guardrail_chain: GuardrailChain,
        config: PipelineConfig,
        llm_caller: Optional[Callable] = None,
    ):
        self.input_sanitizer = input_sanitizer
        self.output_validator = output_validator
        self.content_filter = content_filter
        self.guardrail_chain = guardrail_chain
        self.config = config
        self.llm_caller = llm_caller

    async def process(
        self,
        user_input: str,
        context: PipelineContext,
    ) -> dict:
        start = time.perf_counter()

        # Stage 1: Input Sanitization
        t = time.perf_counter()
        san_result = self.input_sanitizer.sanitize(user_input)
        context.timings["input_sanitize"] = (time.perf_counter() - t) * 1000

        if not san_result.passed:
            context.stages_failed.append("input_sanitize")
            raise PipelineError(
                "input_sanitize", 400,
                "Your message could not be processed. Try a shorter text.",
            )
        context.stages_passed.append("input_sanitize")
        if san_result.issues:
            context.flags.append({"stage": "input_sanitize", "issues": san_result.issues})

        # Stage 2: LLM Call
        t = time.perf_counter()
        try:
            raw_output = await self._call_llm(san_result.sanitized)
            context.stages_passed.append("llm_call")
        except Exception as e:
            context.stages_failed.append("llm_call")
            logger.error(f"[{context.request_id}] LLM error: {e}")
            context.total_time_ms = (time.perf_counter() - start) * 1000
            return self.config.validator.fallback_response
        finally:
            context.timings["llm_call"] = (time.perf_counter() - t) * 1000

        # Stage 3: Output Validation (with retry)
        t = time.perf_counter()
        val_result = self.output_validator.validate(raw_output)

        if not val_result.success:
            for retry in range(self.config.max_output_retries):
                raw_output = await self._call_llm(
                    san_result.sanitized + "\n\nRespond ONLY with valid JSON."
                )
                val_result = self.output_validator.validate(raw_output)
                if val_result.success:
                    context.flags.append({
                        "stage": "output_validate",
                        "retry": retry + 1,
                    })
                    break

        context.timings["output_validate"] = (time.perf_counter() - t) * 1000

        if val_result.success:
            context.stages_passed.append("output_validate")
        else:
            context.stages_failed.append("output_validate")
            context.total_time_ms = (time.perf_counter() - start) * 1000
            return self.config.validator.fallback_response

        # Stage 4: Content Filter
        t = time.perf_counter()
        output_text = json.dumps(val_result.data, ensure_ascii=False)
        filter_result = self.content_filter.filter(output_text)
        context.timings["content_filter"] = (time.perf_counter() - t) * 1000

        if filter_result.verdict == FilterVerdict.BLOCKED:
            context.stages_failed.append("content_filter")
            context.flags.append({"stage": "content_filter", "flags": filter_result.flags})
            context.total_time_ms = (time.perf_counter() - start) * 1000
            return self.config.validator.fallback_response
        context.stages_passed.append("content_filter")

        # Stage 5: Guardrails
        t = time.perf_counter()
        gr_result = self.guardrail_chain.run(
            output_text,
            context={"confidence": val_result.data.get("confidence", 0.5)},
        )
        context.timings["guardrails"] = (time.perf_counter() - t) * 1000

        if not gr_result.passed:
            context.stages_failed.append("guardrails")
            context.total_time_ms = (time.perf_counter() - start) * 1000
            return self.config.validator.fallback_response
        context.stages_passed.append("guardrails")

        # Audit log
        context.total_time_ms = (time.perf_counter() - start) * 1000
        if self.config.enable_audit_log:
            self._audit_log(context)

        return val_result.data

    async def _call_llm(self, user_input: str) -> str:
        if self.llm_caller:
            return await self.llm_caller(user_input)

        from openai import AsyncOpenAI
        client = AsyncOpenAI()
        response = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": self.config.system_prompt},
                {"role": "user", "content": user_input},
            ],
            temperature=0.3,
        )
        return response.choices[0].message.content

    def _audit_log(self, context: PipelineContext):
        entry = {
            "request_id": context.request_id,
            "stages_passed": context.stages_passed,
            "stages_failed": context.stages_failed,
            "flags": len(context.flags),
            "timings": {k: round(v, 2) for k, v in context.timings.items()},
            "total_ms": round(context.total_time_ms, 2),
        }
        logger.info(json.dumps(entry))

Step 8: pipeline/__init__.py

from .sanitizer import InputSanitizer, SanitizationResult, SanitizationAction
from .validator import OutputValidator, ValidationResult, ValidationAction
from .content_filter import ContentFilter, FilterResult, FilterVerdict
from .guardrails import (
    Guardrail, GuardrailChain, GuardrailAction, GuardrailResult, ChainResult,
    LengthGuardrail, TopicGuardrail, ConfidenceGuardrail,
    create_guardrail_chain,
)
from .pipeline import SanitizationPipeline, PipelineContext, PipelineError
from .config import PipelineConfig

Step 9: FastAPI Application

Create app.py:

import logging
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

from pipeline import (
    InputSanitizer, OutputValidator, ContentFilter,
    SanitizationPipeline, PipelineContext, PipelineError, PipelineConfig,
    create_guardrail_chain,
)

logging.basicConfig(level=logging.INFO)

app = FastAPI(title="Sanitized AI API", version="1.0")


class ChatRequest(BaseModel):
    message: str = Field(min_length=1, max_length=10000)
    user_id: str = "anonymous"


class ChatResponse(BaseModel):
    answer: str = Field(min_length=1)
    confidence: float = Field(ge=0.0, le=1.0, default=0.5)


config = PipelineConfig()

pipeline = SanitizationPipeline(
    input_sanitizer=InputSanitizer(config.sanitizer),
    output_validator=OutputValidator(ChatResponse, config.validator),
    content_filter=ContentFilter(config.content_filter),
    guardrail_chain=create_guardrail_chain(config.guardrails),
    config=config,
)


@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    context = PipelineContext(user_id=request.user_id, endpoint="/chat")
    try:
        result = await pipeline.process(request.message, context)
        return ChatResponse(**result)
    except PipelineError as e:
        raise HTTPException(status_code=e.status_code, detail=e.user_message)
    except Exception:
        raise HTTPException(status_code=500, detail="Internal error.")


@app.get("/health")
async def health():
    return {"status": "ok", "pipeline_version": "1.0"}

Step 10: Tests

Create tests/test_sanitizer.py:

from pipeline.sanitizer import InputSanitizer, SanitizationAction
from pipeline.config import SanitizerConfig


def test_clean_input_passes():
    sanitizer = InputSanitizer()
    result = sanitizer.sanitize("¿Cuánto cuesta el iPhone 15?")
    assert result.passed
    assert result.action == SanitizationAction.PASS


def test_empty_input_rejected():
    sanitizer = InputSanitizer()
    result = sanitizer.sanitize("   ")
    assert not result.passed
    assert result.action == SanitizationAction.REJECTED


def test_zero_width_removed():
    sanitizer = InputSanitizer()
    result = sanitizer.sanitize("Hola\u200b mundo\u200d")
    assert result.passed
    assert result.sanitized == "Hola mundo"
    assert result.action == SanitizationAction.CLEANED


def test_html_stripped():
    sanitizer = InputSanitizer()
    result = sanitizer.sanitize("<b>Hello</b> <script>alert(1)</script>")
    assert result.passed
    assert "<" not in result.sanitized
    assert result.action == SanitizationAction.CLEANED


def test_unicode_normalized():
    sanitizer = InputSanitizer()
    result = sanitizer.sanitize("Ｈｅｌｌｏ")
    assert result.passed
    assert result.sanitized == "Hello"


def test_length_truncation():
    sanitizer = InputSanitizer(SanitizerConfig(max_length=10))
    result = sanitizer.sanitize("a" * 100)
    assert result.passed
    assert len(result.sanitized) == 10
    assert result.action == SanitizationAction.TRUNCATED


def test_length_rejection():
    sanitizer = InputSanitizer(SanitizerConfig(max_length=10, on_overlength="reject"))
    result = sanitizer.sanitize("a" * 100)
    assert not result.passed
    assert result.action == SanitizationAction.REJECTED

Create tests/test_validator.py:

from pydantic import BaseModel, Field
from pipeline.validator import OutputValidator, ValidationAction


class TestSchema(BaseModel):
    answer: str = Field(min_length=1)
    confidence: float = Field(ge=0.0, le=1.0, default=0.5)


def test_valid_json():
    validator = OutputValidator(TestSchema)
    result = validator.validate('{"answer": "Paris", "confidence": 0.9}')
    assert result.success
    assert result.action == ValidationAction.VALID


def test_json_in_markdown():
    validator = OutputValidator(TestSchema)
    result = validator.validate('```json\n{"answer": "Paris"}\n```')
    assert result.success


def test_partial_json_repaired():
    validator = OutputValidator(TestSchema)
    result = validator.validate('{"answer": "Paris", "confidence": 0.8')
    assert result.success
    assert result.action == ValidationAction.REPAIRED


def test_no_json_uses_fallback():
    validator = OutputValidator(TestSchema)
    result = validator.validate("No tengo esa información.")
    assert not result.success
    assert result.action == ValidationAction.FALLBACK
    assert result.data is not None


def test_invalid_schema_uses_fallback():
    validator = OutputValidator(TestSchema)
    result = validator.validate('{"answer": "", "confidence": 2.0}')
    assert not result.success
    assert result.action == ValidationAction.FALLBACK

Create tests/test_content_filter.py:

from pipeline.content_filter import ContentFilter, FilterVerdict


def test_clean_text():
    f = ContentFilter()
    result = f.filter("El iPhone 15 cuesta $799.")
    assert result.verdict == FilterVerdict.CLEAN


def test_toxic_blocked():
    f = ContentFilter()
    result = f.filter("Voy a destruir todo y matar a todos.")
    assert result.verdict == FilterVerdict.BLOCKED


def test_pii_flagged():
    f = ContentFilter()
    result = f.filter("Contacta a juan@email.com.")
    assert result.verdict == FilterVerdict.FLAGGED


def test_off_topic_flagged():
    f = ContentFilter()
    result = f.filter("Las próximas elecciones serán en noviembre.")
    assert result.verdict == FilterVerdict.FLAGGED

Create tests/test_guardrails.py:

from pipeline.guardrails import (
    LengthGuardrail, TopicGuardrail, ConfidenceGuardrail,
    GuardrailChain, GuardrailAction,
)


def test_length_pass():
    gr = LengthGuardrail(max_length=100)
    result = gr.execute("Short text")
    assert result.action == GuardrailAction.PASS


def test_length_modify():
    gr = LengthGuardrail(max_length=10)
    result = gr.execute("A very long text that exceeds the limit")
    assert result.action == GuardrailAction.MODIFY
    assert len(result.modified_text) < 20


def test_topic_block():
    gr = TopicGuardrail(blocked_topics=[r"\bpolítica\b"])
    result = gr.execute("La política del gobierno...")
    assert result.action == GuardrailAction.BLOCK


def test_confidence_modify():
    gr = ConfidenceGuardrail(threshold=0.5)
    result = gr.execute("Respuesta", context={"confidence": 0.3})
    assert result.action == GuardrailAction.MODIFY
    assert "⚠️" in result.modified_text


def test_chain_pass():
    chain = GuardrailChain([
        LengthGuardrail(100),
        ConfidenceGuardrail(0.5),
    ])
    result = chain.run("Short text", context={"confidence": 0.9})
    assert result.passed


def test_chain_block():
    chain = GuardrailChain([
        LengthGuardrail(100),
        TopicGuardrail([r"\bpolítica\b"]),
    ])
    result = chain.run("La política del gobierno")
    assert not result.passed
    assert result.blocked_by == "topic_boundary"

Create tests/test_pipeline.py:

import pytest
from pipeline import (
    InputSanitizer, OutputValidator, ContentFilter,
    SanitizationPipeline, PipelineContext, PipelineError, PipelineConfig,
    create_guardrail_chain,
)
from pydantic import BaseModel, Field


class TestResponse(BaseModel):
    answer: str = Field(min_length=1)
    confidence: float = Field(ge=0.0, le=1.0, default=0.5)


@pytest.fixture
def pipeline():
    config = PipelineConfig()

    async def mock_llm(user_input: str) -> str:
        return '{"answer": "Test response", "confidence": 0.85}'

    return SanitizationPipeline(
        input_sanitizer=InputSanitizer(config.sanitizer),
        output_validator=OutputValidator(TestResponse, config.validator),
        content_filter=ContentFilter(config.content_filter),
        guardrail_chain=create_guardrail_chain(config.guardrails),
        config=config,
        llm_caller=mock_llm,
    )


@pytest.mark.asyncio
async def test_pipeline_happy_path(pipeline):
    context = PipelineContext()
    result = await pipeline.process("¿Cuánto cuesta el iPhone?", context)
    assert result["answer"] == "Test response"
    assert "input_sanitize" in context.stages_passed
    assert "output_validate" in context.stages_passed


@pytest.mark.asyncio
async def test_pipeline_empty_input_rejected(pipeline):
    context = PipelineContext()
    with pytest.raises(PipelineError) as exc_info:
        await pipeline.process("   ", context)
    assert exc_info.value.status_code == 400


@pytest.mark.asyncio
async def test_pipeline_llm_failure_uses_fallback(pipeline):
    async def failing_llm(user_input):
        raise Exception("API unavailable")

    pipeline.llm_caller = failing_llm
    context = PipelineContext()
    result = await pipeline.process("Test", context)
    assert result["confidence"] == 0.0
    assert "llm_call" in context.stages_failed

---

Execution

Run the tests

cd sanitization-pipeline
pytest tests/ -v

# Expected output:
# tests/test_sanitizer.py::test_clean_input_passes PASSED
# tests/test_sanitizer.py::test_empty_input_rejected PASSED
# tests/test_sanitizer.py::test_zero_width_removed PASSED
# tests/test_sanitizer.py::test_html_stripped PASSED
# tests/test_sanitizer.py::test_unicode_normalized PASSED
# tests/test_sanitizer.py::test_length_truncation PASSED
# tests/test_sanitizer.py::test_length_rejection PASSED
# tests/test_validator.py::test_valid_json PASSED
# tests/test_validator.py::test_json_in_markdown PASSED
# tests/test_validator.py::test_partial_json_repaired PASSED
# tests/test_validator.py::test_no_json_uses_fallback PASSED
# tests/test_validator.py::test_invalid_schema_uses_fallback PASSED
# tests/test_content_filter.py::test_clean_text PASSED
# tests/test_content_filter.py::test_toxic_blocked PASSED
# tests/test_content_filter.py::test_pii_flagged PASSED
# tests/test_content_filter.py::test_off_topic_flagged PASSED
# tests/test_guardrails.py::test_length_pass PASSED
# ...
# All tests passed!

Run the server

uvicorn app:app --reload --port 8000

Test with curl

# Happy path
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "¿Cuánto cuesta el iPhone 15?", "user_id": "test-user"}'

# Input with HTML (gets cleaned)
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "<script>alert(1)</script> ¿Precio del laptop?", "user_id": "test-user"}'

# Empty input (rejected)
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "   ", "user_id": "test-user"}'

# Health check
curl http://localhost:8000/health

---

Success criteria

Your project is complete when you can verify these points:

• Directory structure with pipeline/, tests/, app.py
• InputSanitizer with Unicode normalization, zero-width removal, HTML strip, length limits
• OutputValidator with JSON extraction, repair, Pydantic validation, fallback
• ContentFilter with toxicity, PII, off-topic detection
• GuardrailChain with 3+ guardrails, fail-fast, timing
• SanitizationPipeline with 5 integrated stages, error handling, audit log
• All tests pass (pytest tests/ -v)
• Functional FastAPI app with /chat and /health endpoints
• Centralized configuration in PipelineConfig
• Generic error messages (they don't reveal security details)

---

Evaluation rubric

Total: 100 points

Category	Points	Key criteria
InputSanitizer	15	Unicode normalization (3), zero-width removal (3), HTML strip (3), length limits (3), configurable (3)
OutputValidator	15	JSON extraction (3), repair (3), Pydantic validation (3), fallback (3), retry concept (3)
ContentFilter	15	Toxicity patterns (4), PII detection (4), off-topic detection (4), configurable (3)
GuardrailChain	15	3+ guardrails (5), chain execution (3), fail-fast (3), timing (2), configurable (2)
Pipeline Integration	20	5 stages integrated (5), error handling per stage (5), audit log (5), fallback strategies (5)
Tests	10	Sanitizer tests (2), validator tests (2), filter tests (2), guardrail tests (2), pipeline tests (2)
Code Quality	10	Clean structure (3), typing (2), config centralized (2), no hardcoded values (3)

Grade distribution

Range	Grade
90-100	Excellent — Production-ready pipeline
80-89	Very good — Solid pipeline with minor improvements
70-79	Good — Covers the basics but needs more robustness
60-69	Acceptable — Missing components or depth
< 60	Needs review — Gaps in the pipeline

---

Common mistakes

1. Hardcoding values instead of using configuration

❌ if len(text) > 4000:
✅ if len(text) > self.config.max_length:

Every limit, threshold, and pattern must come from the configuration. This lets you tune the pipeline without changing code.

2. Revealing block reasons to the user

❌ HTTPException(detail="Input blocked: injection pattern detected")
✅ HTTPException(detail="Your message could not be processed.")

Detailed messages give the attacker information. Generic messages are safer.

3. Not testing edge cases

Tests only for happy paths are not enough. Test: empty inputs, whitespace-only inputs, inputs with only invisible Unicode, outputs without JSON, outputs with partial JSON, toxic outputs.

4. Ignoring the audit log

Without an audit log, you can't know what happens in production. The log must include at least: request_id, stages passed/failed, flags count, total time.

5. Not handling LLM failures

If the OpenAI API fails, your pipeline must return a fallback, not a 500 error. The fallback is a low-quality response, but it's better than an error.

6. Guardrails that can't be disabled

Every guardrail must have an enabled flag. When debugging, you need to disable individual guardrails to isolate problems.

7. Content filter without per-business configuration

The toxicity and PII patterns are generic. The policy rules (competitors, internal prices, disclaimers) are specific to your business. Make sure the policy rules are configurable.

8. Pipeline without a fallback in any stage

If each stage raises exceptions without a fallback, a single failure breaks everything. The output stages (validation, filter, guardrails) must have fallback responses.

---

Connection with the following modules

Your Sanitization Pipeline is the fourth artifact. As you progress:

Module	How it connects
Module 5: Secrets Management	The API keys your pipeline uses (OpenAI) are protected with Vault/KMS instead of env vars
Module 6: PII Protection	Your basic PII detection (regex) is replaced with Presidio for enterprise-grade detection
Module 7: Security Testing	You test your pipeline with adversarial inputs, pen testing, and red team exercises
Module 8: Integration	Your pipeline integrates with Injection Defense (M3) + Secrets (M5) + PII (M6) in the complete system

---

Summary

• The Sanitization Pipeline is the central artifact of Module 4 — it integrates input sanitization, output validation, content filtering, and guardrails into a reusable FastAPI middleware
• 5 components work in sequence: InputSanitizer → OutputValidator → ContentFilter → GuardrailChain → AuditLogger
• Centralized configuration with Pydantic lets you tune the pipeline without changing code
• Per-stage error handling distinguishes between rejecting inputs (400) and using fallbacks for outputs (200 with a safe response)
• Unit tests verify each component in isolation and the integrated pipeline
• The pipeline integrates with the Injection Defense Pipeline from Module 3 and consolidates in Module 8

---

Project resources

1. Pydantic V2 Documentation — Complete reference for validation and configuration schemas
2. FastAPI Documentation — Web framework for the pipeline server
3. OpenAI API Reference — The LLM API the pipeline wraps
4. Pytest Documentation — Testing framework for the pipeline tests
5. OWASP LLM05: Improper Output Handling — The vulnerability the pipeline mitigates
6. Python Logging — Reference for audit logging

---

Created: March 2026 Version: 1.0