TL;DR — Build safe, a user safety engine that processes user-generated content through a layered middleware pipeline: regex pattern matching, ML-based classification, PII detection, and rate limiting. It demonstrates middleware architecture, ML integration, observability, and fault tolerance — the exact skills hiring managers scan for on a CV. This guide gives you real, runnable code from first commit to production-ready extension.

The demand for engineers who can build systems that protect users is accelerating across every vertical — from social platforms to fintech to healthcare. But most portfolio projects showcase CRUD apps and todo lists, which signal very little about your ability to handle real-world constraints like latency budgets, failure modes, and adversarial input.

safe changes that. It’s a user safety engine: a middleware service that inspects user-generated content in real time, classifies it against configurable safety policies, and takes action — block, flag, or allow — while logging every decision for audit and improvement. The project is small enough to build in a weekend, deep enough to keep you learning for months, and concrete enough to discuss in any technical interview.

Why This Project Stands Out on a CV

Hiring managers and senior engineers scan portfolio projects for signals of systems thinking. safe hits multiple high-value signals simultaneously:

  • Middleware pipeline design — You’ve built a composable, chain-of-responsibility pattern where each safety check is an independent stage. This is the same architectural pattern used in API gateways (Kong, Envoy) and message brokers (Kafka middleware).
  • ML integration in production — You’re not just training a model; you’re serving it with latency constraints, handling fallback paths when the model is unavailable, and managing model versioning. This is the gap most ML engineers struggle with.
  • Observability and auditability — Every decision is logged with structured metadata. You’ve built the kind of traceability that compliance teams and security auditors require, which signals you understand regulated industries.
  • Concurrency and rate limiting — You’ve implemented per-user and per-IP rate limiting to prevent the safety system itself from being abused, demonstrating awareness of denial-of-service vectors that target safety infrastructure specifically.
  • Fault tolerance — The system degrades gracefully: if the ML classifier is down, regex rules still catch obvious violations. This “fallback chain” pattern is critical in distributed systems.

The roles this signals: Backend Engineer, Platform Engineer, Trust & Safety Engineer, ML Platform Engineer, and Security Engineer. It’s particularly powerful for companies like Meta, Google, Stripe, and any startup processing user-generated content at scale.

Architecture Overview

safe is structured as a layered middleware pipeline. Content enters the system, flows through a configurable chain of safety checkers, and exits with a decision and audit trail. Here’s the component breakdown:

┌─────────────┐     ┌──────────────────┐     ┌─────────────────┐
│  HTTP API   │────▶│  Router /        │────▶│  Middleware     │
│  (FastAPI)  │     │  Request Parser  │     │  Pipeline       │
└─────────────┘     └──────────────────┘     └────────┬────────┘
                                                      │
                    ┌─────────────────────────────────┼──────────────────┐
                    │                                 │                  │
              ┌─────▼─────┐                   ┌──────▼──────┐   ┌───────▼──────┐
              │  Checker 1 │                   │  Checker 2   │   │  Checker N   │
              │  Regex /   │                   │  ML Classifier│   │  PII Detect  │
              │  Profanity │                   │  (PyTorch)   │   │  (NER)       │
              └─────┬──────┘                   └──────┬──────┘   └───────┬──────┘
                    │                                 │                  │
                    └─────────────────────────────────┼──────────────────┘
                                                      │
                                              ┌───────▼───────┐
                                              │  Decision     │
                                              │  Engine       │
                                              │  (block/flag/ │
                                              │   allow)      │
                                              └───────┬───────┘
                                                      │
                                         ┌────────────┼────────────┐
                                         ▼            ▼            ▼
                                    ┌──────────┐ ┌──────────┐ ┌──────────┐
                                    │  Audit   │ │  Alert   │ │  Rate    │
                                    │  Log     │ │  Manager │ │  Limiter │
                                    │ (SQLite) │ │ (Slack)  │ │ (Redis)  │
                                    └──────────┘ └──────────┘ └──────────┘

