mirror of
https://github.com/tkenaz/breathe-memory.git
synced 2026-08-28 04:24:59 +00:00
4e671a0a83
Context optimization and associative memory for LLM applications. Two-phase system: SYNAPSE (pre-generation memory injection) + GraphCompactor (structured context compression). - Interface-based, storage-agnostic, LLM-agnostic - Memory Nexus: PostgreSQL + pgvector reference backend - Zero mandatory dependencies beyond stdlib - 28 tests passing, clean install verified Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
68 lines
2 KiB
Python
68 lines
2 KiB
Python
"""Abstract MemoryStore interface."""
|
|
from __future__ import annotations
|
|
|
|
from abc import ABC, abstractmethod
|
|
from typing import Optional
|
|
|
|
from .models import Memory, MemoryMetadata
|
|
from breathe.interfaces import RetrievedNode, VectorSearchClient
|
|
|
|
|
|
class MemoryStore(VectorSearchClient, ABC):
|
|
"""
|
|
Abstract memory store — combines VectorSearchClient with CRUD operations.
|
|
|
|
Implements VectorSearchClient so it can be passed directly to Synapse::
|
|
|
|
store = PostgresMemoryStore(dsn="postgresql://...")
|
|
synapse = Synapse(vector_client=store)
|
|
|
|
Implement this to connect Memory Nexus to any backend.
|
|
See ``PostgresMemoryStore`` for the asyncpg + pgvector reference implementation.
|
|
"""
|
|
|
|
@abstractmethod
|
|
async def initialize(self) -> None:
|
|
"""Initialize the store (create tables, load models, etc.)."""
|
|
|
|
@abstractmethod
|
|
async def store(
|
|
self,
|
|
content: str,
|
|
metadata: Optional[MemoryMetadata] = None,
|
|
) -> Memory:
|
|
"""
|
|
Store a memory and return it with generated ID.
|
|
|
|
Args:
|
|
content: The text content to store.
|
|
metadata: Optional metadata (tags, source, importance).
|
|
"""
|
|
|
|
@abstractmethod
|
|
async def get(self, memory_id: str) -> Optional[Memory]:
|
|
"""Retrieve a memory by ID."""
|
|
|
|
@abstractmethod
|
|
async def delete(self, memory_id: str) -> bool:
|
|
"""Delete a memory by ID. Returns True if deleted."""
|
|
|
|
@abstractmethod
|
|
async def search(self, query: str, limit: int = 5) -> list[RetrievedNode]:
|
|
"""
|
|
Semantic search. Implements VectorSearchClient.
|
|
|
|
Args:
|
|
query: Search query text.
|
|
limit: Maximum results.
|
|
|
|
Returns:
|
|
List of RetrievedNode sorted by similarity descending.
|
|
"""
|
|
|
|
@abstractmethod
|
|
async def get_recent(self, limit: int = 20) -> list[Memory]:
|
|
"""Return most recently stored memories."""
|
|
|
|
async def close(self) -> None:
|
|
"""Clean up resources (close connections, etc.)."""
|