VelesDB Python

Python bindings for VelesDB — a high-performance vector database for AI applications. numpy>=1.20 ships as a hard runtime dependency since v1.13.8, so a single pip install velesdb is sufficient.

The Python engine behind VelesDB — the explainable, local-first memory engine for AI agents. It fuses vector + graph + columnar under VelesQL; the why() wedge below returns the evidence path behind every recall.

Agent memory: the why() wedge

Beyond raw vector search, VelesDB ships a high-level MemoryService — local-first agent memory whose differentiator is why(): it answers a question with the best-matching memory plus the connected subgraph reachable through typed links — context that shares no words with the question, which a plain vector recall is blind to. The store is on disk, so it works across process restarts.

from velesdb import MemoryService            # offline, deterministic, no API key

mem = MemoryService("./agent_memory")        # a real on-disk store; survives restarts
reason = mem.remember("Robert is recovering from knee surgery")
mem.remember("Booked the aisle seat on Robert's flight", links=[(reason, "because")])

# A new process, weeks later, reopens the same store and asks why:
mem.why("why the aisle seat on Robert's flight?")   # walks booking → reason — recall() can't

Full runnable demo: examples/agent_memory/why_across_sessions.py. The same wedge ships for Node and as a local MCP server.

mem.feedback(id, success) closes the RL loop: reinforce or weaken a memory after use and recall re-ranks against the updated confidence.

Context Compiler: token-budgeted, provenance-audited prompt context

Your agent burns most of its tokens re-reading redundant context. MemoryService.compile_context compresses it deterministically (no LLM, no cloud, zero new logic in the binding — it delegates straight to the same velesdb_memory::context bridge the MCP server and the Node binding use): the same request always compiles to the same bytes, duplicates drop, repeated log lines collapse with counts, code / URLs / numbers / negative constraints survive verbatim (metadata={"verbatim": True} forces it), a stable cache-friendly prefix can be pinned (metadata={"cache": True}), and over-budget content becomes a recoverable ctx://source/<hash> handle — never a silent loss.

mem = MemoryService("./agent_memory")

compiled = mem.compile_context({
    "query": "deploy pipeline safety",
    "token_budget": 2000,
    "project": "veles",
    "fragments": [
        {"content": "The deploy pipeline runs clippy before tests."},
        {"content": "The deploy pipeline runs clippy before tests."},  # duplicate, dropped
        {"content": "Never restart the primary during a rebalance.",
         "metadata": {"verbatim": True}},
    ],
})
compiled["content"]    # the compiled prompt context (fits the budget)
compiled["risk"]       # "low" | "medium" | "high" -- "high" means critical content did not fit
compiled["decisions"]  # one auditable decision per fragment (rule_id, reason, risk)
compiled["insights"]   # {"tokens_in", "tokens_out", "tokens_saved", ...} -- local estimates

# What did not fit stays recoverable, byte for byte:
handle = compiled["sources"][0]["handle"]        # "ctx://source/18021940868160883968"
mem.retrieve_context_source(handle)              # -> {"content": str, "media"?: ...}

# Aggregate savings across every compile_context call (optionally per project):
mem.context_savings(project="veles")             # {"events", "tokens_saved", ...}

# Persist/reload an agent's distilled working state across sessions:
wid = mem.save_working_context("veles", "session-1", {
    "goal": "ship the release",
    "active_constraints": [{"text": "never restart during rebalance"}],
    "verified_facts": [], "open_hypotheses": [], "decisions": [],
    "exact_evidence": [], "pending_actions": ["run smoke tests"],
})
mem.load_working_context("veles", "session-1")   # -> the same dict, or None if never saved

The request/result JSON matches the MCP compile_context / context_savings tools field for field, with one documented difference: every u64 id (fragment_id, content_hash, memory_id, entries of fragment_ids, and input fragments[].id) crosses as a native Python int (unlimited precision) here, versus a decimal string on the Node binding — both are faithful renderings of the same value, never truncated (ids are FNV-1a 64-bit hashes, uniformly spread over the full u64 range, so roughly half of them exceed i64::MAX — see handle above). tokens_saved is a local estimate, not billed tokens.

from langchain_core.documents import Document
from langchain_core.runnables import RunnableLambda
from velesdb import MemoryService

mem = MemoryService("./agent_memory")


def compile_docs(inputs: dict) -> str:
    fragments = [{"content": d.page_content} for d in inputs["docs"]]
    compiled = mem.compile_context(
        {"query": inputs["query"], "token_budget": 2000, "fragments": fragments}
    )
    return compiled["content"]


