Getting started with Python
The open-source SVS library supports all documented features except our proprietary vector compression (LVQ and Leanvec), which are exclusive to Intel hardware and available via our shared library and PyPI package. This tutorial shows how to use our vector compression and unlock significant performance and memory gains!
We also provide an example using the open-source SVS library only.
Installation
Installing SVS is simple. We test on Ubuntu 22.04 LTS, but it should work on any Linux distribution or macOS.
Prerequisites
Python >= 3.9
Installing
The SVS Python package is available as a Python wheel
pip install scalable-vs
or as a Conda package
conda install -c https://software.repos.intel.com/python/conda/ scalable-vs
Note that the Python package name is scalable-vs while
the Python import name is svs.
Run this command to confirm SVS is installed correctly, it should print the library version.
python3 -c "import svs; print(svs.library_version())"
Step by step example using vector compression
Here is a step by step explanation of the example that showcases the most important features of SVS.
We will use a random dataset generated using svs.generate_test_dataset().
We first load svs and the os module also required for this example.
import os
import svs
Then proceed to generate the test dataset.
# Create a test dataset.
# This will create a directory "example_data_vamana" and populate it with three
# entries:
# - data.fvecs: The test dataset.
# - queries.fvecs: The test queries.
# - groundtruth.fvecs: The groundtruth.
test_data_dir = "./example_data_vamana"
svs.generate_test_dataset(
1000, # Create 1000 vectors in the dataset.
100, # Generate 100 query vectors.
256, # Set the vector dimensionality to 256.
test_data_dir, # The directory where results will be generated.
data_seed = 1234, # Random number seed for reproducibility.
query_seed = 5678, # Random number seed for reproducibility.
num_threads = 4, # Number of threads to use.
distance = svs.DistanceType.L2, # The distance type to use.
)
Compress the data
To boost performance and reduce memory usage, we first compress the data using our vector compression technique LeanVec. See How to Choose Compression Parameters for details.
# We are going to construct a LeanVec dataset on-the-fly from uncompressed data.
# First, we construct a loader for the uncompressed data.
uncompressed_loader = svs.VectorDataLoader(
os.path.join(test_data_dir, "data.fvecs"),
svs.DataType.float32
)
# Next - we construct a LeanVecLoader.
# This loader is configured to perform the following:
# - Reduce dimensionality of the primary dataset to 128 dimensions.
# - Use LVQ4 for the primary dataset.
# - Use LVQ8 for the secondary, unreduced dataset.
leanvec_loader = svs.LeanVecLoader(
uncompressed_loader,
128, # The reduced number of dimensions.
primary_kind = svs.LeanVecKind.lvq4, # The encoding of the primary dataset.
secondary_kind = svs.LeanVecKind.lvq8, # The encoding of the secondary dataset.
)
Building the index
To search effectively, first build a graph-based index linking related data vectors. We’ll keep defaults for hyperparameters,
except for graph_max_degree and build window_size. Other values can be tuned later based on the dataset.
See this example for dynamic indexing (adding and removing vectors over time).
parameters = svs.VamanaBuildParameters(
graph_max_degree = 64,
window_size = 128,
)
index = svs.Vamana.build(
parameters,
leanvec_loader,
svs.DistanceType.L2,
num_threads = 4,
)
Searching the index
The graph is built; we can now query it. Load queries from disk and set search_window_size –larger values boost accuracy but reduce speed (see How to Set the Search Window Size).
n_neighbors = 10
index.search_window_size = 50
index.num_threads = 4
queries = svs.read_vecs(os.path.join(test_data_dir, "queries.fvecs"))
I, D = index.search(queries, n_neighbors)
After searching, we compare the results with the ground-truth and print the obtained recall.
groundtruth = svs.read_vecs(os.path.join(test_data_dir, "groundtruth.ivecs"))
recall = svs.k_recall_at(groundtruth, I, n_neighbors, n_neighbors)
print(f"Recall = {recall}")
Saving and loading the index
If you are satisfied with the performance of the generated index, you can save it to disk to avoid rebuilding it in the future.
index.save(
os.path.join(test_data_dir, "example_config"),
os.path.join(test_data_dir, "example_graph"),
os.path.join(test_data_dir, "example_data"),
)
index = svs.Vamana(
os.path.join(test_data_dir, "example_config"),
os.path.join(test_data_dir, "example_graph"),
os.path.join(test_data_dir, "example_data"),
svs.DistanceType.L2,
num_threads = 4,
)
See svs.Vamana.save() for details about the save function.
Note
The save index function currently uses three folders for saving. All three are needed to be able to reload the index.
One folder for the graph.
One folder for the data.
One folder for metadata.
This is subject to change in the future.
Entire example
We run automated tests to ensure the script stays valid. You can ignore all test-related code.
# Copyright (C) 2025 Intel Corporation
#
# This software and the related documents are Intel copyrighted materials,
# and your use of them is governed by the express license under which they
# were provided to you ("License"). Unless the License provides otherwise,
# you may not use, modify, copy, publish, distribute, disclose or transmit
# this software or the related documents without Intel's prior written
# permission.
#
# This software and the related documents are provided as is, with no
# express or implied warranties, other than those that are expressly stated
# in the License.
# Import `unittest` to allow for automated testing.
import unittest
# [imports]
import os
import svs
# [imports]
DEBUG_MODE = False
def assert_equal(lhs, rhs, message: str = "", epsilon = 0.05):
if DEBUG_MODE:
print(f"{message}: {lhs} == {rhs}")
else:
assert lhs < rhs + epsilon, message
assert lhs > rhs - epsilon, message
test_data_dir = None
def run():
# [generate-dataset]
# Create a test dataset.
# This will create a directory "example_data_vamana" and populate it with three
# entries:
# - data.fvecs: The test dataset.
# - queries.fvecs: The test queries.
# - groundtruth.fvecs: The groundtruth.
test_data_dir = "./example_data_vamana"
svs.generate_test_dataset(
1000, # Create 1000 vectors in the dataset.
100, # Generate 100 query vectors.
256, # Set the vector dimensionality to 256.
test_data_dir, # The directory where results will be generated.
data_seed = 1234, # Random number seed for reproducibility.
query_seed = 5678, # Random number seed for reproducibility.
num_threads = 4, # Number of threads to use.
distance = svs.DistanceType.L2, # The distance type to use.
)
# [generate-dataset]
# [create-loader]
# We are going to construct a LeanVec dataset on-the-fly from uncompressed data.
# First, we construct a loader for the uncompressed data.
uncompressed_loader = svs.VectorDataLoader(
os.path.join(test_data_dir, "data.fvecs"),
svs.DataType.float32
)
# Next - we construct a LeanVecLoader.
# This loader is configured to perform the following:
# - Reduce dimensionality of the primary dataset to 128 dimensions.
# - Use LVQ4 for the primary dataset.
# - Use LVQ8 for the secondary, unreduced dataset.
leanvec_loader = svs.LeanVecLoader(
uncompressed_loader,
128, # The reduced number of dimensions.
primary_kind = svs.LeanVecKind.lvq4, # The encoding of the primary dataset.
secondary_kind = svs.LeanVecKind.lvq8, # The encoding of the secondary dataset.
)
# [create-loader]
# An index can be constructed using a LeanVec dataset.
# [build-parameters]
parameters = svs.VamanaBuildParameters(
graph_max_degree = 64,
window_size = 128,
)
# [build-parameters]
# [build-index]
index = svs.Vamana.build(
parameters,
leanvec_loader,
svs.DistanceType.L2,
num_threads = 4,
)
# [build-index]
# Set the search window size of the index and perform queries and load the queries.
# [perform-queries]
n_neighbors = 10
index.search_window_size = 50
index.num_threads = 4
queries = svs.read_vecs(os.path.join(test_data_dir, "queries.fvecs"))
I, D = index.search(queries, n_neighbors)
# [perform-queries]
# Compare with the groundtruth.
# [recall]
groundtruth = svs.read_vecs(os.path.join(test_data_dir, "groundtruth.ivecs"))
recall = svs.k_recall_at(groundtruth, I, n_neighbors, n_neighbors)
print(f"Recall = {recall}")
# [recall]
assert_equal(recall, 0.856)
# Finally, we can save the index and reload from a previously saved set of files.
# [saving-loading]
index.save(
os.path.join(test_data_dir, "example_config"),
os.path.join(test_data_dir, "example_graph"),
os.path.join(test_data_dir, "example_data"),
)
index = svs.Vamana(
os.path.join(test_data_dir, "example_config"),
os.path.join(test_data_dir, "example_graph"),
os.path.join(test_data_dir, "example_data"),
svs.DistanceType.L2,
num_threads = 4,
)
# [saving-loading]
#####
##### Main Executable
#####
if __name__ == "__main__":
run()
#####
##### As a unit test.
#####
class VamanaExampleTestCase(unittest.TestCase):
def tearDown(self):
if test_data_dir is not None:
print(f"Removing temporary directory {test_data_dir}")
os.rmdir(test_data_dir)
def test_all(self):
run()
Using open-source SVS only
Building and installing
Prerequisites
Python >= 3.9
A C++20 capable compiler:
GCC >= 11.0
Clang >= 13.0
To build and install, clone the repo and run the following pip install command.
# Clone the repository
git clone https://github.com/intel/ScalableVectorSearch
cd ScalableVectorSearch
# Install svs using pip
pip install bindings/python
Run this command to confirm SVS is installed correctly, it should print the library version.
python3 -c "import svs; print(svs.library_version())"
Entire example
We run automated tests to ensure the script stays valid. You can ignore all test-related code.
# Copyright 2025 Intel Corporation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# Import `unittest` to allow for automated testing.
import unittest
# [imports]
import os
import svs
# [imports]
DEBUG_MODE = False
def assert_equal(lhs, rhs, message: str = "", epsilon = 0.05):
if DEBUG_MODE:
print(f"{message}: {lhs} == {rhs}")
else:
assert lhs < rhs + epsilon, message
assert lhs > rhs - epsilon, message
def run_test_float(index, queries, groundtruth):
expected = {
10: 0.5664,
20: 0.7397,
30: 0.8288,
40: 0.8837,
}
for window_size in range(10, 50, 10):
index.search_window_size = window_size
I, D = index.search(queries, 10)
recall = svs.k_recall_at(groundtruth, I, 10, 10)
assert_equal(
recall, expected[window_size], f"Standard Search Check ({window_size})"
)
# Shadow this as a global to make it available to the test-case clean-up.
test_data_dir = None
def run():
# ###
# Generating test data
# ###
# [generate-dataset]
# Create a test dataset.
# This will create a directory "example_data_vamana" and populate it with three
# entries:
# - data.fvecs: The test dataset.
# - queries.fvecs: The test queries.
# - groundtruth.ivecs: The groundtruth.
test_data_dir = "./example_data_vamana"
svs.generate_test_dataset(
10000, # Create 10000 vectors in the dataset.
1000, # Generate 1000 query vectors.
128, # Set the vector dimensionality to 128.
test_data_dir, # The directory where results will be generated.
data_seed = 1234, # Random number seed for reproducibility.
query_seed = 5678, # Random number seed for reproducibility.
num_threads = 4, # Number of threads to use.
distance = svs.DistanceType.L2, # The distance type to use.
)
# [generate-dataset]
# ###
# Building the index
# ###
# [build-parameters]
# Now, we can build a graph index over the data set.
parameters = svs.VamanaBuildParameters(
graph_max_degree = 64,
window_size = 128,
)
# [build-parameters]
# [build-index]
# Build the index.
index = svs.Vamana.build(
parameters,
svs.VectorDataLoader(
os.path.join(test_data_dir, "data.fvecs"), svs.DataType.float32
),
svs.DistanceType.L2,
num_threads = 4,
)
# [build-index]
# [build-index-fromNumpyArray]
# Build the index.
data = svs.read_vecs(os.path.join(test_data_dir, "data.fvecs"))
index = svs.Vamana.build(
parameters,
data,
svs.DistanceType.L2,
num_threads = 4,
)
# [build-index-fromNumpyArray]
# ###
# Searching the index
# ###
# [load-aux]
# Load the queries and ground truth.
queries = svs.read_vecs(os.path.join(test_data_dir, "queries.fvecs"))
groundtruth = svs.read_vecs(os.path.join(test_data_dir, "groundtruth.ivecs"))
# [load-aux]
# [perform-queries]
# Set the search window size of the index and perform queries.
index.search_window_size = 30
I, D = index.search(queries, 10)
# Compare with the groundtruth.
recall = svs.k_recall_at(groundtruth, I, 10, 10)
print(f"Recall = {recall}")
assert_equal(recall, 0.8288)
# [perform-queries]
# [search-window-size]
# We can vary the search window size to demonstrate the trade off in accuracy.
for window_size in range(10, 50, 10):
index.search_window_size = window_size
I, D = index.search(queries, 10)
recall = svs.k_recall_at(groundtruth, I, 10, 10)
print(f"Window size = {window_size}, Recall = {recall}")
# [search-window-size]
##### Begin Test
run_test_float(index, queries, groundtruth)
##### End Test
# ###
# Saving the index
# ###
# [saving-results]
# Finally, we can save the results.
index.save(
os.path.join(test_data_dir, "example_config"),
os.path.join(test_data_dir, "example_graph"),
os.path.join(test_data_dir, "example_data"),
)
# [saving-results]
# ###
# Reloading a saved index
# ###
# [loading]
# We can reload an index from a previously saved set of files.
index = svs.Vamana(
os.path.join(test_data_dir, "example_config"),
svs.GraphLoader(os.path.join(test_data_dir, "example_graph")),
svs.VectorDataLoader(
os.path.join(test_data_dir, "example_data"), svs.DataType.float32
),
svs.DistanceType.L2,
num_threads = 4,
)
# We can rerun the queries to ensure everything works properly.
index.search_window_size = 30
I, D = index.search(queries, 10)
# Compare with the groundtruth.
recall = svs.k_recall_at(groundtruth, I, 10, 10)
print(f"Recall = {recall}")
assert_equal(recall, 0.8288)
# [loading]
##### Begin Test
run_test_float(index, queries, groundtruth)
##### End Test
# [only-loading]
# We can reload an index from a previously saved set of files.
index = svs.Vamana(
os.path.join(test_data_dir, "example_config"),
os.path.join(test_data_dir, "example_graph"),
os.path.join(test_data_dir, "example_data"),
svs.DistanceType.L2,
num_threads = 4,
)
# [only-loading]
# [runtime-nthreads]
index.num_threads = 4
# [runtime-nthreads]
#####
##### Main Executable
#####
if __name__ == "__main__":
run()
#####
##### As a unit test.
#####
class VamanaExampleTestCase(unittest.TestCase):
def tearDown(self):
if test_data_dir is not None:
print(f"Removing temporary directory {test_data_dir}")
os.rmdir(test_data_dir)
def test_all(self):
run()