memblock
HOME

[ SDK REFERENCE ]

MemBlock Documentation

Everything you need to store, query, and reason over structured memory in Python.

pip install memblock

Installation Extras

The base install gives you everything for local memory with SQLite. Add optional extras for additional capabilities.

pip install "memblock[postgres]"

PostgreSQL storage backend for production multi-tenant deployments

pip install "memblock[embeddings]"

Local vector embeddings via FastEmbed — CPU-only, no API key needed

pip install "memblock[llm]"

LLM-powered memory extraction with OpenAI, Anthropic, and Google Gemini

pip install "memblock[reranker-cohere]"

Cohere-based reranker for improved search relevance

pip install "memblock[reranker-cross-encoder]"

HuggingFace cross-encoder reranker — runs locally

pip install "memblock[pool]"

Connection pooling via psycopg_pool for production PostgreSQL deployments

pip install "memblock[pgvector]"

Server-side HNSW vector search via pgvector — replaces brute-force cosine

pip install "memblock[all-cloud]"

All cloud backends, cloud embeddings (OpenAI, Gemini), LLM extraction, Cohere reranker, pooling, and pgvector — works on Python 3.13+

pip install "memblock[all]"

Everything including local FastEmbed & cross-encoder reranker — requires Python ≤ 3.12 (onnxruntime dependency)

Core API

The essential methods for storing, querying, and building context from memory.

MemBlock()

MemBlock(storage="sqlite:///:memory:", encryption_key=None, embeddings=False)

Initialize a MemBlock instance. Supports SQLite (local file or in-memory) and PostgreSQL. Pass an encryption_key to enable AES-256 per-block encryption.

Returns: MemBlock instance

.store()

mem.store(content, type=BlockType.FACT, confidence=1.0, tags=None, encryption_level=EncryptionLevel.NONE, happened_at=None, temporal_precision="exact")

Store a new memory block. Supports 5 types: FACT, PREFERENCE, EVENT, ENTITY, RELATION. Optionally set happened_at for temporal anchoring and encrypt at STANDARD or SENSITIVE level.

Returns: Block

.query()

mem.query(type=None, tags=None, text_search=None, min_confidence=0.0, sort_by="relevance", limit=10)

Query blocks with structured filters. Combines full-text search, tag filtering, type filtering, and optional semantic (vector) hybrid search.

Returns: list[Block]

.build_context()

mem.build_context(query=None, token_budget=4000, strategy="relevance")

Build an LLM-ready context string from relevant memory. Strategies: relevance, graph_walk, type_grouped, adaptive. The new adaptive strategy combines relevance + graph expansion + deduplication.

Returns: str

.build_context_with_confidence()

mem.build_context_with_confidence(query=None, token_budget=4000, strategy="relevance", evidence_threshold=0.3)

Build context with confidence metadata for adversarial robustness. Returns a tuple of (context_string, ContextConfidence). When evidence is below threshold, signals that the question may be unanswerable.

Returns: (str, ContextConfidence)

.multi_hop_query()

mem.multi_hop_query(text_search, limit=10)

Multi-hop iterative retrieval for complex reasoning. Performs 3 passes: hybrid search, entity extraction + focused queries, and graph walk with proximity boosting. Best for questions connecting multiple facts.

Returns: list[Block]

.update()

mem.update(block_id, content=None, confidence=None, tags=None)

Update an existing block by ID. Only provided fields are modified. Triggers on_update hooks.

Returns: Block | None

.delete()

mem.delete(block_id, cascade=False)

Delete a block by ID. With cascade=True, also removes linked child blocks and graph edges.

Returns: bool

Knowledge Graph

Connect memories into a traversable graph. No external graph database required.

.link()

mem.link(source_id, target_id, relation=EdgeRelation.RELATED_TO)

Create a directed edge between two blocks. Supported relations: RELATED_TO, DERIVED_FROM, CONTRADICTS, SUPPORTS, PRECEDES, PART_OF.

Returns: None

.neighbors()

mem.neighbors(block_id, relation=None)

Get all blocks directly connected to a block. Optionally filter by relation type.

Returns: list[Block]

.traverse()

mem.traverse(block_id, max_depth=3)

Walk the knowledge graph from a starting block up to max_depth hops. Returns all reachable blocks.

Returns: list[Block]

Org-Level Analytics

Track what questions users are asking across your organization. Deduplicated, frequency-counted, with noise filtering and time-series breakdowns. Opt in with enable_analytics=True.

.log_question()

mem.log_question(question, user_id=None, org_id=None)

Log a user question for analytics. Automatically normalizes text, deduplicates, and tracks frequency and unique users. Noise (greetings, short strings) is filtered out and returns None.

Returns: QuestionRecord | None

.get_top_questions()

mem.get_top_questions(org_id=None, limit=20, since=None, until=None)

Get the most frequently asked questions for an organization, ordered by frequency. Optionally filter by time range.

Returns: list[QuestionRecord]

.get_trending_questions()

mem.get_trending_questions(org_id=None, window_days=7, limit=10)

Get questions trending upward — compares recent frequency against historical average to surface emerging topics.

Returns: list[QuestionTrend]

.get_question_breakdown()

mem.get_question_breakdown(question, org_id=None, granularity="daily")

Get a time-series breakdown of how often a specific question was asked. Granularity: "daily", "weekly", or "monthly".

Returns: dict[str, int]

.question_stats()

mem.question_stats(org_id=None)

Get aggregate analytics stats: total unique questions, total asks, and the top question with its frequency.

Returns: dict

Advanced

Tamper detection, async support, and data export.