compile_step = RunnableLambda(compile_docs)
docs = [
    Document(page_content="The deploy pipeline runs clippy before tests."),
    Document(page_content="The deploy pipeline runs clippy before tests."),
    Document(page_content="Never restart the primary during a rebalance."),
]
compile_step.invoke({"query": "deploy pipeline safety", "docs": docs})
# -> "The deploy pipeline runs clippy before tests.\n\nNever restart the primary during a rebalance."

from llama_index.core import Document
from velesdb import MemoryService

mem = MemoryService("./agent_memory")
docs = [
    Document(text="The deploy pipeline runs clippy before tests."),
    Document(text="The deploy pipeline runs clippy before tests."),
    Document(text="Never restart the primary during a rebalance."),
]
fragments = [{"content": d.text} for d in docs]
compiled = mem.compile_context(
    {"query": "deploy pipeline safety", "token_budget": 2000, "fragments": fragments}
)
compiled["content"]                    # same deduped, budgeted context
compiled["insights"]["tokens_saved"]   # 13

Features

• Dense + Sparse Vector Search: HNSW index with SIMD-optimized distance, plus inverted-index sparse search and hybrid dense+sparse fusion
• Multi-Query Fusion: Native MQG support with RRF, Weighted, and Relative Score fusion strategies
• Agent Memory SDK: SemanticMemory, EpisodicMemory, ProceduralMemory for AI agent workflows
• Graph Collections: Persistent knowledge graphs with BFS/DFS traversal and optional node embeddings
• In-Memory GraphStore: Lightweight graph for ad-hoc analysis without disk persistence
• VelesQL Parser: Programmatic query introspection with VelesQL.parse() and ParsedStatement
• Multiple Distance Metrics: Cosine, Euclidean, Dot Product, Hamming, Jaccard
• Persistent Storage: Memory-mapped files for efficient disk I/O
• Metadata Collections: Schema-free reference tables with no vector overhead
• Product Quantization: PQ and OPQ training for compressed storage and faster search
• Collection Analytics: analyze_collection() with row counts, size metrics, and index statistics
• NumPy Integration: Native support for NumPy arrays
• Type Hints: Full .pyi stub file for IDE autocompletion

Installation

pip install velesdb

Quick Start

import velesdb

# Open or create a database
db = velesdb.Database("./my_vectors")

# Create a collection for 768-dimensional vectors (e.g., BERT embeddings).
# `create_collection` is the recommended Python entry point. It accepts
# typed dataclasses (`hnsw=HnswOptions(...)`, `auto_reindex=AutoReindexOptions(...)`,
# `limits=LimitsOptions(...)`) for tuning. The Rust core also exposes
# `create_vector_collection` for callers who want the explicit typed API.
collection = db.create_collection(
    name="documents",
    dimension=768,
    metric="cosine"  # Options: "cosine", "euclidean", "dot" (aliases: "dotproduct", "inner", "ip"), "hamming", "jaccard"
)

# Insert vectors with metadata
collection.upsert([
    {
        "id": 1,
        "vector": [0.1, 0.2, ...],  # 768-dim vector
        "payload": {"title": "Introduction to AI", "category": "tech"}
    },
    {
        "id": 2,
        "vector": [0.3, 0.4, ...],
        "payload": {"title": "Machine Learning Basics", "category": "tech"}
    }
])

# Search for similar vectors
results = collection.search_request(
    velesdb.SearchOptions(vector=[0.15, 0.25, ...], top_k=5)  # Query vector
)

for result in results:
    print(f"ID: {result['id']}, Score: {result['score']:.4f}")
    print(f"Payload: {result['payload']}")

Search API. search_request(SearchOptions(...)) is the canonical search entry point and is used throughout this README. The older multi-keyword collection.search(vector=..., top_k=..., filter=...) still works but is deprecated since v1.15 (emits a DeprecationWarning, removed in v2.0). SearchOptions accepts vector, sparse_vector, top_k, filter, sparse_index_name and include_vectors, plus a fluent builder (SearchOptions().with_vector(v).with_top_k(10)).

Engine configuration (VelesConfigOptions)

Database(path, config=...) accepts a typed VelesConfigOptions covering every engine section of the core VelesConfig: limits, search, hnsw, storage and quantization. Build it in code or load it from a velesdb.toml (engine-only semantics: a shell-owned [server]/[logging] table in a shared file is ignored):

from velesdb import (
    Database, VelesConfigOptions, LimitsOptions, SearchConfigOptions,
)

# In code:
cfg = VelesConfigOptions(
    limits=LimitsOptions(max_collections=50),
    search=SearchConfigOptions(default_mode="accurate", max_results=100),
)
db = Database("./tenant1", config=cfg)

