Gemini API で構築するコンテンツモデレーション本番システム

from pathlib import Path
 
def moderate_image(
    image_path: str,
    context_text: Optional[str] = None
) -> ModerationResult:
    """画像（およびオプションの説明文）をモデレーションする"""
    
    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 = "以下の画像コンテンツを審査してください。"
    if context_text:
        prompt_text += f"\n\n投稿者のテキスト説明: {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"処理エラー: {str(e)}",
            escalate_to_human=True
        )

import asyncio
import redis.asyncio as aioredis
import json
import logging
from datetime import datetime
 
logger = logging.getLogger(__name__)
 
class ModerationWorker:
    """非同期モデレーション Worker（二段階 Flash -> Pro 昇格対応）"""
    
    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:
        """キューのアイテムを処理する"""
        content_type = item.get("type", "text")
        content_id = item.get("id")
        
        # 第1段階: Flash で高速スクリーニング
        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="未対応コンテンツタイプ",
                escalate_to_human=True
            )
        
        # 第2段階: 境界ケースは Pro で精査
        if result.confidence < self.pro_escalation_threshold and content_type == "text":
            logger.info(f"[{content_id}] Flash confidence={result.confidence:.2f} -> 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):
        """メインワーカーループ"""
        redis = await aioredis.from_url(self.redis_url)
        logger.info(f"Worker 起動 — キュー: {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)
                
                # 結果を Redis に保存
                pipeline = redis.pipeline()
                for result in results:
                    if isinstance(result, Exception):
                        logger.error(f"処理エラー: {result}")
                        continue
                    pipeline.hset(
                        self.results_key,
                        result["id"],
                        json.dumps(result)
                    )
                await pipeline.execute()
                
                logger.info(f"{len(results)} 件処理完了")
        finally:
            await redis.aclose()

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):
    """人間レビュー待ちのコンテンツ一覧を取得"""
    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):
    """人間モデレーターの判定を受け付ける"""
    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()
        })
    )
    
    # 教師データとして蓄積（将来のプロンプト改善に活用）
    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}

from prometheus_client import Counter, Histogram, Gauge
import time
 
moderation_requests_total = Counter(
    "moderation_requests_total",
    "モデレーションリクエスト総数",
    ["decision", "content_type", "model"]
)
moderation_latency_seconds = Histogram(
    "moderation_latency_seconds",
    "モデレーション処理時間（秒）",
    buckets=[0.5, 1.0, 2.0, 5.0, 10.0, 30.0]
)
human_review_queue_size = Gauge(
    "human_review_queue_size",
    "人間レビュー待ちキューサイズ"
)
 
def moderate_with_metrics(content: str, content_type: str = "text") -> ModerationResult:
    """メトリクス付きモデレーション"""
    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