Spaces:
Running on Zero
Running on Zero
Bot de Telegram + openjev en un contenedor: polling en hilo principal, inferencia en-proceso
cc04911 verified Download rules.py from danielgx300/KD: direct link, hf CLI and curl.
- Browser
- Download file 14.5 kB
-
https://huggingface.co/spaces/danielgx300/KD/resolve/main/rules.py
- Command line
-
hf download hf://spaces/danielgx300/KD/rules.py
-
curl -L -o rules.py https://huggingface.co/spaces/danielgx300/KD/resolve/main/rules.py
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 | |
| # -------------------------------------------------------------------------- | |
| 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)} | |
| 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) | |
| 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 | |
| def violations(self) -> list[RuleVerdict]: | |
| return [f for f in self.flags if f.verdict == "violation"] | |
| def uncertain(self) -> list[RuleVerdict]: | |
| return [f for f in self.flags if f.verdict == "uncertain"] | |
| 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, | |
| ) | |