brain.memory — Система памяти¶

5 видов памяти + MemoryManager + SQLite backend. Аналог гиппокампа и долговременной памяти.

MemoryManager¶

Единая точка доступа ко всем видам памяти. Координирует персистентность через MemoryDatabase.

MemoryManager ¶

MemoryManager(data_dir: str = 'brain/data/memory', working_max_size: int = 20, semantic_max_nodes: int = 10000, episodic_max: int = 5000, auto_consolidate: bool = True, on_event: Optional[Callable[[str, Dict[str, Any]], None]] = None, storage_backend: str = 'auto', brain_logger: Optional[BrainLogger] = None)

Единый интерфейс ко всей системе памяти мозга.

Параметры

data_dir — директория для хранения JSON-файлов working_max_size — максимальный размер рабочей памяти semantic_max_nodes — максимальное количество узлов семантической памяти episodic_max — максимальное количество эпизодов auto_consolidate — запускать ли фоновую консолидацию on_event — callback для логирования событий

store ¶

store(content: Any, modality: str = 'text', importance: float = 0.5, source_ref: str = '', tags: Optional[List[str]] = None, concepts: Optional[List[str]] = None, trace_id: str = '', session_id: str = '', auto_extract_facts: bool = True) -> Dict[str, Any]

Сохранить информацию в память.

Автоматически: 1. Добавляет в рабочую память (всегда) 2. Если importance >= 0.4 → добавляет в эпизодическую 3. Если auto_extract_facts → пытается извлечь факт в семантическую 4. Регистрирует источник

Returns:

Type	Description
`Dict[str, Any]`	Словарь с результатами: {"working": item, "episodic": ep, "semantic": node}

retrieve ¶

retrieve(query: str, memory_types: Optional[List[str]] = None, top_n: int = 5, min_importance: float = 0.0) -> MemorySearchResult

Поиск по всем видам памяти.

Parameters:

Name	Type	Description	Default
`query`	`str`	строка поиска	required
`memory_types`	`Optional[List[str]]`	список видов памяти для поиска ['working', 'semantic', 'episodic'] или None (все)	`None`
`top_n`	`int`	максимальное количество результатов из каждого вида	`5`
`min_importance`	`float`	минимальная важность	`0.0`

Returns:

Type	Description
`MemorySearchResult`	MemorySearchResult с результатами из всех видов памяти

status ¶

status() -> Dict[str, Any]

Полный статус системы памяти.

WorkingMemory¶

Кратковременная рабочая память с ограниченной ёмкостью и TTL.

WorkingMemory ¶

WorkingMemory(max_size: int = DEFAULT_MAX_SIZE, ram_limit_pct: float = RAM_LIMIT_PCT)

Рабочая память мозга — активный контекст текущего цикла.

Параметры

max_size — максимальное количество элементов (по умолчанию 20) importance_threshold — порог важности для защиты от вытеснения (0.8+) ram_limit_pct — при превышении этого % RAM — уменьшить окно

get_all ¶

get_all() -> List[MemoryItem]

Получить все элементы (защищённые + обычные).

clear ¶

clear(keep_important: bool = True)

Очистить рабочую память.

Parameters:

Name	Type	Description	Default
`keep_important`	`bool`	если True — сохранить защищённые элементы	`True`

status ¶

status() -> Dict[str, Any]

Статус рабочей памяти.

SemanticMemory¶

Долговременная семантическая память: граф понятий и связей.

SemanticMemory ¶

SemanticMemory(data_path: str = 'brain/data/memory/semantic.json', max_nodes: int = 10000, decay_rate: float = 0.005, autosave_every: int = 50, storage_backend: str = 'auto', db: Optional['MemoryDatabase'] = None)

Семантическая память — граф понятий и фактов.

Хранит долгосрочные знания: что такое X, как X связано с Y. Работает в RAM, персистируется в JSON.

Параметры

data_path — путь к JSON-файлу для сохранения max_nodes — максимальное количество узлов в памяти decay_rate — скорость затухания уверенности (за цикл) autosave_every — автосохранение каждые N операций записи

deny_fact ¶

deny_fact(concept: str, delta: float = 0.1)

Опровергнуть факт — снизить уверенность.

delete_fact ¶

delete_fact(concept: str) -> bool

Удалить факт из памяти.

apply_decay ¶

apply_decay(rate: Optional[float] = None)

Применить затухание ко всем фактам. Вызывается периодически (например, каждые N циклов).

status ¶

status() -> Dict[str, Any]

Статус семантической памяти.

EpisodicMemory¶

Эпизодическая память: события с временными метками и мультимодальными свидетельствами.

EpisodicMemory ¶

EpisodicMemory(data_path: str = 'brain/data/memory/episodes.json', max_episodes: int = 5000, importance_threshold: float = IMPORTANCE_PROTECT, autosave_every: int = 20, storage_backend: str = 'auto', db: Optional['MemoryDatabase'] = None)

