rpi: tg — кэш векторов (npz) и служба tg-embed: поиск ~6 с вместо ~40

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-18 20:01:45 +03:00
parent 48c129615f
commit 3bdb75d465
4 changed files with 114 additions and 12 deletions
@@ -299,7 +299,7 @@ Docker CE 29.9 + Compose v5.6 из репо download.docker.com (в Debian ст
 - Проект `tg_export` (Telethon) перенесён с Mac: код, юниты и установка — [`stacks/rpi-tg-export/`](../stacks/rpi-tg-export/README.md). На малине — **второй аккаунт владельца**, докачка **только текста** раз в сутки (`tg-export.timer`, 04:30, `run --no-media --archived`), выгрузка в `/mnt/backup/tg/` (tgexport:astra 2750/640 — Astra читает). Старая выгрузка первого аккаунта (~14 ГБ) — на Mac, не переносилась (решение владельца).
 - Отдельный системный пользователь `tgexport` (без шелла): сессия `/var/lib/tg-export/tg.session` (600) = полный доступ к аккаунту — Astra её не видит; ключи приложения `/etc/tg-export/env` (root:tgexport 640, те же `api_id`/`api_hash`, что на Mac). Вход — владелец: `ssh -t rpi sudo -u tgexport tg-export login`.
 - Мониторинг — `backup-metrics backup_tg_export` → `BackupFailed`/`BackupStale` (>30 ч). ⚠️ Первый прогон может идти дольше суток из-за FloodWait.
- **Поиск и сводка** (с 2026-10-18): докачка каждый час (было раз в сутки 04:30) + индекс `/mnt/backup/tg/index.db` (FTS5 + эмбеддинги potion-multilingual-128M, fastembed в venv, модели `/opt/tg-export/models`); `tg search` для Astra, темы `tg topic`, сводка `tg-report.timer` 08:00 Asia/Nicosia через Codex → очередь outbox. Навык и правила — astra-bot `11dd4fc`. Подробно — README стека. Вход во второй аккаунт (@mikpokrovsky) — 2026-10-18; выгружаются только 6 выбранных владельцем чатов (`--only`, см. README стека). ⚠️ 2026-10-18 18:45–18:51 по ошибке шла выгрузка всех групп и каналов (успела только группа Python, 35 500 сообщений) — остановлена, данные удалены по решению владельца. ⚠️ Таймер выгрузки **выключен** до вступления владельца в @cyprusdiy, @cymedic и Cyprus Car – Chat; сводка включена, без тем и индекса молчит.
+- **Поиск и сводка** (с 2026-10-18): докачка каждый час (было раз в сутки 04:30) + индекс `/mnt/backup/tg/index.db` (FTS5 + эмбеддинги potion-multilingual-128M, fastembed в venv, модели `/opt/tg-export/models`); `tg search` для Astra, темы `tg topic`, сводка `tg-report.timer` 08:00 Asia/Nicosia через Codex → очередь outbox. Навык и правила — astra-bot `11dd4fc`. Служба `tg-embed` (astra) держит модель эмбеддингов в памяти для быстрых запросов (~6 с вместо ~40). Первая полная выгрузка 2026-10-18 19:49: 12 чатов, 303 тыс. сообщений с 2024-10-18, индекс 85 580 кусков, 1,2 ГБ на HDD. Подробно — README стека. Вход во второй аккаунт (@mikpokrovsky) — 2026-10-18; выгружаются только 6 выбранных владельцем чатов (`--only`, см. README стека). ⚠️ 2026-10-18 18:45–18:51 по ошибке шла выгрузка всех групп и каналов (успела только группа Python, 35 500 сообщений) — остановлена, данные удалены по решению владельца. ⚠️ Таймер выгрузки **выключен** до вступления владельца в @cyprusdiy, @cymedic и Cyprus Car – Chat; сводка включена, без тем и индекса молчит.
 - Доступ: тексты — только для владельца; Оле и Поле закрыто (правило в `codex/AGENTS.md` и скилл `tg`, astra-bot `27bf8e6`), как почта.

 ## Читалка books → Obsidian (с 2026-10-18)
@@ -45,6 +45,10 @@ sudo systemctl enable --now tg-export.timer
  Сравнение на 8 тыс. сообщений групп 2026-10-18: potion 57 с, `paraphrase-multilingual-MiniLM-L12-v2` 915 с при
  сопоставимом качестве поиска (MiniLM к тому же видит только ~128 токенов куска). e5-large/jina-v3 на Pi 4 —
  дни на историю. Сменить — `TG_EMB_MODEL` + переиндексация (удалить `index.db`).
+- **Скорость** (2026-10-18, 85 тыс. кусков): векторы всех кусков — `index.emb.npz` рядом с индексом (пересобирается после
+  индексации; из SQLite на HDD было ~40 с на запрос); модель держит служба **`tg-embed`** (astra, сокет
+  `/run/tg-embed/embed.sock` 0600, ~140 МБ RSS) — загрузка модели в процесс ~11 с (fastembed/ONNX; model2vec ещё
+  медленнее, 28 с — проверено, векторы те же). Итого `tg search` ~6 с. Нет службы — `tg` грузит модель сам.
 - `tg search ЗАПРОС [--chat] [--since 90d] [-n]` — гибрид BM25 + косинус (RRF), куски со ссылками t.me
  (`t.me/<username>/<id>` или `t.me/c/<id>/<msg>`). ⚠️ Одна формулировка часто промахивается
  («посоветуйте спортзал» не нашло «фитнес в отелях») — Astra по навыку `tg` переформулирует 2–4 раза.
@@ -8,6 +8,7 @@
  tg report [--hours 24] [--dry]   сводка по темам за период → владельцу в Telegram (--dry — только напечатать)
  tg stats            сколько чатов/сообщений/кусков, когда обновлялось
  tg index            (сервис tg-export, юзер tgexport) — дописать в индекс новые сообщения выгрузки
+  tg serve            (служба tg-embed, юзер astra) — держать модель в памяти, векторы запросов по unix-сокету

 Индекс — /mnt/backup/tg/index.db (SQLite: FTS5 + эмбеддинги potion-multilingual-128M), пишет tgexport,
 читают все из группы astra. Кусок = подряд идущие сообщения одного чата с паузами < 20 мин (до ~800 знаков).
@@ -38,13 +39,44 @@ TZ = ZoneInfo("Asia/Nicosia")
 GAP = 20 * 60          # пауза, после которой начинается новый кусок
 CHUNK_CHARS = int(os.environ.get("TG_CHUNK_CHARS", "800"))
 TG_MAX = 3800          # одно сообщение Telegram (лимит 4096) — сводку режем на части
+# векторы всех кусков одним файлом рядом с индексом: чтение 85 тыс. BLOB из SQLite на HDD — ~40 с на запрос,
+# а np.load(mmap) — доли секунды; пересобирается после каждой индексации
+CACHE = DB.with_suffix(".emb.npz")

 _emb = None
+SOCK = os.environ.get("TG_EMBED_SOCK", "/run/tg-embed/embed.sock")


-def embed(texts):
+def _embed_remote(texts):
+    """Векторы от службы tg-embed (модель уже в памяти) — иначе None. Загрузка модели в процесс — ~11 с на Pi 4."""
+    import socket
+    import numpy as np
+    if not os.path.exists(SOCK):
+        return None
+    try:
+        with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s:
+            s.settimeout(30)
+            s.connect(SOCK)
+            s.sendall(json.dumps({"texts": texts}).encode() + b"\n")
+            buf = b""
+            while not buf.endswith(b"\n"):
+                part = s.recv(1 << 20)
+                if not part:
+                    break
+                buf += part
+        r = json.loads(buf)
+        return np.frombuffer(bytes.fromhex(r["vecs"]), dtype=np.float32).reshape(r["n"], r["dim"])
+    except (OSError, ValueError, KeyError):
+        return None
+
+
+def embed(texts, remote=True):
    global _emb
    import numpy as np
+    if remote and len(texts) <= 64:
+        v = _embed_remote(list(texts))
+        if v is not None:
+            return v
    if _emb is None:
        from fastembed import TextEmbedding
        _emb = TextEmbedding(EMB_MODEL, cache_dir=MODELS, threads=4)
@@ -162,9 +194,24 @@ def cmd_index(a):
        total += len(new)
    c.execute("INSERT OR REPLACE INTO meta VALUES ('indexed', ?)", (dt.datetime.now(TZ).isoformat(timespec="seconds"),))
    c.commit()
+    if total or not CACHE.exists():
+        build_cache(c)
    print(f"проиндексировано новых сообщений: {total}")


+def build_cache(c):
+    import numpy as np
+    rows = c.execute("SELECT id, chat_id, ts1, emb FROM chunk ORDER BY id").fetchall()
+    if not rows:
+        return
+    tmp = CACHE.with_name(CACHE.name + ".tmp.npz")
+    np.savez(tmp, ids=np.array([r["id"] for r in rows], dtype=np.int64),
+             chat=np.array([r["chat_id"] for r in rows], dtype=np.int64),
+             ts=np.array([r["ts1"] for r in rows], dtype=np.int64),
+             emb=np.frombuffer(b"".join(r["emb"] for r in rows), dtype=np.float32).reshape(len(rows), -1))
+    os.replace(tmp, CACHE)
+
+
 # ---------------------------------------------------------------- поиск

 def link(chat, msg_id):
@@ -216,16 +263,25 @@ def search(q, chat=None, since=None, n=12, min_sim=None):
            ranks[row["id"]] = ranks.get(row["id"], 0) + 1 / (60 + r)
            by_words.add(row["id"])
    qv = embed([q])[0]
-    ids, mats = [], []
-    for row in c.execute(f"SELECT c.id, c.emb FROM chunk c WHERE {w}", args):   # 384 float32 на кусок — в памяти ок
-        ids.append(row["id"])
-        mats.append(row["emb"])
-    if ids:
-        m = np.frombuffer(b"".join(mats), dtype=np.float32).reshape(len(ids), -1)
-        s = m @ qv
-        for r, i in enumerate(np.argsort(-s)[:60]):
-            sims[ids[i]] = float(s[i])
-            ranks[ids[i]] = ranks.get(ids[i], 0) + 1 / (60 + r)
+    if CACHE.exists():
+        z = np.load(CACHE)
+        ids, emb, mask = z["ids"], z["emb"], np.ones(len(z["ids"]), dtype=bool)
+        if since:
+            mask &= z["ts"] >= since
+        if chat:
+            cids = [r[0] for r in c.execute("SELECT id FROM chat WHERE title LIKE ? OR username LIKE ?", (f"%{chat}%", f"%{chat}%"))]
+            mask &= np.isin(z["chat"], cids)
+        ids, emb = ids[mask], emb[mask]
+    else:                                  # кэша нет (старый индекс) — медленно, из базы
+        rows = c.execute(f"SELECT c.id, c.emb FROM chunk c WHERE {w}", args).fetchall()
+        ids = np.array([r["id"] for r in rows], dtype=np.int64)
+        emb = np.frombuffer(b"".join(r["emb"] for r in rows), dtype=np.float32).reshape(len(rows), -1) if rows else None
+    if len(ids):
+        s = emb @ qv
+        top_i = np.argpartition(-s, min(60, len(s) - 1))[:60]
+        for r, i in enumerate(top_i[np.argsort(-s[top_i])]):
+            sims[int(ids[i])] = float(s[i])
+            ranks[int(ids[i])] = ranks.get(int(ids[i]), 0) + 1 / (60 + r)
    top = sorted(ranks, key=ranks.get, reverse=True)
    if min_sim is not None:      # для сводки: найденное словами — всегда, по смыслу — только достаточно близкое
        top = [i for i in top if i in by_words or sims.get(i, 0) >= min_sim]
@@ -361,6 +417,30 @@ def cmd_report(a):
    print(f"сводка поставлена в очередь: {len(parts)} сообщ.")


+def cmd_serve(a):
+    """Служба tg-embed: модель в памяти, векторы запросов по unix-сокету (одна строка JSON → одна строка JSON)."""
+    import socketserver
+    embed(["прогрев"], remote=False)
+
+    class H(socketserver.StreamRequestHandler):
+        def handle(self):
+            try:
+                req = json.loads(self.rfile.readline())
+                v = embed(req["texts"][:64], remote=False)
+                out = {"n": v.shape[0], "dim": v.shape[1], "vecs": v.astype("float32").tobytes().hex()}
+            except Exception as e:        # плохой запрос не роняет службу
+                out = {"error": str(e)}
+            self.wfile.write(json.dumps(out).encode() + b"\n")
+
+    if os.path.exists(SOCK):
+        os.unlink(SOCK)
+    old = os.umask(0o077)                 # сокет — только своему пользователю (astra)
+    srv = socketserver.UnixStreamServer(SOCK, H)
+    os.umask(old)
+    print(f"tg-embed: {EMB_MODEL} в памяти, сокет {SOCK}", flush=True)
+    srv.serve_forever()
+
+
 def main():
    p = argparse.ArgumentParser(prog="tg", description=__doc__.split("\n")[0])
    sub = p.add_subparsers(dest="cmd", required=True)
@@ -373,6 +453,7 @@ def main():
    s = sub.add_parser("report"); s.add_argument("--hours", type=int, default=24); s.add_argument("--dry", action="store_true")
    sub.add_parser("stats")
    s = sub.add_parser("index"); s.add_argument("--quiet", action="store_true")
+    sub.add_parser("serve")
    a = p.parse_args()
    globals()["cmd_" + a.cmd](a)

@@ -0,0 +1,17 @@
+[Unit]
+Description=Telegram-поиск: модель эмбеддингов в памяти (векторы запросов tg search/report за доли секунды)
+
+[Service]
+User=astra
+Group=astra
+# сокет /run/tg-embed/embed.sock, только astra (0700 каталог, 0600 сокет)
+RuntimeDirectory=tg-embed
+RuntimeDirectoryMode=0700
+ExecStart=/opt/tg-export/venv/bin/python /opt/tg-export/tgrag.py serve
+Restart=on-failure
+RestartSec=30
+Nice=5
+MemoryMax=1500M
+
+[Install]
+WantedBy=multi-user.target