OpenAI ile Otomatik İçerik Moderasyonu Sistemi Nasıl Kurulur

Moderasyon sistemleri kurarken her zaman aynı soruyla karşılaşırsınız: “Kural tabanlı mı yazalım, ML modeli mi kullanalım?” Kural tabanlı sistemler bakımı zor, ML modelleri eğitmesi pahalı. OpenAI’nin Moderation API’si ve GPT modelleri bu ikisi arasında son derece pratik bir orta yol sunuyor. Geçen ay e-ticaret platformu kuran bir müşteri için kurduğumuz sistemin notlarını derledim; umarım işe yarar.

Mimari Kararlar: Neden İki Katmanlı?

Tek bir API çağrısıyla her şeyi çözmeye çalışmak cazip ama gerçek trafikte pahalıya mal olur. Bizim sistemimiz şu şekilde çalışıyor:

Birinci katman: OpenAI’nin ücretsiz Moderation API’si. Hızlı, ucuz, standart kategorileri yakalıyor.

İkinci katman: GPT-4o-mini ile özelleştirilmiş değerlendirme. Sadece birinci katmandan geçen şüpheli içerikler için çalışıyor.

Bu yaklaşımla maliyet %60-70 düştü. Çünkü açık spam ve hakaret zaten Moderation API’de takılıyor, pahalı GPT çağrısına gerek kalmıyor.

Ortam Kurulumu

# Python sanal ortamı oluştur
python3 -m venv moderation-env
source moderation-env/bin/activate

# Gerekli paketler
pip install openai redis fastapi uvicorn pydantic python-dotenv httpx

# Versiyon sabitleme (production'da kritik)
pip freeze > requirements.txt

.env dosyası:

OPENAI_API_KEY=sk-proj-xxxxxxxxxxxx
REDIS_URL=redis://localhost:6379
MODERATION_THRESHOLD=0.7
GPT_MODEL=gpt-4o-mini
MAX_RETRIES=3
RATE_LIMIT_PER_MINUTE=100

Temel Moderation API Entegrasyonu

Önce basit bir wrapper yazalım. OpenAI’nin döndürdüğü skor yapısını anlamak önemli:

import os
import asyncio
from openai import AsyncOpenAI
from dataclasses import dataclass
from typing import Optional
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))

@dataclass
class ModerationResult:
    flagged: bool
    categories: dict
    scores: dict
    action: str  # "allow", "review", "block"
    reason: Optional[str] = None

async def run_moderation_api(content: str) -> ModerationResult:
    """
    OpenAI Moderation API ile ilk katman kontrol.
    Bu endpoint ücretsiz, agresif kullanabilirsiniz.
    """
    try:
        response = await client.moderations.create(
            input=content,
            model="omni-moderation-latest"
        )
        
        result = response.results[0]
        scores = result.category_scores.__dict__
        categories = result.categories.__dict__
        
        # Eşik değeri kontrolü
        threshold = float(os.getenv("MODERATION_THRESHOLD", 0.7))
        high_scores = {k: v for k, v in scores.items() if v > threshold}
        
        if result.flagged:
            action = "block"
            reason = f"Moderation API işaretledi: {list(categories.keys())}"
        elif high_scores:
            action = "review"  # İkinci katmana gönder
            reason = f"Yüksek skor kategorileri: {high_scores}"
        else:
            action = "allow"
            reason = None
            
        return ModerationResult(
            flagged=result.flagged,
            categories=categories,
            scores=scores,
            action=action,
            reason=reason
        )
        
    except Exception as e:
        logger.error(f"Moderation API hatası: {e}")
        # Güvenli taraf: hata durumunda review'a gönder
        return ModerationResult(
            flagged=False,
            categories={},
            scores={},
            action="review",
            reason=f"API hatası: {str(e)}"
        )

İkinci Katman: GPT ile Bağlam Analizi

Moderation API bağlamı anlamıyor. “Müşteri beni öldürsün ama bu ürün harika” gibi bir cümleyi yanlış işaretleyebilir. GPT burada devreye giriyor:

import json
from openai import AsyncOpenAI

SYSTEM_PROMPT = """Sen bir e-ticaret platformu için içerik moderasyon uzmanısın.
Türkçe ve İngilizce içerikleri değerlendiriyorsun.

Değerlendirme kriterlerin:
1. Ürün yorumları samimi şikayet mi, hakaret mi?
2. Müşteri hizmetleri mesajları tehdit içeriyor mu?
3. Kullanıcı profil bilgileri uygunsuz mu?
4. Bağlamı doğru yorumla - "berbat ürün" ile "sizi mahvedeceğim" arasındaki farkı anla.

Yanıtını SADECE JSON formatında ver:
{
    "decision": "allow|block|review",
    "confidence": 0.0-1.0,
    "reason": "kısa açıklama",
    "suggested_action": "açıklama veya düzenleme önerisi"
}"""

