Content moderation

export API_BASE_URL="https://gateway.example.com/v1"
export API_KEY="YOUR_API_KEY"
export API_MODEL="YOUR_MODEL_ID"
.venv/bin/python examples/moderations.py

import math
import os
from openai import OpenAI, APIConnectionError, APIStatusError

texts = ["A friendly greeting.", "An invitation to a community picnic."]
with OpenAI(base_url=os.environ["API_BASE_URL"].rstrip("/"),
            api_key=os.environ["API_KEY"], timeout=60.0, max_retries=0) as client:
    try:
        result = client.post("/moderations", cast_to=dict, body={
            "model": os.environ["API_MODEL"], "input": texts,
        })
        if not isinstance(result, dict) or result.get("error"):
            raise ValueError("Invalid moderation response.")
        rows = result.get("results")
        if not isinstance(rows, list) or len(rows) != len(texts):
            raise ValueError("Missing moderation results.")
        decisions = []
        for row in rows:
            if not isinstance(row, dict) or type(row.get("flagged")) is not bool:
                raise ValueError("Missing moderation decision.")
            categories, scores = row.get("categories"), row.get("category_scores")
            if not isinstance(categories, dict) or not categories or not isinstance(scores, dict):
                raise ValueError("Missing moderation categories.")
            if categories.keys() != scores.keys():
                raise ValueError("Category keys do not match.")
            if any(type(value) is not bool for value in categories.values()):
                raise ValueError("Invalid category decision.")
            if any(type(value) not in (int, float) or not math.isfinite(value)
                   or not 0 <= value <= 1 for value in scores.values()):
                raise ValueError("Invalid moderation score.")
            # This application sends any positive signal to review.
            decisions.append("review" if row["flagged"] or any(categories.values()) else "allow")
        for index, decision in enumerate(decisions):
            print(f"{index}: {decision}")
    except APIStatusError as error:
        raise SystemExit(f"Request failed (HTTP {error.status_code}). No automatic retry.")
    except APIConnectionError:
        raise SystemExit("Connection failed. No moderation decision available.")
    except (ValueError, TypeError, AttributeError):
        raise SystemExit("Invalid moderation response. No moderation decision available.")