# Or from TOML (fail-fast: invalid TOML/values raise ValueError,
# a missing file raises FileNotFoundError):
cfg = VelesConfigOptions.from_toml_path("./velesdb.toml")
db = Database("./tenant1", config=cfg)

wal_batch is intentionally not exposed — the concurrent WAL writer is a VelesDB Enterprise feature (see docs/guides/WRITE_CONCURRENCY.md).

End-to-End: Text to Search Results (RAG Pipeline)

VelesDB stores and searches vectors — it does not generate embeddings. Use any embedding model to convert text to vectors first.

# pip install velesdb sentence-transformers
import velesdb
from sentence_transformers import SentenceTransformer

# 1. Load an embedding model (runs locally, no API key needed)
model = SentenceTransformer("all-MiniLM-L6-v2")  # outputs 384-dim vectors

# 2. Create a collection matching the model's dimension
db = velesdb.Database("./rag_data")
collection = db.create_collection("docs", dimension=384, metric="cosine")

# 3. Embed and store documents
texts = [
    "VelesDB is a local-first vector database written in Rust.",
    "HNSW is an approximate nearest neighbor search algorithm.",
    "RAG combines retrieval with language model generation.",
]
vectors = model.encode(texts).tolist()

collection.upsert([
    {"id": i, "vector": v, "payload": {"text": t}}
    for i, (v, t) in enumerate(zip(vectors, texts))
])

# 4. Search with a natural language query
query_vector = model.encode("How does vector search work?").tolist()
results = collection.search_request(velesdb.SearchOptions(vector=query_vector, top_k=2))

for r in results:
    print(f"Score: {r['score']:.4f} | {r['payload']['text']}")
# Score: 0.5621 | HNSW is an approximate nearest neighbor search algorithm.
# Score: 0.4238 | VelesDB is a local-first vector database written in Rust.

Built-in embedding adapters (optional)

Since v1.16.0, velesdb.embed ships thin, opt-in adapters so you don't have to wire the embedding call yourself. Their backing libraries are soft dependencies loaded lazily, so the base wheel stays light — install the extra you need:

# pip install "velesdb[embed-sentence-transformers]"   # local, no API key
# pip install "velesdb[embed-openai]"                   # OpenAI-compatible API
# pip install "velesdb[embed]"                          # both
import velesdb
from velesdb.embed import SentenceTransformerEmbedder  # or OpenAIEmbedder

embedder = SentenceTransformerEmbedder("all-MiniLM-L6-v2")  # runs on-device
db = velesdb.Database("./rag_data")
collection = db.create_collection("docs", dimension=embedder.dimension, metric="cosine")

texts = ["VelesDB is a local-first vector database.", "HNSW powers fast ANN search."]
vectors = embedder.embed(texts)
collection.upsert([{"id": i, "vector": v, "payload": {"text": t}}
                   for i, (v, t) in enumerate(zip(vectors, texts))])

results = collection.search_request(
    velesdb.SearchOptions(vector=embedder.embed(["fast search"])[0], top_k=2)
)

OpenAIEmbedder(model="text-embedding-3-small", *, api_key=..., base_url=..., dimensions=...) targets OpenAI, Azure OpenAI, vLLM, or any OpenAI-compatible endpoint via base_url. Both adapters satisfy the Embedder protocol (dimension: int, embed(texts) -> list[list[float]]), so you can drop in your own implementation. dimension is inferred after the first embed() call when not known up front.

Full RAG demo with PDF ingestion: demos/rag-pdf-demo/

API Reference

Database

# Create/open database
db = velesdb.Database("./path/to/data")

# List collections
names = db.list_collections()
names = db.get_collections()  # compatibility alias

# Create collection (with optional HNSW tuning via typed options)
collection = db.create_collection("name", dimension=768, metric="cosine")
from velesdb import HnswOptions
collection = db.create_collection(
    "tuned",
    dimension=768,
    hnsw=HnswOptions(m=48, ef_construction=600),
)

# Get existing collection
collection = db.get_collection("name")

# Delete collection
db.delete_collection("name")

# Create a metadata-only collection (no vectors, payload-only CRUD)
products = db.create_metadata_collection("products")

# Create a graph collection (see Graph Collections section)
graph = db.create_graph_collection("knowledge", dimension=768)

# Agent memory for AI workflows (see Agent Memory SDK section)
memory = db.agent_memory(dimension=384)

# Train Product Quantization for compressed search
db.train_pq("name", m=8, k=256)
db.train_pq("name", m=16, k=128, opq=True)  # Optimized PQ