async def gpt_content_analysis(
    content: str, 
    context: dict,
    initial_scores: dict
) -> dict:
    """
    GPT ile derinlemesine içerik analizi.
    Sadece şüpheli içerikler için çalıştır.
    """
    
    user_message = f"""İçerik: {content}

Bağlam bilgisi:
- İçerik türü: {context.get('content_type', 'bilinmiyor')}
- Kullanıcı geçmişi: {context.get('user_history', 'ilk kez')}
- Platform bölümü: {context.get('platform_section', 'genel')}

İlk tarama skorları (yüksek değerler dikkat çekti):
{json.dumps(initial_scores, ensure_ascii=False, indent=2)}

Bu içeriği değerlendir."""

    try:
        response = await client.chat.completions.create(
            model=os.getenv("GPT_MODEL", "gpt-4o-mini"),
            messages=[
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": user_message}
            ],
            temperature=0.1,  # Tutarlılık için düşük tutuyoruz
            max_tokens=300,
            response_format={"type": "json_object"}
        )
        
        result = json.loads(response.choices[0].message.content)
        logger.info(f"GPT kararı: {result['decision']} (güven: {result['confidence']})")
        return result
        
    except json.JSONDecodeError as e:
        logger.error(f"GPT JSON parse hatası: {e}")
        return {"decision": "review", "confidence": 0.5, "reason": "Parse hatası"}
    except Exception as e:
        logger.error(f"GPT analiz hatası: {e}")
        return {"decision": "review", "confidence": 0.0, "reason": str(e)}

Redis ile Rate Limiting ve Cache

Production’da aynı içeriği tekrar tekrar değerlendirmek hem pahalı hem de gereksiz. Redis’te sonuçları cache’liyoruz:

import redis.asyncio as redis
import hashlib
import json
from typing import Optional

redis_client = redis.from_url(os.getenv("REDIS_URL", "redis://localhost:6379"))

async def get_cached_result(content: str) -> Optional[dict]:
    """İçerik hash'i ile önce cache'e bak."""
    content_hash = hashlib.sha256(content.encode()).hexdigest()
    cached = await redis_client.get(f"moderation:{content_hash}")
    
    if cached:
        logger.info(f"Cache hit: {content_hash[:8]}...")
        return json.loads(cached)
    return None

async def cache_result(content: str, result: dict, ttl: int = 3600):
    """Sonucu 1 saat cache'le."""
    content_hash = hashlib.sha256(content.encode()).hexdigest()
    await redis_client.setex(
        f"moderation:{content_hash}",
        ttl,
        json.dumps(result, ensure_ascii=False)
    )

async def check_rate_limit(user_id: str) -> bool:
    """
    Kullanıcı başına dakikada maksimum istek kontrolü.
    Aynı kullanıcı spam içerik göndermeye çalışıyorsa erken yakala.
    """
    key = f"rate_limit:{user_id}"
    limit = int(os.getenv("RATE_LIMIT_PER_MINUTE", 100))
    
    pipe = redis_client.pipeline()
    pipe.incr(key)
    pipe.expire(key, 60)
    results = await pipe.execute()
    
    request_count = results[0]
    
    if request_count > limit:
        logger.warning(f"Rate limit aşıldı: {user_id} ({request_count} istek/dk)")
        return False
    return True

Ana Orkestrasyon Servisi

İki katmanı birleştiren ana servis:

import time
from typing import Optional

