API Reference

Complete API documentation for Remina.

Memory Class

Synchronous interface for memory operations.

from remina import Memory

Constructor

Memory(config: Optional[Union[MemoryConfig, Dict[str, Any]]] = None)

Initialize Remina Memory with optional configuration.

Parameters:

config — Configuration dict or MemoryConfig object. Uses defaults if not provided.

Example:

# Default configuration
memory = Memory()
 
# Custom configuration
memory = Memory({
    "storage": {"provider": "postgres", "config": {...}},
    "llm": {"provider": "openai", "config": {...}},
})

add()

def add(
    messages: Union[str, List[Dict[str, str]]],
    user_id: str,
    metadata: Optional[Dict[str, Any]] = None,
    **kwargs,
) -> Dict[str, Any]

Extract and store memories from messages or text.

Parameters:

messages — Text string or list of message dicts with role and content
user_id — User identifier for namespacing
metadata — Optional metadata to attach to extracted memories
**kwargs — Additional options (agent_id, run_id, etc.)

Returns:

{
    "results": [
        {"id": "abc123", "memory": "Name is John", "event": "ADD"},
        {"id": "def456", "memory": "Works at Google", "event": "ADD"},
    ]
}

Example:

# From text
result = memory.add(
    messages="I'm John, a software engineer.",
    user_id="user_123"
)
 
# From conversation
result = memory.add(
    messages=[
        {"role": "user", "content": "I prefer dark mode"},
        {"role": "assistant", "content": "Noted."},
    ],
    user_id="user_123",
    metadata={"source": "preferences"}
)

search()

def search(
    query: str,
    user_id: str,
    limit: int = 10,
    filters: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]

Search for relevant memories using hybrid retrieval.

Parameters:

query — Search query
user_id — User identifier
limit — Maximum results (default: 10)
filters — Optional metadata filters

Returns:

{
    "results": [
        {
            "id": "abc123",
            "memory": "Works at Google",
            "score": 0.92,
            "metadata": {...},
            "created_at": "2024-01-15T10:30:00",
            "updated_at": "2024-01-15T10:30:00",
        },
        ...
    ]
}

Example:

results = memory.search(
    query="What is the user's profession?",
    user_id="user_123",
    limit=5
)
 
for mem in results["results"]:
    print(f"[{mem['score']:.2f}] {mem['memory']}")

get()

def get(memory_id: str) -> Optional[Dict[str, Any]]

Retrieve a single memory by ID.

Parameters:

memory_id — Unique memory identifier

Returns: Memory dict or None if not found.

{
    "id": "abc123",
    "user_id": "user_123",
    "content": "Works at Google",
    "hash": "...",
    "metadata": {...},
    "tags": [],
    "source": "extraction",
    "importance": 0.6,
    "access_count": 5,
    "created_at": "2024-01-15T10:30:00",
    "updated_at": "2024-01-15T10:30:00",
    "last_accessed_at": "2024-01-16T14:20:00",
}

get_all()

def get_all(user_id: str, limit: int = 100) -> Dict[str, Any]

Retrieve all memories for a user.

Parameters:

user_id — User identifier
limit — Maximum results (default: 100)

Returns:

{
    "results": [
        {"id": "...", "memory": "...", "metadata": {...}, ...},
        ...
    ]
}

update()

def update(memory_id: str, content: str) -> Dict[str, Any]

Update memory content. Re-generates embedding.

Parameters:

memory_id — Memory to update
content — New content

Returns:

{"success": True, "message": "Memory updated"}
# or
{"success": False, "message": "Memory not found"}

delete()

def delete(memory_id: str) -> Dict[str, Any]

Delete a single memory.

Parameters:

memory_id — Memory to delete

Returns:

{"success": True, "message": "Memory deleted"}

delete_all()

def delete_all(user_id: str) -> Dict[str, Any]

Delete all memories for a user.

Parameters:

user_id — User identifier

Returns:

{"success": True, "message": "Deleted 15 memories"}

close()

def close() -> None

Close all connections. Call when done.

AsyncMemory Class

Asynchronous interface with identical methods.

from remina import AsyncMemory
import asyncio
 
async def main():
    memory = AsyncMemory()
    
    await memory.add(
        messages="I prefer async Python",
        user_id="user_123"
    )
    
    results = await memory.search(
        query="programming preferences",
        user_id="user_123"
    )
    
    await memory.close()
 
asyncio.run(main())

All methods are async:

await memory.add(...)
await memory.search(...)
await memory.get(...)
await memory.get_all(...)
await memory.update(...)
await memory.delete(...)
await memory.delete_all(...)
await memory.close()

Data Models

Memory

from remina.models import Memory
 
@dataclass
class Memory:
    id: str
    user_id: str
    content: str
    hash: str
    embedding: Optional[List[float]]
    metadata: Dict[str, Any]
    tags: List[str]
    source: str  # "manual", "conversation", "extraction"
    importance: float  # 0.0 - 1.0
    decay_rate: float
    access_count: int
    created_at: datetime
    updated_at: datetime
    last_accessed_at: datetime
    links: List[str]
    is_consolidated: bool
    consolidated_from: List[str]

SearchResult

from remina.models import SearchResult
 
@dataclass
class SearchResult:
    id: str
    memory: str
    score: float
    metadata: Dict[str, Any]
    created_at: Optional[str]
    updated_at: Optional[str]

Exceptions

from remina.exceptions import (
    ReminaError,
    ConfigurationError,
    StorageError,
    VectorStoreError,
    EmbeddingError,
    LLMError,
    CacheError,
    MemoryNotFoundError,
    RateLimitError,
)

ReminaError (Base)

class ReminaError(Exception):
    message: str
    error_code: str
    details: Dict[str, Any]
    suggestion: str
    debug_info: Dict[str, Any]

Error Codes

Code	Exception	Description
`CFG_*`	ConfigurationError	Configuration issues
`STORAGE_*`	StorageError	L2 storage failures
`VECTOR_*`	VectorStoreError	Vector store failures
`EMBED_*`	EmbeddingError	Embedding failures
`LLM_*`	LLMError	LLM failures
`CACHE_*`	CacheError	Redis cache failures
`MEM_404`	MemoryNotFoundError	Memory not found
`RATE_*`	RateLimitError	Rate limit exceeded

Example

from remina.exceptions import StorageError, ReminaError
 
try:
    memory.add(messages="...", user_id="user_123")
except StorageError as e:
    print(f"Storage error [{e.error_code}]: {e.message}")
    print(f"Suggestion: {e.suggestion}")
except ReminaError as e:
    print(f"Error: {e.message}")

Configuration Classes

from remina.configs.base import MemoryConfig, MemoryItem
from remina.configs.cache import CacheConfig
from remina.configs.storage import StorageConfig
from remina.configs.vectors import VectorStoreConfig
from remina.configs.embeddings import EmbedderConfig
from remina.configs.llms import LLMConfig

See Configuration for detailed documentation.

LLM Providers Overview