# Analyze collection statistics
stats = db.analyze_collection("name")
print(stats["total_points"], stats["total_size_bytes"])

# Query plan cache management
cache_stats = db.plan_cache_stats()
print(f"Hit rate: {cache_stats['hit_rate']:.2%}")
db.clear_plan_cache()

Collection

# Get collection info
info = collection.info()
# {"name": "documents", "dimension": 768, "metric": "cosine", "storage_mode": "full", "point_count": 100, "metadata_only": False}

# Insert/update vectors (with immediate flush)
collection.upsert([
    {"id": 1, "vector": [...], "payload": {"key": "value"}}
])

# Bulk insert (optimized for high-throughput - 3-7x faster)
# Uses parallel HNSW insertion + single flush at the end
collection.upsert_bulk([
    {"id": i, "vector": vectors[i].tolist()} for i in range(10000)
])

# Vector search
results = collection.search_request(velesdb.SearchOptions(vector=[...], top_k=10))

# Search with custom HNSW ef_search (trade speed for recall)
results = collection.search_with_ef(vector=[...], top_k=10, ef_search=256)

# Search returning only IDs and scores (faster, no payload transfer)
results = collection.search_ids(vector=[...], top_k=10)
# [{"id": 1, "score": 0.98}, {"id": 2, "score": 0.95}, ...]

# Batch search (multiple queries in parallel)
batch_results = collection.batch_search([
    {"vector": [0.1, 0.2, ...], "top_k": 5},
    {"vector": [0.3, 0.4, ...], "top_k": 10, "filter": {"condition": ...}},
])

# Multi-query fusion search (MQG pipelines)
from velesdb import FusionStrategy

results = collection.multi_query_search(
    vectors=[query1, query2, query3],  # Multiple reformulations
    top_k=10,
    fusion=FusionStrategy.rrf(k=60)  # RRF, average, maximum, or weighted
)

# Weighted fusion (like SearchXP scoring)
results = collection.multi_query_search(
    vectors=[v1, v2, v3],
    top_k=10,
    fusion=FusionStrategy.weighted(
        avg_weight=0.6,
        max_weight=0.3,
        hit_weight=0.1
    )
)

# Relative Score Fusion (linear combination of dense + sparse scores)
results = collection.multi_query_search(
    vectors=[v1, v2],
    top_k=10,
    fusion=FusionStrategy.relative_score(dense_weight=0.7, sparse_weight=0.3)
)

# Maximum fusion (take the highest score across queries)
results = collection.multi_query_search(
    vectors=[v1, v2],
    top_k=10,
    fusion=FusionStrategy.maximum()
)

# Multi-query search returning only IDs and fused scores
results = collection.multi_query_search_ids(
    vectors=[v1, v2, v3],
    top_k=10,
    fusion=FusionStrategy.rrf()
)
# [{"id": 1, "score": 0.85}, ...]

# Hybrid dense + sparse search (fused with RRF k=60 by default)
results = collection.search_request(velesdb.SearchOptions(
    vector=[0.1, 0.2, ...],
    sparse_vector={0: 1.0, 42: 2.0},
    top_k=10,
))  # uses RRF by default; see Sparse Vector Search section for details

# Text search (BM25)
results = collection.text_search(query="machine learning", top_k=10)

# Hybrid search (vector + text with RRF fusion)
results = collection.hybrid_search(
    vector=[0.1, 0.2, ...],
    query="machine learning",
    top_k=10,
    vector_weight=0.7  # 0.0 = text only, 1.0 = vector only
)

# Get specific points
points = collection.get([1, 2, 3])

# Delete points
collection.delete([1, 2, 3])

# Check if empty
is_empty = collection.is_empty()

# Flush to disk
collection.flush()

# VelesQL query
results = collection.query(
    "SELECT * FROM vectors WHERE category = 'tech' LIMIT 10"
)

# VelesQL with parameters
results = collection.query(
    "SELECT * FROM vectors WHERE VECTOR NEAR $query LIMIT 5",
    params={"query": [0.1, 0.2, ...]}
)

# Search with metadata filter
results = collection.search_with_filter(
    vector=[0.1, 0.2, ...],
    top_k=10,
    filter={"condition": {"type": "eq", "field": "category", "value": "tech"}}
)

# Streaming insert (high-throughput, eventual consistency)
count = collection.stream_insert([
    {"id": 100, "vector": [...], "payload": {"key": "value"}}
])