.verify()

mem.verify()

Run tamper detection across the entire operation log. SHA-256 hash-chain verification catches any unauthorized modification.

Returns: TamperReport

.export_markdown()

mem.export_markdown()

Export all memory blocks as a formatted Markdown string. Useful for debugging, sharing, or archiving.

Returns: str

AsyncMemBlock

async with AsyncMemBlock(storage=...) as mem:

Async wrapper around MemBlock. All methods (store, query, build_context, etc.) are awaitable. Use as an async context manager.

Returns: AsyncMemBlock instance

Production

Connection pooling, per-user instance caching, and server-side vector search for multi-tenant deployments.

MemBlockPool()

MemBlockPool(storage, max_instances=100, **shared_kwargs)

LRU-cached instance factory for multi-tenant deployments. Creates and caches per-user MemBlock instances. Evicts least-recently-used when exceeding max_instances. Use as a context manager.

Returns: MemBlockPool instance

pool.get()

pool.get(user_id, **overrides)

Get or create a MemBlock instance for a user. Returns cached instance if available, creates new one otherwise. LRU eviction keeps memory bounded.

Returns: MemBlock

AsyncMemBlockPool()

AsyncMemBlockPool(storage, max_instances=100, **shared_kwargs)

Async wrapper around MemBlockPool. Returns AsyncMemBlock instances via await pool.get(). Use as an async context manager with FastAPI or aiohttp.

Returns: AsyncMemBlockPool instance

Connection Pooling

MemBlock(storage="postgresql://...", user_id="u_123", pool=connection_pool)

Pass a psycopg_pool.ConnectionPool to share connections across requests. Connections are borrowed on use and returned on close() — no per-request connection overhead.

Returns: MemBlock instance (pooled)

pgvector Search

pip install "memblock[pgvector]"

When the pgvector PostgreSQL extension is installed, MemBlock auto-detects it and uses server-side HNSW vector search instead of brute-force Python cosine. Dual-write to BYTEA + vector column ensures backward compatibility. No code changes needed.

Returns: Automatic — transparent upgrade

Block Types

Every memory block has a type. Use the right type to enable structured queries and context building.

FACTObjective information or knowledge
PREFERENCEUser likes, dislikes, or choices
EVENTSomething that happened at a point in time
ENTITYA person, place, organization, or thing
RELATIONA connection between two entities

Quick Example

from memblock import MemBlock, BlockType

mem = MemBlock(storage="sqlite:///memory.db")

# Store a preference
mem.store("User prefers dark mode", type=BlockType.PREFERENCE)

# Query by text
results = mem.query(text_search="dark mode")

# Build LLM context
context = mem.build_context(
    query="user preferences",
    token_budget=2000
)

Analytics Example

from memblock import MemBlock

mem = MemBlock(
    storage="sqlite:///memory.db",
    enable_analytics=True,
    org_id="my_org"
)

# Log questions from your chatbot users
mem.log_question("How do I reset my password?", user_id="user_42")
mem.log_question("How do I reset my password?", user_id="user_88")
mem.log_question("What is the refund policy?", user_id="user_42")

# See what users are asking most
top = mem.get_top_questions(limit=10)
# → [QuestionRecord("how do i reset my password", freq=2, users=2), ...]

# Get trending questions (rising in the last 7 days)
trending = mem.get_trending_questions(window_days=7)

# Daily breakdown for a specific question
breakdown = mem.get_question_breakdown(
    "How do I reset my password?",
    granularity="daily"
)
# → {"2026-03-16": 2}

# Aggregate stats
stats = mem.question_stats()
# → {"total_unique_questions": 2, "total_asks": 3, ...}

Temporal + Multi-Hop Example

from memblock import MemBlock, BlockType
from datetime import datetime, timezone

mem = MemBlock(storage="sqlite:///memory.db")

# Store events with temporal anchoring
mem.store(
    "User deployed v2.0 to production",
    type=BlockType.EVENT,
    happened_at=datetime(2024, 3, 10, tzinfo=timezone.utc),
    temporal_precision="day",
    tags=["deploy", "v2"]
)

# Temporal queries are parsed automatically
results = mem.query(text_search="What happened last March?")

# Multi-hop retrieval for complex questions
results = mem.multi_hop_query("Who recommended the restaurant?")

# Adaptive context with confidence gating
context, confidence = mem.build_context_with_confidence(
    query="user preferences",
    strategy="adaptive",
    evidence_threshold=0.3
)
if not confidence.sufficient_evidence:
    print("Not enough information to answer")

Production Example

from psycopg_pool import ConnectionPool
from memblock import MemBlockPool, AsyncMemBlockPool, BlockType

# Connection pooling + instance factory
pg_pool = ConnectionPool(
    "postgresql://user:pass@localhost/mydb",
    min_size=4, max_size=20
)

# Sync — each user gets a cached MemBlock instance
with MemBlockPool(
    "postgresql://user:pass@localhost/mydb",
    max_instances=100,
    pool=pg_pool,
) as pool:
    mem = pool.get("user_123")
    mem.store("Prefers dark mode", type=BlockType.PREFERENCE)

# Async — works with FastAPI / aiohttp
async with AsyncMemBlockPool(
    "postgresql://user:pass@localhost/mydb",
    max_instances=100,
    pool=pg_pool,
) as pool:
    mem = await pool.get("user_456")
    await mem.store("Works at Acme", type=BlockType.FACT)

# pgvector is automatic — just install the extra
# pip install "memblock[pgvector]"
# If pgvector extension exists in your DB, vector
# search uses server-side HNSW instead of Python cosine