async def moderate_content(
    content: str,
    user_id: str,
    context: Optional[dict] = None
) -> dict:
    """
    Ana moderasyon fonksiyonu.
    Rate limit -> Cache -> Moderation API -> GPT (gerekirse)
    """
    start_time = time.time()
    context = context or {}
    
    # Rate limit kontrolü
    if not await check_rate_limit(user_id):
        return {
            "decision": "block",
            "reason": "Rate limit aşıldı",
            "processing_time": 0,
            "layers_used": []
        }
    
    # Cache kontrolü
    cached = await get_cached_result(content)
    if cached:
        cached["from_cache"] = True
        return cached
    
    # Katman 1: Moderation API
    mod_result = await run_moderation_api(content)
    layers_used = ["moderation_api"]
    
    final_decision = mod_result.action
    final_reason = mod_result.reason
    confidence = 1.0 if mod_result.flagged else 0.5
    
    # Katman 2: GPT analizi (sadece review olanlar için)
    if mod_result.action == "review":
        high_scores = {k: v for k, v in mod_result.scores.items() if v > 0.3}
        gpt_result = await gpt_content_analysis(content, context, high_scores)
        
        layers_used.append("gpt_analysis")
        final_decision = gpt_result.get("decision", "review")
        final_reason = gpt_result.get("reason", "")
        confidence = gpt_result.get("confidence", 0.5)
    
    processing_time = round((time.time() - start_time) * 1000, 2)
    
    result = {
        "decision": final_decision,
        "reason": final_reason,
        "confidence": confidence,
        "layers_used": layers_used,
        "processing_time_ms": processing_time,
        "from_cache": False,
        "moderation_scores": mod_result.scores
    }
    
    # Sonucu cache'le (block kararlarını daha uzun tut)
    ttl = 7200 if final_decision == "block" else 3600
    await cache_result(content, result, ttl)
    
    logger.info(
        f"Karar: {final_decision} | "
        f"Katmanlar: {layers_used} | "
        f"Süre: {processing_time}ms"
    )
    
    return result

FastAPI ile REST Endpoint

Bunu bir servise dönüştürelim:

from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel, field_validator
import uvicorn

app = FastAPI(title="İçerik Moderasyon Servisi", version="1.0.0")

class ModerationRequest(BaseModel):
    content: str
    user_id: str
    content_type: str = "comment"  # comment, message, profile, review
    platform_section: str = "general"
    
    @field_validator("content")
    @classmethod
    def content_must_not_be_empty(cls, v):
        if not v.strip():
            raise ValueError("İçerik boş olamaz")
        if len(v) > 10000:
            raise ValueError("İçerik 10.000 karakteri geçemez")
        return v.strip()

class ModerationResponse(BaseModel):
    decision: str
    reason: str
    confidence: float
    processing_time_ms: float
    from_cache: bool

@app.post("/moderate", response_model=ModerationResponse)
async def moderate_endpoint(
    request: ModerationRequest,
    background_tasks: BackgroundTasks
):
    context = {
        "content_type": request.content_type,
        "platform_section": request.platform_section
    }
    
    result = await moderate_content(
        content=request.content,
        user_id=request.user_id,
        context=context
    )
    
    # Block kararlarını arka planda logla
    if result["decision"] == "block":
        background_tasks.add_task(
            log_blocked_content,
            request.content,
            request.user_id,
            result
        )
    
    return ModerationResponse(**result)

@app.get("/health")
async def health_check():
    try:
        await redis_client.ping()
        return {"status": "healthy", "redis": "connected"}
    except Exception as e:
        raise HTTPException(status_code=503, detail=f"Redis bağlantı hatası: {e}")

async def log_blocked_content(content: str, user_id: str, result: dict):
    """Engellenen içerikleri ayrı bir Redis listesine yaz."""
    log_entry = {
        "user_id": user_id,
        "content_preview": content[:100] + "..." if len(content) > 100 else content,
        "reason": result.get("reason"),
        "timestamp": time.time()
    }
    await redis_client.lpush("blocked_content_log", json.dumps(log_entry))
    await redis_client.ltrim("blocked_content_log", 0, 9999)  # Son 10.000 kaydı tut

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000, reload=False)

Systemd Servis Olarak Çalıştırma

Geliştirme bitti, production’a taşıyalım:

# Servis dosyasını oluştur
cat > /etc/systemd/system/content-moderation.service << 'EOF'
[Unit]
Description=OpenAI Content Moderation Service
After=network.target redis.service
Wants=redis.service

[Service]
Type=simple
User=www-data
Group=www-data
WorkingDirectory=/opt/content-moderation
Environment="PATH=/opt/content-moderation/moderation-env/bin"
EnvironmentFile=/opt/content-moderation/.env
ExecStart=/opt/content-moderation/moderation-env/bin/uvicorn 
    main:app 
    --host 0.0.0.0 
    --port 8000 
    --workers 4 
    --log-level info 
    --access-log
Restart=always
RestartSec=5
StandardOutput=journal
StandardError=journal
SyslogIdentifier=content-moderation

# Güvenlik kısıtlamaları
NoNewPrivileges=true
ProtectSystem=strict
PrivateTmp=true
ReadWritePaths=/opt/content-moderation/logs

[Install]
WantedBy=multi-user.target
EOF

# Servisi etkinleştir ve başlat
systemctl daemon-reload
systemctl enable content-moderation
systemctl start content-moderation

# Durumu kontrol et
systemctl status content-moderation
journalctl -u content-moderation -f

İzleme ve Alert Sistemi

