Архитектура корпоративного антитоксичного ИИ анализатора

⚙️ Архитектура пайплайна

Каждое сообщение проходит через 7 этапов менее чем за 300 мс. Исходный текст уничтожается после векторизации — хранятся только обезличенные метрики.

🧪 Интерактивный прототип

Попробуй ввести любое сообщение — анализатор в реальном времени покажет: векторный эмбеддинг, вероятности классов и итоговый вердикт.
(В браузере используется упрощённая модель на TensorFlow.js)

🐍 Реализация на Python (FastAPI + ONNX)

Вот как выглядит настоящий сервис-анализатор в продакшене. Обрати внимание на отсутствие сохранения текста — только векторы и метрики.

# main.py — Streaming токсичный анализатор
import numpy as np
from fastapi import FastAPI
from pydantic import BaseModel
import onnxruntime as ort
import redis
import re
import hashlib

app = FastAPI()
redis_client = redis.Redis(host='localhost', decode_responses=True)

# Загружаем модели (замороженные веса)
embedder = ort.InferenceSession('distilbert_embedding.onnx')
classifier = ort.InferenceSession('toxic_classifier.onnx')

class Message(BaseModel):
    text: str
    user_id: str
    channel: str

def preprocess(text: str) -> str:
    # Маскировка PII (имена, email, телефоны)
    text = re.sub(r'\b[A-Z][a-z]+ [A-Z][a-z]+\b', '[PERSON]', text)
    text = re.sub(r'\S+@\S+', '[EMAIL]', text)
    return text.lower().strip()

def get_user_context(user_id: str) -> dict:
    # Счетчики из Redis, а не из базы переписок
    return {
        'toxicity_history': float(redis_client.get(f'tox:{user_id}') or 0.0),
        'msg_count_hour': int(redis_client.get(f'msgs:{user_id}') or 0),
        'is_manager': redis_client.sismember('managers', user_id)
    }

@app.post('/analyze')
def analyze(msg: Message):
    # 1. Предобработка (текст уничтожается после этого этапа)
    clean_text = preprocess(msg.text)
    
    # 2. Векторизация → эмбеддинг (384 числа)
    embedding = embedder.run(None, {'input': np.array([clean_text])})[0]
    
    # 3. Классификация
    logits = classifier.run(None, {'embedding': embedding})[0]
    probs = np.exp(logits) / np.sum(np.exp(logits))  # softmax
    
    # 4. Контекстный корректировщик
    ctx = get_user_context(msg.user_id)
    # Если руководитель — снижаем порог чувствительности на 15%
    penalty = 0.85 if ctx['is_manager'] else 1.0
    toxic_score = float(probs[1] * penalty)  # индекс 1 = класс "токсично"
    
    # 5. Роутинг
    if toxic_score >= 0.85:
        verdict = 'BLOCKED'
        # Только в этом случае сохраняем хеш текста на 72 часа
        text_hash = hashlib.sha256(msg.text.encode()).hexdigest()
        redis_client.setex(f'blocked:{text_hash}', 259200, msg.text)
    else:
        verdict = 'ALLOWED'
    
    # 6. Обновляем счетчики в Redis (без текста)
    redis_client.incr(f'msgs:{msg.user_id}')
    redis_client.expire(f'msgs:{msg.user_id}', 3600)
    if toxic_score > 0.5:
        redis_client.incrbyfloat(f'tox:{msg.user_id}', 0.1)
    
    # 7. Логгируем только обезличенные метрики
    return {
        'verdict': verdict,
        'toxic_score': round(toxic_score, 4),
        'class_probs': {
            'safe': round(float(probs[0]), 4),
            'aggression': round(float(probs[1]), 4),
            'discrimination': round(float(probs[2]), 4),
            'threat': round(float(probs[3]), 4),
        },
        'embedding_preview': embedding[0][:8].tolist(),
        # текст НЕ возвращается и НЕ сохраняется
    }

📊 Как строится аналитика без текста

Вместо сырых переписок мы агрегируем векторные кластеры и временные ряды скоров. Вот SQL-схема для дашборда HR:

-- Таблица агрегатов (хранится 90 дней)
CREATE TABLE daily_toxicity_metrics (
    date DATE,
    department VARCHAR(50),
    avg_toxic_score FLOAT,
    blocked_count INT,
    total_messages INT,
    -- Семантические кластеры токсичных тем (без текста!)
    top_embedding_centroids JSONB,  -- усреднённые векторы
    risk_trend FLOAT  -- скользящее среднее
);

-- Пример запроса для дашборда
SELECT date, department, avg_toxic_score,
       CASE WHEN avg_toxic_score > 0.6 THEN '⚠️ Требуется внимание'
            WHEN avg_toxic_score > 0.75 THEN '🚨 Критический уровень'
            ELSE '✅ Норма' END AS status
FROM daily_toxicity_metrics
WHERE date > NOW() - INTERVAL '7 days'
ORDER BY date DESC;

🛡️ Почему это легально и этично