Эпизодическая память — хронологическая история событий мозга.

Хранит всё, что мозг воспринял, сделал или подумал. Работает в RAM, персистируется в JSON.

Параметры

data_path — путь к JSON-файлу max_episodes — максимальное количество эпизодов в RAM importance_threshold — порог важности для защиты от удаления autosave_every — автосохранение каждые N операций записи

store ¶

store(content: str, modality: str = 'text', source: str = '', importance: float = 0.5, confidence: float = 1.0, tags: Optional[List[str]] = None, concepts: Optional[List[str]] = None, modal_evidence: Optional[List[ModalEvidence]] = None, trace_id: str = '', session_id: str = '') -> Episode

Сохранить новый эпизод.

Returns:

Type	Description
`Episode`	Созданный Episode

retrieve_by_concept ¶

retrieve_by_concept(concept: str, top_n: int = 10, min_importance: float = 0.0) -> List[Episode]

Найти эпизоды, связанные с понятием.

Returns:

Type	Description
`List[Episode]`	Список эпизодов, отсортированных по важности и времени

status ¶

status() -> Dict[str, Any]

Статус эпизодической памяти.

ProceduralMemory¶

Процедурная память: паттерны действий и их эффективность.

ProceduralMemory ¶

ProceduralMemory(data_path: str = 'brain/data/memory/procedures.json', autosave_every: int = 20, storage_backend: str = 'auto', db: Optional['MemoryDatabase'] = None)

Процедурная память — хранилище навыков и стратегий.

Параметры

data_path — путь к JSON-файлу autosave_every — автосохранение каждые N операций

store ¶

store(name: str, steps: List[Dict[str, Any]], description: str = '', trigger_pattern: str = '', tags: Optional[List[str]] = None, priority: float = 0.5) -> Procedure

Сохранить процедуру.

Parameters:

Name	Type	Description	Default
`name`	`str`	уникальное имя	required
`steps`	`List[Dict[str, Any]]`	список шагов [{"action": ..., "params": ...}, ...]	required
`description`	`str`	описание	`''`
`trigger_pattern`	`str`	паттерн ситуации	`''`
`tags`	`Optional[List[str]]`	теги	`None`
`priority`	`float`	приоритет (0.0 — 1.0)	`0.5`

Returns:

Type	Description
`Procedure`	Procedure

retrieve ¶

retrieve(query: str, tags: Optional[List[str]] = None, min_success_rate: float = 0.0, top_n: int = 5) -> List[Procedure]

Найти подходящие процедуры.

Parameters:

Name	Type	Description	Default
`query`	`str`	строка поиска (по имени, описанию, trigger_pattern)	required
`tags`	`Optional[List[str]]`	фильтр по тегам	`None`
`min_success_rate`	`float`	минимальный success rate	`0.0`
`top_n`	`int`	максимальное количество результатов	`5`

Returns:

Type	Description
`List[Procedure]`	Список процедур, отсортированных по эффективности

status ¶

status() -> Dict[str, Any]

Статус процедурной памяти.

SourceMemory¶

Память об источниках: доверие, подтверждения, противоречия, чёрный список.

SourceMemory ¶

SourceMemory(data_path: str = 'brain/data/memory/sources.json', default_trust: float = 0.7, autosave_every: int = 30, storage_backend: str = 'auto', db: Optional['MemoryDatabase'] = None)

Память об источниках знаний — кто сказал что и насколько ему доверять.

Параметры

data_path — путь к JSON-файлу default_trust — начальный уровень доверия для новых источников autosave_every — автосохранение каждые N операций

blacklist ¶

blacklist(source_id: str, reason: str = '')

Добавить источник в чёрный список.

get_trust ¶

get_trust(source_id: str) -> float

Получить уровень доверия к источнику.

Returns:

Name	Type	Description
`float`	`float`	0.0 — 1.0 (0.5 если источник неизвестен)

status ¶

status() -> Dict[str, Any]

Статус памяти об источниках.

ConsolidationEngine¶

Фоновая консолидация памяти: перенос из рабочей в долговременную.

ConsolidationEngine ¶

ConsolidationEngine(working: WorkingMemory, episodic: EpisodicMemory, semantic: SemanticMemory, source: SourceMemory, procedural: ProceduralMemory, config: Optional[ConsolidationConfig] = None, on_event: Optional[Callable[[str, Dict[str, Any]], None]] = None)

Движок консолидации памяти — аналог Гиппокампа.

Запускается в фоновом потоке и периодически: 1. Переносит важные элементы из Working Memory → Episodic/Semantic 2. Применяет decay к семантической памяти 3. Сохраняет все виды памяти на диск 4. При нехватке RAM — агрессивно очищает рабочую память

Параметры