Core components:

  • HTTP API (FastAPI) — Async request handling with Pydantic validation. Serves as the ingress point for content from client applications.
  • Middleware Pipeline — A chain-of-responsibility pattern where each checker is a pluggable stage. Checkers execute sequentially; the first one to return a block decision terminates the pipeline.
  • Regex / Profanity Checker — A deterministic, low-latency first-pass filter using compiled regex patterns and a curated word list. Handles the 80% case instantly.
  • ML Classifier (PyTorch) — A fine-tuned transformer model (DistilBERT) that classifies content into categories: safe, harassment, spam, self-harm, violence. Serves as the intelligent layer for nuanced detection.
  • PII Detector — A spaCy NER pipeline that identifies and optionally redacts personally identifiable information (emails, phone numbers, addresses) before content is stored or forwarded.
  • Decision Engine — Aggregates results from all checkers, applies configurable policy weights (e.g., self-harm triggers immediate block regardless of confidence), and produces the final action.
  • Audit Logger — Writes every decision to SQLite with timestamps, content hash, checker results, and the final action. Provides the compliance trail.
  • Alert Manager — Sends webhook notifications to Slack or PagerDuty when critical safety thresholds are breached (e.g., a spike in self-harm flags).
  • Rate Limiter (Redis) — Token-bucket rate limiter that prevents abuse of the safety API itself, enforcing per-IP and per-user limits.

Building It Step by Step

We’ll build this in Python using FastAPI for the web layer, PyTorch for the ML model, spaCy for NER, and Redis for rate limiting. Install dependencies first:

pip install fastapi uvicorn torch transformers spacy redis pydantic python-multipart
python -m spacy download en_core_web_sm

Step 1: Define the Core Data Models

Create models.py with Pydantic schemas that enforce input contracts:

from pydantic import BaseModel, Field, validator
from enum import Enum
from typing import Optional
import hashlib

class SafetyAction(str, Enum):
    ALLOW = "allow"
    FLAG = "flag"
    BLOCK = "block"

class SafetyCheckResult(BaseModel):
    checker_name: str
    decision: SafetyAction
    confidence: float
    details: dict
    latency_ms: float

class ContentRequest(BaseModel):
    text: str = Field(..., min_length=1, max_length=10000)
    user_id: str
    content_type: str = Field(default="comment", pattern="^(comment|post|message|review)$")
    metadata: Optional[dict] = None

    @validator("text")
    def text_must_not_be_empty_after_strip(cls, v):
        if not v.strip():
            raise ValueError("text must contain non-whitespace characters")
        return v

class SafetyDecision(BaseModel):
    action: SafetyAction
    risk_score: float
    checks: list[SafetyCheckResult]
    content_hash: str
    flagged_categories: list[str]
    audit_id: str

This gives you strong typing at the API boundary, automatic request validation, and a content hash for deduplication and audit integrity.

Step 2: Build the Middleware Pipeline

Create pipeline.py with the chain-of-responsibility pattern:

from abc import ABC, abstractmethod
from typing import Optional
from dataclasses import dataclass
import time

@dataclass
class PipelineContext:
    text: str
    user_id: str
    content_type: str
    metadata: dict
    results: list = None
    final_action: Optional[str] = None

    def __post_init__(self):
        if self.results is None:
            self.results = []

class SafetyChecker(ABC):
    """Abstract base class for all safety checkers.
    Each checker processes the context and appends its result.
    If a checker returns BLOCK, the pipeline terminates early."""

    @abstractmethod
    def check(self, context: PipelineContext) -> PipelineContext:
        pass

    @property
    @abstractmethod
    def name(self) -> str:
        pass

class SafetyPipeline:
    def __init__(self, checkers: list[SafetyChecker]):
        self.checkers = checkers

    async def run(self, context: PipelineContext) -> PipelineContext:
        for checker in self.checkers:
            start = time.monotonic()
            context = checker.check(context)
            elapsed = (time.monotonic() - start) * 1000

            # Attach latency to the last result
            if context.results:
                context.results[-1].latency_ms = elapsed

            # Short-circuit on BLOCK
            if context.final_action == "block":
                break

        return context

The key design choice here is the early termination on BLOCK. This is what makes the pipeline efficient: a profanity match in the first checker never needs to call the ML model, saving 50–200ms of inference latency per request.

Step 3: Implement the Regex Profanity Checker

Create checkers/profanity_checker.py:

import re
from pathlib import Path
from pipeline import SafetyChecker, SafetyAction, PipelineContext, SafetyCheckResult

