Course ES
← back to chapter

Moderation gate

The input/output moderation sandwich as a live app: flagged input is blocked with its categories; clean input gets a model answer that is itself moderated before you see it.

Showcase — Moderation gate

The input/output moderation sandwich as a live app: flagged input is blocked with its categories; clean input gets a model answer that is itself moderated before you see it.

Run

bash bootstrap-secrets.sh              # reads ../../../../.env, writes secrets/
docker compose up --build              # default: PROVIDER=openai

Open http://localhost:3000. Gemini: PROVIDER=gemini docker compose up --build.

What's where

  • backend/ai_openai.py — the moderation endpoint + the sandwich.
  • backend/ai_gemini.py — a model-as-classifier moderator + the sandwich.
  • frontend/app/page.tsx — message box, gated result.

Stop

docker compose down

Run locally

Download the project as a ZIP and run it with Docker. Brings up a FastAPI backend + Next.js frontend on localhost:3000.

Download moderation-gate.zip

unzip moderation-gate.zip
cd moderation-gate
bash bootstrap-secrets.sh   # one-time: pulls API keys into ./secrets
docker compose up --build   # default provider: openai
# or:  PROVIDER=gemini docker compose up --build

Type some input, pick a provider, and run the same code shown in Source against the live API. Sign-in required.


  

The same modules the Run button hits. The whole project (frontend, Dockerfile, compose) is in the ZIP under README.

backend/ai_openai.py

"""Showcase 1 (OpenAI): a moderation gate.

The input/output sandwich as a live app. Your message is moderated; if it's
flagged you see the categories and it's blocked. If it's clean, the model
answers and the answer is moderated too before you ever see it.
"""
from openai import OpenAI

_client = OpenAI()

_MODEL = "gpt-5.4-nano"


def _moderate(text: str) -> tuple[bool, list[str]]:
    result = _client.moderations.create(model="omni-moderation-latest", input=text).results[0]
    return result.flagged, [name for name, on in result.categories.model_dump().items() if on]


def run(text: str) -> str:
    msg = text.strip()
    if not msg:
        return "Type a message to send through the moderation gate."
    flagged, cats = _moderate(msg)
    if flagged:
        return f"[input blocked]\ncategories: {', '.join(cats)}"
    answer = _client.responses.create(model=_MODEL, input=[{"role": "user", "content": msg}]).output_text
    out_flagged, out_cats = _moderate(answer)
    if out_flagged:
        return f"[output withheld]\ncategories: {', '.join(out_cats)}"
    return f"{answer}\n\n(moderation: input clean · output clean)"

backend/ai_gemini.py

"""Showcase 1 (Gemini): a moderation gate.

Same input/output sandwich, using a model-as-classifier moderator.
"""
import os

from google import genai
from google.genai import types

_client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])

_MODEL = "gemini-3.1-flash-lite"

_CATEGORIES = "hate, harassment, self-harm, sexual, violence, dangerous"


def _moderate(text: str) -> tuple[bool, list[str]]:
    r = _client.models.generate_content(
        model=_MODEL,
        contents=[types.Content(role="user", parts=[types.Part(text=text)])],
        config=types.GenerateContentConfig(
            system_instruction=(
                "You are a content moderator. Reply with ONLY a comma-separated list "
                f"of any of these categories that apply: {_CATEGORIES}. If none apply, "
                "reply exactly 'none'."
            ),
        ),
    )
    cats = [c.strip() for c in (r.text or "").lower().split(",") if c.strip() and c.strip() != "none"]
    return (len(cats) > 0), cats


def _ask(text: str) -> str:
    r = _client.models.generate_content(
        model=_MODEL, contents=[types.Content(role="user", parts=[types.Part(text=text)])],
    )
    return r.text or ""


def run(text: str) -> str:
    msg = text.strip()
    if not msg:
        return "Type a message to send through the moderation gate."
    flagged, cats = _moderate(msg)
    if flagged:
        return f"[input blocked]\ncategories: {', '.join(cats)}"
    answer = _ask(msg)
    out_flagged, out_cats = _moderate(answer)
    if out_flagged:
        return f"[output withheld]\ncategories: {', '.join(out_cats)}"
    return f"{answer}\n\n(moderation: input clean · output clean)"

Project files

  • .gitignore
  • README.es.md
  • README.md
  • backend/Dockerfile
  • backend/ai_gemini.py
  • backend/ai_openai.py
  • backend/main.py
  • backend/requirements.txt
  • bootstrap-secrets.sh
  • docker-compose.yml
  • frontend/Dockerfile
  • frontend/app/layout.tsx
  • frontend/app/page.tsx
  • frontend/next.config.ts
  • frontend/package.json
  • frontend/tsconfig.json