Масштабирование модерации контента

Схема модерации, которая справляется с 1 000 сообщений в день, ломается на миллионе: начинают мешать ограничения по частоте запросов, задержка каждого вызова накапливается, а расходы раздуваются. В этом руководстве показано, как встроить модерацию Discuse в высоконагруженные пайплайны — когда блокировать, когда проверять в фоне, что кэшировать и как обходить пиковую нагрузку.

Что ломается, когда модерация масштабируется?

У каждой проблемы масштабирования есть конкретная причина и конкретное решение:

Проблема	Влияние	Решение
Рост объёма	Лимиты API, увеличение затрат	Асинхронная обработка, кэширование
Требования к задержке	Плохой пользовательский опыт	Предмодерация, очереди
Управление затратами	Бюджетные ограничения	Умная маршрутизация, кэширование
Согласованность	Неравномерное применение правил	Централизованная конфигурация
Обработка пиковых нагрузок	Снижение качества работы сервиса	Автомасштабирование, очереди

Какой архитектурный паттерн выбрать?

Паттерн 1: Синхронная премодерация

Лучше всего подходит для: небольшого объёма контента с высокими рисками (платежи, юридические материалы)

User Input → API Call → Moderation → Decision → Response
                ↓
           (Blocking call)

// Simple synchronous flow
async function createPost(content) {
  // Check moderation before saving
  const result = await checkModeration(content);

  if (result.has_violations) {
    throw new ModerationError(result.message);
  }

  // Safe to publish
  return await savePost(content);
}

Плюсы: простота, мгновенная обратная связь Минусы: увеличивает задержку, ограничивает пропускную способность

Паттерн 2: Асинхронная постмодерация

Лучше всего подходит для: большого объёма контента с более низкими рисками (комментарии, сообщения)

User Input → Save (Pending) → Publish → Background Check → Action
                                              ↓
                                    ┌────────┴────────┐
                                    ↓                 ↓
                                  Safe            Violation
                                    ↓                 ↓
                                  Keep             Remove

// Async flow with queue
async function createPost(content) {
  // Save immediately with pending status
  const post = await savePost(content, { status: 'pending' });

  // Queue for async moderation
  await moderationQueue.add('check-content', {
    postId: post.id,
    content: content
  });

  return post;
}

// Background worker
moderationQueue.process('check-content', async (job) => {
  const result = await checkModeration(job.data.content);

  if (result.has_violations) {
    await removePost(job.data.postId);
    await notifyUser(job.data.postId, 'content_removed');
  } else {
    await updatePost(job.data.postId, { status: 'approved' });
  }
});

Плюсы: не блокирует выполнение, справляется с большим объёмом Минусы: вредоносный контент некоторое время остаётся видимым

Паттерн 3: Многоуровневая модерация

Лучше всего подходит для: сбалансированного подхода с оптимизацией затрат

User Input → Quick Check → High Confidence? ──Yes──► Auto-decision
                              │
                              No
                              ↓
                         Full Analysis ──► Decision

async function tieredModeration(content) {
  // Tier 1: Fast local checks (regex, blocklists)
  const localResult = await quickLocalCheck(content);
  if (localResult.definiteViolation) {
    return { action: 'block', source: 'local' };
  }

  // Tier 2: Cached API results
  const cacheKey = hashContent(content);
  const cached = await cache.get(cacheKey);
  if (cached) {
    return { action: cached.action, source: 'cache' };
  }

  // Tier 3: Full API check
  const apiResult = await checkModeration(content);
  await cache.set(cacheKey, apiResult, TTL);

  return { action: determineAction(apiResult), source: 'api' };
}

Паттерн 4: Разветвление с агрегацией

Лучше всего подходит для: нескольких типов проверок с независимым масштабированием

                    ┌──► Text Check ──┐
                    │                  │
User Input ──► Router ──► Image Check ──► Aggregator ──► Decision
                    │                  │
                    └──► Link Check ──┘

async function parallelModeration(content) {
  const checks = [];

  if (content.text) {
    checks.push(checkText(content.text));
  }

  if (content.images?.length) {
    checks.push(checkImages(content.images));
  }

  if (content.links?.length) {
    checks.push(checkLinks(content.links));
  }

  // Run all checks in parallel
  const results = await Promise.all(checks);

  // Aggregate results
  return aggregateResults(results);
}

function aggregateResults(results) {
  // Most severe result wins
  const hasViolation = results.some(r => r.has_violations);
  const maxScore = Math.max(...results.map(r => r.max_score || 0));

  return {
    has_violations: hasViolation,
    max_score: maxScore,
    details: results
  };
}

