miruvor 0.1.1

Python SDK for Zyra SNN Memory Database - Brain-inspired memory for AI agents

Miruvor

Python SDK for Zyra SNN Memory Database - Brain-inspired memory for AI agents.

Features

• Simple API: Three core methods - store(), ingest(), retrieve()
• Pattern Completion: O(L) retrieval via pointer layer - no training required
• Sync & Async: Both MiruvorClient and AsyncMiruvorClient
• Type Safe: Full Pydantic v2 models with validation
• Production Ready: Automatic retries, timeouts, logging, error handling
• Batch Operations: Efficient batch store with concurrency control
• Health Checks: Built-in health monitoring

Installation

pip install miruvor

Quick Start

Sync Usage

from miruvor import MiruvorClient

# Defaults to production. Set MIRUVOR_BASE_URL for dev (e.g. http://localhost:8000).
# Set MIRUVOR_API_KEY in env to skip passing api_key.
client = MiruvorClient(api_key="your-api-key", token="your-jwt-token")  # token optional

# Store a memory
response = client.store(
    text="The user prefers dark mode and likes Python",
    tags=["preferences", "user"],
    metadata={"source": "settings"}
)
print(f"Stored: {response.memory_id} in {response.storage_time_ms:.2f}ms")

# Retrieve memories
results = client.retrieve(query="What does the user prefer?", top_k=5)
for memory in results.results:
    print(f"[{memory.score:.3f}] {memory.data['text']}")

# Ingest with LLM enhancement
ingest_response = client.ingest(content="Long document...", priority="high")
print(f"Queued: {ingest_response.message_id}")

client.close()

Async Usage

import asyncio
from miruvor import AsyncMiruvorClient

async def main():
    async with AsyncMiruvorClient(api_key="your-api-key", token="your-jwt-token") as client:
        response = await client.store(text="User loves async/await", tags=["preferences"])
        results = await client.retrieve(query="What does user love?")
        print(f"Found {results.num_results} memories")

asyncio.run(main())

Batch Operations

import asyncio
from miruvor import AsyncMiruvorClient

async def batch_example():
    async with AsyncMiruvorClient(api_key="...", token="...") as client:
        memories = [
            {"text": "Memory 1", "tags": ["tag1"]},
            {"text": "Memory 2", "tags": ["tag2"]},
            {"text": "Memory 3", "tags": ["tag3"]},
        ]
        responses = await client.store_batch(memories, max_concurrent=10)
        print(f"Stored {len(responses)} memories")

asyncio.run(batch_example())

API Reference

MiruvorClient / AsyncMiruvorClient

Constructor: MiruvorClient(api_key, base_url=None, token=None, timeout=30.0, max_retries=3) — base_url defaults to production; override with MIRUVOR_BASE_URL env.

Methods:

• health() - Check API health status
• store(text, tags=None, metadata=None) - Store a memory
• store_batch(memories, max_concurrent=10) - Store multiple memories (async has concurrency)
• ingest(content, priority="normal", ...) - Ingest with LLM enhancement
• retrieve(query, top_k=5, use_sparse=None) - Retrieve via pattern completion

Error Handling

from miruvor import MiruvorClient, AuthenticationError, RateLimitError, ValidationError

try:
    response = client.store(text="Hello world")
except AuthenticationError as e:
    print(f"Auth failed: {e.message}")
except RateLimitError as e:
    print(f"Retry after {e.retry_after} seconds")
except ValidationError as e:
    print(f"Validation error: {e.message}")

Examples

See the examples/ directory:

• quickstart.py - Basic sync usage
• async_example.py - Async operations
• batch_ingestion.py - Batch operations with concurrency

Development

pip install -e ".[dev]"
pytest
black src/ tests/
ruff check src/ tests/
mypy src/

Requirements

• Python 3.9+
• requests >= 2.31.0
• httpx >= 0.27.0
• pydantic >= 2.0.0
• urllib3 >= 2.0.0

License

MIT License - see LICENSE file for details.