b780c9ba9e
- tools/hermes_history_ingest.py: параметры --hermes-db/--rag-db/--prefix/--cutoff/--sessions-cards; rag.DB перекрывается атрибутом (вендор фиксирует при импорте); cutoff='' = вся история - scripts/migrate_hermes.py: --history-full по умолчанию ON в --apply (rag_index → ingest), --no-history-full / --history-agent <имя> для префикса - docs/migration-hermes.md: раздел 6.5 - e2e: полный apply на смоук-базе — +350 чанков / 75 файлов за 24с, history_full.ok=true
230 lines
10 KiB
Python
230 lines
10 KiB
Python
#!/usr/bin/env python3
|
||
"""Hermes → Simbiot: историческая память (t-261003150347-896567).
|
||
|
||
Тянет сессии из /root/.hermes/state.db (до CUTOFF — старт харнеса 2026-10-03),
|
||
строит выжимки (диалоговые карточки + помесячные роллапы) и индексирует их
|
||
в RAG-базу симбиота /root/rag/rag.db (та же схема chunks, что и у остального
|
||
корпуса: fastembed BAAI/bge-small-en-v1.5, чанк 900/150).
|
||
|
||
Идемпотентно: перед вставкой чанков пути `hermes-history/...` удаляются.
|
||
Сырые логи не индексируются — только цитаты user + ключевые ответы assistant.
|
||
|
||
Usage:
|
||
hermes_history_ingest.py # полная загрузка
|
||
hermes_history_ingest.py --limit 3 # смоук на 3 сессиях
|
||
hermes_history_ingest.py --dry-run # статистика без записи
|
||
"""
|
||
import argparse
|
||
import os
|
||
import sqlite3
|
||
import sys
|
||
import time
|
||
|
||
RAG_DIR = os.environ.get("RAG_DIR", "/root/rag") # где лежит rag.py (вендор)
|
||
RAG_VENV_PY = os.environ.get("RAG_VENV_PY", "/root/rag-venv/bin/python3")
|
||
sys.path.insert(0, RAG_DIR)
|
||
import rag # noqa: E402 (общий индексатор: chunk_text/embed/get_db)
|
||
|
||
HERMES_DB = os.environ.get("HERMES_DB", "/root/.hermes/state.db")
|
||
RAG_DB = os.environ.get("RAG_DB", "/root/rag/rag.db")
|
||
# Старт текущего харнеса: ивенты до этой даты симбиот «не помнит» (для
|
||
# ручного прогона на этом хосте). При миграции нового агента cutoff не нужен —
|
||
# передаётся --cutoff явно или берётся «всё» (пустая строка).
|
||
CUTOFF_UTC = "2026-10-03 00:00:00"
|
||
PREFIX = "hermes-history"
|
||
P_USER, P_ASSIST = 40, 10 # лимиты строк на сессию
|
||
LEN_USER, LEN_ASSIST = 170, 220
|
||
HUMAN_SOURCES = ("telegram", "cli", "matrix")
|
||
|
||
|
||
def iso2epoch(s):
|
||
import calendar
|
||
t = time.strptime(s, "%Y-%m-%d %H:%M:%S")
|
||
return calendar.timegm(t)
|
||
|
||
|
||
CUTOFF_EPOCH = iso2epoch(CUTOFF_UTC)
|
||
|
||
|
||
def clean(s, limit):
|
||
s = " ".join((s or "").split())
|
||
if len(s) > limit:
|
||
s = s[: limit - 1].rstrip() + "…"
|
||
return s
|
||
|
||
|
||
def usable(s):
|
||
s = (s or "").strip()
|
||
if not s or s.startswith("<"): # системные вставки/контейнеры
|
||
return False
|
||
return True
|
||
|
||
|
||
def fetch_sessions(db, cutoff_epoch=None):
|
||
if cutoff_epoch:
|
||
return db.execute(
|
||
"""SELECT id, source, title, started_at, message_count
|
||
FROM sessions
|
||
WHERE started_at < ?
|
||
ORDER BY started_at""",
|
||
(cutoff_epoch,),
|
||
).fetchall()
|
||
return db.execute(
|
||
"""SELECT id, source, title, started_at, message_count
|
||
FROM sessions
|
||
ORDER BY started_at"""
|
||
).fetchall()
|
||
|
||
|
||
def session_card(db, sid):
|
||
"""Карточка сессии: дата/титул + цитаты user и ключевые ответы assistant."""
|
||
msgs = db.execute(
|
||
"""SELECT role, content, timestamp FROM messages
|
||
WHERE session_id=? AND role IN ('user','assistant')
|
||
ORDER BY id""",
|
||
(sid,),
|
||
).fetchall()
|
||
users = [m for m in msgs if m[0] == "user" and usable(m[1])]
|
||
assists = [m for m in msgs if m[0] == "assistant" and usable(m[1]) and len(m[1]) > 60]
|
||
if not users:
|
||
return None
|
||
u_sel = users[: P_USER // 2] + users[-(P_USER // 2):] if len(users) > P_USER else users
|
||
a_sel = sorted(assists, key=lambda m: -len(m[1]))[:P_ASSIST] if assists else []
|
||
a_pos = {id(m): i for i, m in enumerate(a_sel)}
|
||
lines = []
|
||
for m in msgs:
|
||
if m[0] == "user" and any(m is x for x in u_sel):
|
||
lines.append("- U: " + clean(m[1], LEN_USER))
|
||
elif m[0] == "assistant" and id(m) in a_pos:
|
||
lines.append("- A: " + clean(m[1], LEN_ASSIST))
|
||
return "\n".join(lines)
|
||
|
||
|
||
def month_rollups(sessions_meta, db, human_ids):
|
||
"""Помесячный роллап: список сессий с датой, источником, титулом, первой репликой."""
|
||
by_month = {}
|
||
for sid, source, title, started, mcount in sessions_meta:
|
||
month = time.strftime("%Y-%m", time.gmtime(started))
|
||
first = db.execute(
|
||
"""SELECT content FROM messages
|
||
WHERE session_id=? AND role='user'
|
||
ORDER BY id LIMIT 1""",
|
||
(sid,),
|
||
).fetchone()
|
||
first_txt = clean(first[0], 110) if first and usable(first[0]) else ""
|
||
t = (title or "").strip()
|
||
label = t if t and len(t) > 3 and not t.lower().startswith(("йоу", "привет")) else first_txt
|
||
by_month.setdefault(month, []).append(
|
||
(started, source, label or "(без титула)", mcount, sid in human_ids)
|
||
)
|
||
out = {}
|
||
for month, rows in by_month.items():
|
||
rows.sort()
|
||
parts = [f"# Hermes: история {month} (выжимка сессий до {CUTOFF_UTC})",
|
||
f"Сессий за месяц: {len(rows)}."]
|
||
for started, source, label, mcount, is_human in rows:
|
||
d = time.strftime("%d.%m", time.gmtime(started))
|
||
src = {"telegram": "tg", "cli": "cli", "cron": "cron", "matrix": "mx",
|
||
"subagent": "sub"}.get(source, source)
|
||
star = "" if is_human else " [авто]"
|
||
parts.append(f"- {d} ({src}) «{clean(label, 90)}» msgs={mcount}{star}")
|
||
out[month] = "\n".join(parts)
|
||
return out
|
||
|
||
|
||
def index_chunks(rdb, path, text):
|
||
chunks = rag.chunk_text(text)
|
||
rdb.execute("DELETE FROM chunks WHERE path=?", (path,))
|
||
vecs = rag.embed(chunks)
|
||
for i, (c, v) in enumerate(zip(chunks, vecs)):
|
||
import hashlib
|
||
sha = hashlib.sha256(c.encode()).hexdigest()[:16]
|
||
rdb.execute(
|
||
"INSERT OR REPLACE INTO chunks (path, chunk_no, sha, text, vec, access_level) "
|
||
"VALUES (?,?,?,?,?,1)",
|
||
(path, i, sha, c, json_dumps(v)),
|
||
)
|
||
return len(chunks)
|
||
|
||
|
||
def json_dumps(v):
|
||
import json
|
||
return json.dumps(v)
|
||
|
||
|
||
def main(argv=None):
|
||
global HERMES_DB, RAG_DB, PREFIX
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--limit", type=int, default=0, help="только N человеческих сессий (смоук)")
|
||
ap.add_argument("--dry-run", action="store_true")
|
||
ap.add_argument("--hermes-db", default=HERMES_DB, help="state.db Гермеса (источник)")
|
||
ap.add_argument("--rag-db", default=RAG_DB, help="rag.db приёмника (RAG_DB env)")
|
||
ap.add_argument("--prefix", default=PREFIX, help="префикс путей чанков (для нескольких агентов: hermes-history/<agent>)")
|
||
ap.add_argument("--cutoff", default=CUTOFF_UTC, help="UTC 'YYYY-MM-DD HH:MM:SS': только сессии ДО; '' = без cutoff")
|
||
ap.add_argument("--sessions-cards", type=int, default=0, help="карточки только последних N сессий (0 = все человеческие)")
|
||
args = ap.parse_args(argv)
|
||
|
||
HERMES_DB, RAG_DB, PREFIX = args.hermes_db, args.rag_db, args.prefix
|
||
rag.DB = RAG_DB # rag.py фиксирует DB при импорте — перекрываем атрибутом
|
||
cutoff_epoch = iso2epoch(args.cutoff) if args.cutoff.strip() else None
|
||
|
||
db = sqlite3.connect(f"file:{HERMES_DB}?mode=ro", uri=True)
|
||
sessions = fetch_sessions(db, cutoff_epoch)
|
||
human = [s for s in sessions if s[1] in HUMAN_SOURCES and (s[4] or 0) >= 2]
|
||
cron = [s for s in sessions if s[1] not in HUMAN_SOURCES]
|
||
if args.limit:
|
||
human = human[: args.limit]
|
||
if args.sessions_cards:
|
||
human = human[-args.sessions_cards:]
|
||
print(f"src={HERMES_DB} rag={RAG_DB} prefix={PREFIX}")
|
||
print(f"sessions total={len(sessions)} human={len(human)} auto/cron={len(cron)} "
|
||
f"cutoff={args.cutoff or 'нет'}")
|
||
|
||
if args.dry_run:
|
||
for sid, source, title, started, mcount in human[:10]:
|
||
print(f" {time.strftime('%Y-%m-%d %H:%M', time.gmtime(started))} "
|
||
f"[{source}] {(title or '')[:60]} msgs={mcount}")
|
||
return
|
||
|
||
rdb = rag.get_db()
|
||
total_chunks, t0 = 0, time.time()
|
||
|
||
# 1) помесячные роллапы (по всем сессиям, включая cron)
|
||
human_ids = {s[0] for s in human}
|
||
roll = month_rollups(sessions, db, human_ids)
|
||
for month in sorted(roll):
|
||
n = index_chunks(rdb, f"{PREFIX}/monthly/{month}.md", roll[month])
|
||
total_chunks += n
|
||
print(f" monthly/{month}: {n} chunks")
|
||
rdb.commit()
|
||
|
||
# 2) карточки диалоговых сессий
|
||
done = 0
|
||
for sid, source, title, started, mcount in human:
|
||
card = session_card(db, sid)
|
||
done += 1
|
||
if not card:
|
||
print(f" {sid}: пропущена (нет user-сообщений)")
|
||
continue
|
||
d = time.strftime("%Y-%m-%d", time.gmtime(started))
|
||
text = (f"# Hermes-сессия {sid} [{source}] {d} — {(title or '').strip()[:80]}\n"
|
||
f"Сообщений: {mcount}. Цитаты пользователя (U) и ключевые ответы (A):\n"
|
||
+ card)
|
||
n = index_chunks(rdb, f"{PREFIX}/sessions/{sid}.md", text)
|
||
total_chunks += n
|
||
if done % 10 == 0 or done == len(human):
|
||
rdb.commit()
|
||
print(f" sessions {done}/{len(human)} (+{total_chunks} chunks, "
|
||
f"{time.time()-t0:.0f}s)")
|
||
rdb.commit()
|
||
grand = rdb.execute("SELECT COUNT(*) FROM chunks").fetchone()[0]
|
||
hermes = rdb.execute(
|
||
"SELECT COUNT(*), COUNT(DISTINCT path) FROM chunks WHERE path LIKE ?",
|
||
(f"{PREFIX}/%",),
|
||
).fetchone()
|
||
print(f"DONE: +{total_chunks} чанков истории за {time.time()-t0:.0f}s; "
|
||
f"всего в rag.db={grand}, {PREFIX}={hermes[0]} чанков / {hermes[1]} файлов")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|