# Scroll through all points in stable batches (no vector required)
# Useful for export, reindexing, or full-collection inspection.
for batch in collection.scroll(batch_size=100, filter=None):
    for point in batch:
        print(point["id"], point["payload"])

# MATCH graph traversal query (VelesQL)
results = collection.match_query(
    "MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a.name, b.name",
    vector=query_embedding,   # optional: add similarity scoring
    threshold=0.5             # minimum similarity threshold
)

# Query returning only IDs and scores (faster, no payload)
ids = collection.query_ids("SELECT * FROM docs WHERE price > 100 LIMIT 5")

# Explain query execution plan
plan = collection.explain("SELECT * FROM docs WHERE category = 'tech' LIMIT 10")
print(plan["tree"])            # execution plan tree
print(plan["estimated_cost_ms"])
print(plan["filter_strategy"]) # "seq_scan", "index_scan", etc.

# Index management
collection.create_property_index("Document", "category")  # O(1) equality lookup
collection.create_range_index("Document", "price")         # O(log n) range queries
indexes = collection.list_indexes()
collection.drop_index("Document", "category")

# Secondary indexes on a payload field (and their memory footprint)
collection.has_secondary_index("category")                 # -> bool
collection.drop_secondary_index("category")                # -> bool (True if dropped)
collection.indexes_memory_usage()                          # -> int (total bytes)

Sparse Vector Search

VelesDB supports sparse vectors alongside dense vectors. Sparse vectors are useful for learned sparse models (SPLADE, BGE-M3 sparse) and keyword-weighted representations.

# Upsert points with both dense and sparse vectors
collection.upsert([
    {
        "id": 1,
        "vector": [0.1, 0.2, 0.3, 0.4],       # dense embedding
        "sparse_vector": {0: 1.5, 3: 0.8, 42: 2.1},  # {dimension_index: weight}
        "payload": {"title": "Sparse retrieval paper"}
    },
    {
        "id": 2,
        "vector": [0.5, 0.6, 0.7, 0.8],
        "sparse_vector": {3: 1.2, 7: 0.5, 42: 0.9},
        "payload": {"title": "Dense retrieval survey"}
    }
])

# Sparse-only search (no dense vector needed)
results = collection.search_request(velesdb.SearchOptions(
    sparse_vector={0: 1.0, 42: 2.0},
    top_k=5,
))

# Hybrid dense + sparse search (fused with RRF k=60 by default)
results = collection.search_request(velesdb.SearchOptions(
    vector=[0.15, 0.25, 0.35, 0.45],
    sparse_vector={0: 1.0, 42: 2.0},
    top_k=10,
))

# Named sparse indexes (e.g., separate SPLADE and BM25 sparse models)
collection.upsert([
    {
        "id": 3,
        "vector": [0.2, 0.3, 0.4, 0.5],
        "sparse_vector": {
            "splade": {10: 1.5, 20: 0.8},
            "bm25":   {5: 2.0, 15: 1.1}
        },
        "payload": {"title": "Multi-model embeddings"}
    }
])

results = collection.search_request(velesdb.SearchOptions(
    vector=[0.2, 0.3, 0.4, 0.5],
    sparse_vector={10: 1.5, 20: 0.8},
    top_k=10,
    sparse_index_name="splade",  # query a specific named sparse index
))

Sparse vectors also work with scipy sparse objects:

from scipy.sparse import csr_matrix
import numpy as np

sparse_query = csr_matrix(np.array([[0.0, 1.5, 0.0, 0.8]]))
results = collection.search_request(velesdb.SearchOptions(sparse_vector=sparse_query, top_k=5))

Fusion Strategies

All multi-query and hybrid search methods accept a FusionStrategy to control how scores from multiple result sets are combined.

from velesdb import FusionStrategy

# Reciprocal Rank Fusion (default) -- robust to score scale differences
strategy = FusionStrategy.rrf(k=60)       # lower k = more weight to top ranks

# Average -- mean score across all queries
strategy = FusionStrategy.average()

# Maximum -- take the highest score per document
strategy = FusionStrategy.maximum()

# Weighted -- custom combination of avg, max, and hit ratio
strategy = FusionStrategy.weighted(avg_weight=0.6, max_weight=0.3, hit_weight=0.1)
strategy = FusionStrategy.weighted({"avg_weight": 0.6, "max_weight": 0.3, "hit_weight": 0.1})
strategy = FusionStrategy.weighted(0.3, 0.1)  # legacy: max_weight, hit_weight
strategy = FusionStrategy.weighted(max_weight=0.3, hit_weight=0.1)  # legacy kwargs

