AI Content Moderation Pipeline

from mixpeek import Mixpeek

client = Mixpeek(api_key="YOUR_API_KEY", namespace="moderation")

# 1. A bucket for user uploads, and a collection whose Gemini step answers your
#    policy questions as JSON. There is no dedicated NSFW or violence model; the
#    judgment comes from this structured description.
bucket = client.buckets.create(
    bucket_name="user-uploads",
    bucket_schema={"properties": {"upload": {"type": "image"}}},
)
collection = client.collections.create(
    collection_name="user-uploads",
    source={"type": "bucket", "bucket_ids": [bucket["bucket_id"]]},
    feature_extractor={
        "feature_extractor_name": "multimodal_extractor",
        "version": "v1",
        "parameters": {
            "run_video_description": True,
            "description_prompt": "Review this image against a content policy.",
            "response_shape": {
                "type": "object",
                "properties": {
                    "adult": {"type": "boolean"},
                    "violence": {"type": "boolean"},
                    "hate_symbols": {"type": "boolean"},
                    "reason": {"type": "string"},
                },
            },
        },
    },
)

# 2. Upload and process
client.buckets.upload(
    bucket["bucket_id"],
    blobs=[{"property": "upload", "type": "image", "data": "https://example.com/user-upload.jpg"}],
)
client.collections.trigger(collection["collection_id"])

# 3. Flag anything the policy check marked
docs = client.documents.list(collection["collection_id"], page_size=100)
for doc in docs["results"]:
    verdict = doc.get("json_output") or {}
    if verdict.get("adult") or verdict.get("violence") or verdict.get("hate_symbols"):
        print("flagged", doc["document_id"], verdict.get("reason"))