ElebobaandClaude Opus 5.5
b4731fb1d6
CI / ruff · pytest (pull_request) Successful in 22s
Freeze interfaces between lanes A (bot), B (backend), C (dashboard) so three developers work in parallel without touching each other's files: dialog Event/Reply + engine registry with echo stub, integration protocols with in-memory defaults, MonitoringService with noop stub, /api/v1 schemas with fixture-backed routes, frozen main.py wiring, config/modules.yaml + due_scales(), DAR-5, tzdata, Andrey trajectory, lanes in WORKFLOW, ADR-010. Refs #34 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
426 lines
15 KiB
Python
426 lines
15 KiB
Python
"""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),
|
||
)
|