# Relative Score Fusion -- linear blend of dense and sparse scores
strategy = FusionStrategy.relative_score(dense_weight=0.7, sparse_weight=0.3)
strategy = FusionStrategy.rsf(dense_weight=0.7, sparse_weight=0.3)  # alias

Agent Memory SDK

VelesDB provides a built-in memory system for AI agents with three subsystems designed for RAG pipelines, chatbots, and autonomous agents.

import velesdb

db = velesdb.Database("./agent_data")
memory = db.agent_memory(dimension=384)  # default dimension is 384

Semantic Memory -- long-term knowledge facts with vector similarity recall:

# Store knowledge facts with their embeddings
memory.semantic.store(1, "Paris is the capital of France", embedding_paris)
memory.semantic.store(2, "Berlin is the capital of Germany", embedding_berlin)
memory.semantic.store(3, "The Eiffel Tower is in Paris", embedding_eiffel)

# Recall by similarity
results = memory.semantic.query(query_embedding, top_k=3)
for r in results:
    print(f"{r['content']} (score: {r['score']:.3f})")

Episodic Memory -- event timeline with temporal and similarity queries:

import time

# Record events as they happen
memory.episodic.record(1, "User asked about weather", timestamp=int(time.time()))
memory.episodic.record(
    2,
    "Agent retrieved forecast data",
    timestamp=int(time.time()),
    embedding=event_embedding  # optional, enables similarity recall
)

# Get recent events
events = memory.episodic.recent(limit=10)
for e in events:
    print(f"[{e['timestamp']}] {e['description']}")

# Get events since a specific timestamp
recent = memory.episodic.recent(limit=5, since=1700000000)

# Find similar past events by embedding
similar = memory.episodic.recall_similar(query_embedding, top_k=5)
for s in similar:
    print(f"{s['description']} (score: {s['score']:.3f})")

Procedural Memory -- learned patterns with confidence scoring and reinforcement:

# Teach a procedure
memory.procedural.learn(
    procedure_id=1,
    name="greet_user",
    steps=["say hello", "ask for name", "confirm preferences"],
    embedding=greeting_embedding,  # optional, enables similarity recall
    confidence=0.8
)

# Recall procedures by similarity (filtered by minimum confidence)
patterns = memory.procedural.recall(
    query_embedding,
    top_k=5,
    min_confidence=0.5
)
for p in patterns:
    print(f"{p['name']}: {p['steps']} (confidence: {p['confidence']:.2f})")

# Reinforce after success or failure (adjusts confidence +0.1 / -0.05)
memory.procedural.reinforce(procedure_id=1, success=True)
memory.procedural.reinforce(procedure_id=1, success=False)

Graph Collections

Graph collections store typed relationships between nodes with optional vector embeddings. They support persistent storage, BFS/DFS traversal, and node payload management.

import velesdb

db = velesdb.Database("./graph_data")

# Create a graph collection (schemaless by default)
graph = db.create_graph_collection("knowledge")

# With node embeddings for vector search over graph nodes
graph = db.create_graph_collection("kg", dimension=768, metric="cosine")

# With a strict schema (only predefined node/edge types)
from velesdb import GraphSchema
schema = GraphSchema.strict()
graph = db.create_graph_collection("typed_kg", schema=schema)

Adding edges and node data:

# Add edges (nodes are created implicitly by their IDs)
graph.add_edge({
    "id": 1, "source": 10, "target": 20,
    "label": "KNOWS",
    "properties": {"since": 2020, "context": "work"}
})
graph.add_edge({
    "id": 2, "source": 20, "target": 30,
    "label": "LIVES_IN",
    "properties": {"since": 2018}
})
graph.add_edge({
    "id": 3, "source": 10, "target": 30,
    "label": "LIVES_IN"
})

# Store properties on nodes
graph.upsert_node_payload(10, {"name": "Alice", "role": "engineer"})
graph.upsert_node_payload(20, {"name": "Bob", "role": "designer"})
graph.upsert_node_payload(30, {"name": "Paris", "type": "city"})

# Retrieve node properties
payload = graph.get_node_payload(10)
print(payload)  # {"name": "Alice", "role": "engineer"}

Querying the graph:

# Get all edges, or filter by label
all_edges = graph.get_edges()
knows_edges = graph.get_edges(label="KNOWS")

# Get outgoing/incoming edges for a node
outgoing = graph.get_outgoing(10)   # edges from Alice
incoming = graph.get_incoming(30)   # edges into Paris
outgoing = graph.get_outgoing_edges(10)  # alias
incoming = graph.get_incoming_edges(30)  # alias

# Node degree
in_deg, out_deg = graph.node_degree(10)

