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.6.tar.gz (9.9 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.6-py3-none-any.whl (10.2 kB view details)

Uploaded Python 3

File details

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

File metadata

  • Download URL: mantr-1.0.6.tar.gz
  • Upload date:
  • Size: 9.9 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.6.tar.gz
Algorithm Hash digest
SHA256 c48b0ed4b1718514a7bf0aa6f7dbbf6c9598a48ec4ad91bf7258c545e7da9f42
MD5 0db3a051afe132a6a34d286b556e60a1
BLAKE2b-256 2796d5237a63f33bfd8709962975a55094460c5c596f2c4989e8af8ed8eb9630

See more details on using hashes here.

File details

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

File metadata

  • Download URL: mantr-1.0.6-py3-none-any.whl
  • Upload date:
  • Size: 10.2 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.6-py3-none-any.whl
Algorithm Hash digest
SHA256 e8b435aecd2dda9f6cf74e3b77cfae8300baaf5736b20fa0a387c44bd928b087
MD5 0f538f277b1be8a8d22ddf2b5349ccac
BLAKE2b-256 60623dabe5b641a7f278c941667f23e55eba5533079a37554b0308c4bca2fc37

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