Append-only, temporal, provenance-aware triple assertions with deterministic structured queries. SQLite and LanceDB persistence with optional vector and semantic retrieval — the knowledge backbone for durable AI agents.
from abstractmemory import InMemoryTripleStore, TripleAssertion, TripleQuery
store = InMemoryTripleStore()
store.add([TripleAssertion(
subject="Scrooge", predicate="related_to",
object="Christmas", scope="session"
)])
hits = store.query(TripleQuery(subject="scrooge"))AbstractMemory stores knowledge as append-only semantic triples with temporal validity, provenance metadata, and confidence scoring. Every assertion records when it was observed, who asserted it, and under what context.
Knowledge is modeled as TripleAssertion records: subject, predicate, object triples with temporal metadata. Assertions are never mutated — new facts supersede old ones through temporal windows.
Every triple carries a provenance dict linking back to the span, artifact, or extraction pipeline that produced it. Full chain-of-evidence from source text to stored fact.
Optional valid_from and valid_until windows let you query for what was true at any point in time. The active_at query filter makes temporal reasoning natural.
AbstractMemory has no runtime dependency on AbstractCore or AbstractRuntime. For embeddings, it calls an AbstractGateway endpoint via a thin HTTP adapter.
Multiple storage backends, deterministic structured queries, and optional vector search — all through a unified TripleStore protocol.
Zero-dependency, process-local store with optional embedder integration. Ideal for tests, prototypes, and short-lived sessions where persistence is not needed.
Persistent SQLite-backed store using only the Python standard library. Deterministic structured queries with automatic schema creation and indexed columns.
Persistent LanceDB-backed store with native cosine-similarity vector search. Attach retrieval metadata to results and store canonical text plus embeddings.
Filter by subject, predicate, object, scope, owner, and temporal range. Deterministic ordering by observed_at with configurable ascending/descending sort.
Semantic retrieval via query_text or query_vector on vector-capable stores. Cosine similarity with configurable min_score thresholds and retrieval metadata.
Subject, predicate, and object terms are automatically canonicalized (trim + lowercase) for consistent matching regardless of input casing.
Optional confidence float per assertion enables downstream filtering and ranking. Combine with provenance for evidence-weighted knowledge graphs.
The AbstractGatewayTextEmbedder adapter calls an AbstractGateway embeddings endpoint via HTTP, keeping heavy model loading on the server side.
AbstractMemory requires Python 3.10+. LanceDB is an optional extra for persistent vector search.
# From PyPI
pip install AbstractMemory
# With LanceDB vector search
pip install "AbstractMemory[lancedb]"
# From source (editable)
pip install -e ".[dev]"from abstractmemory import InMemoryTripleStore, TripleAssertion, TripleQuery
store = InMemoryTripleStore()
store.add([
TripleAssertion(
subject="Scrooge",
predicate="related_to",
object="Christmas",
scope="session",
owner_id="sess-1",
observed_at="2026-01-01T00:00:00+00:00",
provenance={"span_id": "span_123"},
)
])
hits = store.query(TripleQuery(
subject="scrooge", # canonicalized (case-insensitive)
scope="session",
owner_id="sess-1",
limit=10
))
assert hits[0].object == "christmas"from abstractmemory import (
AbstractGatewayTextEmbedder,
LanceDBTripleStore,
TripleAssertion,
TripleQuery,
)
embedder = AbstractGatewayTextEmbedder(
base_url="http://localhost:8000",
auth_token=os.getenv("ABSTRACTGATEWAY_AUTH_TOKEN"),
)
store = LanceDBTripleStore("data/kg", embedder=embedder)
store.add([TripleAssertion(
subject="e:scrooge", predicate="is_a",
object="person", scope="global",
)])
hits = store.query(TripleQuery(query_text="scrooge", limit=5))All stores implement the TripleStore protocol: add(), query(), and close(). The data model is built around TripleAssertion and TripleQuery.
# Core triple
subject: str # Entity or value (canonicalized)
predicate: str # Relationship type (canonicalized)
object: str # Target entity or value (canonicalized)
# Scoping
scope: str # "run" | "session" | "global"
owner_id: str # Optional scope owner
# Temporal
observed_at: str # When observed (RFC-3339)
valid_from: str # Validity window start (optional)
valid_until: str # Validity window end (optional, exclusive)
# Metadata
confidence: float # 0.0 to 1.0 (optional)
provenance: dict # Span/artifact pointers
attributes: dict # Extractor evidence, retrieval metadata# Exact match (after canonicalization)
subject, predicate, object
scope, owner_id
# Temporal range
since # observed_at >= since
until # observed_at <= until
active_at # valid_from <= active_at < valid_until
# Semantic / vector
query_text # Requires embedder (no keyword fallback)
query_vector # Bypass embedding generation
min_score # Cosine similarity threshold
# Result shaping
order # "asc" | "desc" by observed_at
limit # <= 0 means unbounded# TextEmbedder protocol
class TextEmbedder:
def embed_texts(texts: list[str]) -> list[list[float]]: ...
# Gateway adapter (HTTP)
AbstractGatewayTextEmbedder(
base_url="http://localhost:8000",
auth_token="...",
endpoint_path="/api/gateway/embeddings", # default
)