Spaces:
Running
Running
| """Replay the stream once and write replay.json, which the app plays back. | |
| For each request in order: if a rule set exists and the gate passes, the trusted, not switched-off rules answer; | |
| otherwise the teacher answers and its answer becomes a labelled example. SPOT_EVERY-th rule answers (by request-id | |
| hash) are also sent to the teacher: that is the drift guard's evidence, and those answers are labelled too. A rule | |
| whose spot-check precision falls below GUARD_PREC after GUARD_MIN checked spans is switched off for good. Each time the | |
| labelled set reaches the next size in SCHEDULE, RuleChef's search synthesis (no LLM) re-learns the rules from all | |
| labelled examples; the new set takes over at the next request. Every rule set version is also scored on the | |
| fixed holdout against the teacher and against the TAB annotation. | |
| Teacher answers are read from teacher.jsonl (precomputed with run_teacher.py, the same code the app runs live). | |
| usage: sim.py DATA_DIR OUT.json | |
| """ | |
| import hashlib | |
| import json | |
| import os | |
| import platform | |
| import sys | |
| import time | |
| import rulechef | |
| from rulechef.core import Dataset, Example, Task, TaskType | |
| from rulechef.search import SearchConfig, SearchSynthesizer | |
| import cascade as C | |
| SPOT_EVERY = 10 | |
| GUARD_MIN, GUARD_PREC = int(os.environ.get("GUARD_MIN", 5)), float(os.environ.get("GUARD_PREC", 0.8)) | |
| SCHEDULE = [50, 100] + list(range(200, 5001, 100)) | |
| MIN_SUPPORT = 2 | |
| def spot(rid): | |
| return int(hashlib.sha1(rid.encode()).hexdigest(), 16) % SPOT_EVERY == 0 | |
| def dataset(labelled): | |
| task = Task(name="stream", description="date / code / quantity spans", input_schema={"text": "str"}, | |
| output_schema={"entities": "List[Entity]"}, type=TaskType.NER, text_field="text") | |
| ds = Dataset(name="stream", task=task) | |
| for i, (text, spans) in enumerate(labelled): | |
| ents = [{"text": s["text"], "start": s["start"], "end": s["end"], "type": s["type"]} for s in spans] | |
| ds.examples.append(Example(id=str(i), input={"text": text}, expected_output={"entities": ents}, source="teacher")) | |
| return ds | |
| def rule_key(d): | |
| return d["output_template"]["type"] + "\t" + d["content"] + "\t" + d["output_template"].get("text", "$0") | |
| def rule_stats(rules, labelled): | |
| """Each rule alone on the labelled data, against the teacher: matches and precision.""" | |
| out = {} | |
| for r in rules: | |
| tp = fp = 0 | |
| for text, spans in labelled: | |
| t, f, _ = C.match_counts(C.run_rules([r], text), spans) | |
| tp, fp = tp + t, fp + f | |
| out[r.id] = (tp, fp) | |
| return out | |
| def evaluate(holdout, rules, gate): | |
| """Holdout scores for one rule set (rules = the trusted, active ones).""" | |
| agg = {k: [0, 0, 0] for k in ("rules_t", "casc_t", "rules_g", "casc_g", "teach_g", "routed_t")} | |
| passed = exact = 0 | |
| for r in holdout: | |
| rs = C.run_rules(rules, r["text"]) if gate is not None else [] | |
| ok = gate is not None and not gate.unexplained(r["text"], rs) | |
| casc = rs if ok else r["teacher"] | |
| for k, pred, ref in (("rules_t", rs, r["teacher"]), ("casc_t", casc, r["teacher"]), ("rules_g", rs, r["gold"]), | |
| ("casc_g", casc, r["gold"]), ("teach_g", r["teacher"], r["gold"])): | |
| for j, v in enumerate(C.match_counts(pred, ref)): | |
| agg[k][j] += v | |
| if ok: | |
| passed += 1 | |
| c = C.match_counts(rs, r["teacher"]) | |
| exact += c[1] == 0 and c[2] == 0 | |
| for j, v in enumerate(c): | |
| agg["routed_t"][j] += v | |
| res = {"n": len(holdout), "rules_answer": passed, "exact_agree": exact} | |
| for k, v in agg.items(): | |
| p, rr, f = C.prf(*v) | |
| res[k] = {"tp": v[0], "fp": v[1], "fn": v[2], "P": round(p, 4), "R": round(rr, 4), "F1": round(f, 4)} | |
| return res | |
| def main(data, out): | |
| reqs = [json.loads(line) for line in open(f"{data}/requests.jsonl")] | |
| teach = {} | |
| for line in open(f"{data}/teacher.jsonl"): | |
| t = json.loads(line) | |
| teach[t["id"]] = t | |
| for r in reqs: | |
| r["teacher"] = teach[r["id"]]["spans"] | |
| r["teacher_ms"] = teach[r["id"]]["ms"] | |
| stream = [r for r in reqs if r["part"] == "stream"] | |
| holdout = [r for r in reqs if r["part"] == "holdout"] | |
| labelled, versions, events, guard_log = [], [], [], [] | |
| guard, blocked = {}, set() | |
| rules, rule_dicts, trusted, gate, next_i = [], {}, set(), None, 0 | |
| versions.append({"v": 0, "at": 0, "n_labelled": 0, "synth_s": 0.0, "rules": [], "gate": None, | |
| "holdout": evaluate(holdout, [], None)}) | |
| def active(): | |
| return [r for r in rules if r.id in trusted and rule_key(rule_dicts[r.id]) not in blocked] | |
| for i, r in enumerate(stream): | |
| if next_i < len(SCHEDULE) and len(labelled) >= SCHEDULE[next_i]: | |
| while next_i < len(SCHEDULE) and len(labelled) >= SCHEDULE[next_i]: | |
| next_i += 1 | |
| t0 = time.perf_counter() | |
| syn = SearchSynthesizer(SearchConfig(labels=C.TYPES, min_support=MIN_SUPPORT)) | |
| rules = syn.synthesize(dataset(labelled)) | |
| synth_s = time.perf_counter() - t0 | |
| t0 = time.perf_counter() | |
| gate = C.Gate.learn(labelled) | |
| stats = rule_stats(rules, labelled) | |
| rule_dicts = {x.id: x.to_dict() for x in rules} | |
| trusted = {x.id for x in rules if sum(stats[x.id]) >= C.TRUST_MIN and stats[x.id][0] / sum(stats[x.id]) >= C.TRUST_PREC} | |
| post_s = time.perf_counter() - t0 | |
| v = len(versions) | |
| versions.append({ | |
| "v": v, "at": i, "n_labelled": len(labelled), "synth_s": round(synth_s, 2), "check_s": round(post_s, 2), | |
| "rules": [{"id": x.id, "key": rule_key(rule_dicts[x.id]), "type": rule_dicts[x.id]["output_template"]["type"], | |
| "regex": x.content, "template": rule_dicts[x.id]["output_template"].get("text", "$0"), | |
| "tp": stats[x.id][0], "fp": stats[x.id][1], "trusted": x.id in trusted, "dict": rule_dicts[x.id]} | |
| for x in rules], | |
| "gate": gate.to_dict(), "blocked_at_deploy": sorted(blocked), "report": syn.report, | |
| "holdout": evaluate(holdout, active(), gate)}) | |
| print(f"v{v} at {i} labelled {len(labelled)} rules {len(rules)} trusted {len(trusted)} synth {synth_s:.1f}s " | |
| f"holdout rules-answer {versions[-1]['holdout']['rules_answer']} casc F1 {versions[-1]['holdout']['casc_t']['F1']}", flush=True) | |
| t0 = time.perf_counter() | |
| act = active() | |
| rs = C.run_rules(act, r["text"]) if gate is not None else [] | |
| unexpl = gate.unexplained(r["text"], rs) if gate is not None else ["(no rules yet)"] | |
| rule_ms = (time.perf_counter() - t0) * 1000 | |
| ev = {"id": r["id"], "v": len(versions) - 1, "rule_ms": round(rule_ms, 3), "teacher_ms": r["teacher_ms"], | |
| "fired": sorted({rule_key(rule_dicts[s["rule_id"]]) for s in rs}) if not unexpl else []} | |
| if unexpl: | |
| ev["by"] = "teacher" | |
| labelled.append((r["text"], r["teacher"])) | |
| else: | |
| ev["by"] = "rules" | |
| if spot(r["id"]): | |
| ev["checked"] = True | |
| tp, fp, fn = C.match_counts(rs, r["teacher"]) | |
| ev["agree"] = fp == 0 and fn == 0 | |
| labelled.append((r["text"], r["teacher"])) | |
| tkeys = [C.key(s) for s in r["teacher"]] | |
| for s in rs: | |
| k = rule_key(rule_dicts[s["rule_id"]]) | |
| g = guard.setdefault(k, [0, 0]) | |
| hit = C.key(s) in tkeys | |
| if hit: | |
| tkeys.remove(C.key(s)) | |
| g[0 if hit else 1] += 1 | |
| if k not in blocked and sum(g) >= GUARD_MIN and g[0] / sum(g) < GUARD_PREC: | |
| blocked.add(k) | |
| guard_log.append({"at": i, "key": k, "tp": g[0], "fp": g[1]}) | |
| print("guard: switched off", k.replace("\t", " "), g, flush=True) | |
| events.append(ev) | |
| meta = {"n_stream": len(stream), "n_holdout": len(holdout), "spot_every": SPOT_EVERY, "guard_min": GUARD_MIN, | |
| "guard_prec": GUARD_PREC, "schedule": SCHEDULE, "min_support": MIN_SUPPORT, "trust_min": C.TRUST_MIN, | |
| "trust_prec": C.TRUST_PREC, "word_min": C.WORD_MIN, "word_ratio": C.WORD_RATIO, "skip_min": C.SKIP_MIN, | |
| "skip_ratio": C.SKIP_RATIO, "rulechef": getattr(rulechef, "__version__", "?"), "python": platform.python_version(), | |
| "cpu": platform.processor() or platform.machine()} | |
| json.dump({"meta": meta, "versions": versions, "events": events, "guard": guard_log, "guard_final": guard}, | |
| open(out, "w"), separators=(",", ":")) | |
| n_rules = sum(e["by"] == "rules" for e in events) | |
| calls = sum(e["by"] == "teacher" or e.get("checked", False) for e in events) | |
| print("rules answered", n_rules, "of", len(events), "model calls", calls, "versions", len(versions) - 1, "switched off", len(guard_log)) | |
| if __name__ == "__main__": | |
| main(sys.argv[1], sys.argv[2]) | |