6. Complete Input → Output Pipeline

Overview

In capsules 02-05 you built four individual components: InputSanitizer, OutputValidator, ContentFilter, and GuardrailChain. Each one works independently. But in production, you don't run them separately — you integrate them into a unified pipeline that processes each request from start to finish: from the moment the user's input arrives until the sanitized response goes out.

This capsule connects all the pieces into a FastAPI middleware that implements the complete flow: Input → Sanitize → Validate → LLM → Validate Output → Filter → Sanitize Output → Respond. It's not new code — it's the integration of everything you already built, with error handling at each stage, fallback strategies, and performance logging.

The pipeline you build here is the simplified version of the capsule 08 project. The difference: here you understand the architecture and the integration patterns; there you implement the complete production-ready version with tests and a rubric.

The complete flow

Request (user)
│
├── STAGE 1: Input Sanitization ──────────────── InputSanitizer (Cap 02)
│   ├── Unicode normalization (NFKC)
│   ├── Zero-width character removal
│   ├── HTML/markdown stripping
│   ├── Whitespace normalization
│   └── Length enforcement
│   │
│   └── If it fails → 400 Bad Request with a generic message
│
├── STAGE 2: Input Validation ────────────────── Pydantic (request model)
│   ├── Type checking
│   └── Field constraints
│   │
│   └── If it fails → 422 Validation Error
│
├── STAGE 3: Injection Detection ─────────────── Module 3 Pipeline
│   ├── Pattern matching
│   └── Classifier
│   │
│   └── If it fails → 400 with a generic message (don't reveal detection)
│
├── STAGE 4: LLM Call ────────────────────────── OpenAI API
│   ├── System prompt + sanitized input
│   └── Temperature, max_tokens config
│   │
│   └── If it fails → 503 Service Unavailable + retry
│
├── STAGE 5: Output Validation ───────────────── OutputValidator (Cap 03)
│   ├── JSON extraction
│   ├── Schema validation (Pydantic)
│   └── Type coercion
│   │
│   └── If it fails → retry LLM (max 2) → fallback response
│
├── STAGE 6: Content Filtering ───────────────── ContentFilter (Cap 04)
│   ├── PII detection
│   ├── Toxicity check
│   ├── Off-topic detection
│   └── Policy enforcement
│   │
│   └── If it fails → safe response (don't reveal the reason)
│
├── STAGE 7: Guardrails ──────────────────────── GuardrailChain (Cap 05)
│   ├── Language consistency
│   ├── Confidence calibration
│   ├── Topic boundaries
│   └── Custom business rules
│   │
│   └── If it fails → safe response
│
├── STAGE 8: Output Sanitization ─────────────── OutputSanitizer
│   ├── HTML escaping
│   └── Encoding normalization
│
└── STAGE 9: Audit Log ──────────────────────── Logger
    ├── Request metadata
    ├── Pipeline stages passed/failed
    ├── Timing per stage
    └── Flags triggered

Response (user)

Pipeline Implementation as FastAPI Middleware

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

from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field

logger = logging.getLogger("sanitization_pipeline")


class PipelineStage(Enum):
    INPUT_SANITIZE = "input_sanitize"
    INPUT_VALIDATE = "input_validate"
    INJECTION_DETECT = "injection_detect"
    LLM_CALL = "llm_call"
    OUTPUT_VALIDATE = "output_validate"
    CONTENT_FILTER = "content_filter"
    GUARDRAILS = "guardrails"
    OUTPUT_SANITIZE = "output_sanitize"


@dataclass
class PipelineContext:
    request_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
    user_id: str = "anonymous"
    endpoint: str = ""
    original_input: str = ""
    sanitized_input: Optional[str] = None
    raw_llm_output: Optional[str] = None
    validated_output: Optional[dict] = None
    filtered_output: Optional[dict] = None
    final_output: Optional[dict] = None
    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

    def add_flag(self, stage: str, flag_type: str, detail: str):
        self.flags.append({
            "stage": stage,
            "type": flag_type,
            "detail": detail,
            "timestamp": time.time(),
        })

    def record_timing(self, stage: str, duration_ms: float):
        self.timings[stage] = duration_ms


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


class SanitizationPipeline:
    """Complete input → output sanitization pipeline."""

    def __init__(
        self,
        input_sanitizer,
        output_validator,
        content_filter,
        guardrail_chain,
        llm_caller: Optional[Callable] = None,
        fallback_response: Optional[dict] = None,
        max_output_retries: int = 2,
    ):
        self.input_sanitizer = input_sanitizer
        self.output_validator = output_validator
        self.content_filter = content_filter
        self.guardrail_chain = guardrail_chain
        self.llm_caller = llm_caller
        self.fallback_response = fallback_response or {
            "answer": "I couldn't process your request. Try rephrasing your question.",
            "confidence": 0.0,
        }
        self.max_output_retries = max_output_retries

    async def process(
        self,
        user_input: str,
        context: PipelineContext,
        system_prompt: str = "",
    ) -> dict:
        """Runs the complete pipeline."""

        context.original_input = user_input
        pipeline_start = time.perf_counter()

        # STAGE 1: Input Sanitization
        stage_start = time.perf_counter()
        sanitize_result = self.input_sanitizer.sanitize(user_input)
        context.record_timing(
            PipelineStage.INPUT_SANITIZE.value,
            (time.perf_counter() - stage_start) * 1000,
        )

        if not sanitize_result.passed:
            context.stages_failed.append(PipelineStage.INPUT_SANITIZE.value)
            context.add_flag(
                PipelineStage.INPUT_SANITIZE.value,
                "rejected",
                str(sanitize_result.issues),
            )
            raise PipelineError(
                stage=PipelineStage.INPUT_SANITIZE.value,
                status_code=400,
                message=f"Input rejected: {sanitize_result.issues}",
                user_message="Your message couldn't be processed. Try a shorter text without special characters.",
            )

        context.sanitized_input = sanitize_result.sanitized
        context.stages_passed.append(PipelineStage.INPUT_SANITIZE.value)

        if sanitize_result.issues:
            context.add_flag(
                PipelineStage.INPUT_SANITIZE.value,
                "cleaned",
                str(sanitize_result.issues),
            )

        # STAGE 4: LLM Call (stages 2-3 handled by FastAPI + M3)
        stage_start = time.perf_counter()
        try:
            raw_output = await self._call_llm(
                context.sanitized_input, system_prompt
            )
            context.raw_llm_output = raw_output
            context.stages_passed.append(PipelineStage.LLM_CALL.value)
        except Exception as e:
            context.stages_failed.append(PipelineStage.LLM_CALL.value)
            logger.error(f"[{context.request_id}] LLM call failed: {e}")
            return self.fallback_response
        finally:
            context.record_timing(
                PipelineStage.LLM_CALL.value,
                (time.perf_counter() - stage_start) * 1000,
            )

        # STAGE 5: Output Validation (with retry)
        stage_start = time.perf_counter()
        validation_result = self.output_validator.validate(raw_output)

        if not validation_result.success:
            for retry in range(self.max_output_retries):
                raw_output = await self._call_llm(
                    context.sanitized_input,
                    system_prompt + "\nIMPORTANT: Respond ONLY with valid JSON.",
                )
                validation_result = self.output_validator.validate(raw_output)
                if validation_result.success:
                    context.add_flag(
                        PipelineStage.OUTPUT_VALIDATE.value,
                        "retry_succeeded",
                        f"Retry {retry + 1} succeeded",
                    )
                    break

        context.record_timing(
            PipelineStage.OUTPUT_VALIDATE.value,
            (time.perf_counter() - stage_start) * 1000,
        )

        if validation_result.success:
            context.validated_output = validation_result.data
            context.stages_passed.append(PipelineStage.OUTPUT_VALIDATE.value)
        else:
            context.stages_failed.append(PipelineStage.OUTPUT_VALIDATE.value)
            context.add_flag(
                PipelineStage.OUTPUT_VALIDATE.value,
                "validation_failed",
                "Using fallback after retries exhausted",
            )
            return self.fallback_response

        # STAGE 6: Content Filtering
        stage_start = time.perf_counter()
        output_text = json.dumps(context.validated_output)
        filter_result = self.content_filter.filter(output_text)
        context.record_timing(
            PipelineStage.CONTENT_FILTER.value,
            (time.perf_counter() - stage_start) * 1000,
        )

        if filter_result.verdict.value == "blocked":
            context.stages_failed.append(PipelineStage.CONTENT_FILTER.value)
            context.add_flag(
                PipelineStage.CONTENT_FILTER.value,
                "blocked",
                str(filter_result.flags),
            )
            return self.fallback_response
        context.stages_passed.append(PipelineStage.CONTENT_FILTER.value)

        # STAGE 7: Guardrails
        stage_start = time.perf_counter()
        guardrail_result = self.guardrail_chain.run(
            output_text,
            context={
                "user_id": context.user_id,
                "input_language": "es",
                "confidence": context.validated_output.get("confidence", 0.5),
            },
        )
        context.record_timing(
            PipelineStage.GUARDRAILS.value,
            (time.perf_counter() - stage_start) * 1000,
        )

        if not guardrail_result.passed:
            context.stages_failed.append(PipelineStage.GUARDRAILS.value)
            context.add_flag(
                PipelineStage.GUARDRAILS.value,
                "blocked",
                f"Blocked by {guardrail_result.blocked_by}",
            )
            return self.fallback_response
        context.stages_passed.append(PipelineStage.GUARDRAILS.value)

        # STAGE 8: Output Sanitization
        context.final_output = context.validated_output
        context.stages_passed.append(PipelineStage.OUTPUT_SANITIZE.value)

        # STAGE 9: Audit Log
        context.total_time_ms = (time.perf_counter() - pipeline_start) * 1000
        self._audit_log(context)

        return context.final_output

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

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

    def _audit_log(self, context: PipelineContext):
        log_entry = {
            "request_id": context.request_id,
            "user_id": context.user_id,
            "endpoint": context.endpoint,
            "stages_passed": context.stages_passed,
            "stages_failed": context.stages_failed,
            "flags_count": len(context.flags),
            "timings_ms": context.timings,
            "total_time_ms": round(context.total_time_ms, 2),
        }

        if context.flags:
            log_entry["flags"] = context.flags

        logger.info(json.dumps(log_entry))

FastAPI Integration

Mounting the pipeline as middleware in a FastAPI application:

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

app = FastAPI(title="Sanitized AI API")


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


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


# Initialize the pipeline components
# (using the classes from the previous capsules)

def create_pipeline() -> SanitizationPipeline:
    """Factory that creates the pipeline with a default configuration."""

    # Components from previous capsules (simplified for the example)
    from dataclasses import dataclass

    @dataclass
    class SimpleSanitizer:
        max_length: int = 4000

        def sanitize(self, text):
            import unicodedata
            cleaned = unicodedata.normalize("NFKC", text.strip())
            issues = []
            if cleaned != text:
                issues.append("normalized")
            if len(cleaned) > self.max_length:
                cleaned = cleaned[:self.max_length]
                issues.append("truncated")

            class Result:
                pass
            r = Result()
            r.passed = bool(cleaned)
            r.sanitized = cleaned
            r.issues = issues
            return r

    @dataclass
    class SimpleValidator:
        def validate(self, raw):
            import json, re
            try:
                match = re.search(r"\{[\s\S]*\}", raw)
                if match:
                    data = json.loads(match.group())
                else:
                    data = json.loads(raw)

                class Result:
                    pass
                r = Result()
                r.success = True
                r.data = data
                return r
            except Exception:
                class Result:
                    pass
                r = Result()
                r.success = False
                r.data = None
                return r

    @dataclass
    class SimpleFilter:
        def filter(self, text):
            class Result:
                pass
            r = Result()
            r.verdict = type("V", (), {"value": "clean"})()
            r.flags = []
            return r

    @dataclass
    class SimpleChain:
        def run(self, text, context=None):
            class Result:
                pass
            r = Result()
            r.passed = True
            r.blocked_by = None
            r.results = []
            r.total_time_ms = 0
            return r

    return SanitizationPipeline(
        input_sanitizer=SimpleSanitizer(),
        output_validator=SimpleValidator(),
        content_filter=SimpleFilter(),
        guardrail_chain=SimpleChain(),
        fallback_response={"answer": "I couldn't process your request.", "confidence": 0.0},
    )


pipeline = create_pipeline()

SYSTEM_PROMPT = """You are a customer service assistant for TechStore.
Respond in JSON with fields: answer (string), confidence (float 0-1).
Only answer questions about electronic products."""


@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(
            user_input=request.message,
            context=context,
            system_prompt=SYSTEM_PROMPT,
        )
        return ChatResponse(**result)

    except PipelineError as e:
        raise HTTPException(
            status_code=e.status_code,
            detail=e.user_message,
        )
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        raise HTTPException(
            status_code=500,
            detail="Internal error. Try again.",
        )


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

Error Handling per stage

Each pipeline stage has a different error strategy:

ERROR_STRATEGIES = {
    PipelineStage.INPUT_SANITIZE: {
        "on_failure": "reject",
        "http_status": 400,
        "user_message": "Your message couldn't be processed.",
        "reveal_reason": False,
        "log_level": "warning",
    },
    PipelineStage.INJECTION_DETECT: {
        "on_failure": "reject",
        "http_status": 400,
        "user_message": "Your message couldn't be processed.",
        "reveal_reason": False,
        "log_level": "warning",
    },
    PipelineStage.LLM_CALL: {
        "on_failure": "fallback",
        "http_status": 200,
        "user_message": None,
        "reveal_reason": False,
        "log_level": "error",
    },
    PipelineStage.OUTPUT_VALIDATE: {
        "on_failure": "retry_then_fallback",
        "max_retries": 2,
        "http_status": 200,
        "user_message": None,
        "reveal_reason": False,
        "log_level": "warning",
    },
    PipelineStage.CONTENT_FILTER: {
        "on_failure": "fallback",
        "http_status": 200,
        "user_message": None,
        "reveal_reason": False,
        "log_level": "warning",
    },
    PipelineStage.GUARDRAILS: {
        "on_failure": "fallback",
        "http_status": 200,
        "user_message": None,
        "reveal_reason": False,
        "log_level": "info",
    },
}

print("Error Handling Strategies:")
print(f"{'Stage':<22} {'On Failure':<22} {'HTTP Status':<14} {'Reveal?':<10}")
print("-" * 68)
for stage, strategy in ERROR_STRATEGIES.items():
    print(
        f"{stage.value:<22} "
        f"{strategy['on_failure']:<22} "
        f"{strategy['http_status']:<14} "
        f"{'Yes' if strategy['reveal_reason'] else 'No':<10}"
    )

# Expected output:
# Error Handling Strategies:
# Stage                  On Failure             HTTP Status    Reveal?
# --------------------------------------------------------------------
# input_sanitize         reject                 400            No
# injection_detect       reject                 400            No
# llm_call               fallback               200            No
# output_validate        retry_then_fallback    200            No
# content_filter         fallback               200            No
# guardrails             fallback               200            No

Key principle: never reveal security information

Notice that reveal_reason is False for every stage. This is intentional:

• ❌ "Your message was blocked by the injection detector" → Tells the attacker you have a detector
• ❌ "Output blocked for toxic content" → Tells the attacker how to evade it
• ❌ "PII detected in output" → Confirms that you have PII detection
• ✅ "I couldn't process your request. Try rephrasing your question." → Generic, gives no hints

Fallback Strategies

from dataclasses import dataclass
from typing import Optional


@dataclass
class FallbackConfig:
    strategy: str  # "static", "cached", "degraded", "error"
    static_response: Optional[dict] = None
    cache_ttl_seconds: int = 300
    degraded_system_prompt: Optional[str] = None


FALLBACK_STRATEGIES = {
    "static": FallbackConfig(
        strategy="static",
        static_response={
            "answer": "I couldn't process your request. Try rephrasing your question.",
            "confidence": 0.0,
        },
    ),
    "cached": FallbackConfig(
        strategy="cached",
        cache_ttl_seconds=300,
    ),
    "degraded": FallbackConfig(
        strategy="degraded",
        degraded_system_prompt="Respond very briefly and conservatively.",
    ),
    "error": FallbackConfig(
        strategy="error",
    ),
}


class FallbackHandler:
    def __init__(self, config: FallbackConfig):
        self.config = config
        self.cache: dict[str, dict] = {}

    def get_fallback(self, original_input: str, stage_failed: str) -> dict:
        if self.config.strategy == "static":
            return self.config.static_response

        if self.config.strategy == "cached":
            cache_key = hash(original_input[:100])
            if cache_key in self.cache:
                return self.cache[cache_key]
            return self.config.static_response or {
                "answer": "Service temporarily unavailable.",
                "confidence": 0.0,
            }

        if self.config.strategy == "error":
            return {
                "error": True,
                "answer": "Error processing the request.",
                "confidence": 0.0,
            }

        return {"answer": "Generic fallback.", "confidence": 0.0}


handler = FallbackHandler(FALLBACK_STRATEGIES["static"])
fallback = handler.get_fallback("How much is the iPhone?", "content_filter")
print(f"Fallback: {fallback}")

# Expected output:
# Fallback: {'answer': "I couldn't process your request. Try rephrasing your question.", 'confidence': 0.0}

Performance Optimization

import time

PERFORMANCE_BUDGET = {
    "target_total_ms": 2000,
    "budget_by_stage": {
        PipelineStage.INPUT_SANITIZE: {"target_ms": 5, "max_ms": 20},
        PipelineStage.INJECTION_DETECT: {"target_ms": 10, "max_ms": 50},
        PipelineStage.LLM_CALL: {"target_ms": 1500, "max_ms": 5000},
        PipelineStage.OUTPUT_VALIDATE: {"target_ms": 5, "max_ms": 20},
        PipelineStage.CONTENT_FILTER: {"target_ms": 10, "max_ms": 300},
        PipelineStage.GUARDRAILS: {"target_ms": 10, "max_ms": 50},
        PipelineStage.OUTPUT_SANITIZE: {"target_ms": 2, "max_ms": 10},
    },
}


def analyze_pipeline_performance(context: PipelineContext) -> dict:
    """Analyzes the pipeline's performance against the budget."""
    analysis = {
        "total_ms": context.total_time_ms,
        "within_budget": context.total_time_ms <= PERFORMANCE_BUDGET["target_total_ms"],
        "stages": {},
    }

    for stage_name, timing in context.timings.items():
        try:
            stage = PipelineStage(stage_name)
        except ValueError:
            continue
        budget = PERFORMANCE_BUDGET["budget_by_stage"].get(stage, {})
        target = budget.get("target_ms", 0)
        max_ms = budget.get("max_ms", 0)

        analysis["stages"][stage_name] = {
            "actual_ms": round(timing, 2),
            "target_ms": target,
            "max_ms": max_ms,
            "status": (
                "ok" if timing <= target
                else "warning" if timing <= max_ms
                else "critical"
            ),
        }

    return analysis


# Example with simulated timings
mock_context = PipelineContext()
mock_context.total_time_ms = 1850
mock_context.timings = {
    "input_sanitize": 3.2,
    "llm_call": 1600,
    "output_validate": 4.1,
    "content_filter": 220,
    "guardrails": 8.5,
}

analysis = analyze_pipeline_performance(mock_context)
print(f"Total: {analysis['total_ms']:.0f}ms (within budget: {analysis['within_budget']})")
for stage, info in analysis["stages"].items():
    print(f"  {stage}: {info['actual_ms']}ms [{info['status']}] (target: {info['target_ms']}ms)")

# Expected output:
# Total: 1850ms (within budget: True)
#   input_sanitize: 3.2ms [ok] (target: 5ms)
#   llm_call: 1600ms [warning] (target: 1500ms)
#   output_validate: 4.1ms [ok] (target: 5ms)
#   content_filter: 220ms [warning] (target: 10ms)
#   guardrails: 8.5ms [ok] (target: 10ms)

Complete Request/Response Lifecycle

from datetime import datetime, timezone


def trace_request_lifecycle(context: PipelineContext) -> str:
    """Generates a visual trace of a request's lifecycle."""
    lines = [
        f"=== Request Lifecycle: {context.request_id} ===",
        f"User: {context.user_id}",
        f"Endpoint: {context.endpoint}",
        f"Time: {datetime.now(timezone.utc).isoformat()}",
        "",
    ]

    all_stages = [
        PipelineStage.INPUT_SANITIZE,
        PipelineStage.INPUT_VALIDATE,
        PipelineStage.INJECTION_DETECT,
        PipelineStage.LLM_CALL,
        PipelineStage.OUTPUT_VALIDATE,
        PipelineStage.CONTENT_FILTER,
        PipelineStage.GUARDRAILS,
        PipelineStage.OUTPUT_SANITIZE,
    ]

    for stage in all_stages:
        name = stage.value
        timing = context.timings.get(name, 0)

        if name in context.stages_passed:
            status = "✅ PASS"
        elif name in context.stages_failed:
            status = "❌ FAIL"
        else:
            status = "⏭️ SKIP"

        lines.append(f"  {status} {name:<20} {timing:.1f}ms")

    lines.append("")
    lines.append(f"  Total: {context.total_time_ms:.1f}ms")
    lines.append(f"  Flags: {len(context.flags)}")

    if context.flags:
        lines.append("  Flag details:")
        for flag in context.flags:
            lines.append(f"    - [{flag['stage']}] {flag['type']}: {flag['detail'][:60]}")

    return "\n".join(lines)


# Simulation
demo_context = PipelineContext(
    request_id="abc12345",
    user_id="user-42",
    endpoint="/chat",
)
demo_context.stages_passed = [
    "input_sanitize", "llm_call", "output_validate",
    "content_filter", "guardrails", "output_sanitize",
]
demo_context.timings = {
    "input_sanitize": 2.5,
    "llm_call": 1200,
    "output_validate": 3.8,
    "content_filter": 15.2,
    "guardrails": 5.1,
}
demo_context.total_time_ms = 1226.6
demo_context.flags = [
    {"stage": "input_sanitize", "type": "cleaned", "detail": "['Unicode normalized (NFKC)']", "timestamp": 0},
]

print(trace_request_lifecycle(demo_context))

# Expected output:
# === Request Lifecycle: abc12345 ===
# User: user-42
# Endpoint: /chat
# ...
#   ✅ PASS input_sanitize       2.5ms
#   ⏭️ SKIP input_validate       0.0ms
#   ⏭️ SKIP injection_detect     0.0ms
#   ✅ PASS llm_call              1200.0ms
#   ✅ PASS output_validate       3.8ms
#   ✅ PASS content_filter        15.2ms
#   ✅ PASS guardrails            5.1ms
#   ✅ PASS output_sanitize       0.0ms
#
#   Total: 1226.6ms
#   Flags: 1

Troubleshooting

Problem 1: "The pipeline adds too much total latency"

When all the stages run sequentially, the latency accumulates.

Solution: Identify the slowest stages (usually the LLM call and the Moderation API). The local sanitization stages (<10ms) aren't the problem. For the Moderation API, consider running it in parallel with the guardrails if they're independent. The LLM call is inherently serial — optimize it with streaming (capsule 07).

Problem 2: "Output validation retries double the cost"

Each retry is an additional LLM call with the corresponding token cost.

Solution: Limit retries to 2. Use decreasing temperature (0.7 → 0.3 → 0.0). Monitor your retry rate — if it's > 10%, the problem is your prompt, not the validation. Improve the system prompt before adding retries.

Problem 3: "The fallback response is boring and generic"

Always returning "I couldn't process your request" is a bad user experience.

Solution: Personalize the fallbacks based on the stage that failed and the context:

CONTEXTUAL_FALLBACKS = {
    "output_validate": "Let me try to answer another way: {simplified_answer}",
    "content_filter": "I have information about that topic but I need to rephrase it. Could you be more specific in your question?",
    "guardrails": "That question is outside my specialty. Can I help you with products or technical support?",
}

Problem 4: "The audit trail logs are too verbose"

With thousands of requests, the complete pipeline logs fill the disk fast.

Solution: Always log: request_id, stages_passed/failed, total_time_ms, flag_count. Log conditionally: flag details (only if there are flags), timings (only if total > budget), raw output (only if validation failed).

Exercises

Exercise 1: Pipeline with a circuit breaker

Implement a circuit breaker that temporarily disables the LLM call if it fails more than N consecutive times.

import time


class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, reset_timeout: int = 60):
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.failure_count = 0
        self.last_failure_time = 0.0
        self.state = "closed"

    def can_execute(self) -> bool:
        if self.state == "closed":
            return True
        if self.state == "open":
            if time.time() - self.last_failure_time > self.reset_timeout:
                self.state = "half-open"
                return True
            return False
        return True  # half-open

    def record_success(self):
        self.failure_count = 0
        self.state = "closed"

    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = "open"


cb = CircuitBreaker(failure_threshold=3, reset_timeout=30)

for i in range(5):
    if cb.can_execute():
        cb.record_failure()
        print(f"Attempt {i+1}: failed (state: {cb.state}, failures: {cb.failure_count})")
    else:
        print(f"Attempt {i+1}: circuit open, using fallback")

# Expected output:
# Attempt 1: failed (state: closed, failures: 1)
# Attempt 2: failed (state: closed, failures: 2)
# Attempt 3: failed (state: open, failures: 3)
# Attempt 4: circuit open, using fallback
# Attempt 5: circuit open, using fallback

Exercise 2: Pipeline metrics dashboard

Create a /metrics endpoint that reports pipeline statistics.

from collections import Counter, defaultdict

class PipelineMetrics:
    def __init__(self):
        self.total_requests = 0
        self.stage_pass_counts = Counter()
        self.stage_fail_counts = Counter()
        self.flag_counts = Counter()
        self.total_times: list[float] = []

    def record(self, context: PipelineContext):
        self.total_requests += 1
        for stage in context.stages_passed:
            self.stage_pass_counts[stage] += 1
        for stage in context.stages_failed:
            self.stage_fail_counts[stage] += 1
        for flag in context.flags:
            self.flag_counts[flag["type"]] += 1
        self.total_times.append(context.total_time_ms)

    def summary(self) -> dict:
        avg_time = sum(self.total_times) / max(len(self.total_times), 1)
        return {
            "total_requests": self.total_requests,
            "avg_time_ms": round(avg_time, 1),
            "stage_pass_rates": {
                stage: count / max(self.total_requests, 1)
                for stage, count in self.stage_pass_counts.items()
            },
            "top_flags": dict(self.flag_counts.most_common(5)),
        }


metrics = PipelineMetrics()
# In production: metrics.record(context) after each request
# @app.get("/metrics")
# async def get_metrics():
#     return metrics.summary()

print("Metrics endpoint ready at /metrics")

Exercise 3: A/B testing of pipeline configurations

Implement a system that lets you run two pipeline configurations in parallel and compare the results.

import random

class ABTestPipeline:
    def __init__(self, pipeline_a, pipeline_b, split_ratio: float = 0.5):
        self.pipeline_a = pipeline_a
        self.pipeline_b = pipeline_b
        self.split_ratio = split_ratio
        self.results_a: list[dict] = []
        self.results_b: list[dict] = []

    async def process(self, user_input: str, context: PipelineContext) -> dict:
        variant = "A" if random.random() < self.split_ratio else "B"
        pipeline = self.pipeline_a if variant == "A" else self.pipeline_b

        result = await pipeline.process(user_input, context, system_prompt="")

        record = {
            "variant": variant,
            "passed": len(context.stages_failed) == 0,
            "flags": len(context.flags),
            "time_ms": context.total_time_ms,
        }

        if variant == "A":
            self.results_a.append(record)
        else:
            self.results_b.append(record)

        return result

    def compare(self) -> dict:
        def stats(results):
            if not results:
                return {"n": 0}
            pass_rate = sum(1 for r in results if r["passed"]) / len(results)
            avg_time = sum(r["time_ms"] for r in results) / len(results)
            avg_flags = sum(r["flags"] for r in results) / len(results)
            return {"n": len(results), "pass_rate": pass_rate, "avg_time": avg_time, "avg_flags": avg_flags}

        return {"A": stats(self.results_a), "B": stats(self.results_b)}


print("A/B test framework ready. Use ABTestPipeline(strict_pipeline, permissive_pipeline)")

Exercise 4: Pipeline with per-stage timeout

Implement individual timeouts for each pipeline stage.

import asyncio

STAGE_TIMEOUTS = {
    "input_sanitize": 1.0,
    "llm_call": 10.0,
    "output_validate": 2.0,
    "content_filter": 3.0,
    "guardrails": 2.0,
}


async def execute_with_timeout(
    stage_name: str, coro, fallback_value=None
) -> tuple[Any, bool]:
    timeout = STAGE_TIMEOUTS.get(stage_name, 5.0)
    try:
        result = await asyncio.wait_for(coro, timeout=timeout)
        return result, True
    except asyncio.TimeoutError:
        logger.warning(f"Stage {stage_name} timed out after {timeout}s")
        return fallback_value, False


print("Stage timeouts configured:")
for stage, timeout in STAGE_TIMEOUTS.items():
    print(f"  {stage}: {timeout}s")

# Expected output:
# Stage timeouts configured:
#   input_sanitize: 1.0s
#   llm_call: 10.0s
#   output_validate: 2.0s
#   content_filter: 3.0s
#   guardrails: 2.0s

Summary

• 🔑 The complete pipeline integrates InputSanitizer → OutputValidator → ContentFilter → GuardrailChain in a sequential flow with error handling per stage
• 🔑 Each stage has a different error strategy: input stages reject (400), output stages use a fallback (200 with a safe response)
• 🔑 Never reveal to the user why an input or output was blocked — generic messages prevent attackers from fingerprinting your defenses
• 🔑 Fallback strategies should be contextual: static response, cached response, degraded mode, or error — depending on the stage that failed and the context
• 🔑 The performance budget distributes the ~2 total seconds: ~1.5s for the LLM, ~200ms for the Moderation API, <50ms for local sanitization/validation/guardrails
• 🔑 The audit log records each request with: stages passed/failed, flags, timings — it's the basis for monitoring and calibration
• 🔑 The circuit breaker protects against cascading LLM failures: after N consecutive failures, it returns a fallback without calling the LLM
• 🔑 This pipeline is the simplified version of the capsule 08 project — there you implement the complete version with tests and a rubric

Additional resources

1. FastAPI Middleware — Official documentation of middleware in FastAPI for integrating the pipeline
2. Circuit Breaker Pattern — Resilience pattern for handling failures in external services
3. OWASP LLM05: Improper Output Handling — The vulnerability the complete pipeline mitigates
4. Structured Logging with Python — Guide to structured logging for audit trails
5. OpenTelemetry for Python — Observability framework for distributed tracing of the pipeline
6. A/B Testing Best Practices — A/B testing principles applicable to pipeline configurations
7. Resilience Patterns — Resilience patterns: retry, circuit breaker, bulkhead, timeout

Created: March 2026 Version: 1.0