Memory Operations

from remina import Memory
 
memory = Memory()
 
result = memory.add(
    messages="I'm Sarah, a data scientist. I specialize in NLP and prefer PyTorch.",
    user_id="sarah_001"
)
 
# Result structure
# {
#   "results": [
#     {"id": "abc123", "memory": "Name is Sarah", "event": "ADD"},
#     {"id": "def456", "memory": "Is a data scientist", "event": "ADD"},
#     {"id": "ghi789", "memory": "Specializes in NLP", "event": "ADD"},
#     {"id": "jkl012", "memory": "Prefers PyTorch", "event": "ADD"}
#   ]
# }

result = memory.add(
    messages=[
        {"role": "user", "content": "I relocated to Seattle last month"},
        {"role": "assistant", "content": "How's the transition?"},
        {"role": "user", "content": "Good overall, adjusting to the weather"}
    ],
    user_id="sarah_001"
)

result = memory.add(
    messages="I prefer using PyTorch over TensorFlow",
    user_id="sarah_001",
    metadata={
        "source": "preferences_conversation",
        "confidence": 0.95,
        "category": "tools"
    }
)

results = memory.search(
    query="What ML frameworks does Sarah use?",
    user_id="sarah_001",
    limit=5
)
 
for mem in results['results']:
    print(f"[{mem['score']:.3f}] {mem['memory']}")

results = memory.search(
    query="user preferences",
    user_id="sarah_001",
    limit=10,
    filters={"category": "tools"}
)

memory_data = memory.get(memory_id="abc123")
 
if memory_data:
    print(f"Content: {memory_data['content']}")
    print(f"Created: {memory_data['created_at']}")

all_memories = memory.get_all(user_id="sarah_001", limit=100)
 
print(f"Total: {len(all_memories['results'])}")
for mem in all_memories['results']:
    print(f"  - {mem['memory']}")

result = memory.update(
    memory_id="abc123",
    content="Name is Sarah Chen"
)
 
if result['success']:
    print("Memory updated")

result = memory.delete(memory_id="abc123")

result = memory.delete_all(user_id="sarah_001")
print(result['message'])  # "Deleted 15 memories"

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

# Recommended: unique, identifiable
user_id = "user_12345"
user_id = "org_acme_user_john"
 
# Avoid: generic identifiers
user_id = "user"
user_id = "1"

memory.add(
    messages="...",
    user_id="user_123",
    metadata={
        "session_id": "sess_abc",
        "source": "onboarding",
        "timestamp": "2024-01-15T10:30:00Z"
    }
)

# Always close connections
memory.close()

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