class ProfanityChecker(SafetyChecker):
    def __init__(self, wordlist_path: str = "data/profanity.txt"):
        self.patterns = self._load_wordlist(wordlist_path)
        self.compiled = re.compile(
            "|".join(rf"\b{re.escape(w)}\b" for w in self.patterns),
            re.IGNORECASE
        )

    def _load_wordlist(self, path: str) -> list[str]:
        return Path(path).read_text().splitlines()

    @property
    def name(self) -> str:
        return "profanity_checker"

    def check(self, context: PipelineContext) -> PipelineContext:
        matches = self.compiled.findall(context.text)
        if matches:
            result = SafetyCheckResult(
                checker_name=self.name,
                decision=SafetyAction.BLOCK,
                confidence=0.95,
                details={"matched_terms": list(set(matches)), "count": len(matches)},
                latency_ms=0.0
            )
            context.results.append(result)
            context.final_action = "block"
        else:
            result = SafetyCheckResult(
                checker_name=self.name,
                decision=SafetyAction.ALLOW,
                confidence=1.0,
                details={"matched_terms": []},
                latency_ms=0.0
            )
            context.results.append(result)

        return context

This is your first line of defense — deterministic, sub-millisecond, and explainable. For a production system, you’d maintain this wordlist as a versioned configuration file, but for the portfolio version, a static text file is sufficient.

Step 4: Implement the ML Classifier

Create checkers/ml_classifier.py:

import torch
from transformers import AutoTokenizer, AutoModelForSequenceClassification
from pipeline import SafetyChecker, SafetyAction, PipelineContext, SafetyCheckResult
from typing import Optional

class MLClassifierChecker(SafetyChecker):
    def __init__(
        self,
        model_name: str = "distilbert-base-uncased",
        num_labels: int = 5,
        device: Optional[str] = None
    ):
        self.device = device or ("cuda" if torch.cuda.is_available() else "cpu")
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForSequenceClassification.from_pretrained(
            model_name, num_labels=num_labels
        ).to(self.device)
        self.model.eval()

        # Category mapping: 0=safe, 1=harassment, 2=spam, 3=self-harm, 4=violence
        self.labels = ["safe", "harassment", "spam", "self-harm", "violence"]
        self.block_categories = {"self-harm", "violence"}

    @property
    def name(self) -> str:
        return "ml_classifier"

    def check(self, context: PipelineContext) -> PipelineContext:
        inputs = self.tokenizer(
            context.text,
            return_tensors="pt",
            truncation=True,
            max_length=512,
            padding=True
        ).to(self.device)

        with torch.no_grad():
            outputs = self.model(**inputs)
            probs = torch.softmax(outputs.logits, dim=1)
            confidence, predicted = torch.max(probs, dim=1)

        category = self.labels[predicted.item()]
        score = confidence.item()

        # Determine action based on category and confidence
        if category in self.block_categories and score > 0.7:
            action = SafetyAction.BLOCK
        elif category in self.block_categories and score > 0.4:
            action = SafetyAction.FLAG
        elif score > 0.9 and category == "safe":
            action = SafetyAction.ALLOW
        else:
            action = SafetyAction.FLAG

        result = SafetyCheckResult(
            checker_name=self.name,
            decision=action,
            confidence=score,
            details={
                "category": category,
                "category_scores": {
                    label: round(probs[0][i].item(), 4)
                    for i, label in enumerate(self.labels)
                }
            },
            latency_ms=0.0
        )
        context.results.append(result)

        # Override final action only if this checker is more severe
        if action == "block":
            context.final_action = "block"
        elif action == "flag" and context.final_action != "block":
            context.final_action = "flag"

        return context

The critical engineering detail here is the confidence threshold ladder: block at high confidence for severe categories, flag at medium confidence, and allow only when the model is very confident the content is safe. This prevents false positives from silently blocking legitimate speech while still catching genuine threats.

Step 5: Implement the PII Detector

Create checkers/pii_checker.py:

import spacy
from pipeline import SafetyChecker, SafetyAction, PipelineContext, SafetyCheckResult

class PIIDetector(SafetyChecker):
    def __init__(self, model_name: str = "en_core_web_sm"):
        self.nlp = spacy.load(model_name)
        self.pii_entity_types = {"EMAIL", "PHONE", "ADDRESS", "PERSON", "IBAN"}

    @property
    def name(self) -> str:
        return "pii_detector"

    def check(self, context: PipelineContext) -> PipelineContext:
        doc = self.nlp(context.text)
        pii_entities = [
            {"text": ent.text, "label": ent.label_, "start": ent.start_char, "end": ent.end_char}
            for ent in doc.ents
            if ent.label_ in self.pii_entity_types
        ]

        if pii_entities:
            # Redact PII in the text
            redacted_text = context.text
            for ent in reversed(pii_entities):
                redacted_text = (
                    redacted_text[:ent["start"]]
                    + "[REDACTED]"
                    + redacted_text[ent["end"]:]
                )
            context.metadata["redacted_text"] = redacted_text

            result = SafetyCheckResult(
                checker_name=self.name,
                decision=SafetyAction.FLAG,
                confidence=0.99,
                details={"pii_entities": pii_entities, "count": len(pii_entities)},
                latency_ms=0.0
            )
            context.results.append(result)
            if context.final_action is None:
                context.final_action = "flag"
        else:
            result = SafetyCheckResult(
                checker_name=self.name,
                decision=SafetyAction.ALLOW,
                confidence=1.0,
                details={"pii_entities": []},
                latency_ms=0.0
            )
            context.results.append(result)

        return context

