Python SDK

pip install memvid-sdk

from memvid_sdk import create, use
import os

path = 'knowledge.mv2'

# IMPORTANT: Use create() for NEW files, use() for EXISTING files
if os.path.exists(path):
    mem = use('basic', path)  # Open existing file
else:
    mem = create(path)  # Create new file (kind='basic' is default)

# Lexical search (BM25) is enabled by default - no need to call enable_lex()

# Add documents (no embeddings needed!)
mem.put(
    title='Meeting Notes',
    label='notes',
    metadata={'source': 'slack'},
    text='Alice mentioned she works at Anthropic...'
)

# Search works immediately
results = mem.find('who works at AI companies?', k=5, mode='lex')
print(results['hits'])

# Ask questions
answer = mem.ask('What does Alice do?', k=5, mode='lex')
print(answer['text'])

# Seal when done (commits changes)
mem.seal()

import memvid_sdk as memvid

# Automatically closes when done
with memvid.use('basic', 'memory.mv2') as mem:
    mem.put('Doc', 'test', {}, text='Content')
    results = mem.find('query')

from memvid_sdk import create, use, lock, unlock, info

# Create new memory file
mem = create('project.mv2')

# With options
mem = create(
    'project.mv2',
    enable_vec=True,      # Enable vector index
    enable_lex=True,      # Enable lexical index
    memory_id='mem_abc',  # Bind to dashboard
    api_key='mv_live_...' # Dashboard API key
)

# Open existing memory with adapter
mem = use('basic', 'project.mv2', mode='open')

# Available adapters
mem = use('langchain', 'file.mv2')
mem = use('llamaindex', 'file.mv2')
mem = use('crewai', 'file.mv2')
mem = use('autogen', 'file.mv2')
mem = use('haystack', 'file.mv2')
mem = use('langgraph', 'file.mv2')
mem = use('semantic-kernel', 'file.mv2')
mem = use('openai', 'file.mv2')
mem = use('google-adk', 'file.mv2')

# Get SDK info
sdk_info = info()

# Verify file integrity
result = mem.verify(deep=True)

# Repair and optimize
result = mem.doctor(
    rebuild_time_index=True,
    rebuild_vec_index=True,
    vacuum=True
)

from memvid_sdk import create

# Automatically closes when done
with create('project.mv2') as mem:
    mem.put('Title', 'label', {}, text='Content')
    results = mem.find('query')

mem.put(
    'Document Title',      # title (required, positional)
    'label',               # label (required, positional)
    {},                    # metadata dict (required, can be {})

    # Content (one of these)
    text='Document content...',
    file='/path/to/document.pdf',

    # Optional
    uri='mv2://docs/intro',
    tags=['api', 'v2'],
    labels=['public', 'reviewed'],
    kind='markdown',
    track='documentation',

    # Embeddings
    enable_embedding=True,
    embedding_model='bge-small',   # or 'openai', 'nomic', etc.
    vector_compression=True,       # 16x compression with PQ

    # Behavior
    auto_tag=True,                 # Auto-generate tags
    extract_dates=True             # Extract date mentions
)

docs = [
    {'title': 'Doc 1', 'label': 'kb', 'text': 'First document'},
    {'title': 'Doc 2', 'label': 'kb', 'text': 'Second document'},
    {'title': 'Doc 3', 'label': 'kb', 'text': 'Third document'}
]

frame_ids = mem.put_many(
    docs,
    embedder=openai_embeddings,  # Custom embedder
    opts={
        'compression_level': 3,
        'enable_embedding': True,
        'embedding_model': 'bge-small'
    }
)

# Simple search
results = mem.find('budget projections')

# With options
results = mem.find(
    'financial outlook',
    mode='auto',              # 'lex', 'sem', 'auto'
    k=10,                     # Number of results
    snippet_chars=480,        # Snippet length
    scope='track:meetings',   # Scope filter

    # Adaptive retrieval
    adaptive=True,
    min_relevancy=0.5,
    max_k=100,
    adaptive_strategy='combined',  # 'relative', 'absolute', 'cliff', 'elbow'

    # Time-travel
    as_of_frame=100,
    as_of_ts=1704067200,

    # Custom embeddings
    embedder=custom_embedder,
    query_embedding_model='openai'
)

for hit in results['hits']:
    print(hit['title'], hit['snippet'])