Moderasyon kararlarını izlemek, sistemin ne kadar doğru çalıştığını anlamak için kritik:

# Prometheus metrikleri için basit bir script
cat > /opt/content-moderation/metrics_exporter.sh << 'EOF'
#!/bin/bash
# Her 5 dakikada çalıştır, blocked içerik sayısını izle

REDIS_CLI="redis-cli"
LOG_FILE="/var/log/moderation-metrics.log"
ALERT_THRESHOLD=50  # 5 dakikada 50'den fazla block varsa alert

BLOCKED_COUNT=$($REDIS_CLI LLEN blocked_content_log)
TIMESTAMP=$(date '+%Y-%m-%d %H:%M:%S')

echo "$TIMESTAMP | Toplam blocked: $BLOCKED_COUNT" >> $LOG_FILE

# Son 5 dakikadaki blokları say (basit yaklaşım)
RECENT_BLOCKS=$($REDIS_CLI LRANGE blocked_content_log 0 99 | 
    python3 -c "
import sys, json, time
entries = [json.loads(l) for l in sys.stdin if l.strip()]
cutoff = time.time() - 300
recent = [e for e in entries if e.get('timestamp', 0) > cutoff]
print(len(recent))
")

if [ "$RECENT_BLOCKS" -gt "$ALERT_THRESHOLD" ]; then
    echo "ALERT: Son 5 dakikada $RECENT_BLOCKS içerik engellendi!" | 
        mail -s "Moderasyon Alert" [email protected]
fi
EOF

chmod +x /opt/content-moderation/metrics_exporter.sh

# Cron'a ekle
echo "*/5 * * * * /opt/content-moderation/metrics_exporter.sh" | crontab -

Gerçek Dünya Notları

Bu sistemi birkaç farklı müşteride çalıştırırken öğrendiğim bazı şeyler var.

Türkçe içerik için özel dikkat: Moderation API İngilizce’ye göre optimize edilmiş. Türkçe’de “sikmek” fiili gerçekten hakaret olabilir ya da tamamen masum bir bağlamda kullanılabilir. GPT katmanı bu nüansı yakalıyor ama system prompt’u dikkatlice yazmanız gerekiyor. “Türkçe argo ve idiomlara dikkat et” gibi bir uyarı eklemek işe yarıyor.

Yanlış pozitif oranı: Birinci iterasyonda yanlış pozitif oranımız %8’di. Threshold’u 0.8’e çıkarınca %3’e düştü ama bazı gerçek sorunlu içerikler kaçmaya başladı. 0.7 iyi bir denge noktası gibi görünüyor, ancak platform türüne göre ayarlamanız gerekiyor. Çocuklara yönelik platform ile kurumsal B2B arasında büyük fark var.

Maliyet tahmini: Günde 10.000 içerik değerlendiriyorsanız, Moderation API ücretsiz olduğu için sadece GPT çağrıları maliyetli. Cache hit oranınız %40-50 civarındaysa aylık maliyet 15-20 dolar civarında kalıyor. İlk ay cache soğuk olacak, maliyetler normalin 2-3 katı olabilir.

Hata yönetimi önemli: OpenAI API zaman zaman timeout atıyor. Retry mekanizması olmadan production’da kötü sürprizlerle karşılaşırsınız. tenacity kütüphanesi bu iş için biçilmiş kaftan.

İnsan incelemesi sıfırlanamaz: GPT ne kadar iyi olursa olsun, “review” kararı verilen içerikleri bir admin panelinde manuel olarak onaylamanız gerekiyor. Özellikle hukuki riskler söz konusu olduğunda. Bu sistemi “insan denetçiyi tamamen ortadan kaldırır” diye sunmak yanlış olur.

Sonuç

Bu sistem doğru kurulduğunda, kural tabanlı filtrelere göre çok daha akıllı kararlar veriyor ve ML modeli eğitmeye göre çok daha az başlangıç yatırımı gerektiriyor. İki katmanlı yapı maliyet açısından kritik; Moderation API’nin ücretsiz olmasından maksimum faydalanıp GPT’yi sadece belirsiz vakalar için kullanmak doğru strateji.

Sistemin en zayıf halkası hala insan gözden geçirmesi gerektiren edge case’ler. Türkçe bağlamı anlayan, platformunuza özel fine-tuned bir model uzun vadede çok daha iyi sonuç verir. Ancak oraya ulaşana kadar bu mimari son derece solid bir başlangıç noktası. Kodu olduğu gibi production’a almayın; kendi ortamınıza, trafik yoğunluğunuza ve güvenlik gereksinimlerinize göre uyarlayın.