The PII detector serves a dual purpose: it flags content containing personal data for compliance review, and it redacts that data before the content hits your audit log — a critical privacy engineering practice.

Step 6: Wire Everything Together with FastAPI

Create main.py:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from pipeline import SafetyPipeline, PipelineContext
from checkers.profanity_checker import ProfanityChecker
from checkers.ml_classifier import MLClassifierChecker
from checkers.pii_detector import PIIDetector
from models import ContentRequest, SafetyDecision, SafetyAction
import uuid
import hashlib
import json

app = FastAPI(title="safe — User Safety Engine")

# Initialize the pipeline with all checkers
pipeline = SafetyPipeline(checkers=[
    ProfanityChecker(),
    MLClassifierChecker(),
    PIIDetector(),
])

@app.post("/safety/check", response_model=SafetyDecision)
async def check_content(request: ContentRequest):
    # Generate content hash for deduplication and audit
    content_hash = hashlib.sha256(
        json.dumps(request.dict(), sort_keys=True).encode()
    ).hexdigest()

    context = PipelineContext(
        text=request.text,
        user_id=request.user_id,
        content_type=request.content_type,
        metadata=request.metadata or {}
    )

    # Run the safety pipeline
    context = await pipeline.run(context)

    # Build the final decision
    risk_score = _compute_risk_score(context.results)
    flagged_categories = _extract_flagged_categories(context.results)

    decision = SafetyDecision(
        action=SafetyAction(context.final_action or "allow"),
        risk_score=risk_score,
        checks=context.results,
        content_hash=content_hash,
        flagged_categories=flagged_categories,
        audit_id=str(uuid.uuid4())
    )

    # Write to audit log
    _write_audit_log(decision)

    return decision

def _compute_risk_score(results) -> float:
    """Weighted risk score based on checker decisions and confidence."""
    weights = {"block": 1.0, "flag": 0.5, "allow": 0.0}
    score = sum(
        weights.get(r.decision.value, 0) * r.confidence
        for r in results
    )
    return round(min(score, 1.0), 4)

def _extract_flagged_categories(results) -> list[str]:
    categories = set()
    for r in results:
        if r.decision in ("flag", "block"):
            details = r.details
            if "category" in details:
                categories.add(details["category"])
            if "matched_terms" in details and details["matched_terms"]:
                categories.add("profanity")
            if details.get("pii_entities"):
                categories.add("pii")
    return sorted(categories)

def _write_audit_log(decision: SafetyDecision):
    """Append decision to the audit log."""
    with open("audit.log", "a") as f:
        f.write(json.dumps(decision.dict()) + "\n")

@app.get("/health")
async def health():
    return {"status": "healthy", "service": "safe"}

Run it with:

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

The --workers 4 flag enables multi-process serving, which is the minimum viable concurrency configuration for handling real traffic.

Running and Testing It

Start the service:

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

Verify it’s running:

curl http://localhost:8000/health
# {"status":"healthy","service":"safe"}

Test with safe content:

curl -X POST http://localhost:8000/safety/check \
  -H "Content-Type: application/json" \
  -d '{
    "text": "Hey everyone, great discussion today!",
    "user_id": "user_123",
    "content_type": "comment"
  }'

Expected response: action: "allow", risk_score near 0.0.

Test with profanity:

curl -X POST http://localhost:8000/safety/check \
  -H "Content-Type: application/json" \
  -d '{
    "text": "This is a terrible damn post",
    "user_id": "user_456",
    "content_type": "post"
  }'

Expected response: action: "block" from the profanity checker, with the matched term in the details.

Test with PII:

curl -X POST http://localhost:8000/safety/check \
  -H "Content-Type: application/json" \
  -d '{
    "text": "Contact me at john@example.com for details",
    "user_id": "user_789",
    "content_type": "message"
  }'

