◈ API / SDK/2026-04-11Advanced

Building a Production Content Moderation System with Gemini API: A

A complete guide to building a production-grade content moderation system with the Gemini API. Covers custom safety criteria, multimodal inspection of text and images, async batch processing, Human-in-the-Loop workflows, and cost optimization.

gemini-api²⁷⁷ content-moderation² production¹⁴⁰ multimodal⁴⁴ safety² python¹⁰⁴

✦ Premium Article

A phrase like "knife techniques" means one thing inside a cooking recipe and something else entirely inside a threatening post. For any service handling user-generated content, that kind of judgment is exactly what keyword matching and image classifiers keep missing.

Gemini 2.5 Pro and Flash close that gap with multimodal input and genuine contextual understanding — which is why building a custom moderation layer on Gemini has become a credible alternative to dedicated services like Amazon Rekognition or Azure Content Moderator.

The built-in safety filters (safety_settings), though, stop short of requirements like these:

Industry-specific prohibited content: Compliance requirements unique to healthcare, finance, or education platforms
Custom severity scoring: Weighted word lists or tiered response flows
Multimodal compound decisions: Evaluating text and images together as a unit
Audit logs and explainability: Recording why a particular decision was made

System Architecture

Core Components

A robust production moderation system is organized into four layers.

Layer 1: Ingest Layer Receives incoming user content and writes it to an async buffer (Redis or Cloud Tasks).

Layer 2: Moderation Processing Layer Worker processes that call the Gemini API to analyze content. Designed for horizontal scaling.

Layer 3: Decision and Action Layer A rules engine that routes analysis results to automatic approval, automatic rejection, or human review.

Layer 4: Human-in-the-Loop Layer Sends borderline cases to human moderators and feeds their decisions back as a training signal.

Choosing Your Gemini Model

To balance cost and accuracy, adopt a two-stage moderation strategy. Run Gemini 2.5 Flash for the initial screen -- it is optimized for high throughput and low cost -- and escalate to Gemini 2.5 Pro only when the confidence score falls near the boundary. Pro delivers higher accuracy for complex contextual decisions but at greater cost per token.

Moderating Text Content

Defining Custom Safety Criteria

One of Gemini API's biggest advantages for moderation is the ability to define business-specific judgment criteria in plain language via System Instructions.

import google.generativeai as genai
import json
from enum import Enum
from dataclasses import dataclass
from typing import Optional
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
 
class ModerationDecision(Enum):
    APPROVED = "approved"
    REJECTED = "rejected"
    REVIEW = "review"
 
@dataclass
class ModerationResult:
    decision: ModerationDecision
    confidence: float          # 0.0 to 1.0
    categories: list[str]      # Detected categories
    reason: str                # Decision rationale (for audit logs)
    escalate_to_human: bool    # Whether human review is needed
 
MODERATION_SYSTEM_PROMPT = """
You are a content moderator for a community platform.
Review each piece of content against the criteria below and respond ONLY in the specified JSON format.
 
## Criteria
 
### REJECTED (immediate removal)
- Threats or harassment targeting real individuals
- Exposure of personal information (home address, phone numbers, credit card numbers)
- Promotion of illegal goods or services
- Content inappropriate for minors
 
### REVIEW (send to human review)
- Inflammatory content related to politics or religion
- Medical or legal advice (requires expert verification)
- Borderline cases where confidence is below 0.7
 
### APPROVED
- General user posts that do not fall into the above categories
 
## Response format (return this JSON and nothing else)
{
  "decision": "approved|rejected|review",
  "confidence": numeric value between 0.0 and 1.0,
  "categories": ["list", "of", "detected", "categories"],
  "reason": "Decision rationale in 1 to 2 sentences"
}
"""
 