Стратегии кэширования

Кэширование на основе содержимого

Кэшируйте результаты для идентичного содержимого:

const cache = new Redis();

async function checkWithCache(content) {
  const hash = crypto.createHash('sha256')
    .update(content.text || '')
    .update(JSON.stringify(content.images || []))
    .digest('hex');

  // Check cache first
  const cached = await cache.get(`mod:${hash}`);
  if (cached) {
    return { ...JSON.parse(cached), cached: true };
  }

  // API call
  const result = await callModerationAPI(content);

  // Cache for 24 hours
  await cache.setex(`mod:${hash}`, 86400, JSON.stringify(result));

  return { ...result, cached: false };
}

Кэширование на основе пользователя

Кэшируйте решения для пользователей, которые повторно нарушают правила:

async function checkUserHistory(userId) {
  const recentViolations = await cache.get(`violations:${userId}`);

  if (recentViolations > 5) {
    // Flag for enhanced scrutiny
    return { enhancedModeration: true };
  }

  return { enhancedModeration: false };
}

async function recordViolation(userId) {
  await cache.incr(`violations:${userId}`);
  await cache.expire(`violations:${userId}`, 86400 * 7); // 7 days
}

Управление очередями

Приоритетные очереди

Обрабатывайте разные типы контента с разными приоритетами:

const queues = {
  critical: new Queue('moderation-critical'),  // Reports, appeals
  high: new Queue('moderation-high'),          // Public posts
  normal: new Queue('moderation-normal'),      // Comments, DMs
  low: new Queue('moderation-low')             // Profile updates
};

async function queueForModeration(content, priority = 'normal') {
  const queue = queues[priority] || queues.normal;

  return queue.add('check', {
    contentId: content.id,
    type: content.type,
    data: content
  }, {
    priority: getPriorityNumber(priority),
    timeout: getTimeout(priority)
  });
}

Соблюдение лимитов запросов

У каждого API-ключа есть лимит запросов в минуту; при его превышении API отклоняет вызов с ошибкой "rate limit exceeded". Ограничивайте частоту запросов на своей стороне, чтобы оставаться в рамках лимита и не расходовать квоту на лишние вызовы. Установите для ограничителя максимальное число запросов в минуту в соответствии с RPM, настроенным для вашего ключа:

const Bottleneck = require('bottleneck');

const limiter = new Bottleneck({
  maxConcurrent: 50,                   // Cap parallel requests
  minTime: 20,                         // Min 20ms between requests
  reservoir: YOUR_KEY_RPM,             // Match your API key's per-minute limit
  reservoirRefreshAmount: YOUR_KEY_RPM,
  reservoirRefreshInterval: 60 * 1000  // Refill each minute
});

async function rateLimitedCheck(content) {
  return limiter.schedule(() => checkModeration(content));
}

Квота не связана с лимитом запросов: каждый вызов учитывается в месячном лимите API-запросов вашего тарифа, а объект usage в ответе сообщает api_requests_used, api_requests_limit и api_requests_remaining, чтобы вы могли отслеживать запас.

Очереди недоставленных задач

Обрабатывайте неудачные попытки модерации:

moderationQueue.process('check-content', async (job) => {
  try {
    const result = await checkModeration(job.data.content);
    await applyDecision(job.data.contentId, result);
  } catch (error) {
    if (job.attemptsMade < 3) {
      throw error; // Retry
    }

    // Move to dead letter queue after 3 attempts
    await deadLetterQueue.add('failed-moderation', {
      ...job.data,
      error: error.message,
      attempts: job.attemptsMade
    });

    // Apply safe default (hold for review)
    await holdForManualReview(job.data.contentId);
  }
});

Оптимизация затрат

Умная маршрутизация

Вызывайте платные APIs только тогда, когда это действительно необходимо:

async function smartModeration(content) {
  // Step 1: Free local checks
  const localResult = runLocalFilters(content);
  if (localResult.isDefinitelySpam) {
    return { action: 'block', cost: 0 };
  }

  // Step 2: Check cache (free)
  const cached = await getCachedResult(content);
  if (cached) {
    return { action: cached.action, cost: 0 };
  }

  // Step 3: Risk-based API call
  const riskScore = calculateRiskScore(content, localResult);

  if (riskScore < 0.2) {
    // Low risk: approve without API call
    return { action: 'approve', cost: 0 };
  }

  // Step 4: API call for uncertain content
  const apiResult = await checkModeration(content);
  return {
    action: determineAction(apiResult),
    cost: calculateApiCost(content)
  };
}

