Python API

Python bindings for LatticeDB, an embedded knowledge graph database for AI/RAG applications.

Installation

pip install latticedb

The native shared library (liblattice.dylib / liblattice.so) must be available on the system. Install it via the install script or build from source with zig build shared.

Quick Start

import numpy as np
from latticedb import Database

with Database("knowledge.db", create=True, enable_vectors=True, vector_dimensions=4) as db:
    # Create nodes, edges, and index content
    db.create_node_fts_index("Person", "bio")

    with db.write() as txn:
        alice = txn.create_node(
            labels=["Person"],
            properties={"name": "Alice", "age": 30},
        )
        bob = txn.create_node(
            labels=["Person"],
            properties={"name": "Bob", "age": 25},
        )
        txn.create_edge(alice.id, bob.id, "KNOWS")

        # Index text for full-text search
        txn.set_property(alice.id, "bio", "Alice works on machine learning research")
        txn.set_property(bob.id, "bio", "Bob studies deep learning and neural networks")

        # Store vector embeddings
        txn.set_vector(alice.id, "embedding", np.array([1.0, 0.0, 0.0, 0.0], dtype=np.float32))
        txn.set_vector(bob.id, "embedding", np.array([0.0, 1.0, 0.0, 0.0], dtype=np.float32))

        txn.commit()

    # Query with Cypher
    result = db.query("MATCH (n:Person) WHERE n.age > 20 RETURN n.name, n.age")
    for row in result:
        print(row)

    # Vector similarity search
    query_vec = np.array([0.9, 0.1, 0.0, 0.0], dtype=np.float32)
    for r in db.vector_search(query_vec, k=2):
        print(f"Node {r.node_id}: distance={r.distance:.4f}")

    # Full-text search
    for r in db.fts_search("Person", "bio", "machine learning"):
        print(f"Node {r.node_id}: score={r.score:.4f}")

    # Fuzzy search (typo-tolerant)
    for r in db.fts_search_fuzzy("Person", "bio", "machin lerning"):
        print(f"Node {r.node_id}: score={r.score:.4f}")

API Reference

Database

Database(
    path: str | Path,
    *,
    create: bool = False,        # Create if doesn't exist
    read_only: bool = False,     # Open in read-only mode
    cache_size_mb: int = 100,    # Page cache size
    enable_vectors: bool | None = None, # Preferred vector config flag
    enable_vector: bool | None = None,  # Deprecated compatibility alias
    vector_dimensions: int = 128,# Vector dimensions
    lock: bool = True            # Take the file lock (see below)
)

db = latticedb.Database(":memory:")

In-memory databases

Pass :memory: as the path and the database has no files behind it. Nothing is written to disk and nothing survives closing it, which suits a scratch database, a test, or one you pulled out of object storage and will hand back as bytes.

It behaves like any other database — transactions, the write-ahead log, and serialization all work. The differences are that it disappears when closed, and that nothing locks it, since no other process can reach it.

A database can only be open in one process at a time. Opening takes a lock on the file, so a second process gets LatticeDatabaseLockedError rather than quietly corrupting your data. A read-only handle shares the lock with other readers, but is still refused while a writer holds the database, because what it would read is a stale file that a checkpoint may be rewriting underneath it.

Pass lock=False only on filesystems where locking does not work. It does not make concurrent access safe; it removes the thing that was going to tell you it was not.

Methods

  • open() / close() - Open/close the database (also works as context manager)
  • read() - Start a read-only transaction (context manager)
  • write() - Start a read-write transaction (context manager)
  • query(cypher, parameters=None) - Execute a Cypher query
  • serialize() - Return the whole database as bytes
  • vector_search(vector, k=10, ef_search=64) - k-NN vector search
  • fts_search(label, prop, query, limit=10) - Full-text search of one declared index
  • fts_search_fuzzy(label, prop, query, limit=10, max_distance=0, min_term_length=0) - Fuzzy full-text search
  • create_node_fts_index(label, property) / drop_node_fts_index(...) / has_node_fts_index(...) - Manage node full-text indexes
  • create_edge_fts_index(edge_type, property) / drop_edge_fts_index(...) / has_edge_fts_index(...) - Manage relationship full-text indexes
  • create_node_property_index(label, property_key) / drop_node_property_index(...) - Manage explicit node equality indexes
  • create_edge_property_index(edge_type, property_key) / drop_edge_property_index(...) - Manage explicit edge equality indexes
  • read_stream(stream, after_sequence=0, limit=100, timeout_ms=0) - Read durable stream records by cursor
  • get_stream_offset(stream, consumer) - Read a committed consumer offset
  • changes(after_sequence=0, limit=100, timeout_ms=0) - Read the built-in graph changefeed
  • cache_clear() - Clear the query cache
  • cache_stats() - Get cache hit/miss statistics

Transaction

Read Operations

  • get_node(node_id) - Get a node by ID, returns Node or None
  • node_exists(node_id) - Check if a node exists
  • get_property(node_id, key) - Get a property value
  • get_outgoing_edges(node_id) - Get outgoing edges from a node
  • get_incoming_edges(node_id) - Get incoming edges to a node
  • find_nodes_by_label_property(label, property_key, value, limit=100) - Indexed node equality lookup
  • find_edges_by_type_property(edge_type, property_key, value, limit=100) - Indexed edge equality lookup
  • is_read_only / is_active - Transaction state

