Spaces:
Running
Running
| """The cascade's pieces, shared by the offline replay (sim.py) and the app: the gate, rule execution, scoring. | |
| Gate: a request is answered by rules only when every trigger token in it is explained. Trigger tokens are tokens | |
| with a digit, plus words the teacher often puts inside spans (learned). A trigger is explained when an active rule's | |
| span covers it, or when its context is a learned skip context: (previous word, digit shape) pairs the teacher has | |
| seen at least SKIP_MIN times and labelled at most SKIP_RATIO of the time, such as "Article 9" or "§ 9". | |
| """ | |
| import os | |
| import re | |
| from collections import Counter, defaultdict | |
| from rulechef.core import Rule, TaskType | |
| from rulechef.executor import RuleExecutor | |
| TYPES = ["DATETIME", "CODE", "QUANTITY"] | |
| TOK = re.compile(r"§|[A-Za-z]+|\d+(?:[.,/:-]\d+)*") | |
| _E = os.environ.get | |
| WORD_MIN, WORD_RATIO = int(_E("WORD_MIN", 3)), float(_E("WORD_RATIO", 0.3)) # a word is a trigger if >= 3 teacher spans contain it and >= 30 % of its uses are inside one | |
| SKIP_MIN, SKIP_RATIO = int(_E("SKIP_MIN", 5)), float(_E("SKIP_RATIO", 0.3)) # a (previous word, shape) context is skipped if seen >= 5 times and the teacher labels it <= 30 % of the time | |
| TRUST_MIN, TRUST_PREC = int(_E("TRUST_MIN", 3)), float(_E("TRUST_PREC", 0.8)) # a rule answers only if it matched >= 3 times on the labelled data with precision >= 0.8 | |
| _EX = RuleExecutor() | |
| def shape(tok): | |
| return re.sub(r"\d+", "9", tok) | |
| def tokens(text): | |
| out, prev = [], "" | |
| for m in TOK.finditer(text): | |
| out.append((m.start(), m.end(), m.group(), prev)) | |
| prev = m.group().lower() | |
| return out | |
| def key(s): | |
| return (s["text"].strip(), s["type"]) | |
| class Gate: | |
| def __init__(self, trigger_words=(), skip=()): | |
| self.trigger_words = set(trigger_words) | |
| self.skip = {tuple(k) for k in skip} | |
| def learn(cls, labelled): | |
| words, ctx = defaultdict(lambda: [0, 0]), defaultdict(lambda: [0, 0]) | |
| for text, spans in labelled: | |
| for s, e, tok, prev in tokens(text): | |
| inside = any(sp["start"] < e and s < sp["end"] for sp in spans) | |
| d = ctx[(prev, shape(tok))] if tok[0].isdigit() else (words[tok] if tok.isalpha() else None) | |
| if d is not None: | |
| d[0] += 1 | |
| d[1] += inside | |
| trig = {w for w, (n, k) in words.items() if k >= WORD_MIN and k / n >= WORD_RATIO} | |
| skip = {c for c, (n, k) in ctx.items() if n >= SKIP_MIN and k / n <= SKIP_RATIO} | |
| return cls(trig, skip) | |
| def unexplained(self, text, spans): | |
| out = [] | |
| for s, e, tok, prev in tokens(text): | |
| digit = tok[0].isdigit() | |
| if not (digit or tok in self.trigger_words): | |
| continue | |
| if any(sp["start"] < e and s < sp["end"] for sp in spans): | |
| continue | |
| if digit and (prev, shape(tok)) in self.skip: | |
| continue | |
| out.append(tok) | |
| return out | |
| def to_dict(self): | |
| return {"trigger_words": sorted(self.trigger_words), "skip": sorted(list(k) for k in self.skip)} | |
| def run_rules(rules, text): | |
| """RuleChef's own executor; returns entity dicts with rule attribution (rule_id).""" | |
| if not rules: | |
| return [] | |
| out = _EX.apply_rules(rules, {"text": text}, TaskType.NER, "text") | |
| return [{"start": e["start"], "end": e["end"], "type": e.get("type"), "text": e["text"], "rule_id": e.get("rule_id")} | |
| for e in out.get("entities", [])] | |
| def match_counts(pred, gold): | |
| """RuleChef's TAB scorer: (text, type) multiset match. Returns tp, fp, fn.""" | |
| g = Counter(key(s) for s in gold) | |
| p = Counter(key(s) for s in pred) | |
| tp = sum((p & g).values()) | |
| return tp, sum(p.values()) - tp, sum(g.values()) - tp | |
| def prf(tp, fp, fn): | |
| p = tp / (tp + fp) if tp + fp else 1.0 | |
| r = tp / (tp + fn) if tp + fn else 1.0 | |
| return p, r, (2 * p * r / (p + r) if p + r else 0.0) | |
| def rules_from_dicts(ds): | |
| return [Rule.from_dict(d) for d in ds] | |