praval.memory.memory_manager
MemoryManager - Unified interface for all Praval agent memory systems
This coordinates: - Short-term working memory - Long-term vector memory - Episodic conversation memory - Semantic knowledge memory
Classes
|
Unified memory management system for Praval agents |
- class praval.memory.memory_manager.MemoryManager(agent_id, backend='auto', qdrant_url='http://localhost:6333', storage_path=None, collection_name='praval_memories', short_term_max_entries=1000, short_term_retention_hours=24, knowledge_base_path=None, embedding_provider=None, embedding_model=None, embedding_dimensions=None, embedding_provider_options=None, embedding_runtime=None)[source]
Bases:
objectUnified memory management system for Praval agents
Provides a single interface to: - Store and retrieve memories across all systems - Coordinate between short-term and long-term storage - Manage different types of memory (episodic, semantic, etc.) - Optimize memory access patterns
- Parameters:
agent_id (str)
backend (str)
qdrant_url (str)
storage_path (str | None)
collection_name (str)
short_term_max_entries (int)
short_term_retention_hours (int)
knowledge_base_path (str | None)
embedding_provider (str | None)
embedding_model (str | None)
embedding_dimensions (int | None)
embedding_provider_options (Dict[str, Any] | None)
embedding_runtime (Any | None)
- __init__(agent_id, backend='auto', qdrant_url='http://localhost:6333', storage_path=None, collection_name='praval_memories', short_term_max_entries=1000, short_term_retention_hours=24, knowledge_base_path=None, embedding_provider=None, embedding_model=None, embedding_dimensions=None, embedding_provider_options=None, embedding_runtime=None)[source]
Initialize the unified memory manager
- Parameters:
agent_id (
str) – ID of the agent using this memorybackend (
str) – Memory backend (“auto”, “chromadb”, “qdrant”, “memory”)qdrant_url (
str) – URL for Qdrant vector databasestorage_path (
Optional[str]) – Path for persistent storagecollection_name (
str) – Collection name for vector storageshort_term_max_entries (
int) – Max entries in short-term memoryshort_term_retention_hours (
int) – Short-term memory retention timeknowledge_base_path (
Optional[str]) – Path to knowledge base files to auto-indexembedding_provider (
Optional[str]) – Provider used for vector embeddingsembedding_model (
Optional[str]) – Embedding model identifierembedding_dimensions (
Optional[int]) – Expected embedding vector sizeembedding_provider_options (
Optional[Dict[str,Any]]) – Provider-specific embedding optionsembedding_runtime (
Optional[Any]) – Preconfigured embedding runtime
- store_memory(agent_id, content, memory_type=None, **kwargs)
- retrieve_memory(memory_id)
- search_memories(query)[source]
Search memories across all systems
- Parameters:
query (
MemoryQuery) – The search query- Return type:
- Returns:
Combined search results
- get_conversation_context(agent_id, turns=10)[source]
Get recent conversation context for an agent
- Parameters:
agent_id (
str) – The agent IDturns (
int) – Number of conversation turns
- Return type:
List[MemoryEntry]- Returns:
List of conversation memories
- store_conversation_turn(agent_id, user_message, agent_response, **kwargs)
- store_knowledge(agent_id, knowledge, domain='general', confidence=1.0, knowledge_type='fact')[source]
Store knowledge or facts
- Parameters:
agent_id (
str) – The agent IDknowledge (
str) – The knowledge contentdomain (
str) – Domain of knowledgeconfidence (
float) – Confidence in the knowledgeknowledge_type (
str) – Type of knowledge (fact, concept, rule)
- Return type:
str- Returns:
Memory ID
- get_domain_knowledge(agent_id, domain, limit=20)[source]
Get knowledge in a specific domain
- Parameters:
agent_id (
str) – The agent IDdomain (
str) – The domainlimit (
int) – Maximum results
- Return type:
List[MemoryEntry]- Returns:
List of knowledge entries
- clear_agent_memories(agent_id, memory_types=None)[source]
Clear memories for a specific agent
- Parameters:
agent_id (
str) – The agent IDmemory_types (
Optional[List[MemoryType]]) – Types of memory to clear (all if None)
- get_active_backend()[source]
Get information about the currently active memory backend.
- Returns:
name: Backend name (“chromadb”, “qdrant”, “memory”)
type: “persistent” or “in_memory”
available: Whether the backend is operational
details: Backend-specific information
- Return type:
Dict[str,Any]
- recall_by_id(memory_id)[source]
Recall a specific memory by ID (for spore references)
- Return type:
List[MemoryEntry]- Parameters:
memory_id (str)