def moderate_text(content: str, model_name: str = "gemini-2.5-flash") -> ModerationResult:
    """Moderate a text content item."""
    model = genai.GenerativeModel(
        model_name=model_name,
        system_instruction=MODERATION_SYSTEM_PROMPT
    )
    
    try:
        response = model.generate_content(
            f"Please review the following content:\n\n{content}",
            generation_config=genai.GenerationConfig(
                response_mime_type="application/json",
                temperature=0.1,
                max_output_tokens=512
            )
        )
        
        result_data = json.loads(response.text)
        decision = ModerationDecision(result_data["decision"])
        confidence = float(result_data.get("confidence", 0.5))
        
        return ModerationResult(
            decision=decision,
            confidence=confidence,
            categories=result_data.get("categories", []),
            reason=result_data.get("reason", ""),
            escalate_to_human=(
                decision == ModerationDecision.REVIEW or
                confidence < 0.7
            )
        )
    except (json.JSONDecodeError, KeyError, ValueError) as e:
        return ModerationResult(
            decision=ModerationDecision.REVIEW,
            confidence=0.0,
            categories=["parse_error"],
            reason=f"API response parse error: {str(e)}",
            escalate_to_human=True
        )

The critical detail here is response_mime_type="application/json". Gemini API's JSON mode guarantees structured responses every time, dramatically reducing parse errors in production. Setting temperature=0.1 ensures consistent, repeatable decisions across similar inputs.

Expected output for a benign post:

{
  "decision": "approved",
  "confidence": 0.95,
  "categories": [],
  "reason": "General user inquiry with no harmful content detected."
}

✦

Thank you for reading this far.

Continue Reading

What follows includes implementation code, benchmarks, and practical content we hope you'll find useful. This site runs without ads — server and development costs are supported entirely by members like you. If it's been helpful, we'd be truly grateful for your support.

WHAT YOU'LL LEARN

✦Developers frustrated by built-in safety filters can now implement a custom harmful-content detection system tailored to their business rules -- and have it running today

✦You will get a complete, production-ready architecture and working code for processing multimodal content (text and images) through an async batch pipeline

✦You will be able to apply a practical tuning strategy that optimizes detection accuracy, processing throughput, and cost simultaneously -- plus wire in a Human-in-the-Loop review flow

Secure payment via Stripe · Cancel anytime

✦

Unlock This Article

Get full access to the rest of this article. Buy once, read anytime. This site is ad-free — your support goes directly toward keeping it running.

Unlock all articles with Membership →

Moderating Images and Multimodal Content

from pathlib import Path
 
def moderate_image(
    image_path: str,
    context_text: Optional[str] = None
) -> ModerationResult:
    """Moderate an image, optionally combined with accompanying text."""
    
    model = genai.GenerativeModel(
        model_name="gemini-2.5-flash",
        system_instruction=MODERATION_SYSTEM_PROMPT
    )
    
    with open(image_path, "rb") as f:
        image_data = f.read()
    
    suffix = Path(image_path).suffix.lower()
    mime_map = {".jpg": "image/jpeg", ".jpeg": "image/jpeg",
                ".png": "image/png", ".webp": "image/webp",
                ".gif": "image/gif"}
    mime_type = mime_map.get(suffix, "image/jpeg")
    
    parts = [
        genai.types.Part.from_bytes(data=image_data, mime_type=mime_type),
    ]
    
    prompt_text = "Please review the following image."
    if context_text:
        prompt_text += f"\n\nCaption from the poster: {context_text}"
    parts.append(prompt_text)
    
    try:
        response = model.generate_content(
            parts,
            generation_config=genai.GenerationConfig(
                response_mime_type="application/json",
                temperature=0.1,
                max_output_tokens=512
            )
        )
        
        result_data = json.loads(response.text)
        decision = ModerationDecision(result_data["decision"])
        confidence = float(result_data.get("confidence", 0.5))
        
        return ModerationResult(
            decision=decision,
            confidence=confidence,
            categories=result_data.get("categories", []),
            reason=result_data.get("reason", ""),
            escalate_to_human=decision == ModerationDecision.REVIEW or confidence < 0.7
        )
    except Exception as e:
        return ModerationResult(
            decision=ModerationDecision.REVIEW,
            confidence=0.0,
            categories=["error"],
            reason=f"Processing error: {str(e)}",
            escalate_to_human=True
        )

Sending text and image together enables context-aware decisions. A photo of a kitchen knife is APPROVED on its own, but paired with the caption "how to hurt someone," the combined context warrants REJECTED. This contextual reasoning is Gemini API's greatest advantage for moderation use cases.

