"""Simbiot Harness — ядро Коди: обучаемые ученики решений (как ядро systemone-bench). Цикл бенча (engine42/router42): учитель → дистилляция в маленькую сеть + kNN-память → живой роутер по новизне → дообучение на реальных исходах. Здесь тот же цикл на решениях Коди. Первая голова — `addressing`: отвечать ли на сообщение в группе (harness/addressing.py). Опыт (события aggregate='core', подписаны как всё в сторе): CoreLabel — метка сообщения: source=teacher_backfill (учитель разметил историю), source=human (кнопка Alex в личке); CoreFeedback — неявная обратная связь чата: бот ответил, а следом «не тебе / продолжает / молчи» → 0; бот промолчал, а автор тут же дёрнул бота («@juno42bot», «Коди?») → 1; CorePrediction — теневой прогноз ученика на живом сообщении; CoreModelPromoted — смена чемпиона (метрики, sha256 файла модели). Живые вердикты учителя (MessageTriaged/MessageIgnored rule=decision, p) тоже метки — мягкие, вес 1. Обратная связь — вес 3, человек — вес 5. Признаки: эмбеддинг сообщения + эмбеддинг контекста (3 реплики до него, ответы бота — только на адресованное) через локальный nomic-embed-text (~70 мс, бесплатно; кеш на диске) + 8 флагов. Ученик — логистическая регрессия (numpy, L2): на сотнях примеров она устойчивее MLP; формат модели допускает MLP позже. Новизна — 1 − средняя косинусная близость к 5 ближайшим из обучающей памяти. Роутер (shadow по умолчанию — SIMBIOT_CORE_ADDRESSING=shadow|on|off): знакомо (новизна < NOV_FAMILIAR) и уверенно (|p−0.5| ≥ CONF) → ученик; незнакомо (новизна ≥ NOV_NOVEL) и учитель сомневается → вопрос Alex кнопками в личку (≤ SIMBIOT_CORE_ASK_PER_DAY в сутки, 5); иначе → учитель. В shadow решает учитель, ученик только пишет прогноз. Тренировка — дневной тик (и cli.py core train): временной холдаут (последние 20%), претендент заменяет чемпиона, только если лучше по взвешенному log-loss на том же холдауте; файл модели в /core/. """ import hashlib import json import os import re import sqlite3 import struct import threading import time import urllib.request from datetime import datetime, timedelta, timezone from . import es from . import paths as _paths try: import numpy as np except ImportError: # ядро выключится само: featurize/train упадут, триаж — учитель np = None HEAD = "addressing" EMB_MODEL = "nomic-embed-text" EMB_DIM = 768 FLAG_NAMES = ("prev_is_bot", "same_author_as_last_addressed", "question", "second_person", "names_other_human", "length", "link", "media") # пороги новизны калибруются при обучении (перцентили новизны холдаута # против обучающей памяти: у nomic близость реплик одного чата высокая, # новизна живёт в 0.03–0.09 — фиксированные пороги бессмысленны); это — дефолт NOV_FAMILIAR = 0.35 NOV_NOVEL = 0.55 # уверенность: |p−0.5| ≥ 0.4 — репетиция на истории правления (5 фолдов): # покрытие 25%, совпадение с учителем 95%; 0.35 → 47% / 93%; 0.3 → 60% / 90% CONF = 0.4 MIN_TRAIN = 40 WEIGHTS = {"teacher": 1.0, "teacher_backfill": 1.0, "feedback": 3.0, "human": 5.0} NEG_RE = re.compile(r"не\s+(?:тебе|к\s+тебе|тебя\s+спрашива)|молчи|не\s+встревай|" r"заткн|продолжает|опять\s+(?:встрял|отвечает|лезет)|не\s+лезь", re.IGNORECASE) SECOND_RE = re.compile(r"\b(?:ты|тебе|тебя|твой|твоя|твоё|можешь|сделай|покажи|скинь|" r"пришли|напиши|ответь|глянь|посмотри|расскажи|подскажи|найди|" r"создай|собери|проверь)\b", re.IGNORECASE) LINK_RE = re.compile(r"https?://") _LOCK = threading.Lock() _STATE = {"model": None, "loaded_sha": None} # ────────────────────────── конфиг/пути ────────────────────────── def mode() -> str: from . import decision v = decision._cfg("SIMBIOT_CORE_ADDRESSING", "shadow").lower() return v if v in ("on", "shadow", "off") else "shadow" def core_dir() -> str: d = (os.environ.get("SIMBIOT_CORE_DIR", "").strip() or os.path.join(_paths.home(), "core")) os.makedirs(os.path.join(d, HEAD), mode=0o700, exist_ok=True) return d def _ollama_base() -> str: from . import decision return decision._cfg("SIMBIOT_EMBED_URL", "http://127.0.0.1:11434") # ────────────────────────── эмбеддинги (кеш) ────────────────────────── def _cache(): db = sqlite3.connect(os.path.join(core_dir(), "emb.sqlite"), timeout=30) db.execute("CREATE TABLE IF NOT EXISTS emb (k TEXT PRIMARY KEY, v BLOB)") return db def embed(texts: list) -> list: """Эмбеддинги (нормированные списки float) с дисковым кешем; батч к Ollama.""" import numpy as np keys = [hashlib.sha1(f"{EMB_MODEL}|{t}".encode()).hexdigest() for t in texts] out = [None] * len(texts) db = _cache() try: for i, k in enumerate(keys): row = db.execute("SELECT v FROM emb WHERE k=?", (k,)).fetchone() if row: out[i] = np.frombuffer(row[0], dtype=np.float32) miss = [i for i, v in enumerate(out) if v is None] for s in range(0, len(miss), 32): chunk = miss[s:s + 32] body = json.dumps({"model": EMB_MODEL, "keep_alive": "2h", "input": [f"classification: {texts[i][:2000]}" for i in chunk]}) req = urllib.request.Request(_ollama_base() + "/api/embed", data=body.encode(), headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=60) as r: vecs = json.loads(r.read().decode())["embeddings"] for i, v in zip(chunk, vecs): a = np.asarray(v, dtype=np.float32) a /= (np.linalg.norm(a) or 1.0) out[i] = a db.execute("INSERT OR REPLACE INTO emb VALUES (?, ?)", (keys[i], a.tobytes())) db.commit() finally: db.close() return out # ────────────────────────── опыт из стора ────────────────────────── def _q(sql, args=()): db = sqlite3.connect(f"file:{es.DB_PATH}?mode=ro", uri=True, timeout=30) try: return db.execute(sql, args).fetchall() finally: db.close() def _once(agg_id: str, etype: str, data: dict) -> bool: """Событие ядра ровно один раз на aggregate_id (метки/фидбек не дублируются).""" if _q("SELECT 1 FROM events WHERE aggregate='core' AND aggregate_id=? LIMIT 1", (agg_id,)): return False es.append_event("core", agg_id, etype, data) return True def group_timeline(since: str = "") -> dict: """{chat_id: [msg, ...]} по времени: группы + вердикты + ответы бота + метки ядра.""" from .addressing import is_group msgs, by_id = {}, {} for aid, et, ts, d in _q( "SELECT aggregate_id, event_type, timestamp, data FROM events WHERE " "event_type IN ('MessageReceived','MessageTriaged','MessageIgnored','ReplySent'," "'CoreLabel','CoreFeedback') AND timestamp > ? ORDER BY id", (since,)): d = json.loads(d) if et == "MessageReceived": if not is_group(d): continue m = {**d, "id": aid, "ts": ts} by_id[aid] = m msgs.setdefault(d.get("chat_id"), []).append(m) continue if et in ("CoreLabel", "CoreFeedback"): m = by_id.get(d.get("msg")) if m is not None and d.get("head", HEAD) == HEAD: m.setdefault("labels", []).append( {"source": d.get("source") or "feedback", "y": float(d["label"]), "ts": ts}) continue m = by_id.get(aid) if m is None: continue if et == "MessageTriaged": m.update(addressed=True, rule=d.get("rule"), teacher_p=d.get("p")) elif et == "MessageIgnored" and d.get("reason") == "not_addressed": m.update(addressed=False, rule=d.get("rule"), teacher_p=d.get("p")) elif et == "ReplySent": t = d.get("text") or "" if not t.startswith("(сводный"): m["reply_text"] = t m.setdefault("reply_ts", ts) return msgs def _was_addressed(m: dict) -> bool: if m.get("addressed") is not None: return bool(m["addressed"]) lab = [l for l in m.get("labels", []) if l["source"] == "teacher_backfill"] return bool(lab and lab[-1]["y"] >= 0.5) # ────────────────────────── признаки ────────────────────────── def _ctx_lines(prior: list) -> list: out = [] for x in prior[-3:]: out.append(f"{x.get('first_name') or '?'}: " f"{(x.get('text') or '[' + str(x.get('kind')) + ']')[:300]}") if _was_addressed(x) and x.get("reply_text"): out.append(f"Cody: {x['reply_text'][:200]}") return out def flags(m: dict, prior: list) -> list: text = m.get("text") or "" last = prior[-1] if prior else None addressed_prior = [x for x in prior if _was_addressed(x)] others = {x.get("first_name") for x in prior if x.get("first_name")} - {m.get("first_name")} names_other = any(re.search(rf"(? 2) return [ 1.0 if (last and _was_addressed(last) and last.get("reply_text")) else 0.0, 1.0 if (addressed_prior and addressed_prior[-1].get("first_name") == m.get("first_name")) else 0.0, 1.0 if "?" in text else 0.0, 1.0 if SECOND_RE.search(text) else 0.0, 1.0 if names_other else 0.0, min(len(text) / 300.0, 1.0), 1.0 if LINK_RE.search(text) else 0.0, 0.0 if (m.get("kind") or "text") == "text" else 1.0, ] def featurize(items: list): """items: [(msg, prior_msgs)] → (X, E_msg) numpy; эмбеддинги одним батчем.""" import numpy as np msg_t = [f"{m.get('first_name') or '?'}: " f"{(m.get('text') or '[' + str(m.get('kind')) + ']')[:1500]}" for m, _ in items] ctx_t = ["\n".join(_ctx_lines(p)) or "(начало разговора)" for _, p in items] # эмбеддинг контекста в признаки не идёт: на истории правления он добавлял # 768 шумных измерений (5 фолдов: 0.822 с ним, 0.845 без); полезное из # контекста — во флагах (prev_is_bot, same_author…) del ctx_t vecs = embed(msg_t) n = len(items) em = np.stack(vecs) if n else np.zeros((0, EMB_DIM), np.float32) fl = np.asarray([flags(m, p) for m, p in items], dtype=np.float32).reshape(n, len(FLAG_NAMES)) return np.hstack([em, fl]), em # ────────────────────────── датасет ────────────────────────── def target(m: dict): """(y, вес, источник) — сильнейший источник метки; None — метки нет.""" labs = m.get("labels", []) for src in ("human", "feedback"): got = [l for l in labs if l["source"] == src] if got: return got[-1]["y"], WEIGHTS[src], src if m.get("rule") == "decision" and m.get("teacher_p") is not None: return float(m["teacher_p"]), WEIGHTS["teacher"], "teacher" got = [l for l in labs if l["source"] == "teacher_backfill"] if got: return got[-1]["y"], WEIGHTS["teacher_backfill"], "teacher_backfill" return None def dataset(): """Серая зона (правила её не решают) с меткой → [(msg, prior, y, w, src)] по времени.""" from .addressing import rule_verdict rows = [] for chat, msgs in group_timeline().items(): for i, m in enumerate(msgs): if m.get("rule") not in (None, "decision", "no_decision", "core"): continue # правило решило — ученику не нужно if m.get("rule") is None and rule_verdict(m)[0] is not None: continue t = target(m) if t is None: continue rows.append((m, msgs[max(0, i - 8):i], *t)) rows.sort(key=lambda r: r[0]["ts"]) return rows # ────────────────────────── модель ────────────────────────── EMB_SCALE = 4.0 # нормированные эмбеддинги: геометрию не трогаем, только масштаб NF = len(FLAG_NAMES) def _prep(model, X): """Эмбеддинги — как есть (×EMB_SCALE), стандартизуются только флаги. Стандартизация каждого из 1536 измерений на сотнях примеров раздувала шумовые измерения до масштаба полезных — регрессия учила шум.""" Z = X.astype(np.float64, copy=True) Z[:, :-NF] *= EMB_SCALE Z[:, -NF:] = (Z[:, -NF:] - model["mu"]) / model["sd"] return Z def train_logreg(X, y, w, l2=0.05, epochs=800, lr=0.2): """Логистическая регрессия: взвешенный log-loss + L2, градиентный спуск (Adam).""" fl = X[:, -NF:].astype(np.float64) model = {"mu": fl.mean(0), "sd": fl.std(0) + 1e-6} Z = _prep(model, X) wt = np.zeros(Z.shape[1]); b = 0.0 sw = w / w.sum() m1 = np.zeros_like(wt); v1 = np.zeros_like(wt); mb = vb = 0.0 for t in range(1, epochs + 1): p = 1 / (1 + np.exp(-(Z @ wt + b))) g = p - y gw = Z.T @ (sw * g) + l2 * wt gb = float((sw * g).sum()) m1 = .9 * m1 + .1 * gw; v1 = .999 * v1 + .001 * gw * gw mb = .9 * mb + .1 * gb; vb = .999 * vb + .001 * gb * gb wt -= lr * (m1 / (1 - .9 ** t)) / (np.sqrt(v1 / (1 - .999 ** t)) + 1e-8) b -= lr * (mb / (1 - .9 ** t)) / (np.sqrt(vb / (1 - .999 ** t)) + 1e-8) model.update(w=wt, b=b) return model def predict(model, X): Z = _prep(model, X) return 1 / (1 + np.exp(-(Z @ model["w"] + model["b"]))) def novelty(model, em): """1 − средняя косинусная близость к 5 ближайшим из обучающей памяти.""" import numpy as np mem = model.get("mem") if mem is None or not len(mem): return np.ones(len(em)) sims = em @ mem.T k = min(5, mem.shape[0]) top = -np.sort(-sims, axis=1)[:, :k] return 1 - top.mean(1) def metrics(model, X, y, w): import numpy as np p = np.clip(predict(model, X), 1e-6, 1 - 1e-6) ll = float(-(w * (y * np.log(p) + (1 - y) * np.log(1 - p))).sum() / w.sum()) acc = float((w * ((p >= .5) == (y >= .5))).sum() / w.sum()) return {"logloss": round(ll, 4), "acc": round(acc, 4), "n": int(len(y))} def _model_path(name): return os.path.join(core_dir(), HEAD, name) def _champion_meta(): try: with open(_model_path("champion.json")) as f: return json.load(f) except (OSError, ValueError): return {} def load_champion(): """Чемпион из файла (кеш по sha256) или None.""" import numpy as np meta = _champion_meta() if not meta.get("file"): return None if _STATE["loaded_sha"] == meta.get("sha256") and _STATE["model"] is not None: return _STATE["model"] path = _model_path(meta["file"]) with open(path, "rb") as f: blob = f.read() if hashlib.sha256(blob).hexdigest() != meta.get("sha256"): return None # файл подменён — ученика нет, решает учитель z = np.load(path) model = {k: z[k] for k in ("w", "mu", "sd", "mem")} model["b"] = float(z["b"]) model["nov_q"] = z["nov_q"] if "nov_q" in z else np.asarray([NOV_FAMILIAR, NOV_NOVEL]) _STATE.update(model=model, loaded_sha=meta["sha256"]) return model def _save(model, metrics_, n_total): import io import numpy as np name = f"model-{datetime.now(timezone.utc).strftime('%y%m%d%H%M%S')}.npz" buf = io.BytesIO() np.savez(buf, w=model["w"], b=np.float64(model["b"]), mu=model["mu"], sd=model["sd"], mem=model["mem"], nov_q=np.asarray(model.get("nov_q", [NOV_FAMILIAR, NOV_NOVEL]))) blob = buf.getvalue() path = _model_path(name) with open(path, "wb") as f: f.write(blob) os.chmod(path, 0o600) sha = hashlib.sha256(blob).hexdigest() prev = _champion_meta() meta = {"file": name, "sha256": sha, "metrics": metrics_, "n": n_total, "kind": "logreg", "features": {"emb": EMB_MODEL, "flags": FLAG_NAMES}, "ts": datetime.now(timezone.utc).isoformat(), "history": ([{k: prev[k] for k in ("file", "sha256", "metrics", "ts") if k in prev}] + prev.get("history", []))[:10] if prev.get("file") else []} tmp = _model_path(".champion.tmp") with open(tmp, "w") as f: json.dump(meta, f, ensure_ascii=False, indent=1) os.replace(tmp, _model_path("champion.json")) return meta def train(log=print) -> dict: """Претендент против чемпиона на временном холдауте; лучше — промоут.""" import numpy as np rows = dataset() if len(rows) < MIN_TRAIN: log(f"[core] {HEAD}: примеров {len(rows)} < {MIN_TRAIN} — обучение отложено") return {"promoted": False, "n": len(rows), "reason": "few_samples"} X, em = featurize([(r[0], r[1]) for r in rows]) y = np.asarray([r[2] for r in rows], np.float64) w = np.asarray([r[3] for r in rows], np.float64) cut = max(int(len(rows) * 0.8), len(rows) - max(10, len(rows) // 5)) ch = train_logreg(X[:cut], y[:cut], w[:cut]) ch["mem"] = em[:cut] m_ch = metrics(ch, X[cut:], y[cut:], w[cut:]) nov_h = novelty(ch, em[cut:]) nov_q = np.asarray([np.quantile(nov_h, .75), np.quantile(nov_h, .97)]) p_h = predict(ch, X[cut:]) sure = np.abs(p_h - .5) >= CONF m_ch["confident_share"] = round(float(sure.mean()), 3) m_ch["confident_acc"] = (round(float(((p_h[sure] >= .5) == (y[cut:][sure] >= .5)).mean()), 3) if sure.any() else None) m_ch["nov_q"] = [round(float(v), 4) for v in nov_q] champ = load_champion() m_old = None if champ is not None: try: m_old = metrics(champ, X[cut:], y[cut:], w[cut:]) except Exception: m_old = None # другой формат признаков — сравнивать нельзя, меняем better = m_old is None or m_ch["logloss"] < m_old["logloss"] - 1e-4 srcs = {} for r in rows: srcs[r[4]] = srcs.get(r[4], 0) + 1 rep = {"head": HEAD, "n": len(rows), "holdout": m_ch, "champion_holdout": m_old, "sources": srcs, "promoted": bool(better)} if better: full = train_logreg(X, y, w) full["mem"] = em full["nov_q"] = nov_q meta = _save(full, m_ch, len(rows)) _STATE.update(model=None, loaded_sha=None) es.append_event("core", f"promote:{HEAD}:{meta['file']}", "CoreModelPromoted", { "head": HEAD, "file": meta["file"], "sha256": meta["sha256"], "holdout": m_ch, "champion_holdout": m_old, "n": len(rows), "sources": srcs}) log(f"[core] {HEAD}: n={len(rows)} претендент {m_ch} чемпион {m_old} → " f"{'ПРОМОУТ' if better else 'чемпион остаётся'}") return rep def rollback() -> dict: meta = _champion_meta() hist = meta.get("history") or [] if not hist: return {"ok": False, "reason": "истории нет"} prev = hist[0] new = {**prev, "history": hist[1:], "ts": datetime.now(timezone.utc).isoformat(), "rolled_back_from": meta.get("file")} tmp = _model_path(".champion.tmp") with open(tmp, "w") as f: json.dump(new, f, ensure_ascii=False, indent=1) os.replace(tmp, _model_path("champion.json")) _STATE.update(model=None, loaded_sha=None) es.append_event("core", f"rollback:{HEAD}:{int(time.time())}", "CoreModelPromoted", { "head": HEAD, "file": prev.get("file"), "sha256": prev.get("sha256"), "rollback_from": meta.get("file")}) return {"ok": True, "file": prev.get("file")} # ────────────────────────── живой контур ────────────────────────── def route(p_student, nov, p_teacher, nov_q=None): fam, novel = (float(nov_q[0]), float(nov_q[1])) if nov_q is not None else (NOV_FAMILIAR, NOV_NOVEL) if p_student is not None and nov < fam and abs(p_student - 0.5) >= CONF: return "student" if nov >= novel and (p_teacher is None or 0.3 < p_teacher < 0.7): return "human" return "teacher" def shadow(m: dict, prior: list, teacher_p, final: bool): """Прогноз ученика на живом сообщении серой зоны → CorePrediction (+ вопрос Alex).""" if mode() == "off": return None try: model = load_champion() X, em = featurize([(m, prior)]) p = float(predict(model, X)[0]) if model is not None else None nov = float(novelty(model, em)[0]) if model is not None else 1.0 except Exception as e: return {"error": str(e)[:200]} r = route(p, nov, teacher_p, model.get("nov_q") if model is not None else None) mid = f"m-{m.get('update_id')}" rec = {"head": HEAD, "msg": mid, "p_student": None if p is None else round(p, 4), "novelty": round(nov, 4), "route": r, "teacher_p": teacher_p, "final": bool(final), "mode": mode(), "agree": None if p is None or teacher_p is None else (p >= .5) == (teacher_p >= .5)} _once(f"pred:{HEAD}:{mid}", "CorePrediction", rec) if r == "human": try: ask_human(m, prior) except Exception: pass # вопрос — бонус, не путь решения return rec def ask_human(m: dict, prior: list) -> bool: """Кнопки Alex в личку: «боту / не боту» — самая ценная метка (≤ N в сутки).""" limit = int(os.environ.get("SIMBIOT_CORE_ASK_PER_DAY", "5")) day = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat() asked = _q("SELECT COUNT(*) FROM events WHERE event_type='CoreQuestionAsked' " "AND timestamp > ?", (day,))[0][0] if asked >= limit: return False mid = f"m-{m.get('update_id')}" if _q("SELECT 1 FROM events WHERE aggregate='core' AND aggregate_id=?", (f"ask:{mid}",)): return False from . import tg_inbox chat = tg_inbox._default_chat_id() if chat is None: return False ctx = "\n".join(_ctx_lines(prior)) or "(начало разговора)" text = (f"🧠 Учусь понимать, когда ко мне обращаются в «{m.get('chat_title') or 'группе'}».\n\n" f"Контекст:\n{ctx[-600:]}\n\n" f"Сообщение — {m.get('first_name')}: «{(m.get('text') or '')[:400]}»\n\n" "Это было мне?") kb = {"inline_keyboard": [[ {"text": "Да, Коди", "callback_data": f"core:{HEAD}:{mid}:1"}, {"text": "Нет, не мне", "callback_data": f"core:{HEAD}:{mid}:0"}]]} tg_inbox._load_env() r = tg_inbox._api("sendMessage", {"chat_id": chat, "text": text, "reply_markup": json.dumps(kb)}, timeout=15) if (r or {}).get("ok"): es.append_event("core", f"ask:{mid}", "CoreQuestionAsked", {"head": HEAD, "msg": mid}) return True return False def on_callback(cq: dict, allowed: set) -> bool: """Ответ Alex на кнопку → CoreLabel source=human. Возвращает True, если наш callback.""" data = cq.get("data") or "" if not data.startswith("core:"): return False from . import tg_inbox _, head, mid, label = data.split(":", 3) user = (cq.get("from") or {}).get("id") ok = not allowed or str(user) in allowed if ok: _once(f"label:{head}:{mid}:human", "CoreLabel", {"head": head, "msg": mid, "label": int(label), "source": "human", "by": user}) try: tg_inbox._api("answerCallbackQuery", { "callback_query_id": cq.get("id"), "text": ("Запомнил 👍" if ok else "Нет прав")}, timeout=10) msg = cq.get("message") or {} if ok and msg.get("message_id"): tg_inbox._api("editMessageReplyMarkup", { "chat_id": (msg.get("chat") or {}).get("id"), "message_id": msg["message_id"], "reply_markup": json.dumps({"inline_keyboard": [[{ "text": "✓ " + ("мне" if label == "1" else "не мне"), "callback_data": "core:noop"}]]})}, timeout=10) except Exception: pass return True def harvest_feedback(log=None) -> int: """Неявная обратная связь чата за последние 2 часа → CoreFeedback (один раз).""" since = (datetime.now(timezone.utc) - timedelta(hours=2)).isoformat() n = 0 for chat, msgs in group_timeline(since).items(): for i, m in enumerate(msgs): if m.get("rule") not in ("decision", "no_decision", "core"): continue t0 = datetime.fromisoformat(m.get("reply_ts") or m["ts"]) nxt = [x for x in msgs[i + 1:] if datetime.fromisoformat(x["ts"]) - t0 <= timedelta(minutes=5)] if m.get("addressed") and m.get("reply_text"): neg = next((x for x in nxt if NEG_RE.search(x.get("text") or "")), None) if neg and _once(f"fb:{HEAD}:{m['id']}", "CoreFeedback", { "head": HEAD, "msg": m["id"], "label": 0, "reason": "neg: " + (neg.get("text") or "")[:120]}): n += 1 elif m.get("addressed") is False: poke = next((x for x in nxt if x.get("first_name") == m.get("first_name") and x.get("rule") in ("mention", "name", "reply_to_bot") and _is_poke(x)), None) if poke and _once(f"fb:{HEAD}:{m['id']}", "CoreFeedback", { "head": HEAD, "msg": m["id"], "label": 1, "reason": "poke: " + (poke.get("text") or "")[:120]}): n += 1 if n and log: log(f"[core] обратная связь: {n}") return n def _is_poke(x: dict) -> bool: """«Дёрнул бота» сразу после — короткое «@juno42bot» / «Коди?» без новой темы.""" t = re.sub(r"@\w+|коди|cody", "", x.get("text") or "", flags=re.IGNORECASE) return len(re.sub(r"[\W_]+", "", t)) <= 12 # ────────────────────────── бэкфилл учителем ────────────────────────── def backfill(log=print, limit: int = 2000) -> int: """Учитель размечает историю серой зоны (как живой: ответы бота в истории — только на адресованное по его же меткам) → CoreLabel source=teacher_backfill.""" from .addressing import model_verdict, rule_verdict n = 0 for chat, msgs in group_timeline().items(): inbox = {} for m in msgs: inbox[m["id"]] = m if m.get("rule") is not None or m.get("labels"): continue # живой вердикт или метка уже есть if rule_verdict(m)[0] is not None: continue p = model_verdict(m, {k: {**v, "addressed": _was_addressed(v)} for k, v in inbox.items() if k != m["id"]}) if p is None: continue if _once(f"label:{HEAD}:{m['id']}:teacher", "CoreLabel", { "head": HEAD, "msg": m["id"], "label": round(p, 4), "source": "teacher_backfill"}): m.setdefault("labels", []).append( {"source": "teacher_backfill", "y": p, "ts": m["ts"]}) n += 1 if n >= limit: return n log(f"[core] бэкфилл учителем: {n} меток") return n def status() -> dict: meta = _champion_meta() day = (datetime.now(timezone.utc) - timedelta(days=7)).isoformat() preds = [json.loads(d) for (d,) in _q( "SELECT data FROM events WHERE event_type='CorePrediction' AND timestamp > ?", (day,))] agree = [p["agree"] for p in preds if p.get("agree") is not None] srcs = {} for (d,) in _q("SELECT data FROM events WHERE event_type IN ('CoreLabel','CoreFeedback')"): s = json.loads(d).get("source") or "feedback" srcs[s] = srcs.get(s, 0) + 1 return { "head": HEAD, "mode": mode(), "champion": meta.get("file"), "champion_metrics": meta.get("metrics"), "champion_n": meta.get("n"), "champion_ts": meta.get("ts"), "labels": srcs, "shadow_7d": {"predictions": len(preds), "agree_with_teacher": (round(sum(agree) / len(agree), 3) if agree else None), "would_be_local": (round(sum(p["route"] == "student" for p in preds) / len(preds), 3) if preds else None), "would_ask_human": sum(p["route"] == "human" for p in preds)}, "questions_7d": _q("SELECT COUNT(*) FROM events WHERE event_type='CoreQuestionAsked' " "AND timestamp > ?", (day,))[0][0], }