Контроль параллельности

Эндпоинт /api/v2/check оценивает одну отправку за один вызов. Чтобы выполнять работу «пакетно», группируйте элементы в воркере и запускайте их параллельно в пределах ограниченного пула, а не рассчитывайте на один запрос с несколькими элементами. Проверка, включающая несколько медиа-URL, сама по себе считается одним вызовом и охватывает их все — один вызов с большим количеством изображений обходится дешевле, чем множество вызовов по одному изображению.

// Bounded concurrency: many items, capped parallel calls.
async function processBatch(items, concurrency = 10) {
  const results = [];
  for (let i = 0; i < items.length; i += concurrency) {
    const slice = items.slice(i, i + concurrency);
    const settled = await Promise.allSettled(
      slice.map(item => checkModeration(item.content))
    );
    results.push(...settled);
  }
  return results;
}

Один запрос может содержать до 10 URL изображений, 5 URL GIF, 3 URL видео, 5 URL документов и 20 ссылок, а также текст длиной до 10 000 символов — поэтому объединяйте медиа одной отправки в один вызов, а не разбивайте их на несколько.

Мониторинг и оповещения

Дашборд ключевых метрик

const metrics = {
  // Volume metrics
  total_requests: new Counter('moderation_requests_total'),
  requests_per_second: new Gauge('moderation_rps'),

  // Latency metrics
  latency_p50: new Histogram('moderation_latency_p50'),
  latency_p99: new Histogram('moderation_latency_p99'),

  // Queue metrics
  queue_depth: new Gauge('moderation_queue_depth'),
  processing_time: new Histogram('moderation_processing_time'),

  // Cost metrics
  api_calls: new Counter('moderation_api_calls'),
  estimated_cost: new Counter('moderation_estimated_cost'),

  // Accuracy metrics
  violations_detected: new Counter('moderation_violations'),
  false_positives: new Counter('moderation_false_positives')
};

Правила оповещения

# Alert on queue backup
- alert: ModerationQueueBackup
  expr: moderation_queue_depth > 10000
  for: 5m
  labels:
    severity: warning
  annotations:
    summary: Moderation queue backing up

# Alert on high latency
- alert: ModerationLatencyHigh
  expr: moderation_latency_p99 > 5000
  for: 2m
  labels:
    severity: critical
  annotations:
    summary: Moderation latency exceeding 5 seconds

# Alert on high error rate
- alert: ModerationErrorRate
  expr: rate(moderation_errors_total[5m]) / rate(moderation_requests_total[5m]) > 0.01
  for: 5m
  labels:
    severity: critical
  annotations:
    summary: Moderation error rate above 1%

Масштабирование инфраструктуры

Горизонтальное масштабирование

# Kubernetes HPA for moderation workers
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: moderation-worker
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: moderation-worker
  minReplicas: 3
  maxReplicas: 50
  metrics:
  - type: External
    external:
      metric:
        name: queue_depth
      target:
        type: AverageValue
        averageValue: 100

Снижение задержек для пользователей по всему миру

API Discuse доступен по единому базовому URL, https://api.discuse.com. Чтобы обеспечить низкую задержку для глобального трафика, запускайте свои воркеры модерации ближе к пользователям и позволяйте им обращаться к API, вместо того чтобы выполнять синхронный вызов из пользовательского маршрута запроса в удалённом регионе. Сочетайте это с описанными выше паттернами асинхронной постмодерации и кэширования, чтобы медленный обмен данными туда-обратно никогда не блокировал пользователя.

Краткий обзор рекомендаций

1. Активно используйте кэширование: для одинакового содержимого возвращается результат из кэша. API также устраняет дублирование одинакового содержимого в рамках одного проекта в течение короткого периода и возвращает cached: true.
2. Используйте очереди: асинхронная обработка для больших объёмов.
3. Внедрите уровни обработки: локально → кэш → API.
4. Отслеживайте всё: глубину очереди, задержку и расход квоты.
5. Продумайте отказоустойчивость: очереди недоставленных сообщений и безопасные резервные сценарии.
6. Масштабируйтесь горизонтально: больше воркеров, а не более мощные машины.
7. Оптимизируйте затраты: умная маршрутизация, меньше лишних повторных вызовов и объединение медиа в одном запросе.

Следующие шаги

• Руководство по модерации контента с помощью AI - Разберитесь, как работает модерация с AI
• Настройка пороговых значений - Точная настройка обнаружения
• Краткое руководство по началу работы - Базовая реализация