d06a4e45bc
- бэкфилл индекса шёл 13 мин при 4 с CPU: _sync_once звал run_projections() (пишет в стор) и ждал блокировку живой лестницы — теперь только чтение; - круги людей — из событий TrustSet, задачи — из своих событий: отставшая проекция дала бы дефолтный круг 4 — ошибка в сторону утечки; - пороги релевантности по прод-индексу (nomic): релевантное 0.78–0.88, посторонний запрос ≤ 0.75 → MIN_SCORE 0.76 + окно 0.07 от лучшего; «рецепт борща» → пусто, «миграция Gennady» → нужная задача; - прод: индекс 1650 разговоров (личка 1 / правление 2 / агенты 3), 270 задач, 91 дайджест; поиск круга 2 не отдаёт записей круга 1. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
382 lines
21 KiB
Python
382 lines
21 KiB
Python
"""Simbiot Harness — динамическая память Коди: глубина D0–D3 и подгрузка по смыслу.
|
||
|
||
Третья ось куба 4×4×4 (круг доверия × тик × глубина). Раньше память тика —
|
||
12 последних реплик чата + свои сообщения за 48 ч + 1 запрос к базе знаний
|
||
кооператива: на «чем ты занимался последний месяц» Коди отвечал по базе
|
||
кооператива и наугад — поиска по СОБСТВЕННОЙ истории не было.
|
||
|
||
Индекс (<home>/memory/index.sqlite) — инкрементально из стора:
|
||
conv — сообщение + ответ Коди (ReplySent; батч — ответ головы);
|
||
out — собственные рассылки бота (OutboundMessage, не ответы);
|
||
task — задача и её итог (TaskCompleted);
|
||
digest — содержательные дайджесты час/день (заглушки — нет).
|
||
Каждая запись хранит КРУГ ПРИВАТНОСТИ: сказанное в чате видно всем его
|
||
участникам → круг записи = самый недоверенный участник чата (личка Alex —
|
||
1, «Правление» — 2); задача — её круг; дайджест — 1 (внутреннее). Поиск
|
||
отдаёт только записи с кругом ≥ круга тика: тик с правлением не поднимет
|
||
личку владельца. Эмбеддинги — nomic-embed-text (search_document/query).
|
||
|
||
Глубина (choose_depth): D0 — только текущее (односложные «ок/спасибо»);
|
||
D1 — разговор (как было); D2 — + эпизоды по смыслу (8 записей); D3 —
|
||
+ архив (15 записей + последние дайджесты). Правила: уровень (час — D2,
|
||
день — D3), слова о прошлом («помнишь», «вчера», «за месяц», «как
|
||
договаривались»…), ответ на старое сообщение бота, задачи в батче;
|
||
потолок — круг (trust.max_depth). Решение пишется (DepthChosen) — сырьё
|
||
для головы ядра «depth».
|
||
"""
|
||
import json
|
||
import os
|
||
import re
|
||
import sqlite3
|
||
import threading
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
from . import es
|
||
from . import paths as _paths
|
||
|
||
try:
|
||
import numpy as np
|
||
except ImportError:
|
||
np = None
|
||
|
||
KINDS = ("conv", "out", "task", "digest")
|
||
BUDGET = {2: (8, 5000), 3: (15, 12000)} # глубина → (записей, символов)
|
||
SYNC_BATCH = 400
|
||
# калибровка на прод-индексе (nomic, 2026-10-17): релевантное 0.78–0.88,
|
||
# посторонний запрос («рецепт борща») — шум ≤ 0.75 → абсолютный порог 0.76 +
|
||
# окно от лучшего совпадения (берём только близкое к лучшему)
|
||
MIN_SCORE = float(os.environ.get("SIMBIOT_MEMORY_MIN_SCORE", "0.76"))
|
||
REL_WINDOW = float(os.environ.get("SIMBIOT_MEMORY_REL_WINDOW", "0.07"))
|
||
_LOCK = threading.Lock()
|
||
_LAST = {} # уровень → последнее решение о глубине (для DepthChosen)
|
||
|
||
PAST_RE = re.compile(
|
||
r"помни|вспомн|вчера|позавчера|на прошлой|прошл(ый|ую|ой|ом|ые)\s+(раз|недел|месяц|встреч)|"
|
||
r"как\s+договаривал|мы\s+же|мы\s+обсуждал|ранее|раньше|в\s+прошлый|"
|
||
r"за\s+(последн\w+\s+)?(месяц|недел|дн)|что\s+(ты\s+)?(делал|сделал|было)|"
|
||
r"история|итог[иа]?\b|на\s+чём\s+останов|напомни", re.IGNORECASE)
|
||
ARCHIVE_RE = re.compile(r"за\s+(последн\w+\s+)?(месяц|недел|квартал|год)|"
|
||
r"чем\s+(ты\s+)?занимал|что\s+(ты\s+)?делал|отч[её]т|итоги|"
|
||
r"стратег|план\s+на", re.IGNORECASE)
|
||
TRIVIAL_RE = re.compile(r"^\W*(ок|окей|ok|ага|да|нет|спасибо|спс|понял|принято|супер|"
|
||
r"класс|норм|👍|🙏|❤️|🔥)\W*$", re.IGNORECASE)
|
||
|
||
|
||
# ────────────────────────── хранилище ──────────────────────────
|
||
|
||
def enabled() -> bool:
|
||
"""SIMBIOT_MEMORY=on (env > .env инстанса); дефолт off — тесты герметичны."""
|
||
from . import decision
|
||
return decision._cfg("SIMBIOT_MEMORY", "off").lower() == "on"
|
||
|
||
|
||
def _dir() -> str:
|
||
d = (os.environ.get("SIMBIOT_MEMORY_DIR", "").strip()
|
||
or os.path.join(_paths.home(), "memory"))
|
||
os.makedirs(d, mode=0o700, exist_ok=True)
|
||
return d
|
||
|
||
|
||
def _db():
|
||
db = sqlite3.connect(os.path.join(_dir(), "index.sqlite"), timeout=30)
|
||
db.execute("CREATE TABLE IF NOT EXISTS items (id TEXT PRIMARY KEY, kind TEXT, ts TEXT, "
|
||
"chat_id TEXT, circle INTEGER, who TEXT, text TEXT, emb BLOB)")
|
||
db.execute("CREATE TABLE IF NOT EXISTS meta (k TEXT PRIMARY KEY, v TEXT)")
|
||
return db
|
||
|
||
|
||
def _store_ro():
|
||
return sqlite3.connect(f"file:{es.DB_PATH}?mode=ro", uri=True, timeout=30)
|
||
|
||
|
||
def _embed(texts, prefix):
|
||
from . import kodi_core
|
||
return kodi_core.embed(texts, prefix)
|
||
|
||
|
||
# ────────────────────────── круг приватности записи ──────────────────────────
|
||
|
||
def chat_circle(chat_id, model: dict, senders: dict) -> int:
|
||
"""Круг чата = самый недоверенный известный участник; личка — круг собеседника."""
|
||
from . import trust
|
||
known = senders.get(str(chat_id)) or set()
|
||
if not known:
|
||
try:
|
||
cid = int(chat_id)
|
||
if cid > 0:
|
||
return trust.circle_of(f"tg:{cid}", model)
|
||
except (TypeError, ValueError):
|
||
pass
|
||
return trust.circle_of(trust.subject_of({"chat_id": chat_id}), model)
|
||
return max(trust.circle_of(s, model) for s in known)
|
||
|
||
|
||
# ────────────────────────── синхронизация ──────────────────────────
|
||
|
||
def sync(limit: int = SYNC_BATCH, log=None) -> int:
|
||
"""Новые события стора → записи индекса с эмбеддингами. Возвращает число записей."""
|
||
return _sync_once(limit, log)[1]
|
||
|
||
|
||
def sync_all(log=None) -> int:
|
||
"""Догнать стор целиком (бэкфилл): пачками, пока есть события."""
|
||
total = 0
|
||
while True:
|
||
nrows, nitems = _sync_once(SYNC_BATCH, log)
|
||
total += nitems
|
||
if nrows == 0:
|
||
return total
|
||
|
||
|
||
def _sync_once(limit: int, log=None) -> tuple:
|
||
"""(событий обработано, записей добавлено)."""
|
||
from . import trust
|
||
with _LOCK:
|
||
db = _db()
|
||
try:
|
||
last = int((db.execute("SELECT v FROM meta WHERE k='last'").fetchone() or ["0"])[0])
|
||
src = _store_ro()
|
||
try:
|
||
rows = src.execute(
|
||
"SELECT id, aggregate_id, event_type, timestamp, data FROM events WHERE id > ? "
|
||
"AND event_type IN ('MessageReceived','ReplySent','OutboundMessage',"
|
||
"'TaskSubmitted','TaskCompleted','DigestWritten') ORDER BY id LIMIT ?",
|
||
(last, limit)).fetchall()
|
||
if not rows:
|
||
return 0, 0
|
||
# круги — прямо из событий TrustSet, только чтение: run_projections()
|
||
# пишет в стор и ждал блокировку живой лестницы (бэкфилл 13 мин
|
||
# при 4 с CPU), а отставшая проекция дала бы дефолтный круг 4 —
|
||
# ошибку в сторону утечки (запись видна шире, чем надо)
|
||
tmap = {}
|
||
for (td,) in src.execute("SELECT data FROM events WHERE event_type='TrustSet' ORDER BY id"):
|
||
td = json.loads(td)
|
||
tmap[td["subject"]] = {"circle": td["circle"]}
|
||
model = {"trust": tmap}
|
||
senders = json.loads((db.execute("SELECT v FROM meta WHERE k='senders'")
|
||
.fetchone() or ["{}"])[0])
|
||
senders = {k: set(v) for k, v in senders.items()}
|
||
pending = [] # (id, kind, ts, chat, circle, who, text)
|
||
for eid, aid, et, ts, d in rows:
|
||
d = json.loads(d)
|
||
if et == "MessageReceived":
|
||
if d.get("chat_id") is not None:
|
||
senders.setdefault(str(d["chat_id"]), set()).add(trust.subject_of(d))
|
||
elif et == "ReplySent":
|
||
rec = _conv_item(src, db, aid, d, ts, model, senders, pending)
|
||
if rec:
|
||
pending.append(rec)
|
||
# исходящее с тем же текстом — это ответ, не рассылка
|
||
db.execute("DELETE FROM items WHERE kind='out' AND chat_id=? "
|
||
"AND substr(text,1,200)=?",
|
||
(str(rec[3]), (d.get("text") or "")[:200]))
|
||
pending[:] = [p for p in pending if not (
|
||
p[1] == "out" and p[3] == rec[3]
|
||
and p[6][:200] == (d.get("text") or "")[:200])]
|
||
elif et == "OutboundMessage":
|
||
t = (d.get("text") or "").strip()
|
||
# ответ на входящее (ReplySent мог прийти раньше) — это conv, не рассылка
|
||
is_reply = any(p[1] == "conv" and p[3] == str(d.get("chat_id"))
|
||
and p[6].split("\nКоди: ", 1)[-1][:200] == t[:200]
|
||
for p in pending) or db.execute(
|
||
"SELECT 1 FROM items WHERE kind='conv' AND chat_id=? AND "
|
||
"instr(text, ?) > 0 LIMIT 1", (str(d.get("chat_id")),
|
||
"\nКоди: " + t[:200])).fetchone()
|
||
if t and not t.startswith("(сводный") and not is_reply:
|
||
pending.append((f"out:{aid}", "out", ts, str(d.get("chat_id")),
|
||
chat_circle(d.get("chat_id"), model, senders),
|
||
"Коди", t[:1500]))
|
||
elif et == "TaskCompleted":
|
||
t = _task_info(src, aid)
|
||
ans = (d.get("answer") or d.get("result") or "")
|
||
if t.get("title") or ans:
|
||
pending.append((f"task:{aid}", "task", ts, "", trust.norm(t.get("circle")),
|
||
"задача", f"Задача: {t.get('title') or aid}\n"
|
||
f"Итог: {str(ans)[:1200]}"))
|
||
elif et == "DigestWritten":
|
||
txt = (d.get("text") or "").strip()
|
||
if (d.get("level") in ("hour", "day") and not d.get("stub") and txt
|
||
and not txt.startswith("[gate:quiet]") and "LLM не запускался" not in txt):
|
||
pending.append((f"digest:{d.get('level')}:{d.get('period')}", "digest",
|
||
ts, "", 1, f"дайджест {d.get('level')}", txt[:1500]))
|
||
finally:
|
||
src.close()
|
||
if pending:
|
||
vecs = _embed([p[6] for p in pending], "search_document")
|
||
db.executemany("INSERT OR REPLACE INTO items VALUES (?,?,?,?,?,?,?,?)",
|
||
[(*p, np.asarray(v, np.float32).tobytes())
|
||
for p, v in zip(pending, vecs)])
|
||
db.execute("INSERT OR REPLACE INTO meta VALUES ('last', ?)", (str(rows[-1][0]),))
|
||
db.execute("INSERT OR REPLACE INTO meta VALUES ('senders', ?)",
|
||
(json.dumps({k: sorted(v) for k, v in senders.items()}),))
|
||
db.commit()
|
||
if log and pending:
|
||
log(f"memory: +{len(pending)} записей")
|
||
return len(rows), len(pending)
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def _task_info(src, tid: str) -> dict:
|
||
"""Заголовок и круг задачи — из её событий (по индексу агрегата)."""
|
||
t = {}
|
||
for et, d in src.execute("SELECT event_type, data FROM events WHERE aggregate='task' "
|
||
"AND aggregate_id=? AND event_type IN ('TaskSubmitted','TaskCircleSet') "
|
||
"ORDER BY id", (tid,)):
|
||
d = json.loads(d)
|
||
if et == "TaskSubmitted":
|
||
t.update(title=d.get("title"), circle=d.get("circle"))
|
||
else:
|
||
t["circle"] = d.get("circle")
|
||
return t
|
||
|
||
|
||
def _conv_item(src, db, aid, d, ts, model, senders, pending=()):
|
||
row = src.execute("SELECT data FROM events WHERE aggregate='inbox' AND aggregate_id=? "
|
||
"AND event_type='MessageReceived' LIMIT 1", (aid,)).fetchone()
|
||
if not row:
|
||
return None
|
||
m = json.loads(row[0])
|
||
text = (m.get("text") or "").strip()
|
||
if not text:
|
||
return None
|
||
reply = d.get("text") or ""
|
||
if reply.startswith("(сводный"):
|
||
head = ((d.get("batch") or {}).get("head"))
|
||
hp = next((p[6] for p in pending if p[0] == f"conv:{head}"), None) # голова в этой пачке
|
||
hr = (hp,) if hp else db.execute("SELECT text FROM items WHERE id=?", (f"conv:{head}",)).fetchone()
|
||
reply = (hr[0].split("\nКоди: ", 1)[-1] if hr else "(общий ответ на несколько сообщений)")
|
||
who = m.get("first_name") or m.get("username") or str(m.get("user_id"))
|
||
return (f"conv:{aid}", "conv", ts, str(m.get("chat_id")),
|
||
chat_circle(m.get("chat_id"), model, senders), who,
|
||
f"{who}: {text[:700]}\nКоди: {reply[:700]}")
|
||
|
||
|
||
# ────────────────────────── поиск ──────────────────────────
|
||
|
||
def search(query: str, circle: int, k: int = 8, kinds=KINDS, exclude_chat=None,
|
||
exclude_since: str = "") -> list:
|
||
"""Записи по смыслу, видимые кругу тика (круг записи ≥ круга тика)."""
|
||
if np is None or not query.strip():
|
||
return []
|
||
db = _db()
|
||
try:
|
||
rows = db.execute(
|
||
f"SELECT id, kind, ts, chat_id, circle, who, text, emb FROM items WHERE circle >= ? "
|
||
f"AND kind IN ({','.join('?' * len(kinds))})", (int(circle), *kinds)).fetchall()
|
||
finally:
|
||
db.close()
|
||
if exclude_chat is not None and exclude_since:
|
||
# свежий разговор этого чата уже в контексте (D1) — не дублируем
|
||
rows = [r for r in rows if not (r[3] == str(exclude_chat) and r[2] >= exclude_since)]
|
||
if not rows:
|
||
return []
|
||
q = np.asarray(_embed([query[:2000]], "search_query")[0], np.float32)
|
||
M = np.stack([np.frombuffer(r[7], np.float32) for r in rows])
|
||
sims = M @ q
|
||
order = np.argsort(-sims)[:k]
|
||
return [{"id": rows[i][0], "kind": rows[i][1], "ts": rows[i][2], "circle": rows[i][4],
|
||
"text": rows[i][6], "score": round(float(sims[i]), 3)} for i in order]
|
||
|
||
|
||
def stats() -> dict:
|
||
db = _db()
|
||
try:
|
||
by = dict(db.execute("SELECT kind, COUNT(*) FROM items GROUP BY kind").fetchall())
|
||
circ = dict(db.execute("SELECT circle, COUNT(*) FROM items GROUP BY circle").fetchall())
|
||
last = (db.execute("SELECT v FROM meta WHERE k='last'").fetchone() or ["0"])[0]
|
||
finally:
|
||
db.close()
|
||
return {"items": by, "by_circle": circ, "last_event": int(last)}
|
||
|
||
|
||
# ────────────────────────── глубина ──────────────────────────
|
||
|
||
def choose_depth(level: str, inbox: list, batch: list, circle: int) -> tuple:
|
||
"""(глубина 0..3, причины). Правила + потолок круга; голова ядра — в тени."""
|
||
from . import trust
|
||
reasons = []
|
||
base = {"min": 1, "hour": 2, "day": 3}.get(level, 1)
|
||
d = base
|
||
reasons.append(f"уровень {level} → D{base}")
|
||
texts = [(m.get("text") or "") for m in inbox or []]
|
||
if inbox and not batch and all(TRIVIAL_RE.match(t.strip() or "x") for t in texts) \
|
||
and not any(m.get("reply_to_full") for m in inbox):
|
||
d = 0
|
||
reasons = ["односложные реплики → D0"]
|
||
joined = "\n".join(texts + [f"{t.get('title') or ''} {t.get('details') or ''}" for t in batch or []])
|
||
if PAST_RE.search(joined):
|
||
d = max(d, 2)
|
||
reasons.append("отсылка к прошлому → D2")
|
||
if ARCHIVE_RE.search(joined):
|
||
d = max(d, 3)
|
||
reasons.append("обзор периода → D3")
|
||
old = (datetime.now(timezone.utc) - timedelta(hours=48)).isoformat()
|
||
for m in inbox or []:
|
||
if m.get("reply_to_full") and (m.get("reply_to_ts") or "") < old:
|
||
d = max(d, 2)
|
||
reasons.append("ответ на давнее сообщение бота → D2")
|
||
if batch:
|
||
d = max(d, 2)
|
||
reasons.append("задачи в батче → D2 (прошлые похожие)")
|
||
# голова ядра «depth» (harness/kodi_core.py): shadow — прогноз в причинах;
|
||
# on — уверенное «нужна глубина» поднимает D0/D1 до D2
|
||
student = None
|
||
try:
|
||
from . import kodi_core
|
||
student = kodi_core.depth_predict({"query": joined[:1500], "level": level,
|
||
"circle": circle, "reasons": " ".join(reasons)})
|
||
except Exception:
|
||
student = None
|
||
if student:
|
||
p, sure = student
|
||
reasons.append(f"ученик depth: p={p:.2f}{' уверен' if sure else ''}")
|
||
if sure and p >= .5 and d < 2 and kodi_core.mode(kodi_core.DEPTH_HEAD) == "on":
|
||
d = 2
|
||
reasons.append("ученик → D2")
|
||
cap = trust.max_depth(circle)
|
||
if d > cap:
|
||
reasons.append(f"потолок круга {circle} → D{cap}")
|
||
d = cap
|
||
_LAST[level] = {"depth": d, "reasons": reasons, "circle": circle, "query": joined[:500]}
|
||
return d, reasons
|
||
|
||
|
||
def context_block(depth: int, level: str, inbox: list, batch: list, circle: int) -> str:
|
||
"""Текст памяти для контекста тика на глубине ≥ 2 (пусто — для D0/D1)."""
|
||
if depth < 2 or not enabled():
|
||
return ""
|
||
k, budget = BUDGET[min(depth, 3)]
|
||
query = "\n".join([(m.get("text") or "") + " " + (m.get("reply_to_full") or "")[:300]
|
||
for m in inbox or []]
|
||
+ [f"{t.get('title') or ''}\n{(t.get('details') or '')[:400]}" for t in batch or []])
|
||
if not query.strip():
|
||
query = {"hour": "итоги последнего часа, задачи, решения",
|
||
"day": "итоги дня, решения, задачи, договорённости"}.get(level, "недавние дела")
|
||
chat = inbox[0].get("chat_id") if inbox else None
|
||
since = (datetime.now(timezone.utc) - timedelta(hours=48)).isoformat()
|
||
try:
|
||
hits = search(query, circle, k=k, exclude_chat=chat, exclude_since=since)
|
||
except Exception as e:
|
||
return f"ПАМЯТЬ: поиск недоступен ({str(e)[:80]}) — отвечай из контекста тика."
|
||
lines, used = [], 0
|
||
top = hits[0]["score"] if hits else 0.0
|
||
for h in hits:
|
||
if h["score"] < MIN_SCORE or h["score"] < top - REL_WINDOW:
|
||
continue
|
||
line = f"- [{h['kind']} {h['ts'][:10]}] {h['text'][:900]}"
|
||
if used + len(line) > budget:
|
||
break
|
||
lines.append(line)
|
||
used += len(line)
|
||
if _LAST.get(level):
|
||
_LAST[level]["hits"] = len(lines)
|
||
if not lines:
|
||
return ""
|
||
head = (f"ПАМЯТЬ — НАЙДЕНО ПО СМЫСЛУ (глубина D{depth}; записи, видимые кругу {circle}; "
|
||
"это прошлое — опирайся, но проверяй актуальность):")
|
||
return head + "\n" + "\n".join(lines)
|
||
|
||
|
||
def last_choice(level: str) -> dict:
|
||
return dict(_LAST.get(level) or {})
|