CueMap Python SDK

High-performance temporal-associative memory store designed for dynamic contextual retrieval.

Overview

CueMap implements a Continuous Gradient Algorithm optimized for associative data structures:

1. Intersection (Context Filter): Triangulates relevant memories by overlapping cues
2. Local semantic reranking: Uses bundled qint8 MiniLM-L3 by default, or q4 MiniLM-L3 with the edge profile, for bounded semantic ranking inside the engine.
3. Recency & Salience (Signal Dynamics): Balances fresh data with salient, high-signal events prioritized by an adaptive impact scoring module.
4. Reinforcement (Access-based Learning): Frequently accessed memories gain signal strength, remaining highly accessible even as they age.
5. Deterministic Facets & Intent Routing: Extracts synchronous source, evidence, temporal, type, and entity facets, then uses sparse intent cues and reranking during recall.

As of v0.7.2, CueMap keeps deterministic lexical candidate discovery and adds bundled qint8 all-MiniLM-L3-v2 for bounded hybrid semantic and intent reranking. The edge engine profile uses a q4 build of the same model. No runtime model download is required, and callers can disable the encoder or provide their own vectors.

v0.7.2 also preserves numeric per-project memory IDs everywhere. If callers need deterministic upsert/dedupe identity, pass source_key; memory IDs remain compact runtime addresses.

Use this SDK to talk to the Rust engine from Python applications.

Installation

pip install cuemap

Quick Start

1. Start the Engine

docker run -p 8080:8080 cuemap/engine:latest

2. Basic Usage

from cuemap import CueMap

client = CueMap()

# Add a memory with deterministic cue extraction
client.add("The server password is abc123")

# Recall by natural language
results = client.recall("server credentials")
print(results[0].content)
# Output: "The server password is abc123"

Core API

Add Memory

# Manual cues
client.add(
    "Meeting with John at 3pm",
    cues=["meeting", "john", "calendar"]
)

# Deterministic cues are derived when cues are omitted
client.add("The payments service is down due to a timeout")

Recall Memories

# Natural language search
results = client.recall(
    "payments failure",
    limit=10,
    explain=True # See how the query was expanded
)

print(results[0].explain)
# Shows normalized cues, intent cues, and reranking details.

# Explicit Cue Search
results = client.recall(
    cues=["meeting", "john"],
    min_intersection=2
)

Grounded Recall (Hallucination Guardrails)

Get verifiable context for LLMs with a strict token budget.

response = client.recall_grounded(
    query="Why is the payment failing?",
    token_budget=500
)

print(response["verified_context"])
# [VERIFIED CONTEXT] ...
print(response["proof"])
# Cryptographic proof of context retrieval

v0.7.2 Recall Controls

CueMap v0.7.2 adds local semantic query signals alongside temporal query intent and optional reconstruction passes for longer conversational/codebase context.

results = client.recall(
    "what did we decide about auth retries?",
    query_time="2026-07-06",
    ordered_reconstruction="auto",
    evidence_coverage="auto",
    parent_fusion="auto",
    semantic_mode="hybrid",
    explain=True,
)

Use semantic_mode="lexical" for a semantic-reranker-disabled comparison, "semantic" for vector candidate discovery, or "hybrid" (the engine default) to rerank lexical candidates. query_embedding supplies a precomputed vector when the application owns the embedding provider.

Classify query or memory intent with the same local engine model. Returned scores are ranking signals, not calibrated probabilities:

classification = client.classify_intent(
    "What did we decide about auth retries?",
    target="query",
)
print(classification["primary_intent"], classification["recall_eligible"])

Cloud Backup (v0.6.1)

Manage project snapshots in the cloud (S3, GCS, Azure).

# Upload current project snapshot
client.backup_upload("default")

# Download and restore snapshot
client.backup_download("default")

# List available backups
backups = client.backup_list()

Ingestion (v0.6+)

Ingest content from various sources directly.

# Ingest URL
client.ingest_url("https://example.com/docs")

# Ingest File (PDF, DOCX, etc.)
client.ingest_file("/path/to/document.pdf")

# Ingest Raw Content with v0.7 logical-block chunking
client.ingest_content(
    "Raw text content...",
    filename="notes.md",
    source_key="docs:notes",
    structural_cues=["source_type:docs"],
    segmenter="logical_block",
)

When an application chunks content itself, pass exactly one vector per produced chunk with embeddings=[[...], [...]].

Preview and persist a repository ingestion scope:

preview = client.preview_directory("/work/my-app", included_paths=["src"])
client.set_project_watch_dir(
    "repo-my-app",
    "/work/my-app",
    included_paths=["src", "README.md"],
)
scope = client.get_project_watch_dir("repo-my-app")

Lexicon Management (v0.6+)

Inspect and wire the brain's associations manually.

# Inspect a cue's relationships
data = client.lexicon_inspect("service:payment")
print(f"Synonyms: {data['outgoing']}")
print(f"Triggers: {data['incoming']}")

# Manually wire a token to a concept
client.lexicon_wire("stripe", "service:payment")

Job Status (v0.6+)

Check the progress of background ingestion tasks.

status = client.jobs_status()
print(f"Ingested: {status['writes_completed']} / {status['writes_total']}")
print(f"Intent ready: {status.get('intent_ready', False)}")

Async Support

from cuemap import AsyncCueMap

async with AsyncCueMap() as client:
    await client.add("Note")
    await client.recall(cues=["note"])

License

MIT