Merge pull request 'feat(core): safety-детектор, лексикон риска и кризисный скрипт' (#33) from feat/4-safety into dev

app/core/safety.py

@@ -0,0 +1,429 @@
+"""Crisis detector (guardrail #2, issue #4): runs BEFORE the LLM on every incoming message.
+
+Algorithm of `check()`:
+
+1. Normalise the (already scrubbed) text, see `normalise()`:
+   lowercase, ё→е, Latin look-alike letters inside Russian words → Cyrillic, all-Latin words transliterated
+   ("hochu umeret" → "хочу умерет"), punctuation → spaces, runs of 3+ identical letters → one ("жииить").
+   Clause boundaries (punctuation) are remembered per token: negation never crosses them.
+2. Run every precompiled regex of `config/lexicon/risk_ru.yaml` over the normalised string.
+   Regexes are written over word stems to cover Russian morphology and frequent typos.
+3. Negation (only for `acute` rules): a small window around the match decides whether the phrase is denied
+   ("не думаю о суициде", "мыслей о суициде нет"). A denied acute match is downgraded to POSSIBLE —
+   never to NONE: when in doubt we choose the higher risk. Window rules are documented in the YAML.
+4. Combinations raise the level (weapon access + any risk signal → ACUTE).
+5. Final level = max(lexicon level, llm_signal). The LLM can only raise the risk, never lower it.
+
+The result carries rule ids and category names only — never the raw or normalised text (guardrail #8).
+"""
+
+import bisect
+import re
+from dataclasses import dataclass
+from functools import lru_cache
+from pathlib import Path
+from typing import Literal
+
+import yaml
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
+
+from app.config import ROOT_DIR
+from app.core.types import RiskLevel, ScrubbedText
+
+LEXICON_PATH = ROOT_DIR / "config" / "lexicon" / "risk_ru.yaml"
+CRISIS_PATH = ROOT_DIR / "config" / "texts" / "crisis.yaml"
+
+Audience = Literal["patient", "family"]
+_LEVELS: dict[str, RiskLevel] = {"acute": RiskLevel.ACUTE, "possible": RiskLevel.POSSIBLE}
+
+
+# ---------------------------------------------------------------------------------------------------------
+# Public result types
+# ---------------------------------------------------------------------------------------------------------
+
+
+class SafetyResult(BaseModel):
+    """Outcome of `check()`. Contains no message text by design."""
+
+    model_config = ConfigDict(frozen=True)
+
+    level: RiskLevel
+    categories: list[str] = Field(default_factory=list)
+    matched_rules: list[str] = Field(default_factory=list)  # rule ids, e.g. "suicide.keyword"
+    negated_rules: list[str] = Field(default_factory=list)  # subset of matched_rules downgraded by negation
+    source: Literal["lexicon", "llm", "both", "none"] = "none"
+
+
+class CrisisResource(BaseModel):
+    model_config = ConfigDict(frozen=True)
+
+    name: str
+    phone: str
+    note: str = ""
+
+
+class CrisisScript(BaseModel):
+    """Fixed crisis reply (no LLM). `text` is the full ready-to-send message."""
+
+    model_config = ConfigDict(frozen=True)
+
+    variant: str
+    audience: Audience
+    body: str
+    resources: list[CrisisResource]
+    doctor_notified: str
+    grounding: str
+    status: str  # "черновик, утверждает психиатр (#8)" until clinically approved
+    text: str
+
+
+# ---------------------------------------------------------------------------------------------------------
+# Lexicon config schema
+# ---------------------------------------------------------------------------------------------------------
+
+
+class _RuleCfg(BaseModel):
+    id: str
+    pattern: str
+    level: Literal["acute", "possible"]
+    negation: Literal["pre", "pre_post", "none"] = "pre"
+
+    @field_validator("pattern")
+    @classmethod
+    def _compiles(cls, v: str) -> str:
+        re.compile(v)
+        return v
+
+
+class _CategoryCfg(BaseModel):
+    description: str = ""
+    rules: list[_RuleCfg]
+
+
+class _ComboCfg(BaseModel):
+    id: str
+    all_of: list[str]
+    any_of: list[str] = Field(default_factory=list)
+    level: Literal["acute", "possible"]
+
+
+class _NegationCfg(BaseModel):
+    window: int = 4
+    negators: set[str]
+    bridge_words: set[str]
+    post_negators: set[str]
+    post_not_followers: set[str]
+
+
+class _LexiconCfg(BaseModel):
+    version: int
+    status: str = ""
+    categories: dict[str, _CategoryCfg]
+    combinations: list[_ComboCfg] = Field(default_factory=list)
+    negation: _NegationCfg
+
+    @model_validator(mode="after")
+    def _unique_ids(self) -> "_LexiconCfg":
+        ids = [r.id for c in self.categories.values() for r in c.rules] + [c.id for c in self.combinations]
+        dupes = {i for i in ids if ids.count(i) > 1}
+        if dupes:
+            raise ValueError(f"duplicate lexicon rule ids: {sorted(dupes)}")
+        unknown = {cat for c in self.combinations for cat in c.all_of + c.any_of} - set(self.categories)
+        if unknown:
+            raise ValueError(f"combinations reference unknown categories: {sorted(unknown)}")
+        return self
+
+
+@dataclass(frozen=True, slots=True)
+class _Rule:
+    id: str
+    category: str
+    regex: re.Pattern[str]
+    level: RiskLevel
+    negation: str
+
+
+@dataclass(frozen=True, slots=True)
+class Lexicon:
+    """Compiled lexicon. Build once via `load_lexicon()`; matching is then pure regex work (< 1 ms)."""
+
+    rules: tuple[_Rule, ...]
+    combinations: tuple[_ComboCfg, ...]
+    negation: _NegationCfg
+    status: str
+
+
+def load_lexicon(path: Path = LEXICON_PATH) -> Lexicon:
+    cfg = _LexiconCfg.model_validate(yaml.safe_load(path.read_text(encoding="utf-8")))
+    rules = tuple(
+        _Rule(r.id, cat, re.compile(r.pattern), _LEVELS[r.level], r.negation)
+        for cat, c in cfg.categories.items()
+        for r in c.rules
+    )
+    return Lexicon(rules, tuple(cfg.combinations), cfg.negation, cfg.status)
+
+
+@lru_cache(maxsize=1)
+def _default_lexicon() -> Lexicon:
+    return load_lexicon()
+
+
+# ---------------------------------------------------------------------------------------------------------
+# Normalisation
+# ---------------------------------------------------------------------------------------------------------
+
+# Latin letters that look like Cyrillic ones: "сyицид" typed with a Latin "y" in a Russian word.
+_HOMOGLYPHS = str.maketrans("aeopcxykmthb", "аеорсхукмтнв")
+# Light transliteration for words typed entirely in Latin. Multi-letter sequences go first.
+_TRANSLIT_MULTI = (
+    ("shch", "щ"),
+    ("sch", "щ"),
+    ("yo", "е"),
+    ("zh", "ж"),
+    ("kh", "х"),
+    ("ts", "ц"),
+    ("ch", "ч"),
+    ("sh", "ш"),
+    ("yu", "ю"),
+    ("ju", "ю"),
+    ("ya", "я"),
+    ("ja", "я"),
+)
+_TRANSLIT_SINGLE = str.maketrans(
+    "abvgdezijklmnoprstufhcwxq'",
+    "абвгдезийклмнопрстуфхцвхкь",
+)
+_LATIN_RE = re.compile(r"[a-z]")
+_CYR_RE = re.compile(r"[а-я]")
+_LATIN_Y_AFTER_VOWEL = re.compile(r"(?<=[aeiou])y(?![aeiou])")
+_WORD_RE = re.compile(r"[0-9a-zа-я']+")
+_REPEAT_RE = re.compile(r"([а-яa-z])\1{2,}")
+_BOUNDARY_RE = re.compile(r"[.,;:!?()\[\]\"«»…—–\n]")
+
+
+def _translit(word: str) -> str:
+    word = _LATIN_Y_AFTER_VOWEL.sub("й", word)
+    for lat, cyr in _TRANSLIT_MULTI:
+        word = word.replace(lat, cyr)
+    return word.translate(_TRANSLIT_SINGLE).replace("y", "ы")
+
+
+def _fix_word(word: str) -> str:
+    if _LATIN_RE.search(word):
+        word = word.translate(_HOMOGLYPHS) if _CYR_RE.search(word) else _translit(word)
+    return word.replace("'", "")
+
+
+@dataclass(frozen=True, slots=True)
+class _Normalised:
+    text: str  # tokens joined by single spaces
+    tokens: tuple[str, ...]
+    starts: tuple[int, ...]  # char offset of each token in `text`
+    boundary_before: tuple[bool, ...]  # True if a clause boundary (punctuation) precedes the token
+
+
+def _normalise(raw: str) -> _Normalised:
+    s = raw.lower().replace("ё", "е")
+    s = _REPEAT_RE.sub(r"\1", s)
+    tokens: list[str] = []
+    boundaries: list[bool] = []
+    # Split into clauses first so negation lookup can stop at punctuation.
+    for clause in _BOUNDARY_RE.split(s):
+        first = True
+        for m in _WORD_RE.finditer(clause):
+            word = _fix_word(m.group())
+            if not word:
+                continue
+            tokens.append(word)
+            boundaries.append(first)
+            first = False
+    starts, pos = [], 0
+    for t in tokens:
+        starts.append(pos)
+        pos += len(t) + 1
+    return _Normalised(" ".join(tokens), tuple(tokens), tuple(starts), tuple(boundaries))
+
+
+def normalise(text: str) -> str:
+    """Normalised form used for matching (exposed for tests and lexicon authoring)."""
+    return _normalise(text).text
+
+
+# ---------------------------------------------------------------------------------------------------------
+# Negation
+# ---------------------------------------------------------------------------------------------------------
+
+
+def _negated_before(n: _Normalised, first_tok: int, neg: _NegationCfg) -> bool:
+    """Walk left from the match over bridge words; a negator found before any other word denies it."""
+    i = first_tok
+    for _ in range(neg.window):
+        if i == 0 or n.boundary_before[i]:
+            return False
+        i -= 1
+        tok = n.tokens[i]
+        if tok in neg.negators:
+            return True
+        if tok not in neg.bridge_words:
+            return False
+    return False
+
+
+def _negated_after(n: _Normalised, last_tok: int, neg: _NegationCfg) -> bool:
+    """Walk right over bridge words looking for "нет" / "не было"-style denial ("мыслей о суициде нет")."""
+    i = last_tok
+    for _ in range(neg.window):
+        i += 1
+        if i >= len(n.tokens) or n.boundary_before[i]:
+            return False
+        tok = n.tokens[i]
+        if tok in neg.post_negators:
+            return True
+        if tok == "не":
+            nxt = i + 1
+            if nxt >= len(n.tokens) or n.boundary_before[nxt]:
+                return False
+            return n.tokens[nxt] in neg.post_not_followers
+        if tok not in neg.bridge_words:
+            return False
+    return False
+
+
+def _is_negated(n: _Normalised, m: re.Match[str], mode: str, neg: _NegationCfg) -> bool:
+    if mode == "none":
+        return False
+    first_tok = bisect.bisect_right(n.starts, m.start()) - 1
+    last_tok = bisect.bisect_right(n.starts, max(m.end() - 1, m.start())) - 1
+    if _negated_before(n, first_tok, neg):
+        return True
+    return mode == "pre_post" and _negated_after(n, last_tok, neg)
+
+
+def _has_affirmed_match(n: _Normalised, rule: _Rule, neg: _NegationCfg) -> bool:
+    """True if at least one reading of the rule in the message is NOT denied.
+
+    We try the regex anchored at every token start, not only the leftmost `finditer` hits: in
+    "ничего не хочу хочу умереть" the leftmost match starts at the first "хочу" (preceded by "не"),
+    while the reading starting at the second "хочу" is affirmed. Any affirmed reading keeps ACUTE, so a
+    denied mention can never hide an affirmed one elsewhere in the message.
+    """
+    for pos in n.starts:
+        m = rule.regex.match(n.text, pos)
+        if m is not None and not _is_negated(n, m, rule.negation, neg):
+            return True
+    # Safety net for a pattern without a leading \b that only matches mid-token.
+    return any(not _is_negated(n, m, rule.negation, neg) for m in rule.regex.finditer(n.text))
+
+
+# ---------------------------------------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------------------------------------
+
+
+def check(
+    text: ScrubbedText, llm_signal: RiskLevel | None = None, *, lexicon: Lexicon | None = None
+) -> SafetyResult:
+    """Classify crisis risk of one message. Deterministic; the LLM signal can only raise the level."""
+    lex = lexicon or _default_lexicon()
+    n = _normalise(text)
+
+    lexicon_level = RiskLevel.NONE
+    matched: list[str] = []
+    negated: list[str] = []
+    categories: set[str] = set()
+
+    if n.text:
+        for rule in lex.rules:
+            if rule.regex.search(n.text) is None:
+                continue
+            level = rule.level
+            if level == RiskLevel.ACUTE and not _has_affirmed_match(n, rule, lex.negation):
+                level = RiskLevel.POSSIBLE
+                negated.append(rule.id)
+            matched.append(rule.id)
+            categories.add(rule.category)
+            lexicon_level = max(lexicon_level, level)
+
+        for combo in lex.combinations:
+            if set(combo.all_of) <= categories and (not combo.any_of or categories & set(combo.any_of)):
+                matched.append(combo.id)
+                lexicon_level = max(lexicon_level, _LEVELS[combo.level])
+
+    llm_level = llm_signal if llm_signal is not None else RiskLevel.NONE
+    from_lex, from_llm = lexicon_level > RiskLevel.NONE, llm_level > RiskLevel.NONE
+    source: Literal["lexicon", "llm", "both", "none"] = (
+        "both" if from_lex and from_llm else "lexicon" if from_lex else "llm" if from_llm else "none"
+    )
+    return SafetyResult(
+        level=max(lexicon_level, llm_level),
+        categories=sorted(categories),
+        matched_rules=matched,
+        negated_rules=negated,
+        source=source,
+    )
+
+
+# ---------------------------------------------------------------------------------------------------------
+# Crisis script
+# ---------------------------------------------------------------------------------------------------------
+
+
+class _VariantCfg(BaseModel):
+    text: str
+    resources: list[str]
+    doctor_notified: str
+    grounding: str
+
+
+class _CrisisCfg(BaseModel):
+    status: str
+    category_priority: list[str] = Field(default_factory=list)
+    resources: dict[str, CrisisResource]
+    variants: dict[str, _VariantCfg]
+
+    @model_validator(mode="after")
+    def _consistent(self) -> "_CrisisCfg":
+        for aud in ("patient", "family"):
+            if f"{aud}.default" not in self.variants:
+                raise ValueError(f"crisis.yaml: missing variant {aud}.default")
+        missing = {r for v in self.variants.values() for r in v.resources} - set(self.resources)
+        if missing:
+            raise ValueError(f"crisis.yaml: unknown resources {sorted(missing)}")
+        return self
+
+
+@lru_cache(maxsize=4)
+def _crisis_cfg(path: Path = CRISIS_PATH) -> _CrisisCfg:
+    return _CrisisCfg.model_validate(yaml.safe_load(path.read_text(encoding="utf-8")))
+
+
+def _render(body: str, resources: list[CrisisResource], doctor_notified: str, grounding: str) -> str:
+    lines = [body.strip(), ""]
+    for r in resources:
+        lines.append(f"• {r.name}: {r.phone}" + (f" ({r.note})" if r.note else ""))
+    lines += ["", doctor_notified, grounding]
+    return "\n".join(lines)
+
+
+def crisis_reply(
+    categories: list[str], audience: Audience = "patient", *, path: Path = CRISIS_PATH
+) -> CrisisScript:
+    """Fixed crisis script for an ACUTE result. Never generated by the LLM."""
+    cfg = _crisis_cfg(path)
+    variant = f"{audience}.default"
+    for cat in cfg.category_priority:
+        if cat in categories and f"{audience}.{cat}" in cfg.variants:
+            variant = f"{audience}.{cat}"
+            break
+    v = cfg.variants[variant]
+    resources = [cfg.resources[r] for r in v.resources]
+    return CrisisScript(
+        variant=variant,
+        audience=audience,
+        body=v.text.strip(),
+        resources=resources,
+        doctor_notified=v.doctor_notified,
+        grounding=v.grounding,
+        status=cfg.status,
+        text=_render(v.text, resources, v.doctor_notified, v.grounding),
+    )