answer = mem.ask(
    'What was decided about the budget?',
    k=8,
    mode='auto',

    # LLM settings
    model='gpt-4o-mini',
    api_key=os.environ['OPENAI_API_KEY'],
    llm_context_chars=120000,

    # Privacy
    mask_pii=True,

    # Time filters
    since=1704067200,
    until=1706745600,

    # Options
    context_only=False,    # Set True to skip synthesis

    # Adaptive retrieval
    adaptive=True,
    min_relevancy=0.5
)

print(answer['text'])
print(answer['hits'])  # Source documents

answer = mem.ask(
    'What is the API endpoint?',
    model='gpt-4o-mini',
    api_key=os.environ['OPENAI_API_KEY']
)

# Check grounding quality
print(answer['grounding'])
# {
#   'score': 0.85,
#   'label': 'HIGH',           # 'LOW', 'MEDIUM', or 'HIGH'
#   'sentence_count': 3,
#   'grounded_sentences': 3,
#   'has_warning': False,
#   'warning_reason': None
# }

# Check if follow-up is needed
follow_up = answer.get('follow_up')
if follow_up and follow_up['needed']:
    print('Low confidence:', follow_up['reason'])
    print('Try these instead:', follow_up['suggestions'])

# Store a correction
frame_id = mem.correct('Ben Koenig reported to Chloe Nguyen before 2025')

# With options
frame_id = mem.correct(
    'The API rate limit is 1000 req/min',
    source='Engineering Team - Jan 2025',
    topics=['API', 'rate limiting'],
    boost=2.5  # Higher retrieval priority (default: 2.0)
)

# Batch corrections
frame_ids = mem.correct_many([
    {'statement': 'OAuth tokens expire after 24 hours', 'topics': ['auth', 'OAuth']},
    {'statement': 'Production DB is db.prod.example.com', 'source': 'Ops Team'}
])

# Verify correction is retrievable
results = mem.find('Ben Koenig reported to')
print(results['hits'][0]['snippet'])  # Should show the correction

# Extract facts using rules engine (fast, offline)
result = mem.enrich(engine='rules')

# Extract with LLM (more accurate)
result = mem.enrich(engine='openai')

# View extracted cards
memories = mem.memories()
print(memories['cards'])

# Filter by entity
alice_cards = mem.memories(entity='Alice')

# Get entity state (O(1) lookup)
alice = mem.state('Alice')
print(alice['slots'])
# {'employer': 'Anthropic', 'role': 'Engineer', 'location': 'SF'}

# Get stats
stats = mem.memories_stats()
print(stats['entityCount'], stats['cardCount'])

# List all entities
entities = mem.memory_entities()

# Add SPO triplets directly
result = mem.add_memory_cards([
    {'entity': 'Alice', 'slot': 'employer', 'value': 'Anthropic'},
    {'entity': 'Alice', 'slot': 'role', 'value': 'Senior Engineer'},
    {'entity': 'Bob', 'slot': 'team', 'value': 'Infrastructure'}
])

print(result['added'], result['ids'])

# Export to JSON
json_data = mem.export_facts(format='json')

# Export to CSV
csv_data = mem.export_facts(format='csv', entity='Alice')

# Export to N-Triples (RDF)
ntriples = mem.export_facts(format='ntriples')

# Extract tables from PDF
result = mem.put_pdf_tables('financial-report.pdf', embed_rows=True)
print(f"Extracted {result['tables_count']} tables")

# List all tables
tables = mem.list_tables()
for table in tables:
    print(table['table_id'], table['n_rows'], table['n_cols'])

# Get table data
data = mem.get_table('tbl_001', format='dict')
csv_data = mem.get_table('tbl_001', format='csv')

timeline = mem.timeline(
    limit=50,
    since=1704067200,
    until=1706745600,
    reverse=True,
    as_of_frame=100
)

# Start recording
session_id = mem.session_start('qa-test')

# Perform operations
mem.find('test query')
mem.ask('What happened?')

# Add checkpoint
mem.session_checkpoint()

# End session
summary = mem.session_end()

# List sessions
sessions = mem.session_list()

# Replay session with different params
replay = mem.session_replay(session_id, top_k=10, adaptive=True)
print(replay['match_rate'])

# Delete session
mem.session_delete(session_id)

from memvid_sdk import lock, unlock, lock_who, lock_nudge

# Encrypt to .mv2e capsule
encrypted_path = lock(
    'project.mv2',
    password='secret',
    force=True
)