working: WorkingMemory episodic: EpisodicMemory semantic: SemanticMemory source: SourceMemory procedural: ProceduralMemory config: ConsolidationConfig on_event: callback для логирования событий

start ¶

start()

Запустить фоновый поток консолидации.

stop ¶

stop(save_all: bool = True)

Остановить фоновый поток консолидации.

status ¶

status() -> Dict[str, Any]

Статус движка консолидации.

MemoryDatabase¶

SQLite WAL backend для всей системы памяти. Опциональное шифрование через SQLCipher (P3-12).

MemoryDatabase ¶

MemoryDatabase(db_path: str = 'brain/data/memory/memory.db', wal_mode: bool = True, encryption_key: Optional[str] = None)

SQLite persistence backend для всей системы памяти.

Thread-safe через RLock. WAL mode для concurrent reads. Единый .db файл для всех видов памяти. Опциональное шифрование через SQLCipher (P3-12).

Параметры

db_path: путь к файлу базы данных (или ":memory:" для тестов) wal_mode: включить WAL mode (рекомендуется для production) encryption_key: ключ шифрования для SQLCipher (None = без шифрования). Требует установки: pip install cognitive-core[encrypted]

schema_version `property` ¶

schema_version: int

Текущая версия схемы.

upsert_semantic_node ¶

upsert_semantic_node(concept: str, data: Dict[str, Any])

Вставить или обновить узел семантической памяти.

Parameters:

Name	Type	Description	Default
`concept`	`str`	ключ понятия (нормализованный)	required
`data`	`Dict[str, Any]`	словарь из SemanticNode.to_dict()	required

load_all_semantic_nodes ¶

load_all_semantic_nodes() -> List[Dict[str, Any]]

Загрузить все узлы семантической памяти.

delete_semantic_node ¶

delete_semantic_node(concept: str)

Удалить узел и его связи.

upsert_episode ¶

upsert_episode(episode_id: str, data: Dict[str, Any])

Вставить или обновить эпизод.

load_all_episodes ¶

load_all_episodes() -> List[Dict[str, Any]]

Загрузить все эпизоды.

delete_episode ¶

delete_episode(episode_id: str)

Удалить эпизод и его evidence.

upsert_source ¶

upsert_source(source_id: str, data: Dict[str, Any])

Вставить или обновить источник.

load_all_sources ¶

load_all_sources() -> List[Dict[str, Any]]

Загрузить все источники.

upsert_procedure ¶

upsert_procedure(name: str, data: Dict[str, Any])

Вставить или обновить процедуру.

load_all_procedures ¶

load_all_procedures() -> List[Dict[str, Any]]

Загрузить все процедуры.

commit ¶

commit()

Зафиксировать транзакцию.

rollback ¶

rollback()

Откатить транзакцию.

close ¶

close()

Закрыть соединение с БД.

status ¶

status() -> Dict[str, Any]

Полный статус базы данных.

table_counts ¶

table_counts() -> Dict[str, int]

Количество записей в каждой таблице.

Шифрование (SQLCipher)¶

# Установка: pip install cognitive-core[encrypted]
from brain.memory.storage import MemoryDatabase

# Зашифрованная база данных
db = MemoryDatabase(
    db_path="brain/data/memory/secure.db",
    encryption_key="my-secret-key-32-chars-minimum!!"
)

# Без шифрования (по умолчанию)
db = MemoryDatabase("brain/data/memory/memory.db")

Требования

Для шифрования необходим пакет sqlcipher3:

pip install cognitive-core[encrypted]

На Windows может потребоваться предварительная установка SQLCipher C-библиотеки.

brain.memory — Система памяти¶

MemoryManager¶

MemoryManager ¶

store ¶

retrieve ¶

status ¶

WorkingMemory¶

WorkingMemory ¶

get_all ¶

clear ¶

status ¶

SemanticMemory¶

SemanticMemory ¶

deny_fact ¶

delete_fact ¶

apply_decay ¶

status ¶

EpisodicMemory¶

EpisodicMemory ¶

store ¶

retrieve_by_concept ¶

status ¶

ProceduralMemory¶

ProceduralMemory ¶

store ¶

retrieve ¶

status ¶

SourceMemory¶

SourceMemory ¶

blacklist ¶

get_trust ¶

status ¶

ConsolidationEngine¶

ConsolidationEngine ¶

start ¶

stop ¶

status ¶

MemoryDatabase¶

MemoryDatabase ¶

schema_version property ¶

upsert_semantic_node ¶

load_all_semantic_nodes ¶

delete_semantic_node ¶

upsert_episode ¶

load_all_episodes ¶

delete_episode ¶

upsert_source ¶

load_all_sources ¶

upsert_procedure ¶

load_all_procedures ¶

commit ¶

rollback ¶

close ¶

status ¶

table_counts ¶

Шифрование (SQLCipher)¶

schema_version `property` ¶