"""Долговременная память ассистента — файл-конспект между сессиями. Принцип (идея юзера 2026-09-11): - в конце разговора («до свидания») модель просится сформулировать конспект, он ДОПОЛНЯЕТСЯ в текстовый файл data/memory.md; - при старте конспект кладётся в системный промпт — ассистент «помнит» прошлые разговоры (кто отец, что обсуждали, договорённости). Файл — простой читаемый текст: можно посмотреть и поправить руками. Хранить сам диалог не нужно — конспект компактнее и токены дешевле. """ from __future__ import annotations import re import time from pathlib import Path from typing import List, Optional class MemoryStore: def __init__(self, path: str | Path = "data/memory.md", max_chars: int = 8000) -> None: self._path = Path(path) self._max_chars = max_chars # ограничение конспекта (защита от бесконечного роста) self._path.parent.mkdir(parents=True, exist_ok=True) # ------------------------------------------------------------------ чтение @property def path(self) -> Path: return self._path def load(self) -> str: """Конспект целиком (пустая строка, если памяти ещё нет).""" if not self._path.is_file(): return "" return self._path.read_text(encoding="utf-8").strip() def as_prompt_block(self) -> str: """Блок для системного промпта (или пустая строка).""" memory = self.load() if not memory: return "" return ( "Вот что ты помнишь из прошлых разговоров с этим человеком " "(его долговременная память):\n" f"{memory}\n" "Используй это естественно, без упоминания «записей» и «памяти»." ) # ------------------------------------------------------------------ запись def append(self, summary: str) -> None: """Дописать новый конспект с датой, не превышая лимит объёма.""" summary = summary.strip() if not summary: return header = f"\n\n## {time.strftime('%d.%m.%Y')}\n" current = self.load() new_total = len(current) + len(header) + len(summary) if new_total > self._max_chars: # Файл распух: сжимаем старые записи в одну итоговую строку current = self._condense(current) with self._path.open("a", encoding="utf-8") as f: if current and not current.endswith("\n"): f.write("\n") f.write(header + summary) def clear(self) -> None: """Стереть память (для отладки).""" self._path.write_text("", encoding="utf-8") # ------------------------------------------------------------------ внутреннее def _condense(self, text: str) -> str: """Сжать конспект: оставить последние записи целиком, старые — краткой выжимкой. Сжатие делаем механически (первые предложения старых записей); качество semantique-сжатия критично только для очень длинной памяти, а это редкость. """ sections: List[str] = [s.strip() for s in re.split(r"\n## ", "\n" + text) if s.strip()] if len(sections) <= 1: return text[: self._max_chars // 2] keep_recent = sections[-4:] # последние 4 записи — целиком old = "\n".join(sections[:-4]) condensed = "\n".join( "• " + " ".join(re.split(r"(?<=[.!?]) ", chunk)[:2]) for chunk in old.split("•") if chunk.strip() ) merged = ("— Итоги прошлых записей: " + condensed + "\n\n" + "\n## ".join(keep_recent)) return merged[: self._max_chars]