Building an Async Batch Processing System

Single synchronous calls will not scale in production. Here is an async worker using a Redis queue.

import asyncio
import redis.asyncio as aioredis
import json
import logging
from datetime import datetime
 
logger = logging.getLogger(__name__)
 
class ModerationWorker:
    """Async moderation worker with two-stage Flash to Pro escalation."""
    
    def __init__(self, redis_url: str, queue_name: str = "moderation_queue"):
        self.redis_url = redis_url
        self.queue_name = queue_name
        self.results_key = "moderation_results"
        self.flash_model = "gemini-2.5-flash"
        self.pro_model = "gemini-2.5-pro"
        self.pro_escalation_threshold = 0.3
    
    async def process_item(self, item: dict) -> dict:
        """Process a single queue item."""
        content_type = item.get("type", "text")
        content_id = item.get("id")
        
        # Stage 1: Flash for fast screening
        if content_type == "text":
            result = moderate_text(item["content"], self.flash_model)
        elif content_type == "image":
            result = moderate_image(item["path"], item.get("caption"))
        else:
            result = ModerationResult(
                decision=ModerationDecision.REVIEW,
                confidence=0.0,
                categories=["unknown_type"],
                reason="Unsupported content type",
                escalate_to_human=True
            )
        
        # Stage 2: Escalate borderline text to Pro
        if result.confidence < self.pro_escalation_threshold and content_type == "text":
            logger.info(
                f"[{content_id}] Flash confidence={result.confidence:.2f} -> escalating to Pro"
            )
            result = moderate_text(item["content"], self.pro_model)
        
        model_used = (self.pro_model
                      if result.confidence < self.pro_escalation_threshold
                      else self.flash_model)
        
        return {
            "id": content_id,
            "decision": result.decision.value,
            "confidence": result.confidence,
            "categories": result.categories,
            "reason": result.reason,
            "escalate_to_human": result.escalate_to_human,
            "processed_at": datetime.utcnow().isoformat(),
            "model_used": model_used
        }
    
    async def run(self, batch_size: int = 10):
        """Main worker loop."""
        redis = await aioredis.from_url(self.redis_url)
        logger.info(f"Worker started -- queue: {self.queue_name}")
        
        try:
            while True:
                items = await redis.lmpop(batch_size, self.queue_name, direction="left")
                
                if not items:
                    await asyncio.sleep(0.5)
                    continue
                
                _, raw_items = items
                tasks = [self.process_item(json.loads(raw)) for raw in raw_items]
                results = await asyncio.gather(*tasks, return_exceptions=True)
                
                pipeline = redis.pipeline()
                for result in results:
                    if isinstance(result, Exception):
                        logger.error(f"Processing error: {result}")
                        continue
                    pipeline.hset(self.results_key, result["id"], json.dumps(result))
                await pipeline.execute()
                
                logger.info(f"Processed {len(results)} items")
        finally:
            await redis.aclose()

asyncio.gather runs items in each batch in parallel, letting you maximize Gemini API throughput within your rate limits while keeping latency low.

Integrating Human-in-the-Loop

Full automation is the goal, but human judgment remains essential for edge cases. Here is a simple HiTL pattern with FastAPI.

from fastapi import FastAPI
from pydantic import BaseModel
 
app = FastAPI()
 
class ReviewDecision(BaseModel):
    content_id: str
    decision: str
    reviewer_id: str
    override_reason: str
 
@app.get("/review/queue")
async def get_review_queue(limit: int = 20):
    """Fetch items awaiting human review."""
    redis = await aioredis.from_url("redis://localhost")
    items = await redis.lrange("human_review_queue", 0, limit - 1)
    return [json.loads(item) for item in items]
 
@app.post("/review/decision")
async def submit_review_decision(decision: ReviewDecision):
    """Accept a human moderator verdict."""
    redis = await aioredis.from_url("redis://localhost")
    
    await redis.hset(
        "final_decisions",
        decision.content_id,
        json.dumps({
            "decision": decision.decision,
            "reviewer_id": decision.reviewer_id,
            "override_reason": decision.override_reason,
            "reviewed_at": datetime.utcnow().isoformat()
        })
    )
    
    # Accumulate as training signal for prompt improvements
    await redis.rpush(
        "training_feedback",
        json.dumps({
            "content_id": decision.content_id,
            "human_decision": decision.decision,
            "timestamp": datetime.utcnow().isoformat()
        })
    )
    
    return {"status": "ok", "content_id": decision.content_id}

Storing human decisions in training_feedback lets you periodically update the few-shot examples in your System Instructions, continuously improving model accuracy without any model retraining.

Optimizing Accuracy, Cost, and Throughput

Cost Optimization

Beyond the two-stage Flash to Pro approach, these strategies make a meaningful difference.

Caching: Cache moderation results for identical or near-identical content. Gemini API's Context Caching feature can dramatically cut the token cost of your System Instructions on repeated calls.

Content preprocessing: Resize images to a maximum of 1024x1024 before sending. Strip unnecessary whitespace and duplicates from text. These steps reduce token consumption without meaningfully affecting decision quality.

Batch sizing: Cap your worker batch size at roughly 80% of your per-minute request limit (RPM) to stay safely within rate limits.

For a deeper look at cost management, see the Gemini API Cost Optimization Complete Guide.

Improving Accuracy

Few-shot examples: Adding approved and rejected examples to your System Instructions can substantially boost accuracy. This is especially effective for industry-specific edge cases such as medication names on a healthcare platform.

Threshold tuning: The confidence < 0.7 escalation threshold should be reviewed monthly using accumulated logs to find the precision/recall balance that fits your platform.

Error Handling

For rate-limit error handling specifically, see Gemini API Error Handling and Rate Limit Patterns. The golden rule for moderation systems: always fail safe -- when in doubt, route to REVIEW rather than APPROVED.

Production Monitoring and Alerting

Track these metrics to keep your moderation system healthy.

from prometheus_client import Counter, Histogram, Gauge
import time
 
moderation_requests_total = Counter(
    "moderation_requests_total",
    "Total moderation requests",
    ["decision", "content_type", "model"]
)
moderation_latency_seconds = Histogram(
    "moderation_latency_seconds",
    "Moderation processing time in seconds",
    buckets=[0.5, 1.0, 2.0, 5.0, 10.0, 30.0]
)
human_review_queue_size = Gauge(
    "human_review_queue_size",
    "Number of items waiting for human review"
)
 
def moderate_with_metrics(content: str, content_type: str = "text") -> ModerationResult:
    """Moderation call with Prometheus instrumentation."""
    start = time.time()
    result = moderate_text(content)
    elapsed = time.time() - start
    
    moderation_requests_total.labels(
        decision=result.decision.value,
        content_type=content_type,
        model="gemini-2.5-flash"
    ).inc()
    
    moderation_latency_seconds.observe(elapsed)
    return result

Three metrics deserve the closest attention. REVIEW rate should sit at 15 to 25% of all decisions -- higher means thresholds are too strict, lower means you may be missing edge cases. Latency P95 should stay under 5 seconds; if it exceeds this, reduce batch sizes or add workers. Human review queue depth that grows consistently over time signals a need to tighten your automatic thresholds.

Summary

The heart of a Gemini API content moderation system is the combination of context-aware judgment and scalable async processing. Key takeaways from this guide:

Custom System Instructions let you define business-specific criteria in plain language
JSON mode (response_mime_type="application/json") guarantees structured responses every time
A Flash to Pro two-stage strategy optimizes both accuracy and cost
Async batch processing combined with Human-in-the-Loop handles production-scale volume
Prometheus metrics enable ongoing threshold tuning and system health monitoring

Content moderation is never set it and forget it. As your user base and platform evolve, so will the types of content you need to handle. Gemini API's flexible prompting and strong multilingual capabilities make it well-suited for building a moderation foundation that keeps pace with your product's growth.

Thank You for Reading

Gemini Lab is ad-free, supported entirely by members like you. We publish practical guides daily with implementation code, benchmarks, and production-ready patterns. If you've found it useful, we'd love to have you on board.