Write Operations

  • create_node(labels=[], properties=None) - Create a node
  • delete_node(node_id) - Delete a node
  • set_property(node_id, key, value) - Set a property on a node
  • set_vector(node_id, key, vector) - Set a vector embedding
  • batch_insert_vectors(label, vectors) - Insert vector-bearing nodes in one call
  • batch_insert(label, vectors) - Deprecated compatibility alias for batch_insert_vectors
  • db.create_node_fts_index(label, property) - Declare a full-text index; writing that property keeps it current
  • create_edge(source_id, target_id, edge_type, properties=None) - Create an edge and return its stable edge ID on the Edge
  • delete_edge(source_id, target_id, edge_type) - Delete an edge
  • set_edge_property(edge_id, key, value) - Set an edge property by stable edge ID
  • get_edge_property(edge_id, key) - Get an edge property by stable edge ID
  • remove_edge_property(edge_id, key) - Remove an edge property by stable edge ID
  • publish_stream(stream, payload, kind="message") - Publish a durable stream record
  • set_stream_offset(stream, consumer, sequence) - Commit a durable consumer offset
  • trim_stream(stream, through_sequence) - Delete stream records through a sequence
  • commit() / rollback() - Commit or rollback the transaction

Bulk Vector Insertion

Insert many nodes with vectors in a single efficient call:

import numpy as np

with Database("vectors.db", create=True, enable_vectors=True, vector_dimensions=128) as db:
    with db.write() as txn:
        vectors = np.random.rand(1000, 128).astype(np.float32)
        node_ids = txn.batch_insert_vectors("Document", vectors)
        print(f"Created {len(node_ids)} nodes")
        txn.commit()

Full-Text Search

results = db.fts_search("Person", "bio", "machine learning", limit=10)
for r in results:
    print(f"Node {r.node_id}: score={r.score:.4f}")

Fuzzy Search (Typo-Tolerant)

# Finds "machine learning" even with typos
results = db.fts_search_fuzzy("Person", "bio", "machne lerning", limit=10)

# Control fuzzy matching sensitivity
results = db.fts_search_fuzzy(
    "machne",
    limit=10,
    max_distance=2,      # Max edit distance (default: 2)
    min_term_length=4,   # Min term length for fuzzy matching (default: 4)
)

Embeddings

LatticeDB includes a built-in hash embedding function and an HTTP client for external embedding services.

Hash Embeddings (Built-in)

Deterministic, no external service needed. Useful for testing or simple keyword-based similarity:

from latticedb.embedding import hash_embed

vec = hash_embed("hello world", dimensions=128)
print(vec.shape)  # (128,)

HTTP Embedding Client

Connect to Ollama, OpenAI, or compatible APIs:

from latticedb.embedding import EmbeddingClient, EmbeddingApiFormat

# Ollama (default)
with EmbeddingClient("http://localhost:11434") as client:
    vec = client.embed("hello world")

# OpenAI-compatible API
with EmbeddingClient(
    "https://api.openai.com/v1",
    model="text-embedding-3-small",
    api_format=EmbeddingApiFormat.OPENAI,
    api_key="sk-...",
) as client:
    vec = client.embed("hello world")

Edge Traversal

with db.read() as txn:
    outgoing = txn.get_outgoing_edges(node_id)
    for edge in outgoing:
        print(f"{edge.id}: {edge.source_id} --[{edge.edge_type}]--> {edge.target_id}")

    incoming = txn.get_incoming_edges(node_id)
    for edge in incoming:
        print(f"{edge.id}: {edge.source_id} --[{edge.edge_type}]--> {edge.target_id}")

Edge Properties

Edge properties are addressed by stable edge ID. create_edge returns an Edge object containing that ID, and traversal results include it.

with db.write() as txn:
    edge = txn.create_edge(alice.id, bob.id, "KNOWS")
    txn.set_edge_property(edge.id, "since", 2020)
    assert txn.get_edge_property(edge.id, "since") == 2020
    txn.commit()

Durable Streams

with db.write() as txn:
    txn.publish_stream("jobs", {"id": 1, "status": "queued"}, kind="job.queued")
    txn.set_stream_offset("jobs", "worker-a", 1)
    txn.commit()

records = db.read_stream("jobs", after_sequence=0, limit=100)
changes = db.changes(after_sequence=0, limit=100)

Cypher Queries

# Pattern matching
result = db.query("MATCH (n:Person) RETURN n.name")

# With parameters
result = db.query(
    "MATCH (n:Person) WHERE n.name = $name RETURN n",
    parameters={"name": "Alice"},
)

# Vector similarity in Cypher
result = db.query(
    "MATCH (n:Document) WHERE n.embedding <=> $vec < 0.5 RETURN n.title",
    parameters={"vec": query_vector},
)

# Full-text search in Cypher
result = db.query(
    'MATCH (n:Document) WHERE n.content @@ "machine learning" RETURN n.title'
)

# Data mutation
db.query("CREATE (n:Person {name: 'Charlie', age: 35})")
db.query("MATCH (n:Person {name: 'Charlie'}) SET n.age = 36")
db.query("MATCH (n:Person {name: 'Charlie'}) DETACH DELETE n")

Query Cache

# Get cache statistics
stats = db.cache_stats()
print(f"Entries: {stats['entries']}, Hits: {stats['hits']}, Misses: {stats['misses']}")

# Clear the cache
db.cache_clear()

Supported Property Types

  • None - Null value
  • bool - Boolean
  • int - 64-bit integer
  • float - 64-bit float
  • str - UTF-8 string
  • bytes - Binary data

Error Handling

from latticedb import LatticeError, LatticeNotFoundError, LatticeIOError

try:
    with Database("nonexistent.db") as db:
        pass
except LatticeNotFoundError:
    print("Database not found")
except LatticeIOError:
    print("I/O error")
except LatticeError as e:
    print(f"Error: {e}")

Requirements

  • Python 3.9+
  • NumPy (for vector operations)
  • The native LatticeDB library (liblattice.dylib / liblattice.so)