# Decrypt back to .mv2
decrypted_path = unlock(
    'project.mv2e',
    password='secret'
)

# Check who has the lock
lock_info = lock_who('project.mv2')

# Nudge stale lock
released = lock_nudge('project.mv2')

# Get current capacity
capacity = mem.get_capacity()

# Get current ticket info
ticket = mem.current_ticket()

# Sync tickets from dashboard
result = mem.sync_tickets('mem_abc123', api_key)

# Apply ticket manually
mem.apply_ticket(ticket_string)

# Get memory binding
binding = mem.get_memory_binding()

# Unbind from dashboard
mem.unbind_memory()

from memvid_sdk.embeddings import (
    OpenAIEmbeddings,
    GeminiEmbeddings,
    MistralEmbeddings,
    CohereEmbeddings,
    VoyageEmbeddings,
    NvidiaEmbeddings,
    HuggingFaceEmbeddings,
    get_embedder
)

# OpenAI
openai = OpenAIEmbeddings(
    api_key=os.environ['OPENAI_API_KEY'],
    model='text-embedding-3-small'  # or 'text-embedding-3-large'
)

# Gemini
gemini = GeminiEmbeddings(
    api_key=os.environ['GEMINI_API_KEY'],
    model='text-embedding-004'
)

# Mistral
mistral = MistralEmbeddings(
    api_key=os.environ['MISTRAL_API_KEY']
)

# Cohere
cohere = CohereEmbeddings(
    api_key=os.environ['COHERE_API_KEY'],
    model='embed-english-v3.0'
)

# Voyage
voyage = VoyageEmbeddings(
    api_key=os.environ['VOYAGE_API_KEY'],
    model='voyage-3'
)

# NVIDIA
nvidia = NvidiaEmbeddings(
    api_key=os.environ['NVIDIA_API_KEY']
)

# HuggingFace (local)
hf = HuggingFaceEmbeddings(model='all-MiniLM-L6-v2')

# Factory function
embedder = get_embedder('openai', api_key='...')

# Use with put_many
mem.put_many(docs, embedder=openai)

# Use with find
mem.find('query', embedder=gemini)

from memvid_sdk.embeddings import LOCAL_EMBEDDING_MODELS

mem.put(
    'Title', 'label', {},
    text='content',
    enable_embedding=True,
    embedding_model=LOCAL_EMBEDDING_MODELS['BGE_SMALL']  # 384d, fast
)

# Available local models
LOCAL_EMBEDDING_MODELS['BGE_SMALL']   # 384d - fastest
LOCAL_EMBEDDING_MODELS['BGE_BASE']    # 768d - balanced
LOCAL_EMBEDDING_MODELS['NOMIC']       # 768d - general purpose
LOCAL_EMBEDDING_MODELS['GTE_LARGE']   # 1024d - highest quality

from memvid_sdk import (
    MemvidError,
    CapacityExceededError,      # MV001
    TicketInvalidError,         # MV002
    TicketReplayError,          # MV003
    LexIndexDisabledError,      # MV004
    TimeIndexMissingError,      # MV005
    VerifyFailedError,          # MV006
    LockedError,                # MV007
    ApiKeyRequiredError,        # MV008
    MemoryAlreadyBoundError,    # MV009
    FrameNotFoundError,         # MV010
    VecIndexDisabledError,      # MV011
    CorruptFileError,           # MV012
    VecDimensionMismatchError,  # MV014
    EmbeddingFailedError,       # MV015
    EncryptionError,            # MV016
    NerModelNotAvailableError,  # MV017
    ClipIndexDisabledError      # MV018
)

try:
    mem.put('Large file', 'data', {}, file='huge.bin')
except CapacityExceededError:
    print('Storage capacity exceeded (MV001)')
except LockedError:
    print('File locked by another process (MV007)')
except VecIndexDisabledError:
    print('Enable vector index first (MV011)')

# Get frame by URI
frame = mem.frame('mv2://docs/intro')

# Get raw binary data
data = mem.blob('mv2://docs/intro')

from memvid_sdk import info, flush_analytics, is_telemetry_enabled, verify_single_file

# Get SDK info
sdk_info = info()
print(sdk_info['version'], sdk_info['platform'])

# Flush analytics
flush_analytics()

# Check telemetry status
enabled = is_telemetry_enabled()

# Verify no auxiliary files
verify_single_file('project.mv2')