beaver/beaver-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
58e343aae016dbda116a11237dd26929a6dfc597
beaver-agent/scripts/seed_memory.py
T

507 lines
20 KiB
Python
Raw Blame History

#!/usr/bin/env python3
# ruff: noqa: E501, EXE001, S603, S108, PLR0915 - одноразовый инструмент, не сервис
"""Одноразовый засев памяти агента из истории vault (решение h, 2026-09-01).
Гоняет `claude -p` на маке (подписка, как бенч) по трём источникам и пишет в
локальный vault - Sync донесёт до dell:
- `💬 чаты/` (не тесты, ещё не в индексе) → выжимка в `мета/бобер/выжимки/`
в формате дистиллятора + строка в `индекс.md`, плюс наблюдения по людям;
- `📅 дни/` (заполненные, помесячно) → наблюдения по людям и о Бобре;
- `🧠 мысли/` (по файлу) → наблюдения, в основном о Бобре.
Наблюдения - `мета/бобер/наблюдения/<Имя> - наблюдения.md`, строка =
`- дата · факт · [[источник]]`; имена только из карточек `👤 люди/` плюс
«Бобёр». В карточку добавляется строка `наблюдения: [[Имя - наблюдения]]`.
Сырые ответы модели лежат в `scripts/seed_out/` - повторный запуск их
переиспользует (resume), файлы vault дописываются без дублей.
python3 scripts/seed_memory.py --dry-run
python3 scripts/seed_memory.py --only chats --limit 2
python3 scripts/seed_memory.py # всё
"""
from __future__ import annotations
import argparse
import json
import re
import subprocess
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from datetime import date
from pathlib import Path
HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE.parent))
from beaver_gateway.core.prompt import assemble # noqa: E402
from recall import People # noqa: E402
VAULT = Path.home() / "Documents/obsidian/hh"
BEAVER = VAULT / "мета/бобер"
CHATS = VAULT / "💬 чаты"
DIARY = VAULT / "📅 дни"
THOUGHTS = VAULT / "🧠 мысли"
PEOPLE = VAULT / "👤 люди"
DIGESTS = BEAVER / "выжимки"
INDEX = BEAVER / "индекс.md"
NOTES = BEAVER / "наблюдения"
OUT = HERE / "seed_out"
MODEL = "claude-opus-5"
MAX_CHARS = 350_000
BOBR = "Бобёр"
PROMPTS = BEAVER / "промпты"
GRANULES = PROMPTS / "гранулы"
VOICE = tuple(
(tag, GRANULES / "голос" / f"{tag}.md")
for tag in (
"role",
"philosophy",
"user_profile",
"operating_modes",
"how_you_operate",
"interaction_guidelines",
)
)
# та же сборка, что у мастера в config.py, только окружение - джоб
SEED_BUILD = (
*VOICE,
("profile", PROMPTS / "профиль.md"),
("corrections", PROMPTS / "поправки.md"),
("vault_map", GRANULES / "карта-vault.md"),
("environment", GRANULES / "окружения" / "джоб.md"),
("dispatcher", GRANULES / "диспетчер.md"),
)
DIGEST_MARK = "=== ВЫЖИМКА ==="
NOTES_MARK = "=== НАБЛЮДЕНИЯ ==="
END_MARK = "=== КОНЕЦ ==="
@dataclass(frozen=True)
class Source:
kind: str # chats | days | thoughts
id: str # имя для seed_out и [[ссылки]]
title: str
text: str
when: str # дата источника YYYY-MM-DD (для наблюдений без даты в тексте)
link: str # что ставить в [[...]] у наблюдений
# ---------------------------------------------------------------- prompts --
def system_prompt(names: list[str]) -> str:
"""Настоящие гранулы Менеджера Бобрения + блок задания на засев."""
distiller = (GRANULES / "дистиллятор.md").read_text(encoding="utf-8")
keep = distiller[distiller.find("Что сохранять:") :].strip()
known = ", ".join(names)
seed = f"""Это засев памяти, одноразовый джоб (решение Бобра, 2026-09-01): тебе по одному отдают
документы из истории vault - старые чаты с тобой, месяцы дневника, заметки-мысли, - и ты
отдаёшь то, что из них должно остаться у тебя в памяти. Тулз нет, файлы запишет gateway из
твоего ответа. Ответ - строго в формате ниже, ничего вне маркеров.
{DIGEST_MARK}
(только для чата; для дневника и мысли блок пропускается целиком)
# тема
## контекст
2-4 строки: зачем открывали.
## решили
- решения и их причины; цифры, имена, даты, [[ссылки]] дословно
## открытые вопросы
- что не закрыто
## ссылки
- [[люди]], [[проекты]], соседние чаты - только реально упомянутые
{NOTES_MARK}
Имя :: YYYY-MM-DD · факт
{END_MARK}
Выжимка - по правилам дистиллятора: нейтральный язык, третье лицо («Бобёр решил»), без
мнения, которого в разговоре не было. {keep}
Наблюдения - твои записки из блока «Память» оверлея: то, чего нет в vault и что понадобится,
чтобы заметить повтор - реакция человека, отговорка, обещание, оценка, состояние, паттерн,
событие в отношениях. Не биография и не пересказ карточки (возраст, где живёт), не общие слова.
Имя - ровно одно из списка ниже, как написано, или «{BOBR}» для самого Бобра; людей не из
списка не записывай. Дата - из текста, если она там есть, иначе дата документа. Одна строка,
до 200 знаков, коротко и по факту; на человека до 6 строк с документа, на {BOBR} - до 12.
Нет наблюдений - блок пустой.
Известные имена: {known}"""
parts = list(SEED_BUILD)
text = assemble(parts)
return f"{text}\n<seed>\n{seed}\n</seed>\n"
def user_prompt(src: Source) -> str:
what = {
"chats": "чат с агентом",
"days": "месяц дневника",
"thoughts": "заметка-мысль",
}[src.kind]
return f"Документ: {what}, «{src.title}», дата {src.when}.\n\n{src.text}"
# ---------------------------------------------------------------- sources --
CALLOUT_RE = re.compile(r"^> \[!(tool|thinking)\]")
def strip_chat(text: str) -> str:
"""Без фронтматтера, тул-коллов и thinking: только реплики."""
if text.startswith("---"):
end = text.find("\n---", 3)
if end > 0:
text = text[end + 4 :]
out: list[str] = []
skipping = False
for line in text.splitlines():
if CALLOUT_RE.match(line):
skipping = True
continue
if skipping and line.startswith(">"):
continue
skipping = False
out.append(line)
cleaned = re.sub(r"\n{3,}", "\n\n", "\n".join(out)).strip()
return cleaned[:MAX_CHARS]
def indexed_chats() -> set[str]:
if not INDEX.exists():
return set()
return set(re.findall(r"\[\[([^\]]+)\]\] →", INDEX.read_text(encoding="utf-8")))
def chat_sources() -> list[Source]:
done = indexed_chats()
out: list[Source] = []
for path in sorted(CHATS.rglob("*.md")):
if (
"тест" in path.stem.lower()
or path.stat().st_size < 5_000
or path.stem in done
):
continue
text = strip_chat(path.read_text(encoding="utf-8", errors="ignore"))
if len(text) < 1_500:
continue
when = (
path.stem[:10]
if re.match(r"\d{4}-\d{2}-\d{2}", path.stem)
else date.today().isoformat()
)
out.append(
Source("chats", f"chat_{path.stem}", path.stem, text, when, path.stem)
)
return out
def day_sources() -> list[Source]:
months: dict[str, list[Path]] = {}
for path in sorted(DIARY.glob("????-??-??.md")):
if path.stat().st_size > 700:
months.setdefault(path.stem[:7], []).append(path)
out: list[Source] = []
for month, files in sorted(months.items()):
parts = []
for f in files:
body = f.read_text(encoding="utf-8", errors="ignore")
body = body.split("\n# планы", 1)[0] # хвост с tasks-запросами
parts.append(f"### [[{f.stem}]]\n{body.strip()}")
text = "\n\n".join(parts)[:MAX_CHARS]
out.append(
Source(
"days",
f"days_{month}",
f"дневник {month}",
text,
f"{month}-01",
"дневник",
)
)
return out
def thought_sources() -> list[Source]:
out: list[Source] = []
for path in sorted(THOUGHTS.rglob("*.md")):
if path.stat().st_size < 400:
continue
m = re.match(r"(\d{4}-\d{2}-\d{2})", path.stem)
when = m.group(1) if m else date.today().isoformat()
text = path.read_text(encoding="utf-8", errors="ignore")[:MAX_CHARS]
out.append(
Source("thoughts", f"thought_{path.stem}", path.stem, text, when, path.stem)
)
return out
# ------------------------------------------------------------------ model --
def _result(payload: str) -> dict:
"""`claude -p --output-format json` отдаёт либо объект, либо массив сообщений."""
data = json.loads(payload)
if isinstance(data, list):
results = [x for x in data if isinstance(x, dict) and x.get("type") == "result"]
return results[-1] if results else {}
return data if isinstance(data, dict) else {}
def ask(src: Source, system: str) -> str:
raw = OUT / f"{src.id}.json"
if raw.exists():
try:
data = _result(raw.read_text(encoding="utf-8"))
except json.JSONDecodeError:
data = {}
if data.get("result") and not data.get("is_error"):
return data["result"]
cmd = [
"claude",
"-p",
user_prompt(src),
"--system-prompt",
system,
"--model",
MODEL,
"--effort",
"medium",
"--tools",
"",
"--setting-sources",
"",
"--strict-mcp-config",
"--disable-slash-commands",
"--no-session-persistence",
"--output-format",
"json",
]
proc = subprocess.run(
cmd, capture_output=True, text=True, timeout=1200, cwd="/tmp", check=False
)
OUT.mkdir(exist_ok=True)
raw.write_text(
proc.stdout or json.dumps({"error": proc.stderr[-2000:]}), encoding="utf-8"
)
if proc.returncode != 0 or not proc.stdout:
msg = f"{src.id}: claude exit {proc.returncode}: {proc.stderr[-500:]}"
raise RuntimeError(msg)
data = _result(proc.stdout)
if data.get("is_error") or not data.get("result"):
msg = f"{src.id}: {data.get('result') or data}"
raise RuntimeError(msg)
return data["result"]
def parse(answer: str) -> tuple[str | None, list[tuple[str, str, str]]]:
digest = None
if DIGEST_MARK in answer and NOTES_MARK in answer:
digest = answer.split(DIGEST_MARK, 1)[1].split(NOTES_MARK, 1)[0].strip() or None
if digest and digest.startswith("("): # «(только для чата ...)» пропуск
digest = None
notes_part = answer.split(NOTES_MARK, 1)[1] if NOTES_MARK in answer else ""
notes_part = notes_part.split(END_MARK, 1)[0]
notes: list[tuple[str, str, str]] = []
for line in notes_part.splitlines():
m = re.match(
r"^\s*-?\s*(.+?)\s*::\s*(\d{4}-\d{2}-\d{2})\s*[·\-]\s*(.+?)\s*$", line
)
if m:
notes.append((m.group(1).strip(), m.group(2), m.group(3).strip()))
return digest, notes
# ------------------------------------------------------------------ write --
SLUG_BAD = re.compile(r"[^\w\s-]", re.UNICODE)
def slugify(text: str, maxlen: int = 40) -> str:
cleaned = re.sub(r"[\s-]+", "-", SLUG_BAD.sub(" ", text).strip()).strip("-")
return cleaned[:maxlen].rstrip("-") or "untitled"
def write_digest(src: Source, body: str, today: str) -> str:
DIGESTS.mkdir(parents=True, exist_ok=True)
stem = f"{today} - {slugify(src.title)}"
path = DIGESTS / f"{stem}.md"
n = 2
while path.exists() and f"[[{src.title}]]" not in path.read_text(encoding="utf-8"):
path = DIGESTS / f"{stem} ({n}).md"
n += 1
path.write_text(
f'---\ntype: выжимка\nsource: "[[{src.title}]]"\ndate: {today}\n---\n\n{body.strip()}\n',
encoding="utf-8",
)
line = f"- {today} [[{src.title}]] → [[{path.stem}]]"
text = (
INDEX.read_text(encoding="utf-8")
if INDEX.exists()
else "# индекс\n\nстрока на выжимку: чат → его выжимка.\n"
)
if line not in text.splitlines():
INDEX.write_text(text.rstrip("\n") + "\n" + line + "\n", encoding="utf-8")
return path.stem
def note_file(name: str) -> Path:
return NOTES / f"{name} - наблюдения.md"
def append_notes(name: str, lines: list[str]) -> int:
NOTES.mkdir(parents=True, exist_ok=True)
path = note_file(name)
if not path.exists():
path.write_text(
f"# {name} - наблюдения\n\n> папка терапевта: пишет менеджер бобрения. "
"Строка = дата · факт · источник. Бобёр по умолчанию не читает, но может.\n\n",
encoding="utf-8",
)
existing = set(path.read_text(encoding="utf-8").splitlines())
fresh = [ln for ln in lines if ln not in existing]
if fresh:
with path.open("a", encoding="utf-8") as fh:
fh.write("\n".join(fresh) + "\n")
return len(fresh)
def sort_notes() -> None:
for path in NOTES.glob("* - наблюдения.md"):
text = path.read_text(encoding="utf-8")
head, bullets = [], []
for ln in text.splitlines():
(bullets if ln.startswith("- ") else head).append(ln)
bullets = sorted(dict.fromkeys(bullets), key=lambda ln: ln[2:12])
while head and not head[-1].strip():
head.pop()
path.write_text("\n".join([*head, "", *bullets]) + "\n", encoding="utf-8")
def link_card(person_path: Path, name: str) -> bool:
text = person_path.read_text(encoding="utf-8")
link = f"[[{name} - наблюдения]]"
if link in text:
return False
person_path.write_text(
text.rstrip("\n") + f"\n\nнаблюдения: {link}\n", encoding="utf-8"
)
return True
# ------------------------------------------------------------------- main --
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--only", choices=["chats", "days", "thoughts"])
ap.add_argument("--limit", type=int)
ap.add_argument("--dry-run", action="store_true")
ap.add_argument("--jobs", type=int, default=4)
ap.add_argument("--pick", help="подстрока в названии источника")
args = ap.parse_args()
# хабы групп («бывшие», «рандомы») - не люди, наблюдения им не пишем
people = [
p for p in People(PEOPLE, VAULT, ttl=10_000).all() if "/группы/" not in p.path
]
by_name = {p.name: p for p in people}
for p in people: # алиасы тоже принимаем, но пишем под именем карточки
for a in p.aliases:
by_name.setdefault(a, p)
names = sorted(p.name for p in people)
system = system_prompt(names)
sources: list[Source] = []
if args.only in (None, "chats"):
sources += chat_sources()
if args.only in (None, "days"):
sources += day_sources()
if args.only in (None, "thoughts"):
sources += thought_sources()
if args.pick:
sources = [s for s in sources if args.pick.lower() in s.title.lower()]
if args.limit:
sources = sources[: args.limit]
total_chars = sum(len(s.text) for s in sources)
print(
f"источников: {len(sources)}, символов: {total_chars:,} (~{total_chars // 3.5:,.0f} токенов)"
)
for s in sources:
print(f" {s.kind:8} {s.title[:60]:60} {len(s.text):>8}")
if args.dry_run:
return
today = date.today().isoformat()
written: dict[str, int] = {}
linked: list[str] = []
failures: list[str] = []
with ThreadPoolExecutor(max_workers=args.jobs) as pool:
futs = {pool.submit(ask, s, system): s for s in sources}
for fut in as_completed(futs):
src = futs[fut]
try:
answer = fut.result()
except Exception as exc: # noqa: BLE001
failures.append(str(exc))
print(f"FAIL {src.id}: {exc}"[:300])
continue
digest, notes = parse(answer)
digest_stem = (
write_digest(src, digest, today)
if src.kind == "chats" and digest
else None
)
grouped: dict[str, list[str]] = {}
unknown: set[str] = set()
for raw_name, when, fact in notes:
name = (
BOBR
if raw_name.casefold() in (BOBR.casefold(), "бобер", "h", "андрей")
else None
)
if name is None:
person = by_name.get(raw_name)
if person is None:
unknown.add(raw_name)
continue
name = person.name
link = f"[[{src.link}]]" if src.kind != "days" else ""
if src.kind == "days":
link = (
f"[[{when}]]"
if (DIARY / f"{when}.md").exists()
else "[[дневник]]"
)
grouped.setdefault(name, []).append(f"- {when} · {fact} · {link}")
for name, lines in grouped.items():
n = append_notes(name, lines)
written[name] = written.get(name, 0) + n
if (
name != BOBR
and name in by_name
and link_card(VAULT / by_name[name].path, name)
):
linked.append(name)
print(
f"ok {src.id}: digest={digest_stem or '-'} notes={sum(len(v) for v in grouped.values())}"
+ (f" unknown={sorted(unknown)}" if unknown else "")
)
sort_notes()
print("\nнаблюдений добавлено по людям:")
for name, n in sorted(written.items(), key=lambda kv: -kv[1]):
print(f" {n:4} {name}")
if linked:
print("ссылка добавлена в карточки:", ", ".join(sorted(set(linked))))
if failures:
print(f"\nошибок: {len(failures)}")
sys.exit(1)
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink