Skip to main content

Official Python SDK for Mantr - Deterministic Semantic Memory

Project description

Mantr Python SDK

Deterministic Semantic Memory for AI Agents

PyPI version Python 3.8+ License: MIT


What is Mantr?

Mantr is a deterministic semantic memory system built on Sanskrit phonetic ontology. Unlike vector databases that use probabilistic embeddings, Mantr provides 100% reproducible knowledge retrieval through graph-based semantic walks.

Perfect for:

  • 🤖 AI agents that need reliable memory
  • 📚 RAG systems requiring deterministic context
  • 🔍 Enterprise search with zero hallucinations
  • 💬 Multi-turn conversations with stable context
  • 🧠 Knowledge graphs for LLM grounding

Quick Start

Installation

pip install mantr

30-Second Example

from mantr import MantrClient

# Initialize
client = MantrClient(api_key='vak_live_...')

# Walk the semantic graph
result = client.walk(phonemes=['dharma', 'karma', 'yoga'])

# Get connections
for path in result.paths:
    print(f"{' → '.join(path.nodes)} (score: {path.score})")

Get Your API Key

  1. Sign up at mantr.net/signup
  2. Get 5,000 free walks/month
  3. No credit card required

Installation & Setup

From PyPI

pip install mantr

From Source

git clone https://github.com/Mantrnet/python-sdk
cd python-sdk
pip install -e .

Environment Setup

import os
os.environ['MANTR_API_KEY'] = 'vak_live_...'

# Or pass directly
from mantr import MantrClient
client = MantrClient(api_key='vak_live_...')

Core Concepts

Phonemes

Sanskrit phonetic units that represent semantic concepts:

# English concepts → Sanskrit phonemes
client.walk(['knowledge'])    # Jnana (ज्ञान)
client.walk(['stability'])    # Sthiti (स्थिति)
client.walk(['power'])        # Shakti (शक्ति)

Semantic Walks

Graph traversals that discover connections:

result = client.walk(
    phonemes=['authentication', 'security'],
    depth=3,      # How deep to traverse
    limit=100     # Max results to return
)

Determinism

Same input = Same output, always:

walk1 = client.walk(['test'])
walk2 = client.walk(['test'])
assert walk1.paths == walk2.paths  # ✓ Always True

AI & Agentic Usage Patterns

1. RAG (Retrieval-Augmented Generation)

Use Mantr for deterministic context retrieval before LLM calls:

from mantr import MantrClient
import openai

client = MantrClient(api_key='vak_live_...')

def rag_query(user_question: str) -> str:
    """RAG pattern with Mantr as retrieval layer"""
    
    # Step 1: Get deterministic context from Mantr
    context = client.walk(
        phonemes=extract_concepts(user_question),
        depth=4,
        limit=50
    )
    
    # Step 2: Format context for LLM
    context_str = "\n".join([
        f"- {' → '.join(path.nodes)}"
        for path in context.paths[:10]
    ])
    
    # Step 3: Pass to LLM
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": f"Knowledge: {context_str}"},
            {"role": "user", "content": user_question}
        ]
    )
    
    return response.choices[0].message.content

# Usage
answer = rag_query("How do I implement OAuth?")

2. Multi-Agent Memory

Shared deterministic memory across multiple agents:

class AgentMemory:
    def __init__(self, agent_id: str):
        self.agent_id = agent_id
        self.client = MantrClient(api_key='vak_live_...')
        self.pod = f"agent_{agent_id}"  # Isolated memory per agent
    
    def remember(self, concept: str) -> dict:
        """Agent recalls from its memory"""
        return self.client.walk(
            phonemes=[concept],
            pod=self.pod,
            depth=3
        )
    
    def share_with(self, other_agent_id: str, concept: str):
        """Share knowledge with another agent"""
        # Both agents can access shared concepts
        return self.client.walk(
            phonemes=[concept],
            pod=f"shared_{self.agent_id}_{other_agent_id}"
        )

# Multi-agent system
researcher = AgentMemory("researcher")
writer = AgentMemory("writer")

# Researcher finds info
research = researcher.remember("machine_learning")

# Writer accesses shared knowledge
context = researcher.share_with("writer", "machine_learning")

3. Conversational Context Tracking

Maintain conversation state across turns:

class ConversationMemory:
    def __init__(self, user_id: str):
        self.client = MantrClient(api_key='vak_live_...')
        self.user_id = user_id
        self.history = []
    
    def process_turn(self, user_message: str) -> str:
        """Process a conversation turn with context"""
        
        # Extract concepts from all history
        all_concepts = self.history + [user_message]
        
        # Get relevant context
        context = self.client.walk(
            phonemes=extract_concepts(all_concepts),
            pod=f"user_{self.user_id}",
            depth=5
        )
        
        # Generate response with context
        response = generate_llm_response(user_message, context)
        
        # Update history
        self.history.append(user_message)
        self.history.append(response)
        
        return response

# Usage
conv = ConversationMemory("user_123")
print(conv.process_turn("Tell me about Python"))
print(conv.process_turn("How does it compare to Java?"))  # Has context!

4. Hybrid Retrieval (Mantr + Vectors)

Combine deterministic + fuzzy search:

from mantr import MantrClient
import pinecone

mantr = MantrClient(api_key='vak_live_...')
pinecone.init(api_key='...')
index = pinecone.Index('docs')

def hybrid_search(query: str, top_k: int = 10):
    """Combine deterministic graph + vector similarity"""
    
    # 1. Deterministic graph walk
    graph_results = mantr.walk(
        phonemes=extract_concepts(query),
        depth=4,
        limit=top_k * 2
    )
    
    # 2. Vector similarity search
    vector_results = index.query(
        vector=embed(query),
        top_k=top_k * 2
    )
    
    # 3. Merge and rerank
    combined = merge_results(graph_results, vector_results)
    
    # 4. Use both for context
    return {
        'structured': graph_results.paths[:top_k],
        'fuzzy': vector_results.matches[:top_k],
        'combined': combined[:top_k]
    }

5. Intent Classification

Deterministic routing for multi-intent systems:

class IntentRouter:
    def __init__(self):
        self.client = MantrClient(api_key='vak_live_...')
        self.intent_map = {
            'billing': ['payment', 'invoice', 'charge'],
            'support': ['help', 'issue', 'problem'],
            'sales': ['pricing', 'plan', 'upgrade']
        }
    
    def classify(self, user_message: str) -> str:
        """Route to correct handler"""
        
        result = self.client.walk(
            phonemes=extract_concepts(user_message),
            depth=2
        )
        
        # Match to intent categories
        for intent, keywords in self.intent_map.items():
            for path in result.paths:
                if any(k in path.nodes for k in keywords):
                    return intent
        
        return 'general'

# Usage
router = IntentRouter()
intent = router.classify("How do I cancel my subscription?")
# → 'billing'

6. Knowledge Graph Construction

Build domain-specific knowledge graphs:

class KnowledgeGraph:
    def __init__(self, domain: str):
        self.client = MantrClient(api_key='vak_live_...')
        self.domain = domain
    
    def explore_topic(self, root_concept: str, max_depth: int = 5):
        """Build graph from a starting concept"""
        
        explored = set()
        to_explore = [root_concept]
        graph = {}
        
        while to_explore and len(explored) < 1000:
            current = to_explore.pop(0)
            if current in explored:
                continue
            
            # Walk from this concept
            result = self.client.walk(
                phonemes=[current],
                pod=self.domain,
                depth=1
            )
            
            # Add edges to graph
            graph[current] = [p.nodes for p in result.paths]
            
            # Add new nodes to explore
            for path in result.paths:
                for node in path.nodes:
                    if node not in explored:
                        to_explore.append(node)
            
            explored.add(current)
        
        return graph

# Build ML knowledge graph
kg = KnowledgeGraph("machine_learning")
ml_graph = kg.explore_topic("neural_network")

7. Semantic Caching

Cache LLM responses based on semantic similarity:

class SemanticCache:
    def __init__(self):
        self.client = MantrClient(api_key='vak_live_...')
        self.cache = {}
    
    def get_or_generate(self, query: str, generator):
        """Get cached response or generate new"""
        
        # Find semantically similar queries
        similar = self.client.walk(
            phonemes=extract_concepts(query),
            depth=2,
            limit=5
        )
        
        # Check if we have a cached response
        for path in similar.paths:
            cache_key = tuple(path.nodes)
            if cache_key in self.cache and path.score > 0.9:
                return self.cache[cache_key]
        
        # Generate new response
        response = generator(query)
        
        # Cache it
        concepts = extract_concepts(query)
        self.cache[tuple(concepts)] = response
        
        return response

Advanced Features

Context Pods (Isolated Knowledge)

# Create separate knowledge domains
docs_client = MantrClient(api_key='vak_live_...', pod='documentation')
support_client = MantrClient(api_key='vak_live_...', pod='support_tickets')

# Each pod maintains isolated context
docs_result = docs_client.walk(['api'])
support_result = support_client.walk(['api'])  # Different results!

Batch Processing

from concurrent.futures import ThreadPoolExecutor

def batch_walk(concepts_list: list) -> list:
    """Process multiple walks in parallel"""
    
    client = MantrClient(api_key='vak_live_...')
    
    with ThreadPoolExecutor(max_workers=10) as executor:
        futures = [
            executor.submit(client.walk, concepts)
            for concepts in concepts_list
        ]
        return [f.result() for f in futures]

# Process 100 queries in parallel
results = batch_walk([
    ['concept_1'], ['concept_2'], ... ['concept_100']
])

Error Handling

from mantr import (
    MantrClient,
    AuthenticationError,
    InsufficientCreditsError,
    RateLimitError
)

client = MantrClient(api_key='vak_live_...')

try:
    result = client.walk(['test'])
except AuthenticationError:
    print("Invalid API key")
except InsufficientCreditsError:
    print("Out of credits - upgrade plan")
except RateLimitError as e:
    print(f"Rate limited - retry after {e.retry_after}s")

API Reference

MantrClient

client = MantrClient(
    api_key: str,              # Required: Your API key
    base_url: str = "https://api.mantr.net",  # Optional: Custom endpoint
    timeout: int = 30          # Optional: Request timeout (seconds)
)

walk()

result = client.walk(
    phonemes: List[str],       # Required: Concepts to explore
    pod: Optional[str] = None, # Optional: Isolated context
    depth: int = 3,            # Optional: Traversal depth (1-10)
    limit: int = 100           # Optional: Max results (1-1000)
) -> WalkResponse

Returns:

WalkResponse(
    paths: List[PathResult],  # Discovered connections
    latency_us: int,          # Execution time (microseconds)
    credits_used: int         # Credits consumed
)

PathResult(
    nodes: List[str],         # Sequence of concepts
    score: float,             # Relevance (0.0-1.0)
    depth: int                # Path depth
)

Examples

See examples/ directory for complete examples:

  • quickstart.py - Basic usage
  • rag_chatbot.py - RAG implementation
  • multi_agent.py - Multi-agent memory
  • hybrid_search.py - Mantr + vector search
  • intent_router.py - Classification system

Performance & Limits

Tier Walks/Month Latency Rate Limit
Sadhaka (Free) 5,000 <100ms 10/min
Rishi ($49/mo) 2,000,000 <50ms 100/min
Brahma ($2,500/mo) Unlimited <10ms No limit

Support


License

MIT License - see LICENSE for details.


Built with ❤️ for the AI community

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

mantr-1.0.3.tar.gz (9.2 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

mantr-1.0.3-py3-none-any.whl (9.4 kB view details)

Uploaded Python 3

File details

Details for the file mantr-1.0.3.tar.gz.

File metadata

  • Download URL: mantr-1.0.3.tar.gz
  • Upload date:
  • Size: 9.2 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.11.14

File hashes

Hashes for mantr-1.0.3.tar.gz
Algorithm Hash digest
SHA256 4a2a09b19b2ca5be541cd235c4ef5dcc78df77c5b642c88627a914d87d27a20b
MD5 53fd2efc0fddaf7ea5436ccc995b14bc
BLAKE2b-256 e6050874c9d37b2cb3ccecd8037d8ebeaf495de07fea17d5b9a9cb2d20b10a75

See more details on using hashes here.

File details

Details for the file mantr-1.0.3-py3-none-any.whl.

File metadata

  • Download URL: mantr-1.0.3-py3-none-any.whl
  • Upload date:
  • Size: 9.4 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.11.14

File hashes

Hashes for mantr-1.0.3-py3-none-any.whl
Algorithm Hash digest
SHA256 0029034cd1092573ddd4a0546b1f51039faa4273643d81a10b8061edadb121a8
MD5 0007cada79a520629747a96b596b7467
BLAKE2b-256 acf43db0d1a7cd988fd7889c1a5eeb5b7f52ae55abb3ef6c10c84a88db455f34

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page