KD / rules.py
danielgx300's picture
Bot de Telegram + openjev en un contenedor: polling en hilo principal, inferencia en-proceso
cc04911 verified
Raw History Blame Contribute Delete
14.5 kB
"""Esquema de preguntas derivado de las reglas del grupo, y logica de veredicto.
DECISIONES DE DISENO (importantes, no son arbitrarias)
-----------------------------------------------------
1. NO usamos `noul`. El model card documenta que `noul` puede seguir las
etiquetas `false:`/`true:` en vez del estado y devolver un "no" confiado
para texto claramente positivo (issue #156). El workaround oficial es un
`choice` de dos opciones con CLAVES NEUTRALES y la respuesta si/no en las
DESCRIPCIONES. Eso es exactamente lo que hacemos: claves "A"/"B".
2. NO usamos `score`. Es el primitive mas debil (0.372 en SST-5). La severidad
la calculamos en Python a partir de las probabilidades por regla, que es
mas controlable y auditable.
3. Espejo para `menores`. Hacemos la MISMA pregunta con las descripciones
invertidas y exigimos acuerdo. Si discrepan, el veredicto es `uncertain` y
escala a humano. Cuesta casi nada: Laya responde TODAS las preguntas en una
sola pasada forward, asi que la pregunta espejo no cuesta una llamada extra.
4. No nos apoyamos en `confidence`. Ese campo es entropia normalizada
(1 - H(p)/log(k)): mide concentracion, NO es probabilidad calibrada. El
propio repo da un ejemplo de 0.000 de accuracy con 0.952 de confianza.
Gateamos con la probabilidad de la opcion, que si es una probabilidad, y
encima exigimos acuerdo entre preguntas.
"""
from __future__ import annotations
import math
from dataclasses import dataclass, field
from typing import Any, Literal
import config
Verdict = Literal["violation", "ok", "uncertain"]
# --------------------------------------------------------------------------
# Preguntas
# --------------------------------------------------------------------------
# Descripciones cortas a proposito: TODAS las opciones de TODAS las preguntas
# comparten `head_max_len` (ver config.py). Con 256 tokens y 16 opciones cada
# una recibiria ~15 tokens y se volverian indistinguibles.
QUESTIONS: dict[str, dict[str, Any]] = {
# --- REGLA CRITICA: menores (tolerancia cero) ---
"menores": {
"type": "choice",
"instructions": "¿El mensaje muestra, comparte o habla de alguien menor de 18 años?",
"criteria": {
"A": "si: el mensaje trata de una persona que no cumple 18 años",
"B": "no: el mensaje no trata de ninguna persona menor de 18 años",
},
},
# Espejo: mismas dos opciones con las descripciones invertidas, para
# detectar el sesgo de posicion. Si no coincide con `menores` -> uncertain.
"menores_espejo": {
"type": "choice",
"instructions": "¿El mensaje muestra, comparte o habla de alguien menor de 18 años?",
"criteria": {
"A": "no: el mensaje no trata de ninguna persona menor de 18 años",
"B": "si: el mensaje trata de una persona que no cumple 18 años",
},
},
# --- Publicidad molesta ---
"publicidad": {
"type": "choice",
"instructions": "¿El mensaje es publicidad o spam no solicitado?",
"criteria": {
"A": "si: anuncia un negocio, una promo o un servicio no pedido",
"B": "no: es comentario normal del grupo, sin publicidad",
},
},
# --- Venta / intercambio de contactos ---
"negocios": {
"type": "choice",
"instructions": "¿El mensaje vende, regala o intercambia contactos a cambio de algo?",
"criteria": {
"A": "si: ofrece contactos, presentaciones o.compiler",
"B": "no: solo comparte informacion, sin venta ni intercambio",
},
},
# --- Pedir contenido especifico de una chica ---
"pedido_contenido": {
"type": "choice",
"instructions": "¿El mensaje le pide a alguien material concreto de una chica?",
"criteria": {
"A": "si: pide fotos, videos o contenido especifico de una chica",
"B": "no: no pide contenido especifico de ninguna chica",
},
},
# --- Especulacion sobre chicas sin pruebas ---
"especulacion": {
"type": "choice",
"instructions": "¿El mensaje especula o consulta sobre una chica sin pruebas?",
"criteria": {
"A": "si: pregunta o afirma cosas de una chica sin saber quien es",
"B": "no: habla con datos concretos o no habla de ninguna chica",
},
},
# --- Pelea entre usuarios ---
"pelea": {
"type": "choice",
"instructions": "Leyendo la conversacion, ¿el mensaje es agresivo o mantiene una pelea?",
"criteria": {
"A": "si: esta tenso, insulta, amenaza o responde con bronca",
"B": "no: es tranquilo, respeta a los demas y no pelea",
},
},
# --- Permiso: para NO castigar contenido permitido ---
# Reseñas, dudas, contenido y contactos son legales en este grupo. Si esto
# sale "si", neutraliza las reglas blandas marcadas con overridden_by.
"contenido_permitido": {
"type": "choice",
"instructions": "¿El mensaje es una reseña, una duda, o comparte contenido o contactos?",
"criteria": {
"A": "si: es una reseña, una duda, o comparte contenido o contactos",
"B": "no: no es ninguna de esas cosas",
},
},
}
QUESTION_IDS: tuple[str, ...] = tuple(QUESTIONS)
# --------------------------------------------------------------------------
# Declaracion de reglas
# --------------------------------------------------------------------------
@dataclass(frozen=True)
class Rule:
"""Una regla del grupo traducida a una (o dos) preguntas de Laya."""
key: str
title: str
question_id: str
violation_option: str
threshold: float
severity: int # 1 = leve, 2 = media, 3 = grave
action: str # "delete" | "warn" | "escalate"
mirror_question_id: str | None = None
mirror_violation_option: str | None = None
# Si se activa, la regla queda anulada (contenido permitido).
overridden_by: str | None = None
# Los miembros nuevos no sufrem borrado automatico.
exempt_new_members: bool = True
RULES: tuple[Rule, ...] = (
Rule(
key="menores",
title="Menores de edad",
question_id="menores",
violation_option="A",
threshold=config.THRESHOLD_MENORES,
severity=3,
# NUNCA borra, ni siquiera en enforce. Escalar a humano siempre:
# borrar evidencia de material de menores es lo peor que puedes hacer.
action="escalate",
mirror_question_id="menores_espejo",
mirror_violation_option="B",
exempt_new_members=False,
),
Rule(
key="publicidad",
title="Spam / publicidad molesta",
question_id="publicidad",
violation_option="A",
threshold=config.THRESHOLD_PUBLICIDAD,
severity=2,
action="delete",
overridden_by="contenido_permitido",
),
Rule(
key="negocios",
title="Venta o intercambio de contactos",
question_id="negocios",
violation_option="A",
threshold=config.THRESHOLD_NEGOCIOS,
severity=2,
action="delete",
overridden_by="contenido_permitido",
),
Rule(
key="pedido_contenido",
title="Pedido de contenido especifico",
question_id="pedido_contenido",
violation_option="A",
threshold=config.THRESHOLD_PEDIDO,
severity=2,
action="warn",
),
Rule(
key="especulacion",
title="Especulacion sobre chicas sin pruebas",
question_id="especulacion",
violation_option="A",
threshold=config.THRESHOLD_ESPECULACION,
severity=1,
action="warn",
),
Rule(
key="pelea",
title="Pelea / escalada entre usuarios",
question_id="pelea",
violation_option="A",
threshold=config.THRESHOLD_PELEA,
severity=2,
action="warn",
# Sin override: pelearse no se lava con "es contenido permitido".
overridden_by=None,
),
)
RULES_BY_KEY: dict[str, Rule] = {r.key: r for r in RULES}
# --------------------------------------------------------------------------
# Lectura y calibracion de respuestas
# --------------------------------------------------------------------------
def _probs(answer: Any) -> dict[str, float]:
"""Saca el dict de probabilidades de una respuesta.
Acepta DOS formas, a proposito:
- nativo de Laya: {"probabilities": {"A": 0.9, "B": 0.1}}
- plano del Space: {"A": 0.9, "B": 0.1} <-- lo que cruza el proceso
El Space devuelve la forma plana, asi que si solo acceptas la nativa el bot
se queda ciego sin avisar. Las claves planas que no sean numeros se
ignoran, asi que tambien tolera que se cuele metadata.
"""
if not isinstance(answer, dict):
return {}
raw = answer.get("probabilities")
if not isinstance(raw, dict):
# Forma plana: usa el dict entero y descarta lo que no sea numerico.
raw = answer
out: dict[str, float] = {}
for k, v in raw.items():
if isinstance(v, bool) or not isinstance(v, (int, float)):
continue
out[str(k)] = float(v)
return out
def apply_temperature(probs: dict[str, float], t: float | None) -> dict[str, float]:
"""p' = softmax(log(p) / t).
Escalar temperatura es dividir los logits por T y renormalizar. T > 1 deja
la distribucion mas plana (menos confiada), que es justo lo que necesita un
modelo que sale sobrecconfiado. `t` es POR PREGUNTA, no por opcion.
"""
if t is None or t <= 0 or t == 1.0 or len(probs) < 2:
return probs
keys = list(probs)
logits = [math.log(max(probs[k], 1e-9)) / t for k in keys]
m = max(logits)
exps = [math.exp(z - m) for z in logits]
denom = sum(exps)
if denom <= 0:
return probs
return {k: e / denom for k, e in zip(keys, exps)}
@dataclass
class RuleVerdict:
key: str
title: str
verdict: Verdict
probability: float
threshold: float
severity: int
action: str
reason: str
probabilities: dict[str, float] = field(default_factory=dict)
@dataclass
class ModerationDecision:
flags: list[RuleVerdict]
is_violation: bool
max_severity: int
escalate: bool
reason: str
allowed: bool
routing: dict[str, Any] = field(default_factory=dict)
elapsed_ms: int = 0
@property
def violations(self) -> list[RuleVerdict]:
return [f for f in self.flags if f.verdict == "violation"]
@property
def uncertain(self) -> list[RuleVerdict]:
return [f for f in self.flags if f.verdict == "uncertain"]
@property
def worst(self) -> RuleVerdict | None:
cand = [f for f in self.flags if f.verdict in ("violation", "uncertain")]
if not cand:
return None
return max(cand, key=lambda f: (f.severity, f.probability))
def _summarize(flags: list[RuleVerdict]) -> str:
hits = [f"{f.title} ({f.probability:.2f})" for f in flags if f.verdict == "violation"]
doubt = [f.title for f in flags if f.verdict == "uncertain"]
if doubt:
hits.append("duda: " + ", ".join(doubt))
return "; ".join(hits) or "sin infracciones"
def evaluate(
answers: dict[str, Any],
temps: dict[str, float] | None = None,
*,
is_new_member: bool = False,
) -> ModerationDecision:
"""Convierte las respuestas crudas de Laya en veredictos por regla.
`answers` es el dict `answers` de la respuesta de `Router.predict`.
`temps` son temperaturas por question_id, producidas por calibrate.py.
"""
temps = temps or {}
answers = answers or {}
calibrated: dict[str, dict[str, float]] = {}
for qid in QUESTION_IDS:
p = _probs(answers.get(qid))
if p:
calibrated[qid] = apply_temperature(p, temps.get(qid))
# Regla de seguridad: contenido permitido.
allowed_p = calibrated.get("contenido_permitido", {})
allowed = bool(allowed_p) and allowed_p.get("A", 0.0) > allowed_p.get("B", 0.0)
flags: list[RuleVerdict] = []
for rule in RULES:
p = calibrated.get(rule.question_id)
if not p:
continue
prob = p.get(rule.violation_option, 0.0)
verdict: Verdict = "violation" if prob >= rule.threshold else "ok"
reason = f"p={prob:.2f} vs t={rule.threshold:.2f}"
# Espejo: exigimos acuerdo entre la pregunta y su espejo invertido.
if rule.mirror_question_id:
mp = calibrated.get(rule.mirror_question_id)
if mp:
mprob = mp.get(rule.mirror_violation_option or "A", 0.0)
agree = (mprob >= rule.threshold) == (verdict == "violation")
reason += f" | espejo={mprob:.2f}"
if not agree:
# Una ve infraccion y la otra no. No nos fiamos de ninguna.
verdict = "uncertain"
reason += " (desacuerdo)"
# Contenido permitido neutraliza las reglas blandas.
if verdict == "violation" and rule.overridden_by and allowed:
verdict = "ok"
reason += " | neutralizado por contenido permitido"
# Miembros nuevos: sin borrado automatico.
if (
verdict == "violation"
and is_new_member
and rule.exempt_new_members
and rule.action == "delete"
):
verdict = "uncertain"
reason += " | miembro nuevo, sin borrado"
flags.append(
RuleVerdict(
key=rule.key,
title=rule.title,
verdict=verdict,
probability=prob,
threshold=rule.threshold,
severity=rule.severity,
action=rule.action,
reason=reason,
probabilities=p,
)
)
has_violation = any(f.verdict == "violation" for f in flags)
# La regla de menores escala aunque venga como `uncertain`: es la unica
# donde un falso positivo justificaria el tiempo de un humano.
escalate = has_violation or any(
f.key == "menores" and f.verdict == "uncertain" for f in flags
)
return ModerationDecision(
flags=flags,
is_violation=has_violation,
max_severity=max((f.severity for f in flags if f.verdict == "violation"), default=0),
escalate=escalate,
reason=_summarize(flags),
allowed=allowed,
)