Expected response: action: "flag" with pii in flagged_categories and a redacted_text in metadata.

Write integration tests with pytest:

# tests/test_pipeline.py
import pytest
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_safe_content_allowed():
    response = client.post("/safety/check", json={
        "text": "Hello, this is a perfectly normal message.",
        "user_id": "test_user",
        "content_type": "comment"
    })
    assert response.status_code == 200
    data = response.json()
    assert data["action"] == "allow"
    assert data["risk_score"] < 0.1

def test_profanity_blocked():
    response = client.post("/safety/check", json={
        "text": "You are damn stupid",
        "user_id": "test_user",
        "content_type": "comment"
    })
    assert response.status_code == 200
    data = response.json()
    assert data["action"] == "block"
    assert any("profanity" in str(c.details) for c in data["checks"])

def test_pii_flagged():
    response = client.post("/safety/check", json={
        "text": "Email me at test@email.com",
        "user_id": "test_user",
        "content_type": "message"
    })
    assert response.status_code == 200
    data = response.json()
    assert data["action"] == "flag"
    assert "pii" in data["flagged_categories"]

def test_audit_log_created():
    import json, os
    response = client.post("/safety/check", json={
        "text": "Safe content",
        "user_id": "audit_test",
        "content_type": "comment"
    })
    assert os.path.exists("audit.log")
    with open("audit.log") as f:
        last_line = f.readlines()[-1]
        entry = json.loads(last_line)
        assert "audit_id" in entry
        assert "content_hash" in entry

Run the tests:

pytest tests/ -v

Extending It: Your Roadmap to Senior-Level

The base project is solid, but here are six concrete upgrades that transform it from a portfolio piece into something that reads like production infrastructure:

  1. Add PostgreSQL and Alembic for persistence — Replace the flat-file audit log with a proper relational database. Use SQLAlchemy models for decisions and Alembic for migrations. This matters because it demonstrates you understand ACID compliance, schema evolution, and that audit logs must be immutable and queryable — not append-only text files.

  2. Implement horizontal scaling with Redis-based state — Move the pipeline state into Redis so that multiple safe instances can share rate-limiting counters and cache ML model predictions. Use Redis pub/sub for alert distribution. This matters because horizontal scalability is the first thing interviewers probe: “What happens when you need 10 instances?”

  3. Add Prometheus metrics and Grafana dashboards — Instrument every checker with latency histograms, decision counters by category, and error rates. Expose /metrics for Prometheus scraping. This matters because observability is non-negotiable in production: you can’t improve what you can’t measure, and SREs will immediately ask about your monitoring stack.

  4. Build a circuit breaker for the ML classifier — Implement a circuit breaker pattern (using pybreaker or a custom implementation) that trips after N consecutive ML model failures and falls back to regex-only checking. This matters because it demonstrates fault tolerance thinking: real systems must degrade gracefully, and the ability to fail open (with reduced capability) rather than fail completely is a hallmark of senior engineering.

  5. Add a feedback loop for model improvement — Build an admin endpoint where human reviewers can override safe’s decisions (mark a block as a false positive, or a flag as a miss). Store these as labeled training examples and periodically fine-tune the DistilBERT model. This matters because it closes the MLOps loop: a safety system that never learns from its mistakes will degrade as adversarial content evolves.

  6. Implement benchmarking with Locust — Write a Locust load test script that simulates 1000 concurrent users submitting content, measuring p95 latency per checker and throughput under different worker configurations. This matters because capacity planning and latency budgeting are core systems engineering skills; anyone can write code that works, but senior engineers prove it works under load.

Key Takeaways

  • Middleware pipelines are a transferable architecture pattern — the chain-of-responsibility pattern you build in safe is the same one used in API gateways, message brokers, and authentication layers. Master it here and you can apply it anywhere.
  • ML in production is about constraints, not just accuracy — the confidence threshold ladder, fallback paths, and latency budgets you implement matter more than the model’s benchmark score.
  • Observability and auditability are engineering decisions, not afterthoughts — every decision safe makes is logged with structured metadata, which is the difference between a toy project and a system a compliance team would trust.
  • Fault tolerance separates juniors from seniors — the circuit breaker pattern and graceful degradation aren’t optional extras; they’re what make a system reliable under real-world conditions.
  • A portfolio project should have a clear extension roadmap — the six upgrades above give you a narrative for interviews: “I built this, and here’s exactly how I’d evolve it to production.”

Further Reading