# List all nodes that have stored data
node_ids = graph.all_node_ids()

Graph traversal (BFS and DFS):

# BFS from Alice, max 3 hops, up to 100 results
results = graph.traverse_bfs(source_id=10, max_depth=3, limit=100)
for r in results:
    print(f"Reached node {r['target_id']} at depth {r['depth']}")

# Multi-source parallel BFS (starts from multiple nodes, deduplicates)
results = graph.traverse_bfs_parallel(
    source_ids=[10, 20, 30],
    max_depth=3,
    limit=100
)

# DFS with relationship type filter
results = graph.traverse_dfs(
    source_id=10,
    max_depth=2,
    rel_types=["KNOWS"]  # only follow KNOWS edges
)

# Vector search over graph nodes (requires dimension at creation)
results = graph.search_by_embedding(query_vector, k=10)
for r in results:
    print(f"Node {r['id']}: score {r['score']:.4f}")

VelesQL MATCH queries (Cypher-like graph pattern matching):

# Important: nodes must have _labels in their payload for label-based matching
graph.upsert_node_payload(10, {"_labels": ["Person"], "name": "Alice"})
graph.upsert_node_payload(20, {"_labels": ["Person"], "name": "Bob"})
graph.upsert_node_payload(30, {"_labels": ["City"], "name": "Paris"})

# MATCH query: find who Alice knows
results = graph.match_query(
    "MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a.name, b.name LIMIT 10"
)
for r in results:
    print(f"Node {r['node_id']} at depth {r['depth']}, projected: {r['projected']}")

# Hybrid: MATCH + vector similarity (requires embeddings)
results = graph.match_query(
    "MATCH (a:Person)-[:KNOWS]->(b) RETURN a, b LIMIT 10",
    vector=query_embedding,  # score each result by similarity
    threshold=0.5            # minimum similarity threshold
)

# VelesQL SELECT also works on GraphCollection
results = graph.query("SELECT * FROM kg WHERE name = 'Alice' LIMIT 5")

# Explain MATCH execution plan
plan = graph.explain("MATCH (a)-[:KNOWS]->(b) RETURN a, b LIMIT 10")
print(plan["tree"])

Persistence:

graph.flush()         # persist all changes to disk
print(graph.edge_count())  # total edges in the graph

VelesQL Parser API

VelesDB (since v1.7.2) exposes the VelesQL parser as a standalone Python API for query introspection, validation, and tooling integration. Parse any VelesQL statement into a ParsedStatement object and inspect its structure without executing it.

from velesdb import VelesQL

# Parse a query and inspect its structure
parsed = VelesQL.parse("SELECT id, title FROM documents WHERE category = 'tech' ORDER BY date DESC LIMIT 20")

print(parsed.collection_name)  # "documents"
print(parsed.columns)          # ["id", "title"]
print(parsed.limit)            # 20
print(parsed.offset)           # None
print(parsed.has_where_clause())   # True
print(parsed.has_order_by())       # True
print(parsed.has_vector_search())  # False
print(parsed.order_by)            # [("date", "DESC")]
print(parsed.is_select())         # True
print(parsed.is_match())          # False

Validate queries without parsing:

# Fast validation (no full parse tree)
VelesQL.is_valid("SELECT * FROM docs LIMIT 10")     # True
VelesQL.is_valid("SELEC * FROM docs")                # False

Inspect advanced query features:

# Vector search detection
parsed = VelesQL.parse("SELECT * FROM docs WHERE vector NEAR $q LIMIT 5")
print(parsed.has_vector_search())  # True

# MATCH (graph) queries
parsed = VelesQL.parse("MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a.name")
print(parsed.is_match())    # True
print(parsed.is_select())   # False

# GROUP BY, HAVING, JOINs, DISTINCT
parsed = VelesQL.parse("SELECT DISTINCT category, COUNT(*) FROM products GROUP BY category")
print(parsed.has_distinct())   # True
print(parsed.has_group_by())   # True
print(parsed.group_by)         # ["category"]

# JOIN inspection
parsed = VelesQL.parse(
    "SELECT * FROM orders JOIN products ON orders.product_id = products.id"
)
print(parsed.has_joins())   # True
print(parsed.join_count)    # 1

Error handling with typed exceptions:

from velesdb import VelesQL, VelesQLSyntaxError

try:
    parsed = VelesQL.parse("SELEC * FROM docs")
except VelesQLSyntaxError as e:
    print(f"Syntax error: {e}")

Key parameters for ParsedStatement:

In-Memory GraphStore

For ad-hoc graph analysis that does not require disk persistence, use the in-memory GraphStore. It supports the same edge operations and traversal algorithms as persistent GraphCollection but runs entirely in memory.

from velesdb import GraphStore, StreamingConfig

# Create an in-memory graph
store = GraphStore()

# Add edges
store.add_edge({"id": 1, "source": 100, "target": 200, "label": "KNOWS"})
store.add_edge({"id": 2, "source": 200, "target": 300, "label": "KNOWS"})
store.add_edge({"id": 3, "source": 100, "target": 300, "label": "FOLLOWS"})

# Query edges by label (O(1) index lookup)
knows_edges = store.get_edges_by_label("KNOWS")

# Outgoing / incoming edges
outgoing = store.get_outgoing(100)
incoming = store.get_incoming(300)

# Filtered outgoing by label
friends_of_100 = store.get_outgoing_by_label(100, "KNOWS")

# Node degree
print(store.out_degree(100))  # 2
print(store.in_degree(300))   # 2

# Check edge existence
print(store.has_edge(1))      # True
store.remove_edge(1)
print(store.has_edge(1))      # False

# Total edges
print(store.edge_count())     # 2

BFS streaming traversal:

# Configure traversal bounds
config = StreamingConfig(
    max_depth=3,              # maximum hops from start node
    max_visited=10000,        # memory bound: max nodes to visit
    relationship_types=["KNOWS"]  # optional: filter by edge label
)

# Traverse the graph from node 100
results = store.traverse_bfs_streaming(100, config)
for r in results:
    print(f"Depth {r.depth}: {r.source} --[{r.label}]--> {r.target} (edge {r.edge_id})")

DFS traversal:

config = StreamingConfig(max_depth=2, max_visited=500)
results = store.traverse_dfs(100, config)
for r in results:
    print(f"Depth {r.depth}: {r.source} -> {r.target}")

TraversalResult attributes:

Storage Modes (Quantization)

Reduce memory usage with vector quantization:

# Full precision (default) - 4 bytes per dimension
collection = db.create_collection("full", dimension=768, storage_mode="full")

# SQ8 quantization - 1 byte per dimension (4x compression)
collection = db.create_collection("sq8", dimension=768, storage_mode="sq8")

# Binary quantization - 1 bit per dimension (32x compression)
collection = db.create_collection("binary", dimension=768, storage_mode="binary")

# Product quantization - 8-32x compression, best for large-scale datasets
collection = db.create_collection("pq", dimension=768, storage_mode="pq")

# RaBitQ - 32x compression with scalar correction, best for high-compression with good recall
collection = db.create_collection("rabitq", dimension=768, storage_mode="rabitq")

Canonical names and aliases are interchangeable: storage_mode="f32" is equivalent to storage_mode="full".

Bulk Loading Performance

For large-scale data import, use upsert_bulk() instead of upsert():

# Recommended for bulk import
import numpy as np

vectors = np.random.rand(10000, 768).astype('float32')
points = [{"id": i, "vector": v.tolist()} for i, v in enumerate(vectors)]

collection.upsert_bulk(points)  # Batch-optimized: parallel HNSW + single flush

Distance Metrics

Performance

VelesDB is built in Rust with explicit SIMD optimizations:

System Benchmarks (Native Rust Engine)

Numbers match docs/reference/promise-contract.json (the single source of truth for the README perf claims).

Measured with Criterion.rs on i9-14900KF. See benchmarks/ for methodology.

Connecting to velesdb-server

The velesdb Python package provides embedded (in-process) access to VelesDB. To connect to a remote velesdb-server instance (with optional API key authentication), use standard HTTP requests:

import requests

API_URL = "http://localhost:8080"
API_KEY = "my-secret-key"  # Only needed when server has auth enabled

headers = {"Authorization": f"Bearer {API_KEY}"}

# Search for similar vectors
response = requests.post(
    f"{API_URL}/collections/documents/search",
    json={"vector": [0.1, 0.2, ...], "top_k": 5},
    headers=headers,
)
results = response.json()

When the server has TLS enabled, use https:// and optionally pass verify=False for self-signed certificates.

See SERVER_SECURITY.md for server authentication and TLS setup.

Requirements

• Python 3.9+
• NumPy >= 1.20 — a hard runtime dependency, installed automatically by pip install velesdb
• No build toolchain required: prebuilt wheels bundle a self-contained Rust engine

License

Licensed under the VelesDB Core License 1.0 (source-available). The compiled wheel embeds the VelesDB engine and is governed by the Core License.

See LICENSE for details.

Links

• GitHub Repository
• Documentation
• Issue Tracker