Repository: voxel51/fiftyone-brain (language: python)

Issue:
Embeddings API cleanup



Discussion / hints:

Fix patch:
diff --git a/fiftyone/brain/__init__.py b/fiftyone/brain/__init__.py
index 24515d2b..1d3d1010 100644
--- a/fiftyone/brain/__init__.py
+++ b/fiftyone/brain/__init__.py
@@ -8,20 +8,26 @@
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
+import fiftyone.brain.config as _foc
+
 from .similarity import (
+    Similarity,
     SimilarityConfig,
-    SimilarityResults,
+    SimilarityIndex,
 )
 from .visualization import (
     VisualizationConfig,
+    VisualizationResults,
     UMAPVisualizationConfig,
     TSNEVisualizationConfig,
     PCAVisualizationConfig,
     ManualVisualizationConfig,
-    VisualizationResults,
 )
 
 
+brain_config = _foc.load_brain_config()
+
+
 def compute_hardness(samples, label_field, hardness_field="hardness"):
     """Adds a hardness field to each sample scoring the difficulty that the
     specified label field observed in classifying the sample.
@@ -264,7 +270,7 @@ def compute_visualization(
     patches that can be interactively visualized.
 
     The representation can be visualized by calling the
-    :meth:`sort_by_similarity() <fiftyone.brain.similarity.SimilarityResults.sort_by_similarity>`
+    :meth:`visualize() <fiftyone.brain.visualization.VisualizationResults.visualize>`
     method of the returned
     :class:`fiftyone.brain.visualization.VisualizationResults` object.
 
@@ -299,11 +305,17 @@ def compute_visualization(
         embeddings (None): if no ``model`` is provided, this argument specifies
             pre-computed embeddings to use, which can be any of the following:
 
+            -   a dict mapping sample IDs to embedding vectors
             -   a ``num_samples x num_embedding_dims`` array of embeddings
+                corresponding to the samples in ``samples``
+            -   if ``patches_field`` is specified, a dict mapping label IDs to
+                to embedding vectors
             -   if ``patches_field`` is specified,  a dict mapping sample IDs
                 to ``num_patches x num_embedding_dims`` arrays of patch
                 embeddings
             -   the name of a dataset field containing the embeddings to use
+            -   a :class:`fiftyone.brain.similarity.SimilarityIndex` from which
+                to retrieve embeddings for all samples/patches in ``samples``
 
             If a ``model`` is provided, this argument specifies the name of a
             field in which to store the computed embeddings. In either case,
@@ -311,10 +323,14 @@ def compute_visualization(
             fully-qualified path to the patch embeddings or just the name of
             the label attribute in ``patches_field``
         points (None): a pre-computed low-dimensional representation to use. If
-            provided, no embeddings will be computed. Can be any of the
+            provided, no embeddings will be used/computed. Can be any of the
             following:
 
-            -   a ``num_samples x num_dims`` array of points
+            -   a dict mapping sample IDs to points vectors
+            -   a ``num_samples x num_dims`` array of points corresponding to
+                the samples in ``samples``
+            -   if ``patches_field`` is specified, a dict mapping label IDs to
+                points vectors
             -   if ``patches_field`` is specified, a ``num_patches x num_dims``
                 array of points whose rows correspond to the flattened list of
                 patches whose IDs are shown below::
@@ -382,31 +398,34 @@ def compute_similarity(
     patches_field=None,
     embeddings=None,
     brain_key=None,
-    metric="euclidean",
     model=None,
     force_square=False,
     alpha=None,
     batch_size=None,
     num_workers=None,
     skip_failures=True,
+    backend=None,
+    **kwargs,
 ):
     """Uses embeddings to index the samples or their patches so that you can
-    query/sort by visual similarity.
+    query/sort by similarity.
+
+    Calling this method only creates the index. You can then call the methods
+    exposed on the retuned :class:`fiftyone.brain.similarity.SimilarityIndex`
+    object to perform the following operations:
 
-    Calling this method (or loading existing results) only generates the index.
-    You can then call the methods exposed on the retuned
-    :class:`fiftyone.brain.similarity.SimilarityResults` object to perform the
-    following operations:
+    -   :meth:`sort_by_similarity() <fiftyone.brain.similarity.SimilarityIndex.sort_by_similarity>`:
+        Sort the samples in the collection by similarity to a specific example
+        or example(s)
 
-    -   :meth:`sort_by_similarity() <fiftyone.brain.similarity.SimilarityResults.sort_by_similarity>`:
-        Sort the samples in the collection by visual similarity to a specific
-        example or example(s)
+    In addition, if the backend supports it, you can call the following
+    duplicate detection methods:
 
-    -   :meth:`find_duplicates() <fiftyone.brain.similarity.SimilarityResults.find_duplicates>`:
+    -   :meth:`find_duplicates() <fiftyone.brain.similarity.DuplicatesMixin.find_duplicates>`:
         Query the index to find all examples with near-duplicates in the
         collection
 
-    -   :meth:`find_unique() <fiftyone.brain.similarity.SimilarityResults.find_unique>`:
+    -   :meth:`find_unique() <fiftyone.brain.similarity.DuplicatesMixin.find_unique>`:
         Query the index to select a subset of examples of a specified size that
         are maximally unique with respect to each other
 
@@ -425,23 +444,31 @@ def compute_similarity(
             :class:`fiftyone.core.labels.Detections`,
             :class:`fiftyone.core.labels.Polyline`, or
             :class:`fiftyone.core.labels.Polylines`
-        embeddings (None): if no ``model`` is provided, this argument specifies
-            pre-computed embeddings to use. Can be any of the following:
+        embeddings (None): embeddings to feed the index. This argument's
+            behavior depends on whether a ``model`` is provided, as described
+            below.
+
+            If no ``model`` is provided, this argument specifies precomputed
+            embeddings to use:
 
             -   a ``num_samples x num_dims`` array of embeddings
             -   if ``patches_field`` is specified,  a dict mapping sample IDs
                 to ``num_patches x num_dims`` arrays of patch embeddings
-            -   the name of a dataset field containing the embeddings to use
+            -   the name of a dataset field from which to load embeddings
+            -   ``None``: use the default model to compute embeddings
+            -   ``False``: **do not** compute embeddings right now
 
-            If a ``model`` is provided, this argument specifies the name of a
-            field in which to store the computed embeddings. In either case,
-            when working with patch embeddings, you can provide either the
-            fully-qualified path to the patch embeddings or just the name of
-            the label attribute in ``patches_field``
+            If a ``model`` is provided, this argument specifies where to store
+            the model's embeddings:
+
+            -   the name of a field in which to store the computed embeddings
+            -   ``False``: **do not** compute or store embeddings right now
+
+            In either case, when working with patch embeddings, you can provide
+            either the fully-qualified path to the patch embeddings or just the
+            name of the label attribute in ``patches_field``
         brain_key (None): a brain key under which to store the results of this
             method
-        metric ("euclidean"): the embedding distance metric to use. See
-            ``sklearn.metrics.pairwise_distance`` for supported values
         model (None): a :class:`fiftyone.core.models.Model` or the name of a
             model from the
             `FiftyOne Model Zoo <https://docs.voxel51.com/user_guide/model_zoo/index.html>`_
@@ -464,24 +491,32 @@ def compute_similarity(
             embeddings
         skip_failures (True): whether to gracefully continue without raising an
             error if embeddings cannot be generated for a sample
+        backend (None): the similarity backend to use. The supported values are
+            ``fiftyone.brain.brain_config.similarity_backends.keys()`` and the
+            default is
+            ``fiftyone.brain.brain_config.default_similarity_backend``
+        **kwargs: keyword arguments for the
+            :class:`fiftyone.brian.SimilarityConfig` subclass of the backend
+            being used
 
     Returns:
-        a :class:`fiftyone.brain.similarity.SimilarityResults`
+        a :class:`fiftyone.brain.similarity.SimilarityIndex`
     """
-    import fiftyone.brain.internal.core.similarity as fbs
+    import fiftyone.brain.similarity as fbs
 
     return fbs.compute_similarity(
         samples,
         patches_field,
         embeddings,
         brain_key,
-        metric,
         model,
         force_square,
         alpha,
         batch_size,
         num_workers,
         skip_failures,
+        backend,
+        **kwargs,
     )
 
 
diff --git a/fiftyone/brain/config.py b/fiftyone/brain/config.py
new file mode 100644
index 00000000..a8c08988
--- /dev/null
+++ b/fiftyone/brain/config.py
@@ -0,0 +1,160 @@
+"""
+Brain config.
+
+| Copyright 2017-2023, Voxel51, Inc.
+| `voxel51.com <https://voxel51.com/>`_
+|
+"""
+import os
+
+from fiftyone.core.config import EnvConfig
+
+
+class BrainConfig(EnvConfig):
+    """FiftyOne brain configuration settings."""
+
+    _BUILTIN_SIMILARITY_BACKENDS = {
+        "sklearn": {
+            "config_cls": "fiftyone.brain.internal.core.sklearn.SklearnSimilarityConfig",
+        },
+        "pinecone": {
+            "config_cls": "fiftyone.brain.internal.core.pinecone.PineconeSimilarityConfig",
+        },
+        "qdrant": {
+            "config_cls": "fiftyone.brain.internal.core.qdrant.QdrantSimilarityConfig",
+        },
+    }
+
+    def __init__(self, d=None):
+        if d is None:
+            d = {}
+
+        self.default_similarity_backend = self.parse_string(
+            d,
+            "default_similarity_backend",
+            env_var="FIFTYONE_BRAIN_DEFAULT_SIMILARITY_BACKEND",
+            default="sklearn",
+        )
+
+        self.similarity_backends = self._parse_similarity_backends(d)
+
+    def _parse_similarity_backends(self, d):
+        d = d.get("similarity_backends", {})
+        env_vars = dict(os.environ)
+
+        #
+        # `FIFTYONE_BRAIN_SIMILARITY_BACKENDS` can be used to declare which
+        # backends are exposed. This may exclude builtin backends and/or
+        # declare new backends
+        #
+
+        if "FIFTYONE_BRAIN_SIMILARITY_BACKENDS" in env_vars:
+            backends = env_vars["FIFTYONE_BRAIN_SIMILARITY_BACKENDS"].split(
+                ","
+            )
+
+            # Declare new backends and omit any others not in `backends`
+            d = {backend: d.get(backend, {}) for backend in backends}
+        else:
+            backends = sorted(self._BUILTIN_SIMILARITY_BACKENDS.keys())
+
+            # Declare builtin backends if necessary
+            for backend in backends:
+                if backend not in d:
+                    d[backend] = {}
+
+        #
+        # Extract parameters from any environment variables of the form
+        # `FIFTYONE_BRAIN_SIMILARITY_<BACKEND>_<PARAMETER>`
+        #
+
+        for backend, parameters in d.items():
+            prefix = "FIFTYONE_BRAIN_SIMILARITY_%s_" % backend.upper()
+            for env_name, env_value in env_vars.items():
+                if env_name.startswith(prefix):
+                    name = env_name[len(prefix) :].lower()
+                    value = _parse_env_value(env_value)
+                    parameters[name] = value
+
+        #
+        # Set default parameters for builtin similarity backends
+        #
+
+        for backend, defaults in self._BUILTIN_SIMILARITY_BACKENDS.items():
+            if backend not in d:
+                continue
+
+            d_backend = d[backend]
+            for name, value in defaults.items():
+                if name not in d_backend:
+                    d_backend[name] = value
+
+        return d
+
+
+def locate_brain_config():
+    """Returns the path to the :class:`BrainConfig` on disk.
+
+    The default location is ``~/.fiftyone/brain_config.json``, but you can
+    override this path by setting the ``FIFTYONE_BRAIN_CONFIG_PATH``
+    environment variable.
+
+    Note that a config file may not actually exist on disk in the default
+    location, in which case the default config settings will be used.
+
+    Returns:
+        the path to the :class:`BrainConfig` on disk
+
+    Raises:
+        OSError: if the brain config path has been customized but the file does
+            not exist on disk
+    """
+    if "FIFTYONE_BRAIN_CONFIG_PATH" not in os.environ:
+        return os.path.join(
+            os.path.expanduser("~"), ".fiftyone", "brain_config.json"
+        )
+
+    config_path = os.environ["FIFTYONE_BRAIN_CONFIG_PATH"]
+    if not os.path.isfile(config_path):
+        raise OSError("Brain config file '%s' not found" % config_path)
+
+    return config_path
+
+
+def load_brain_config():
+    """Loads the FiftyOne brain config.
+
+    Returns:
+        a :class:`BrainConfig` instance
+    """
+    brain_config_path = locate_brain_config()
+    if os.path.isfile(brain_config_path):
+        return BrainConfig.from_json(brain_config_path)
+
+    return BrainConfig()
+
+
+def _parse_env_value(value):
+    try:
+        return int(value)
+    except:
+        pass
+
+    try:
+        return float(value)
+    except:
+        pass
+
+    if value in ("True", "true"):
+        return True
+
+    if value in ("False", "false"):
+        return False
+
+    if value in ("None", ""):
+        return None
+
+    if "," in value:
+        return [_parse_env_value(v) for v in value.split(",")]
+
+    return value
diff --git a/fiftyone/brain/internal/core/pinecone.py b/fiftyone/brain/internal/core/pinecone.py
new file mode 100644
index 00000000..ff5e1254
--- /dev/null
+++ b/fiftyone/brain/internal/core/pinecone.py
@@ -0,0 +1,581 @@
+"""
+Piencone similarity backend.
+
+| Copyright 2017-2023, Voxel51, Inc.
+| `voxel51.com <https://voxel51.com/>`_
+|
+"""
+import logging
+
+import numpy as np
+
+import eta.core.utils as etau
+
+import fiftyone.core.utils as fou
+from fiftyone.brain.similarity import (
+    SimilarityConfig,
+    Similarity,
+    SimilarityIndex,
+)
+import fiftyone.brain.internal.core.utils as fbu
+
+pinecone = fou.lazy_import("pinecone")
+
+
+logger = logging.getLogger(__name__)
+
+_SUPPORTED_METRICS = ("cosine", "dotproduct", "euclidean")
+
+
+class PineconeSimilarityConfig(SimilarityConfig):
+    """Configuration for the Pinecone similarity backend.
+
+    Args:
+        embeddings_field (None): the sample field containing the embeddings,
+            if one was provided
+        model (None): the :class:`fiftyone.core.models.Model` or name of the
+            zoo model that was used to compute embeddings, if known
+        patches_field (None): the sample field defining the patches being
+            analyzed, if any
+        supports_prompts (None): whether this run supports prompt queries
+        index_name (None): the name of a Pinecone index to use or create. If
+            none is provided, a new index will be created
+        index_type (None): the index type to use when creating a new index
+        namespace (None): a namespace under which to store vectors added to the
+            index
+        metric (None): the embedding distance metric to use when creating a
+            new index. Supported values are
+            ``("cosine", "dotproduct", "euclidean")``
+        replicas (None): an optional number of replicas when creating a new
+            index
+        shards (None): an optional number of shards when creating a new index
+        pods (None): an optional number of pods when creating a new index
+        pod_type (None): an optional pod type when creating a new index
+        api_key (None): a Pinecone API key to use
+        environment (None): a Pinecone environment to use
+        project_name (None): a Pinecone project to use
+    """
+
+    def __init__(
+        self,
+        embeddings_field=None,
+        model=None,
+        patches_field=None,
+        supports_prompts=None,
+        index_name=None,
+        index_type=None,
+        namespace=None,
+        metric=None,
+        replicas=None,
+        shards=None,
+        pods=None,
+        pod_type=None,
+        api_key=None,
+        environment=None,
+        project_name=None,
+        **kwargs,
+    ):
+        if metric is not None and metric not in _SUPPORTED_METRICS:
+            raise ValueError(
+                "Unsupported metric '%s'. Supported values are %s"
+                % (metric, _SUPPORTED_METRICS)
+            )
+
+        super().__init__(
+            embeddings_field=embeddings_field,
+            model=model,
+            patches_field=patches_field,
+            supports_prompts=supports_prompts,
+            **kwargs,
+        )
+
+        self.index_name = index_name
+        self.index_type = index_type
+        self.namespace = namespace
+        self.metric = metric
+        self.replicas = replicas
+        self.shards = shards
+        self.pods = pods
+        self.pod_type = pod_type
+
+        # store privately so these aren't serialized
+        self._api_key = api_key
+        self._environment = environment
+        self._project_name = project_name
+
+    @property
+    def method(self):
+        return "pinecone"
+
+    @property
+    def api_key(self):
+        return self._api_key
+
+    @api_key.setter
+    def api_key(self, value):
+        self._api_key = value
+
+    @property
+    def environment(self):
+        return self._environment
+
+    @environment.setter
+    def environment(self, value):
+        self._environment = value
+
+    @property
+    def project_name(self):
+        return self._project_name
+
+    @project_name.setter
+    def project_name(self, value):
+        self._project_name = value
+
+    @property
+    def max_k(self):
+        return 10000  # Pinecone limit
+
+    @property
+    def supports_least_similarity(self):
+        return False
+
+    @property
+    def supported_aggregations(self):
+        return ("mean",)
+
+    def load_credentials(
+        self, api_key=None, environment=None, project_name=None
+    ):
+        self._load_parameters(
+            api_key=api_key, environment=environment, project_name=project_name
+        )
+
+
+class PineconeSimilarity(Similarity):
+    """Pinecone similarity factory.
+
+    Args:
+        config: a :class:`PineconeSimilarityConfig`
+    """
+
+    def ensure_requirements(self):
+        fou.ensure_package("pinecone-client")
+
+    def ensure_usage_requirements(self):
+        fou.ensure_package("pinecone-client")
+
+    def initialize(self, samples, brain_key):
+        return PineconeSimilarityIndex(
+            samples, self.config, brain_key, backend=self
+        )
+
+
+class PineconeSimilarityIndex(SimilarityIndex):
+    """Class for interacting with Pinecone similarity indexes.
+
+    Args:
+        samples: the :class:`fiftyone.core.collections.SampleCollection` used
+        config: the :class:`PineconeSimilarityConfig` used
+        brain_key: the brain key
+        backend (None): a :class:`PineconeSimilarity` instance
+    """
+
+    def __init__(self, samples, config, brain_key, backend=None):
+        super().__init__(samples, config, brain_key, backend=backend)
+        self._index = None
+        self._initialize()
+
+    def _initialize(self):
+        pinecone.init(
+            api_key=self.config.api_key,
+            environment=self.config.environment,
+            project_name=self.config.project_name,
+        )
+
+        try:
+            index_names = pinecone.list_indexes()
+        except Exception as e:
+            raise ValueError(
+                "Failed to connect to Pinecone backend at environment '%s'. "
+                "Refer to https://docs.voxel51.com/integrations/pinecone.html "
+                "for more information" % self.config.environment
+            ) from e
+
+        if self.config.index_name is None:
+            root = "fiftyone-" + fou.to_slug(self.samples._root_dataset.name)
+            index_name = fbu.get_unique_name(root, index_names)
+
+            self.config.index_name = index_name
+            self.save_config()
+
+        if self.config.index_name in index_names:
+            index = pinecone.Index(self.config.index_name)
+        else:
+            index = None
+
+        self._index = index
+
+    def _create_index(self, dimension):
+        kwargs = dict(
+            index_type=self.config.index_type,
+            metric=self.config.metric,
+            replicas=self.config.replicas,
+            shards=self.config.shards,
+            pods=self.config.pods,
+            pod_type=self.config.pod_type,
+        )
+        kwargs = {k: v for k, v in kwargs.items() if v is not None}
+
+        pinecone.create_index(
+            self.config.index_name,
+            dimension=dimension,
+            **kwargs,
+        )
+
+        self._index = pinecone.Index(self.config.index_name)
+
+    @property
+    def index(self):
+        """The ``pinecone.Index`` instance for this index."""
+        return self._index
+
+    @property
+    def total_index_size(self):
+        if self._index is None:
+            return None
+
+        return self._index.describe_index_stats()["total_vector_count"]
+
+    def add_to_index(
+        self,
+        embeddings,
+        sample_ids,
+        label_ids=None,
+        overwrite=True,
+        allow_existing=True,
+        warn_existing=False,
+        batch_size=100,
+        namespace=None,
+    ):
+        if namespace is None:
+            namespace = self.config.namespace
+
+        if self._index is None:
+            self._create_index(embeddings.shape[1])
+
+        if label_ids is not None:
+            ids = label_ids
+        else:
+            ids = sample_ids
+
+        if warn_existing or not allow_existing or not overwrite:
+            existing_ids = self._get_existing_ids(ids)
+            num_existing = len(existing_ids)
+
+            if num_existing > 0:
+                if not allow_existing:
+                    raise ValueError(
+                        "Found %d IDs (eg %s) that already exist in the index"
+                        % (num_existing, next(iter(existing_ids)))
+                    )
+
+                if warn_existing:
+                    if overwrite:
+                        logger.warning(
+                            "Overwriting %d IDs that already exist in the "
+                            "index",
+                            num_existing,
+                        )
+                    else:
+                        logger.warning(
+                            "Skipping %d IDs that already exist in the index",
+                            num_existing,
+                        )
+        else:
+            existing_ids = set()
+
+        if existing_ids and not overwrite:
+            del_inds = [i for i, _id in enumerate(ids) if _id in existing_ids]
+
+            embeddings = np.delete(embeddings, del_inds)
+            sample_ids = np.delete(sample_ids, del_inds)
+            if label_ids is not None:
+                label_ids = np.delete(label_ids, del_inds)
+
+        embeddings = [e.tolist() for e in embeddings]
+        sample_ids = list(sample_ids)
+        if label_ids is not None:
+            ids = list(label_ids)
+        else:
+            ids = list(sample_ids)
+
+        for _embeddings, _ids, _sample_ids in zip(
+            fou.iter_batches(embeddings, batch_size),
+            fou.iter_batches(ids, batch_size),
+            fou.iter_batches(sample_ids, batch_size),
+        ):
+            _id_dicts = [
+                {"id": _id, "sample_id": _sid}
+                for _id, _sid in zip(_ids, _sample_ids)
+            ]
+            self._index.upsert(
+                list(zip(_ids, _embeddings, _id_dicts)),
+                namespace=namespace,
+            )
+
+    def remove_from_index(
+        self,
+        sample_ids=None,
+        label_ids=None,
+        allow_missing=True,
+        warn_missing=False,
+    ):
+        if label_ids is not None:
+            ids = label_ids
+        else:
+            ids = sample_ids
+
+        if not allow_missing or warn_missing:
+            existing_ids = self._index.fetch(ids).vectors.keys()
+            missing_ids = set(existing_ids) - set(ids)
+            num_missing = len(missing_ids)
+
+            if num_missing > 0:
+                if not allow_missing:
+                    raise ValueError(
+                        "Found %d IDs (eg %s) that are not present in the "
+                        "index" % (num_missing, missing_ids[0])
+                    )
+
+                if warn_missing:
+                    logger.warning(
+                        "Ignoring %d IDs that are not present in the index",
+                        num_missing,
+                    )
+
+        self._index.delete(ids=ids)
+
+    def get_embeddings(
+        self,
+        sample_ids=None,
+        label_ids=None,
+        allow_missing=True,
+        warn_missing=False,
+    ):
+        if label_ids is not None:
+            if self.config.patches_field is None:
+                raise ValueError("This index does not support label IDs")
+
+            if sample_ids is not None:
+                logger.warning(
+                    "Ignoring sample IDs when label IDs are provided"
+                )
+
+        if sample_ids is not None and self.config.patches_field is not None:
+            (
+                embeddings,
+                sample_ids,
+                label_ids,
+                missing_ids,
+            ) = self._get_patch_embeddings_from_sample_ids(sample_ids)
+        elif self.config.patches_field is not None:
+            (
+                embeddings,
+                sample_ids,
+                label_ids,
+                missing_ids,
+            ) = self._get_patch_embeddings_from_label_ids(label_ids)
+        else:
+            (
+                embeddings,
+                sample_ids,
+                label_ids,
+                missing_ids,
+            ) = self._get_sample_embeddings(sample_ids)
+
+        num_missing_ids = len(missing_ids)
+        if num_missing_ids > 0:
+            if not allow_missing:
+                raise ValueError(
+                    "Found %d IDs (eg %s) that do not exist in the index"
+                    % (num_missing_ids, missing_ids[0])
+                )
+
+            if warn_missing:
+                logger.warning(
+                    "Skipping %d IDs that do not exist in the index",
+                    num_missing_ids,
+                )
+
+        embeddings = np.array(embeddings)
+        sample_ids = np.array(sample_ids)
+        if label_ids is not None:
+            label_ids = np.array(label_ids)
+
+        return embeddings, sample_ids, label_ids
+
+    def cleanup(self):
+        pinecone.delete_index(self.config.index_name)
+        self._index = None
+
+    def _get_existing_ids(self, ids, batch_size=1000):
+        existing_ids = set()
+        for batch_ids in fou.iter_batches(ids, batch_size):
+            response = self._index.fetch(ids=list(batch_ids))["vectors"]
+            existing_ids.update(response.keys())
+
+        return existing_ids
+
+    def _get_sample_embeddings(self, sample_ids, batch_size=1000):
+        found_embeddings = []
+        found_sample_ids = []
+
+        if sample_ids is None:
+            raise ValueError(
+                "Pinecone does not support retrieving all vectors in an index"
+            )
+
+        for batch_ids in fou.iter_batches(sample_ids, batch_size):
+            response = self._index.fetch(ids=list(batch_ids))["vectors"]
+
+            for r in response.values():
+                found_embeddings.append(r["values"])
+                found_sample_ids.append(r["id"])
+
+        missing_ids = list(set(sample_ids) - set(found_sample_ids))
+
+        return found_embeddings, found_sample_ids, None, missing_ids
+
+    def _get_patch_embeddings_from_label_ids(self, label_ids, batch_size=1000):
+        found_embeddings = []
+        found_sample_ids = []
+        found_label_ids = []
+
+        if label_ids is None:
+            raise ValueError(
+                "Pinecone does not support retrieving all vectors in an index"
+            )
+
+        for batch_ids in fou.iter_batches(label_ids, batch_size):
+            response = self._index.fetch(ids=list(batch_ids))["vectors"]
+
+            for r in response.values():
+                found_embeddings.append(r["values"])
+                found_sample_ids.append(r["metadata"]["sample_id"])
+                found_label_ids.append(r["id"])
+
+        missing_ids = list(set(label_ids) - set(found_label_ids))
+
+        return found_embeddings, found_sample_ids, found_label_ids, missing_ids
+
+    def _get_patch_embeddings_from_sample_ids(
+        self, sample_ids, batch_size=100
+    ):
+        found_embeddings = []
+        found_sample_ids = []
+        found_label_ids = []
+
+        query_vector = [0.0] * self._index.describe_index_stats().dimension
+        top_k = min(batch_size, self.config.max_k)
+
+        for batch_ids in fou.iter_batches(sample_ids, batch_size):
+            response = self._index.query(
+                vector=query_vector,
+                filter={"sample_id": {"$in": list(batch_ids)}},
+                top_k=top_k,
+                include_values=True,
+                include_metadata=True,
+            )
+
+            for r in response["matches"]:
+                found_embeddings.append(r["values"])
+                found_sample_ids.append(r["metadata"]["sample_id"])
+                found_label_ids.append(r["id"])
+
+        missing_ids = list(set(sample_ids) - set(found_sample_ids))
+
+        return found_embeddings, found_sample_ids, found_label_ids, missing_ids
+
+    def _kneighbors(
+        self,
+        query=None,
+        k=None,
+        reverse=False,
+        aggregation=None,
+        return_dists=False,
+    ):
+        if query is None:
+            raise ValueError("Pinecone does not support full index neighbors")
+
+        if reverse is True:
+            raise ValueError(
+                "Pinecone does not support least similarity queries"
+            )
+
+        if k is None or k > self.config.max_k:
+            raise ValueError("Pincone requires k<=%s" % self.config.max_k)
+
+        if aggregation not in (None, "mean"):
+            raise ValueError("Unsupported aggregation '%s'" % aggregation)
+
+        query = self._parse_neighbors_query(query)
+        if aggregation == "mean" and query.ndim == 2:
+            query = query.mean(axis=0)
+
+        single_query = query.ndim == 1
+        if single_query:
+            query = [query]
+
+        if self.config.patches_field is not None:
+            index_ids = self.current_label_ids
+        else:
+            index_ids = self.current_sample_ids
+
+        _filter = {"id": {"$in": list(index_ids)}}
+
+        ids = []
+        dists = []
+        for q in query:
+            response = self._index.query(
+                vector=q.tolist(), top_k=k, filter=_filter
+            )
+            ids.append([r["id"] for r in response["matches"]])
+            if return_dists:
+                dists.append([r["score"] for r in response["matches"]])
+
+        if single_query:
+            ids = ids[0]
+            if return_dists:
+                dists = dists[0]
+
+        if return_dists:
+            return ids, dists
+
+        return ids
+
+    def _parse_neighbors_query(self, query):
+        if etau.is_str(query):
+            query_ids = [query]
+            single_query = True
+        else:
+            query = np.asarray(query)
+
+            # Query by vector(s)
+            if np.issubdtype(query.dtype, np.number):
+                return query
+
+            query_ids = list(query)
+            single_query = False
+
+        # Query by ID(s)
+        response = self._index.fetch(query_ids)["vectors"]
+        query = np.array([response[_id]["values"] for _id in query_ids])
+
+        if single_query:
+            query = query[0, :]
+
+        return query
+
+    @classmethod
+    def _from_dict(cls, d, samples, config, brain_key):
+        return cls(samples, config, brain_key)
diff --git a/fiftyone/brain/internal/core/qdrant.py b/fiftyone/brain/internal/core/qdrant.py
new file mode 100644
index 00000000..0f3c244f
--- /dev/null
+++ b/fiftyone/brain/internal/core/qdrant.py
@@ -0,0 +1,612 @@
+"""
+Qdrant similarity backend.
+
+| Copyright 2017-2023, Voxel51, Inc.
+| `voxel51.com <https://voxel51.com/>`_
+|
+"""
+import logging
+
+import numpy as np
+
+import eta.core.utils as etau
+
+import fiftyone.core.utils as fou
+from fiftyone.brain.similarity import (
+    SimilarityConfig,
+    Similarity,
+    SimilarityIndex,
+)
+import fiftyone.brain.internal.core.utils as fbu
+
+qdrant = fou.lazy_import("qdrant_client")
+qmodels = fou.lazy_import("qdrant_client.http.models")
+
+
+logger = logging.getLogger(__name__)
+
+_SUPPORTED_METRICS = {
+    "cosine": qmodels.Distance.COSINE,
+    "dotproduct": qmodels.Distance.DOT,
+    "euclidean": qmodels.Distance.EUCLID,
+}
+
+
+class QdrantSimilarityConfig(SimilarityConfig):
+    """Configuration for the Qdrant similarity backend.
+
+    Args:
+        embeddings_field (None): the sample field containing the embeddings,
+            if one was provided
+        model (None): the :class:`fiftyone.core.models.Model` or name of the
+            zoo model that was used to compute embeddings, if known
+        patches_field (None): the sample field defining the patches being
+            analyzed, if any
+        supports_prompts (None): whether this run supports prompt queries
+        collection_name (None): the name of a Qdrant collection to use or
+            create. If none is provided, a new collection will be created
+        metric (None): the embedding distance metric to use when creating a
+            new index. Supported values are
+            ``("cosine", "dotproduct", "euclidean")``
+        replication_factor (None): an optional replication factor to use when
+            creating a new index
+        shard_number (None): an optional number of shards to use when creating
+            a new index
+        write_consistency_factor (None): an optional write consistsency factor
+            to use when creating a new index
+        hnsw_config (None): an optional dict of HNSW config parameters to use
+            when creating a new index
+        optimizers_config (None): an optional dict of optimizer parameters to
+            use when creating a new index
+        wal_config (None): an optional dict of WAL config parameters to use
+            when creating a new index
+    """
+
+    def __init__(
+        self,
+        embeddings_field=None,
+        model=None,
+        patches_field=None,
+        supports_prompts=None,
+        collection_name=None,
+        metric=None,
+        replication_factor=None,
+        shard_number=None,
+        write_consistency_factor=None,
+        hnsw_config=None,
+        optimizers_config=None,
+        wal_config=None,
+        url=None,
+        api_key=None,
+        **kwargs,
+    ):
+        if metric is not None and metric not in _SUPPORTED_METRICS:
+            raise ValueError(
+                "Unsupported metric '%s'. Supported values are %s"
+                % (metric, tuple(_SUPPORTED_METRICS.keys()))
+            )
+
+        super().__init__(
+            embeddings_field=embeddings_field,
+            model=model,
+            patches_field=patches_field,
+            supports_prompts=supports_prompts,
+            **kwargs,
+        )
+
+        self.collection_name = collection_name
+        self.metric = metric
+        self.replication_factor = replication_factor
+        self.shard_number = shard_number
+        self.write_consistency_factor = write_consistency_factor
+        self.hnsw_config = hnsw_config
+        self.optimizers_config = optimizers_config
+        self.wal_config = wal_config
+
+        # store privately so these aren't serialized
+        self._url = url
+        self._api_key = api_key
+
+    @property
+    def method(self):
+        return "qdrant"
+
+    @property
+    def url(self):
+        return self._url
+
+    @url.setter
+    def url(self, value):
+        self._url = value
+
+    @property
+    def api_key(self):
+        return self._api_key
+
+    @api_key.setter
+    def api_key(self, value):
+        self._api_key = value
+
+    @property
+    def max_k(self):
+        return None
+
+    @property
+    def supports_least_similarity(self):
+        return False
+
+    @property
+    def supported_aggregations(self):
+        return ("mean",)
+
+    def load_credentials(self, url=None, api_key=None):
+        self._load_parameters(url=url, api_key=api_key)
+
+
+class QdrantSimilarity(Similarity):
+    """Qdrant similarity factory.
+
+    Args:
+        config: a :class:`QdrantSimilarityConfig`
+    """
+
+    def ensure_requirements(self):
+        fou.ensure_package("qdrant-client")
+
+    def ensure_usage_requirements(self):
+        fou.ensure_package("qdrant-client")
+
+    def initialize(self, samples, brain_key):
+        return QdrantSimilarityIndex(
+            samples, self.config, brain_key, backend=self
+        )
+
+
+class QdrantSimilarityIndex(SimilarityIndex):
+    """Class for interacting with Qdrant similarity indexes.
+
+    Args:
+        samples: the :class:`fiftyone.core.collections.SampleCollection` used
+        config: the :class:`QdrantSimilarityConfig` used
+        brain_key: the brain key
+        backend (None): a :class:`QdrantSimilarity` instance
+    """
+
+    def __init__(self, samples, config, brain_key, backend=None):
+        super().__init__(samples, config, brain_key, backend=backend)
+        self._client = None
+        self._initialize()
+
+    def _initialize(self):
+        self._client = qdrant.QdrantClient(
+            url=self.config.url, api_key=self.config.api_key
+        )
+
+        try:
+            collection_names = self._get_collection_names()
+        except Exception as e:
+            raise ValueError(
+                "Failed to connect to Qdrant backend at URL '%s'. Refer to "
+                "https://docs.voxel51.com/integrations/qdrant.html for more "
+                "information" % self.config.url
+            ) from e
+
+        if self.config.collection_name is None:
+            root = "fiftyone-" + fou.to_slug(self.samples._root_dataset.name)
+            collection_name = fbu.get_unique_name(root, collection_names)
+
+            self.config.collection_name = collection_name
+            self.save_config()
+
+    def _get_collection_names(self):
+        return [c.name for c in self._client.get_collections().collections]
+
+    def _create_collection(self, dimension):
+        if self.config.metric:
+            metric = self.config.metric
+        else:
+            metric = "cosine"
+
+        vectors_config = qmodels.VectorParams(
+            size=dimension,
+            distance=_SUPPORTED_METRICS[metric],
+        )
+
+        if self.config.hnsw_config:
+            hnsw_config = qmodels.HnswConfig(**self.config.hnsw_config)
+        else:
+            hnsw_config = None
+
+        if self.config.optimizers_config:
+            optimizers_config = qmodels.OptimizersConfig(
+                **self.config.optimizers_config
+            )
+        else:
+            optimizers_config = None
+
+        if self.config.wal_config:
+            wal_config = qmodels.WalConfig(**self.config.wal_config)
+        else:
+            wal_config = None
+
+        self._client.recreate_collection(
+            collection_name=self.config.collection_name,
+            vectors_config=vectors_config,
+            shard_number=self.config.shard_number,
+            replication_factor=self.config.replication_factor,
+            hnsw_config=hnsw_config,
+            optimizers_config=optimizers_config,
+            wal_config=wal_config,
+        )
+
+    def _get_index_ids(self, batch_size=1000):
+        ids = []
+
+        offset = 0
+        while offset is not None:
+            response = self._client.scroll(
+                collection_name=self.config.collection_name,
+                offset=offset,
+                limit=batch_size,
+                with_payload=True,
+                with_vectors=False,
+            )
+            ids.extend([self._to_fiftyone_id(r.id) for r in response[0]])
+            offset = response[-1]
+
+        return ids
+
+    @property
+    def total_index_size(self):
+        return self._client.count(self.config.collection_name).count
+
+    @property
+    def client(self):
+        """The ``qdrant.QdrantClient`` instance for this index."""
+        return self._client
+
+    def add_to_index(
+        self,
+        embeddings,
+        sample_ids,
+        label_ids=None,
+        overwrite=True,
+        allow_existing=True,
+        warn_existing=False,
+        batch_size=1000,
+    ):
+        if self.config.collection_name not in self._get_collection_names():
+            self._create_collection(embeddings.shape[1])
+
+        if label_ids is not None:
+            ids = label_ids
+        else:
+            ids = sample_ids
+
+        if warn_existing or not allow_existing or not overwrite:
+            index_ids = self._get_index_ids()
+
+            existing_ids = set(ids) & set(index_ids)
+            num_existing = len(existing_ids)
+
+            if num_existing > 0:
+                if not allow_existing:
+                    raise ValueError(
+                        "Found %d IDs (eg %s) that already exist in the index"
+                        % (num_existing, next(iter(existing_ids)))
+                    )
+
+                if warn_existing:
+                    if overwrite:
+                        logger.warning(
+                            "Overwriting %d IDs that already exist in the "
+                            "index",
+                            num_existing,
+                        )
+                    else:
+                        logger.warning(
+                            "Skipping %d IDs that already exist in the index",
+                            num_existing,
+                        )
+        else:
+            existing_ids = set()
+
+        if existing_ids and not overwrite:
+            del_inds = [i for i, _id in enumerate(ids) if _id in existing_ids]
+
+            embeddings = np.delete(embeddings, del_inds)
+            sample_ids = np.delete(sample_ids, del_inds)
+            if label_ids is not None:
+                label_ids = np.delete(label_ids, del_inds)
+
+        embeddings = [e.tolist() for e in embeddings]
+        sample_ids = list(sample_ids)
+        if label_ids is not None:
+            ids = list(label_ids)
+        else:
+            ids = list(sample_ids)
+
+        for _embeddings, _ids, _sample_ids in zip(
+            fou.iter_batches(embeddings, batch_size),
+            fou.iter_batches(ids, batch_size),
+            fou.iter_batches(sample_ids, batch_size),
+        ):
+            self._client.upsert(
+                collection_name=self.config.collection_name,
+                points=qmodels.Batch(
+                    ids=self._to_qdrant_ids(_ids),
+                    payloads=[{"sample_id": _id} for _id in _sample_ids],
+                    vectors=_embeddings,
+                ),
+            )
+
+    def remove_from_index(
+        self,
+        sample_ids=None,
+        label_ids=None,
+        allow_missing=True,
+        warn_missing=False,
+    ):
+        if label_ids is not None:
+            ids = label_ids
+        else:
+            ids = sample_ids
+
+        qids = self._to_qdrant_ids(ids)
+
+        if warn_missing or not allow_missing:
+            response = self._retrieve_points(qids, with_vectors=False)
+
+            existing_ids = self._to_fiftyone_ids([r.id for r in response])
+            missing_ids = list(set(ids) - set(existing_ids))
+            num_missing_ids = len(missing_ids)
+
+            if num_missing_ids > 0:
+                if not allow_missing:
+                    raise ValueError(
+                        "Found %d IDs (eg %s) that do not exist in the index"
+                        % (num_missing_ids, missing_ids[0])
+                    )
+                if warn_missing and not allow_missing:
+                    logger.warning(
+                        "Skipping %d IDs that do not exist in the index",
+                        num_missing_ids,
+                    )
+
+        self._client.delete(
+            collection_name=self.config.collection_name,
+            points_selector=qmodels.PointIdsList(points=qids),
+        )
+
+    def get_embeddings(
+        self,
+        sample_ids=None,
+        label_ids=None,
+        allow_missing=True,
+        warn_missing=False,
+    ):
+        if label_ids is not None:
+            if self.config.patches_field is None:
+                raise ValueError("This index does not support label IDs")
+
+            if sample_ids is not None:
+                logger.warning(
+                    "Ignoring sample IDs when label IDs are provided"
+                )
+
+        if sample_ids is not None and self.config.patches_field is not None:
+            (
+                embeddings,
+                sample_ids,
+                label_ids,
+                missing_ids,
+            ) = self._get_patch_embeddings_from_sample_ids(sample_ids)
+        elif self.config.patches_field is not None:
+            (
+                embeddings,
+                sample_ids,
+                label_ids,
+                missing_ids,
+            ) = self._get_patch_embeddings_from_label_ids(label_ids)
+        else:
+            (
+                embeddings,
+                sample_ids,
+                label_ids,
+                missing_ids,
+            ) = self._get_sample_embeddings(sample_ids)
+
+        num_missing_ids = len(missing_ids)
+        if num_missing_ids > 0:
+            if not allow_missing:
+                raise ValueError(
+                    "Found %d IDs (eg %s) that do not exist in the index"
+                    % (num_missing_ids, missing_ids[0])
+                )
+
+            if warn_missing:
+                logger.warning(
+                    "Skipping %d IDs that do not exist in the index",
+                    num_missing_ids,
+                )
+
+        embeddings = np.array(embeddings)
+        sample_ids = np.array(sample_ids)
+        if label_ids is not None:
+            label_ids = np.array(label_ids)
+
+        return embeddings, sample_ids, label_ids
+
+    def cleanup(self):
+        self._client.delete_collection(self.config.collection_name)
+
+    def _retrieve_points(self, qids, with_vectors=True, with_payload=True):
+        # @todo add batching?
+        return self._client.retrieve(
+            collection_name=self.config.collection_name,
+            ids=qids,
+            with_vectors=with_vectors,
+            with_payload=with_payload,
+        )
+
+    def _get_sample_embeddings(self, sample_ids):
+        if sample_ids is None:
+            sample_ids = self._get_index_ids()
+
+        response = self._retrieve_points(
+            self._to_qdrant_ids(sample_ids),
+            with_vectors=True,
+        )
+
+        found_embeddings = [r.vector for r in response]
+        found_sample_ids = self._to_fiftyone_ids([r.id for r in response])
+        missing_ids = list(set(sample_ids) - set(found_sample_ids))
+
+        return found_embeddings, found_sample_ids, None, missing_ids
+
+    def _get_patch_embeddings_from_label_ids(self, label_ids):
+        if label_ids is None:
+            label_ids = self._get_index_ids()
+
+        response = self._retrieve_points(
+            self._to_qdrant_ids(label_ids),
+            with_vectors=True,
+            with_payload=True,
+        )
+
+        found_embeddings = [r.vector for r in response]
+        found_sample_ids = [r.payload["sample_id"] for r in response]
+        found_label_ids = self._to_fiftyone_ids([r.id for r in response])
+        missing_ids = list(set(label_ids) - set(found_label_ids))
+
+        return found_embeddings, found_sample_ids, found_label_ids, missing_ids
+
+    def _get_patch_embeddings_from_sample_ids(self, sample_ids):
+        _filter = qmodels.Filter(
+            should=[
+                qmodels.FieldCondition(
+                    key="sample_id", match=qmodels.MatchValue(value=sid)
+                )
+                for sid in sample_ids
+            ]
+        )
+
+        response = self._client.scroll(
+            collection_name=self.config.collection_name,
+            scroll_filter=_filter,
+            with_vectors=True,
+            with_payload=True,
+        )[0]
+
+        found_embeddings = [r.vector for r in response]
+        found_sample_ids = [r.payload["sample_id"] for r in response]
+        found_label_ids = [r.id for r in response]
+        missing_ids = list(set(sample_ids) - set(found_sample_ids))
+
+        return found_embeddings, found_sample_ids, found_label_ids, missing_ids
+
+    def _kneighbors(
+        self,
+        query=None,
+        k=None,
+        reverse=False,
+        aggregation=None,
+        return_dists=False,
+    ):
+        if query is None:
+            raise ValueError("Qdrant does not support full index neighbors")
+
+        if reverse is True:
+            raise ValueError(
+                "Qdrant does not support least similarity queries"
+            )
+
+        if aggregation not in (None, "mean"):
+            raise ValueError("Unsupported aggregation '%s'" % aggregation)
+
+        if k is None:
+            k = self.index_size
+
+        query = self._parse_neighbors_query(query)
+        if aggregation == "mean" and query.ndim == 2:
+            query = query.mean(axis=0)
+
+        single_query = query.ndim == 1
+        if single_query:
+            query = [query]
+
+        if self.config.patches_field is not None:
+            index_ids = self.current_label_ids
+        else:
+            index_ids = self.current_sample_ids
+
+        _filter = qmodels.Filter(
+            must=[
+                qmodels.HasIdCondition(has_id=self._to_qdrant_ids(index_ids))
+            ]
+        )
+
+        ids = []
+        dists = []
+        for q in query:
+            results = self._client.search(
+                collection_name=self.config.collection_name,
+                query_vector=q,
+                query_filter=_filter,
+                with_payload=False,
+                limit=k,
+            )
+
+            ids.append(self._to_fiftyone_ids([r.id for r in results]))
+            if return_dists:
+                dists.append([r.score for r in results])
+
+        if single_query:
+            ids = ids[0]
+            if return_dists:
+                dists = dists[0]
+
+        if return_dists:
+            return ids, dists
+
+        return ids
+
+    def _parse_neighbors_query(self, query):
+        if etau.is_str(query):
+            query_ids = [query]
+            single_query = True
+        else:
+            query = np.asarray(query)
+
+            # Query by vector(s)
+            if np.issubdtype(query.dtype, np.number):
+                return query
+
+            query_ids = list(query)
+            single_query = False
+
+        # Query by ID(s)
+        qids = self._to_qdrant_ids(query_ids)
+        response = self._retrieve_points(qids, with_vectors=True)
+        query = np.array([r.vector for r in response])
+
+        if single_query:
+            query = query[0, :]
+
+        return query
+
+    def _to_qdrant_id(self, _id):
+        return _id + "00000000"
+
+    def _to_qdrant_ids(self, ids):
+        return [self._to_qdrant_id(_id) for _id in ids]
+
+    def _to_fiftyone_id(self, qid):
+        return qid.replace("-", "")[:-8]
+
+    def _to_fiftyone_ids(self, qids):
+        return [self._to_fiftyone_id(qid) for qid in qids]
+
+    @classmethod
+    def _from_dict(cls, d, samples, config, brain_key):
+        return cls(samples, config, brain_key)
diff --git a/fiftyone/brain/internal/core/similarity.py b/fiftyone/brain/internal/core/similarity.py
deleted file mode 100644
index b6a1ad80..00000000
--- a/fiftyone/brain/internal/core/similarity.py
+++ /dev/null
@@ -1,999 +0,0 @@
-"""
-Similarity methods.
-
-| Copyright 2017-2023, Voxel51, Inc.
-| `voxel51.com <https://voxel51.com/>`_
-|
-"""
-from collections import defaultdict
-import logging
-
-from bson import ObjectId
-import numpy as np
-import sklearn.metrics as skm
-import sklearn.neighbors as skn
-import sklearn.preprocessing as skp
-
-import eta.core.utils as etau
-
-from fiftyone import ViewField as F
-import fiftyone.core.brain as fob
-import fiftyone.core.context as foc
-import fiftyone.core.fields as fof
-import fiftyone.core.labels as fol
-import fiftyone.core.patches as fop
-import fiftyone.core.stages as fos
-import fiftyone.core.validation as fov
-import fiftyone.zoo as foz
-
-from fiftyone.brain.similarity import SimilarityConfig, SimilarityResults
-import fiftyone.brain.internal.core.utils as fbu
-
-
-logger = logging.getLogger(__name__)
-
-_AGGREGATIONS = {"mean": np.mean, "min": np.min, "max": np.max}
-
-_DEFAULT_MODEL = "mobilenet-v2-imagenet-torch"
-_DEFAULT_BATCH_SIZE = None
-
-_MAX_PRECOMPUTE_DISTS = 15000  # ~1.7GB to store distance matrix in-memory
-_COSINE_HACK_ATTR = "_cosine_hack"
-
-
-def compute_similarity(
-    samples,
-    patches_field,
-    embeddings,
-    brain_key,
-    metric,
-    model,
-    force_square,
-    alpha,
-    batch_size,
-    num_workers,
-    skip_failures,
-):
-    """See ``fiftyone/brain/__init__.py``."""
-
-    fov.validate_collection(samples)
-
-    if model is None and embeddings is None:
-        model = _DEFAULT_MODEL
-        if batch_size is None:
-            batch_size = _DEFAULT_BATCH_SIZE
-
-    if etau.is_str(model):
-        _model = foz.load_zoo_model(model)
-        try:
-            supports_prompts = _model.can_embed_prompts
-        except:
-            supports_prompts = None
-    else:
-        _model = model
-        supports_prompts = None
-
-    if etau.is_str(embeddings):
-        embeddings_field = fbu.parse_embeddings_field(
-            samples,
-            embeddings,
-            patches_field=patches_field,
-            allow_embedded=model is None,
-        )
-        embeddings = None
-    else:
-        embeddings_field = None
-
-    config = SimilarityConfig(
-        embeddings_field=embeddings_field,
-        model=model,
-        patches_field=patches_field,
-        metric=metric,
-        supports_prompts=supports_prompts,
-    )
-    brain_method = config.build()
-    brain_method.ensure_requirements()
-
-    if brain_key is not None:
-        brain_method.register_run(samples, brain_key)
-
-    embeddings, sample_ids, label_ids = fbu.get_embeddings(
-        samples,
-        model=_model,
-        patches_field=patches_field,
-        embeddings_field=embeddings_field,
-        embeddings=embeddings,
-        force_square=force_square,
-        alpha=alpha,
-        batch_size=batch_size,
-        num_workers=num_workers,
-        skip_failures=skip_failures,
-    )
-
-    results = SimilarityResults(
-        samples,
-        config,
-        embeddings=embeddings,
-        sample_ids=sample_ids,
-        label_ids=label_ids,
-    )
-
-    brain_method.save_run_results(samples, brain_key, results)
-
-    return results
-
-
-class Similarity(fob.BrainMethod):
-    def ensure_requirements(self):
-        pass
-
-    def get_fields(self, samples, brain_key):
-        fields = []
-        if self.config.patches_field is not None:
-            fields.append(self.config.patches_field)
-
-        return fields
-
-    def cleanup(self, samples, brain_key):
-        pass
-
-
-class NeighborsHelper(object):
-    def __init__(self, embeddings, metric):
-        self.embeddings = embeddings
-        self.metric = metric
-
-        self._initialized = False
-        self._full_dists = None
-        self._curr_keep_inds = None
-        self._curr_dists = None
-        self._curr_neighbors = None
-
-    def get_distances(self, keep_inds=None):
-        self._init()
-
-        if self._same_keep_inds(keep_inds):
-            return self._curr_dists
-
-        if keep_inds is not None:
-            dists, _ = self._build(keep_inds=keep_inds, build_neighbors=False)
-        else:
-            dists = self._full_dists
-
-        self._curr_keep_inds = keep_inds
-        self._curr_dists = dists
-        self._curr_neighbors = None
-
-        return dists
-
-    def get_neighbors(self, keep_inds=None):
-        self._init()
-
-        if self._curr_neighbors is not None and self._same_keep_inds(
-            keep_inds
-        ):
-            return self._curr_neighbors, self._curr_dists
-
-        dists, neighbors = self._build(keep_inds=keep_inds)
-
-        self._curr_keep_inds = keep_inds
-        self._curr_dists = dists
-        self._curr_neighbors = neighbors
-
-        return neighbors, dists
-
-    def _same_keep_inds(self, keep_inds):
-        if keep_inds is None and self._curr_keep_inds is None:
-            return True
-
-        if (
-            isinstance(keep_inds, np.ndarray)
-            and isinstance(self._curr_keep_inds, np.ndarray)
-            and keep_inds.size == self._curr_keep_inds.size
-            and (keep_inds == self._curr_keep_inds).all()
-        ):
-            return True
-
-        return False
-
-    def _init(self):
-        if self._initialized:
-            return
-
-        # Pre-compute all pairwise distances if number of embeddings is small
-        if len(self.embeddings) <= _MAX_PRECOMPUTE_DISTS:
-            dists, _ = self._build_precomputed(
-                self.embeddings, build_neighbors=False
-            )
-        else:
-            dists = None
-
-        self._initialized = True
-        self._full_dists = dists
-        self._curr_keep_inds = None
-        self._curr_dists = dists
-        self._curr_neighbors = None
-
-    def _build(self, keep_inds=None, build_neighbors=True):
-        # Use full distance matrix if available
-        if self._full_dists is not None:
-            if keep_inds is not None:
-                dists = self._full_dists[keep_inds, :][:, keep_inds]
-            else:
-                dists = self._full_dists
-
-            if build_neighbors:
-                neighbors = skn.NearestNeighbors(metric="precomputed")
-                neighbors.fit(dists)
-            else:
-                neighbors = None
-
-            return dists, neighbors
-
-        # Must build index
-        embeddings = self.embeddings
-
-        if keep_inds is not None:
-            embeddings = embeddings[keep_inds]
-
-        if len(embeddings) <= _MAX_PRECOMPUTE_DISTS:
-            dists, neighbors = self._build_precomputed(
-                embeddings, build_neighbors=build_neighbors
-            )
-        else:
-            dists = None
-            neighbors = self._build_graph(embeddings)
-
-        return dists, neighbors
-
-    def _build_precomputed(self, embeddings, build_neighbors=True):
-        logger.info("Generating index...")
-
-        # Center embeddings
-        embeddings = np.asarray(embeddings)
-        embeddings -= embeddings.mean(axis=0, keepdims=True)
-
-        dists = skm.pairwise_distances(embeddings, metric=self.metric)
-
-        if build_neighbors:
-            neighbors = skn.NearestNeighbors(metric="precomputed")
-            neighbors.fit(dists)
-        else:
-            neighbors = None
-
-        logger.info("Index complete")
-
-        return dists, neighbors
-
-    def _build_graph(self, embeddings):
-        logger.info(
-            "Generating neighbors graph for %d embeddings; this may take "
-            "awhile...",
-            len(embeddings),
-        )
-
-        # Center embeddings
-        embeddings = np.asarray(embeddings)
-        embeddings -= embeddings.mean(axis=0, keepdims=True)
-
-        metric = self.metric
-
-        if metric == "cosine":
-            # Nearest neighbors does not directly support cosine distance, so
-            # we approximate via euclidean distance on unit-norm embeddings
-            cosine_hack = True
-            embeddings = skp.normalize(embeddings, axis=1)
-            metric = "euclidean"
-        else:
-            cosine_hack = False
-
-        neighbors = skn.NearestNeighbors(metric=metric)
-        neighbors.fit(embeddings)
-
-        setattr(neighbors, _COSINE_HACK_ATTR, cosine_hack)
-
-        logger.info("Index complete")
-
-        return neighbors
-
-
-def plot_distances(results, bins, log, backend, **kwargs):
-    _ensure_neighbors(results)
-
-    keep_inds = results._curr_keep_inds
-    neighbors, _ = results._neighbors_helper.get_neighbors(keep_inds=keep_inds)
-    metric = results.config.metric
-    thresh = results.thresh
-
-    dists, _ = neighbors.kneighbors(n_neighbors=1)
-
-    if backend == "matplotlib":
-        return _plot_distances_mpl(dists, metric, thresh, bins, log, **kwargs)
-
-    return _plot_distances_plotly(dists, metric, thresh, bins, log, **kwargs)
-
-
-def sort_by_similarity(
-    results, query, k, reverse, aggregation, dist_field, mongo
-):
-    _ensure_neighbors(results)
-
-    samples = results.view
-    sample_ids = results.current_sample_ids
-    label_ids = results.current_label_ids
-    keep_inds = results._curr_keep_inds
-    patches_field = results.config.patches_field
-    metric = results.config.metric
-
-    selecting_samples = patches_field is None or isinstance(
-        samples, fop.PatchesView
-    )
-
-    if aggregation not in _AGGREGATIONS:
-        raise ValueError(
-            "Unsupported aggregation method '%s'. Supported values are %s"
-            % (aggregation, tuple(_AGGREGATIONS.keys()))
-        )
-
-    query = _parse_query(results, query)
-
-    if isinstance(query, np.ndarray):
-        # Query by vectors
-        query_embeddings = query
-
-        index_embeddings = results.embeddings
-        if keep_inds is not None:
-            index_embeddings = index_embeddings[keep_inds]
-
-        dists = skm.pairwise_distances(
-            index_embeddings, query_embeddings, metric=metric
-        )
-    else:
-        # Query by IDs
-        query_ids = query
-
-        # Parse query IDs (always using full index)
-        if patches_field is None:
-            ids = results.sample_ids
-        else:
-            ids = results.label_ids
-
-        bad_ids = []
-        query_inds = []
-        for query_id in query_ids:
-            _inds = np.where(ids == query_id)[0]
-            if _inds.size == 0:
-                bad_ids.append(query_id)
-            else:
-                query_inds.append(_inds[0])
-
-        if bad_ids:
-            raise ValueError(
-                "Query IDs %s were not included in this index" % bad_ids
-            )
-
-        # Perform query
-        dists = results._neighbors_helper.get_distances()
-        if dists is not None:
-            # Use pre-computed distances
-            if keep_inds is not None:
-                dists = dists[keep_inds, :]
-
-            dists = dists[:, query_inds]
-        else:
-            # Compute distances from embeddings
-            index_embeddings = results.embeddings
-            if keep_inds is not None:
-                index_embeddings = index_embeddings[keep_inds]
-
-            query_embeddings = results.embeddings[query_inds]
-            dists = skm.pairwise_distances(
-                index_embeddings, query_embeddings, metric=metric
-            )
-
-    agg_fcn = _AGGREGATIONS[aggregation]
-    dists = agg_fcn(dists, axis=1)
-
-    inds = np.argsort(dists)
-    if reverse:
-        inds = np.flip(inds)
-
-    if k is not None:
-        inds = inds[:k]
-
-    if patches_field is not None:
-        ids = label_ids
-    else:
-        ids = sample_ids
-
-    #
-    # Store query distances
-    #
-
-    if dist_field is not None:
-        if selecting_samples:
-            values = {ids[ind]: dists[ind] for ind in inds}
-            samples.set_values(dist_field, values, key_field="id")
-        else:
-            label_type, path = samples._get_label_field_path(
-                patches_field, dist_field
-            )
-            if issubclass(label_type, fol._LABEL_LIST_FIELDS):
-                samples._set_list_values_by_id(
-                    path,
-                    sample_ids[inds],
-                    label_ids[inds],
-                    dists[inds],
-                    path.rsplit(".", 1)[0],
-                )
-            else:
-                values = {sample_ids[ind]: dists[ind] for ind in inds}
-                samples.set_values(path, values, key_field="id")
-
-    #
-    # Construct sorted view
-    #
-
-    stages = []
-
-    if selecting_samples:
-        stage = fos.Select(ids[inds], ordered=True)
-        stages.append(stage)
-    else:
-        # We're sorting by object similarity but this is not a patches view, so
-        # arrange the samples in order of their first occuring label
-        result_sample_ids = _unique_no_sort(sample_ids[inds])
-        stage = fos.Select(result_sample_ids, ordered=True)
-        stages.append(stage)
-
-        if k is not None:
-            _ids = [ObjectId(_id) for _id in label_ids[inds]]
-            stage = fos.FilterLabels(patches_field, F("_id").is_in(_ids))
-            stages.append(stage)
-
-    if mongo:
-        pipeline = []
-        for stage in stages:
-            stage.validate(samples)
-            pipeline.extend(stage.to_mongo(samples))
-
-        return pipeline
-
-    view = samples
-    for stage in stages:
-        view = view.add_stage(stage)
-
-    return view
-
-
-def _parse_query(results, query):
-    if isinstance(query, np.ndarray):
-        # Query by vector(s)
-        if query.size == 0:
-            raise ValueError("At least one query vector must be provided")
-
-        if query.ndim == 1:
-            query = query[np.newaxis, :]
-
-        return query
-
-    if etau.is_str(query):
-        query = [query]
-    else:
-        query = list(query)
-
-    if not query:
-        raise ValueError("At least one query must be provided")
-
-    try:
-        ObjectId(query[0])
-        is_prompts = False
-    except:
-        is_prompts = True
-
-    if is_prompts:
-        if not results.config.supports_prompts:
-            raise ValueError(
-                "Invalid query ID '%s' (this model does not support prompts)"
-                % query[0]
-            )
-
-        if results._model is None:
-            results._model = foz.load_zoo_model(results.config.model)
-
-        return results._model.embed_prompts(query)
-
-    return query
-
-
-def find_duplicates(results, thresh, fraction):
-    _ensure_neighbors(results)
-
-    keep_inds = results._curr_keep_inds
-    embeddings = results.embeddings
-    metric = results.config.metric
-    patches_field = results.config.patches_field
-
-    neighbors, dists = results._neighbors_helper.get_neighbors(
-        keep_inds=keep_inds
-    )
-
-    if keep_inds is not None:
-        embeddings = embeddings[keep_inds]
-
-    if patches_field is not None:
-        ids = results.current_label_ids
-        logger.info("Computing duplicate patches...")
-    else:
-        ids = results.current_sample_ids
-        logger.info("Computing duplicate samples...")
-
-    num_embeddings = len(embeddings)
-
-    #
-    # Detect duplicates
-    #
-
-    if fraction is not None:
-        num_keep = int(round(min(max(0, 1.0 - fraction), 1) * num_embeddings))
-        keep, thresh = _remove_duplicates_count(
-            neighbors,
-            num_keep,
-            num_embeddings,
-            init_thresh=thresh,
-        )
-    else:
-        keep = _remove_duplicates_thresh(neighbors, thresh, num_embeddings)
-
-    unique_ids = [_id for idx, _id in enumerate(ids) if idx in keep]
-    duplicate_ids = [_id for idx, _id in enumerate(ids) if idx not in keep]
-
-    #
-    # Locate nearest non-duplicate for each duplicate
-    #
-
-    if unique_ids and duplicate_ids:
-        unique_inds = np.array(sorted(keep))
-        dup_inds = np.array(
-            [idx for idx in range(num_embeddings) if idx not in keep]
-        )
-
-        if dists is not None:
-            # Use pre-computed distances
-            _dists = dists[unique_inds, :][:, dup_inds]
-            min_inds = np.argmin(_dists, axis=0)
-            min_dists = _dists[min_inds, range(len(dup_inds))]
-        else:
-            neighbors = skn.NearestNeighbors(metric=metric)
-            neighbors.fit(embeddings[unique_inds, :])
-            min_dists, min_inds = neighbors.kneighbors(
-                embeddings[dup_inds, :], n_neighbors=1
-            )
-            min_dists = min_dists.ravel()
-            min_inds = min_inds.ravel()
-
-        neighbors_map = defaultdict(list)
-        for dup_id, min_ind, min_dist in zip(
-            duplicate_ids, min_inds, min_dists
-        ):
-            nearest_id = ids[unique_inds[min_ind]]
-            neighbors_map[nearest_id].append((dup_id, min_dist))
-
-        neighbors_map = {
-            k: sorted(v, key=lambda t: t[1]) for k, v in neighbors_map.items()
-        }
-    else:
-        neighbors_map = {}
-
-    results._thresh = thresh
-    results._unique_ids = unique_ids
-    results._duplicate_ids = duplicate_ids
-    results._neighbors_map = neighbors_map
-
-    logger.info("Duplicates computation complete")
-
-
-def find_unique(results, count):
-    _ensure_neighbors(results)
-
-    keep_inds = results._curr_keep_inds
-    neighbors, _ = results._neighbors_helper.get_neighbors(keep_inds=keep_inds)
-    patches_field = results.config.patches_field
-    num_embeddings = results.index_size
-
-    if patches_field is not None:
-        ids = results.current_label_ids
-        logger.info("Computing unique patches...")
-    else:
-        ids = results.current_sample_ids
-        logger.info("Computing unique samples...")
-
-    # Find uniques
-    keep, thresh = _remove_duplicates_count(neighbors, count, num_embeddings)
-
-    unique_ids = [_id for idx, _id in enumerate(ids) if idx in keep]
-    duplicate_ids = [_id for idx, _id in enumerate(ids) if idx not in keep]
-
-    results._thresh = thresh
-    results._unique_ids = unique_ids
-    results._duplicate_ids = duplicate_ids
-    results._neighbors_map = None
-
-    logger.info("Uniqueness computation complete")
-
-
-def duplicates_view(
-    results, type_field, id_field, dist_field, sort_by, reverse
-):
-    samples = results.view
-    patches_field = results.config.patches_field
-    neighbors_map = results.neighbors_map
-
-    if patches_field is not None and not isinstance(samples, fop.PatchesView):
-        samples = samples.to_patches(patches_field)
-
-    if sort_by == "distance":
-        key = lambda kv: min(e[1] for e in kv[1])
-    elif sort_by == "count":
-        key = lambda kv: len(kv[1])
-    else:
-        raise ValueError(
-            "Invalid sort_by='%s'; supported values are %s"
-            % (sort_by, ("distance", "count"))
-        )
-
-    existing_ids = set(samples.values("id"))
-    neighbors = [(k, v) for k, v in neighbors_map.items() if k in existing_ids]
-
-    ids = []
-    types = []
-    nearest_ids = []
-    dists = []
-    for _id, duplicates in sorted(neighbors, key=key, reverse=reverse):
-        ids.append(_id)
-        types.append("nearest")
-        nearest_ids.append(_id)
-        dists.append(0)
-
-        for dup_id, dist in duplicates:
-            ids.append(dup_id)
-            types.append("duplicate")
-            nearest_ids.append(_id)
-            dists.append(dist)
-
-    dups_view = samples.select(ids, ordered=True)
-
-    if type_field is not None:
-        dups_view._dataset.add_sample_field(type_field, fof.StringField)
-        dups_view.set_values(type_field, types)
-
-    if id_field is not None:
-        dups_view._dataset.add_sample_field(id_field, fof.StringField)
-        dups_view.set_values(id_field, nearest_ids)
-
-    if dist_field is not None:
-        dups_view._dataset.add_sample_field(dist_field, fof.FloatField)
-        dups_view.set_values(dist_field, dists)
-
-    return dups_view
-
-
-def unique_view(results):
-    samples = results.view
-    patches_field = results.config.patches_field
-    unique_ids = results.unique_ids
-
-    if patches_field is not None and not isinstance(samples, fop.PatchesView):
-        samples = samples.to_patches(patches_field)
-
-    return samples.select(unique_ids)
-
-
-def values(results, path_or_expr):
-    samples = results.view
-    patches_field = results.config.patches_field
-    if patches_field is not None:
-        ids = results.current_label_ids
-    else:
-        ids = results.current_sample_ids
-
-    return fbu.get_values(
-        samples, path_or_expr, ids, patches_field=patches_field
-    )
-
-
-def visualize_duplicates(results, visualization, backend, **kwargs):
-    visualization = _ensure_visualization(results, visualization)
-
-    samples = results.view
-    duplicate_ids = results.duplicate_ids
-    neighbors_map = results.neighbors_map
-    patches_field = results.config.patches_field
-
-    dup_ids = set(duplicate_ids)
-    nearest_ids = set(neighbors_map.keys())
-
-    with visualization.use_view(samples, allow_missing=True):
-        if patches_field is not None:
-            ids = visualization.current_label_ids
-        else:
-            ids = visualization.current_sample_ids
-
-        labels = []
-        for _id in ids:
-            if _id in dup_ids:
-                label = "duplicate"
-            elif _id in nearest_ids:
-                label = "nearest"
-            else:
-                label = "unique"
-
-            labels.append(label)
-
-        if backend == "plotly":
-            kwargs["edges"] = _build_edges(ids, neighbors_map)
-            kwargs["edges_title"] = "neighbors"
-            kwargs["labels_title"] = "type"
-
-        return visualization.visualize(
-            labels=labels,
-            classes=["unique", "nearest", "duplicate"],
-            backend=backend,
-            **kwargs,
-        )
-
-
-def visualize_unique(results, visualization, backend, **kwargs):
-    visualization = _ensure_visualization(results, visualization)
-
-    samples = results.view
-    unique_ids = results.unique_ids
-    patches_field = results.config.patches_field
-
-    unique_ids = set(unique_ids)
-
-    with visualization.use_view(samples, allow_missing=True):
-        if patches_field is not None:
-            ids = visualization.current_label_ids
-        else:
-            ids = visualization.current_sample_ids
-
-        labels = []
-        for _id in ids:
-            if _id in unique_ids:
-                label = "unique"
-            else:
-                label = "other"
-
-            labels.append(label)
-
-        return visualization.visualize(
-            labels=labels,
-            classes=["other", "unique"],
-            backend=backend,
-            **kwargs,
-        )
-
-
-def _unique_no_sort(values):
-    seen = set()
-    return [v for v in values if v not in seen and not seen.add(v)]
-
-
-def _ensure_neighbors(results):
-    if results._neighbors_helper is not None:
-        return
-
-    embeddings = results.embeddings
-    metric = results.config.metric
-    results._neighbors_helper = NeighborsHelper(embeddings, metric)
-
-
-def _ensure_visualization(results, visualization):
-    if visualization is not None:
-        return visualization
-
-    import fiftyone.brain as fb
-
-    samples = results._samples
-    embeddings = results.embeddings
-    patches_field = results.config.patches_field
-
-    if embeddings.shape[1] in {2, 3}:
-        config = fb.VisualizationConfig(
-            patches_field=patches_field, num_dims=2
-        )
-        return fb.VisualizationResults(samples, config, embeddings)
-
-    return fb.compute_visualization(
-        samples,
-        patches_field=patches_field,
-        embeddings=embeddings,
-        num_dims=2,
-        seed=51,
-        verbose=True,
-    )
-
-
-def _build_edges(ids, neighbors_map):
-    inds_map = {_id: idx for idx, _id in enumerate(ids)}
-
-    edges = []
-    for nearest_id, duplicates in neighbors_map.items():
-        nearest_ind = inds_map[nearest_id]
-        for dup_id, _ in duplicates:
-            dup_ind = inds_map[dup_id]
-            edges.append((dup_ind, nearest_ind))
-
-    return np.array(edges)
-
-
-def _remove_duplicates_thresh(neighbors, thresh, num_embeddings):
-    # When not using brute force, we approximate cosine distance by computing
-    # Euclidean distance on unit-norm embeddings. ED = sqrt(2 * CD), so we need
-    # to scale the threshold appropriately
-    if getattr(neighbors, _COSINE_HACK_ATTR, False):
-        thresh = np.sqrt(2.0 * thresh)
-
-    inds = neighbors.radius_neighbors(radius=thresh, return_distance=False)
-
-    keep = set(range(num_embeddings))
-    for ind in range(num_embeddings):
-        if ind in keep:
-            keep -= {i for i in inds[ind] if i > ind}
-
-    return keep
-
-
-def _remove_duplicates_count(
-    neighbors, num_keep, num_embeddings, init_thresh=None
-):
-    if init_thresh is not None:
-        thresh = init_thresh
-    else:
-        thresh = 1
-
-    if num_keep <= 0:
-        logger.info("threshold: -, kept: %d, target: %d", num_keep, num_keep)
-        return set(), None
-
-    if num_keep >= num_embeddings:
-        logger.info("threshold: -, kept: %d, target: %d", num_keep, num_keep)
-        return set(range(num_embeddings)), None
-
-    thresh_lims = [0, None]
-    num_target = num_keep
-    num_keep = -1
-
-    while True:
-        keep = _remove_duplicates_thresh(neighbors, thresh, num_embeddings)
-        num_keep_last = num_keep
-        num_keep = len(keep)
-
-        logger.info(
-            "threshold: %f, kept: %d, target: %d",
-            thresh,
-            num_keep,
-            num_target,
-        )
-
-        if num_keep == num_target or (
-            num_keep == num_keep_last
-            and thresh_lims[1] is not None
-            and thresh_lims[1] - thresh_lims[0] < 1e-6
-        ):
-            break
-
-        if num_keep < num_target:
-            # Need to decrease threshold
-            thresh_lims[1] = thresh
-            thresh = 0.5 * (thresh_lims[0] + thresh)
-        else:
-            # Need to increase threshold
-            thresh_lims[0] = thresh
-            if thresh_lims[1] is not None:
-                thresh = 0.5 * (thresh + thresh_lims[1])
-            else:
-                thresh *= 2
-
-    return keep, thresh
-
-
-def _plot_distances_plotly(dists, metric, thresh, bins, log, **kwargs):
-    import plotly.graph_objects as go
-    import fiftyone.core.plots.plotly as fopl
-
-    counts, edges = np.histogram(dists, bins=bins)
-    left_edges = edges[:-1]
-    widths = edges[1:] - edges[:-1]
-    customdata = np.stack((edges[:-1], edges[1:]), axis=1)
-
-    hover_lines = [
-        "<b>count: %{y}</b>",
-        "distance: [%{customdata[0]:.2f}, %{customdata[1]:.2f}]",
-    ]
-    hovertemplate = "<br>".join(hover_lines) + "<extra></extra>"
-
-    bar = go.Bar(
-        x=left_edges,
-        y=counts,
-        width=widths,
-        customdata=customdata,
-        offset=0,
-        marker_color="#FF6D04",
-        hovertemplate=hovertemplate,
-        showlegend=False,
-    )
-
-    traces = [bar]
-
-    if thresh is not None:
-        line = go.Scatter(
-            x=[thresh, thresh],
-            y=[0, max(counts)],
-            mode="lines",
-            line=dict(color="#17191C", width=3),
-            hovertemplate="<b>thresh: %{x}</b><extra></extra>",
-            showlegend=False,
-        )
-        traces.append(line)
-
-    figure = go.Figure(traces)
-
-    figure.update_layout(
-        xaxis_title="nearest neighbor distance (%s)" % metric,
-        yaxis_title="count",
-        hovermode="x",
-        yaxis_rangemode="tozero",
-    )
-
-    if log:
-        figure.update_layout(yaxis_type="log")
-
-    figure.update_layout(**fopl._DEFAULT_LAYOUT)
-    figure.update_layout(**kwargs)
-
-    if foc.is_jupyter_context():
-        figure = fopl.PlotlyNotebookPlot(figure)
-
-    return figure
-
-
-def _plot_distances_mpl(
-    dists, metric, thresh, bins, log, ax=None, figsize=None, **kwargs
-):
-    import matplotlib.pyplot as plt
-
-    if ax is None:
-        fig, ax = plt.subplots()
-    else:
-        fig = ax.figure
-
-    counts, edges = np.histogram(dists, bins=bins)
-    left_edges = edges[:-1]
-    widths = edges[1:] - edges[:-1]
-
-    ax.bar(
-        left_edges,
-        counts,
-        width=widths,
-        align="edge",
-        color="#FF6D04",
-        **kwargs,
-    )
-
-    if thresh is not None:
-        ax.vlines(thresh, 0, max(counts), color="#17191C", linewidth=3)
-
-    if log:
-        ax.set_yscale("log")
-
-    ax.set_xlabel("nearest neighbor distance (%s)" % metric)
-    ax.set_ylabel("count")
-
-    if figsize is not None:
-        fig.set_size_inches(*figsize)
-
-    plt.tight_layout()
-
-    return fig
diff --git a/fiftyone/brain/internal/core/sklearn.py b/fiftyone/brain/internal/core/sklearn.py
new file mode 100644
index 00000000..a63f17d6
--- /dev/null
+++ b/fiftyone/brain/internal/core/sklearn.py
@@ -0,0 +1,839 @@
+"""
+Sklearn similarity backend.
+
+| Copyright 2017-2023, Voxel51, Inc.
+| `voxel51.com <https://voxel51.com/>`_
+|
+"""
+import logging
+
+import numpy as np
+import sklearn.metrics as skm
+import sklearn.neighbors as skn
+import sklearn.preprocessing as skp
+
+import eta.core.utils as etau
+
+from fiftyone.brain.similarity import (
+    DuplicatesMixin,
+    SimilarityConfig,
+    Similarity,
+    SimilarityIndex,
+)
+import fiftyone.brain.internal.core.utils as fbu
+
+
+logger = logging.getLogger(__name__)
+
+_AGGREGATIONS = {
+    "mean": np.mean,
+    "post-mean": np.mean,
+    "post-min": np.min,
+    "post-max": np.max,
+}
+
+_MAX_PRECOMPUTE_DISTS = 15000  # ~1.7GB to store distance matrix in-memory
+_COSINE_HACK_ATTR = "_cosine_hack"
+
+
+class SklearnSimilarityConfig(SimilarityConfig):
+    """Configuration for the sklearn similarity backend.
+
+    Args:
+        embeddings_field (None): the sample field containing the embeddings,
+            if one was provided
+        model (None): the :class:`fiftyone.core.models.Model` or name of the
+            zoo model that was used to compute embeddings, if known
+        patches_field (None): the sample field defining the patches being
+            analyzed, if any
+        supports_prompts (None): whether this run supports prompt queries
+        metric ("cosine"): the embedding distance metric to use. See
+            ``sklearn.metrics.pairwise_distance`` for supported values
+    """
+
+    def __init__(
+        self,
+        embeddings_field=None,
+        model=None,
+        patches_field=None,
+        supports_prompts=None,
+        metric="cosine",
+        **kwargs,
+    ):
+        super().__init__(
+            embeddings_field=embeddings_field,
+            model=model,
+            patches_field=patches_field,
+            supports_prompts=supports_prompts,
+            **kwargs,
+        )
+        self.metric = metric
+
+    @property
+    def method(self):
+        return "sklearn"
+
+    @property
+    def max_k(self):
+        return None
+
+    @property
+    def supports_least_similarity(self):
+        return True
+
+    @property
+    def supported_aggregations(self):
+        return tuple(_AGGREGATIONS.keys())
+
+
+class SklearnSimilarity(Similarity):
+    """Sklearn similarity factory.
+
+    Args:
+        config: an :class:`SklearnSimilarityConfig`
+    """
+
+    def initialize(self, samples, brain_key):
+        return SklearnSimilarityIndex(
+            samples, self.config, brain_key, backend=self
+        )
+
+
+class SklearnSimilarityIndex(SimilarityIndex, DuplicatesMixin):
+    """Class for interacting with sklearn similarity indexes.
+
+    Args:
+        samples: the :class:`fiftyone.core.collections.SampleCollection` used
+        config: the :class:`SklearnSimilarityConfig` used
+        brain_key: the brain key
+        embeddings (None): a ``num_embeddings x num_dims`` array of embeddings
+        sample_ids (None): a ``num_embeddings`` array of sample IDs
+        label_ids (None): a ``num_embeddings`` array of label IDs, if
+            applicable
+        backend (None): a :class:`SklearnSimilarity` instance
+    """
+
+    def __init__(
+        self,
+        samples,
+        config,
+        brain_key,
+        embeddings=None,
+        sample_ids=None,
+        label_ids=None,
+        backend=None,
+    ):
+        embeddings, sample_ids, label_ids = self._parse_data(
+            samples,
+            config,
+            embeddings=embeddings,
+            sample_ids=sample_ids,
+            label_ids=label_ids,
+        )
+
+        self._embeddings = embeddings
+        self._sample_ids = sample_ids
+        self._label_ids = label_ids
+
+        self._ids_to_inds = None
+        self._curr_ids_to_inds = None
+        self._neighbors_helper = None
+
+        SimilarityIndex.__init__(
+            self, samples, config, brain_key, backend=backend
+        )
+        DuplicatesMixin.__init__(self)
+
+    @property
+    def embeddings(self):
+        return self._embeddings
+
+    @property
+    def sample_ids(self):
+        return self._sample_ids
+
+    @property
+    def label_ids(self):
+        return self._label_ids
+
+    @property
+    def total_index_size(self):
+        return len(self._sample_ids)
+
+    def add_to_index(
+        self,
+        embeddings,
+        sample_ids,
+        label_ids=None,
+        overwrite=True,
+        allow_existing=True,
+        warn_existing=False,
+    ):
+        _sample_ids, _label_ids, ii, jj = fbu.add_ids(
+            sample_ids,
+            label_ids,
+            self._sample_ids,
+            self._label_ids,
+            patches_field=self.config.patches_field,
+            overwrite=overwrite,
+            allow_existing=allow_existing,
+            warn_existing=warn_existing,
+        )
+
+        if ii.size == 0:
+            return
+
+        _embeddings = embeddings[ii, :]
+
+        if self.config.embeddings_field is not None:
+            fbu.add_embeddings(
+                self._samples,
+                _embeddings,
+                _sample_ids,
+                _label_ids,
+                self.config.embeddings_field,
+                patches_field=self.config.patches_field,
+            )
+
+        _e = self._embeddings
+
+        n = _e.shape[0]
+        if n == 0:
+            _e = np.empty((0, embeddings.shape[1]), dtype=embeddings.dtype)
+
+        d = _e.shape[1]
+        m = jj[-1] - n + 1
+
+        if m > 0:
+            if _e.size > 0:
+                _e = np.concatenate((_e, np.empty((m, d), dtype=_e.dtype)))
+            else:
+                _e = np.empty_like(_embeddings)
+
+        _e[jj, :] = _embeddings
+
+        self._embeddings = _e
+        self._sample_ids = _sample_ids
+        self._label_ids = _label_ids
+
+    def remove_from_index(
+        self,
+        sample_ids=None,
+        label_ids=None,
+        allow_missing=True,
+        warn_missing=False,
+    ):
+        _sample_ids, _label_ids, rm_inds = fbu.remove_ids(
+            sample_ids,
+            label_ids,
+            self._sample_ids,
+            self._label_ids,
+            patches_field=self.config.patches_field,
+            allow_missing=allow_missing,
+            warn_missing=warn_missing,
+        )
+
+        if rm_inds.size == 0:
+            return
+
+        if self.config.embeddings_field is not None:
+            fbu.remove_embeddings(
+                self._samples,
+                self.config.embeddings_field,
+                sample_ids=_sample_ids,
+                label_ids=_label_ids,
+                patches_field=self.config.patches_field,
+            )
+
+        _embeddings = np.delete(self._embeddings, rm_inds)
+
+        self._embeddings = _embeddings
+        self._sample_ids = _sample_ids
+        self._label_ids = _label_ids
+
+    def use_view(self, *args, **kwargs):
+        self._curr_ids_to_inds = None
+        return super().use_view(*args, **kwargs)
+
+    def get_embeddings(
+        self,
+        sample_ids=None,
+        label_ids=None,
+        allow_missing=True,
+        warn_missing=False,
+    ):
+        if label_ids is not None:
+            if self.config.patches_field is None:
+                raise ValueError("This index does not support label IDs")
+
+            if sample_ids is not None:
+                logger.warning(
+                    "Ignoring sample IDs when label IDs are provided"
+                )
+
+            inds = _get_inds(
+                label_ids,
+                self.label_ids,
+                "label",
+                allow_missing,
+                warn_missing,
+            )
+
+            embeddings = self._embeddings[inds, :]
+            sample_ids = self.sample_ids[inds]
+            label_ids = np.asarray(label_ids)
+        elif sample_ids is not None:
+            if etau.is_str(sample_ids):
+                sample_ids = [sample_ids]
+
+            if self.config.patches_field is not None:
+                sample_ids = set(sample_ids)
+                bools = [_id in sample_ids for _id in self.sample_ids]
+                inds = np.nonzero(bools)[0]
+            else:
+                inds = _get_inds(
+                    sample_ids,
+                    self.sample_ids,
+                    "sample",
+                    allow_missing,
+                    warn_missing,
+                )
+
+            embeddings = self._embeddings[inds, :]
+            sample_ids = self.sample_ids[inds]
+            if self.config.patches_field is not None:
+                label_ids = self.label_ids[inds]
+            else:
+                label_ids = None
+        else:
+            embeddings = self._embeddings.copy()
+            sample_ids = self.sample_ids.copy()
+            if self.config.patches_field is not None:
+                label_ids = self.label_ids.copy()
+            else:
+                label_ids = None
+
+        return embeddings, sample_ids, label_ids
+
+    def reload(self):
+        if self.config.embeddings_field is not None:
+            embeddings, sample_ids, label_ids = self._parse_data(
+                self._samples, self.config
+            )
+
+            self._embeddings = embeddings
+            self._sample_ids = sample_ids
+            self._label_ids = label_ids
+
+            self._ids_to_inds = None
+            self._curr_ids_to_inds = None
+            self._neighbors_helper = None
+
+        super().reload()
+
+    def cleanup(self):
+        pass
+
+    def attributes(self):
+        attrs = super().attributes()
+
+        if self.config.embeddings_field is None:
+            attrs.extend(["embeddings", "sample_ids", "label_ids"])
+
+        return attrs
+
+    def _kneighbors(
+        self,
+        query=None,
+        k=None,
+        reverse=False,
+        aggregation=None,
+        return_dists=False,
+    ):
+        if aggregation is not None:
+            return self._kneighbors_aggregate(
+                query, k, reverse, aggregation, return_dists
+            )
+
+        (
+            query,
+            query_inds,
+            full_index,
+            single_query,
+        ) = self._parse_neighbors_query(query)
+
+        can_use_dists = full_index or query_inds is not None
+        neighbors, dists = self._get_neighbors(can_use_dists=can_use_dists)
+
+        if dists is not None:
+            # Use pre-computed distances
+            if query_inds is not None:
+                _dists = dists[query_inds, :]
+                _rows = range(len(query_inds))
+            else:
+                _dists = dists
+                _rows = range(dists.shape[0])
+
+            inds = np.argmin(_dists, axis=1)
+            if return_dists:
+                dists = _dists[_rows, inds]
+            else:
+                dists = None
+        else:
+            if return_dists:
+                dists, inds = neighbors.kneighbors(
+                    X=query, n_neighbors=k, return_distance=True
+                )
+            else:
+                inds = neighbors.kneighbors(
+                    X=query, n_neighbors=k, return_distance=False
+                )
+                dists = None
+
+        return self._format_output(
+            inds, dists, full_index, single_query, return_dists
+        )
+
+    def _radius_neighbors(self, query=None, thresh=None, return_dists=False):
+        (
+            query,
+            query_inds,
+            full_index,
+            single_query,
+        ) = self._parse_neighbors_query(query)
+
+        can_use_dists = full_index or query_inds is not None
+        neighbors, dists = self._get_neighbors(can_use_dists=can_use_dists)
+
+        # When not using brute force, we approximate cosine distance by
+        # computing Euclidean distance on unit-norm embeddings.
+        # ED = sqrt(2 * CD), so we need to scale the threshold appropriately
+        if getattr(neighbors, _COSINE_HACK_ATTR, False):
+            thresh = np.sqrt(2.0 * thresh)
+
+        if dists is not None:
+            # Use pre-computed distances
+            if query_inds is not None:
+                _dists = dists[query_inds, :]
+            else:
+                _dists = dists
+
+            inds = [np.nonzero(d <= thresh)[0] for d in _dists]
+            if return_dists:
+                dists = [d[i] for i, d in zip(inds, _dists)]
+            else:
+                dists = None
+        else:
+            if return_dists:
+                dists, inds = neighbors.radius_neighbors(
+                    X=query, radius=thresh, return_distance=True
+                )
+            else:
+                dists = None
+                inds = neighbors.radius_neighbors(
+                    X=query, radius=thresh, return_distance=False
+                )
+
+        return self._format_output(
+            inds, dists, full_index, single_query, return_dists
+        )
+
+    def _kneighbors_aggregate(
+        self, query, k, reverse, aggregation, return_dists
+    ):
+        if query is None:
+            raise ValueError("Full index queries do not support aggregation")
+
+        if aggregation not in _AGGREGATIONS:
+            raise ValueError(
+                "Unsupported aggregation method '%s'. Supported values are %s"
+                % (aggregation, tuple(_AGGREGATIONS.keys()))
+            )
+
+        query, query_inds, _, _ = self._parse_neighbors_query(query)
+
+        # Pre-aggregation
+        if aggregation == "mean":
+            if query.shape[0] > 1:
+                query = query.mean(axis=0, keepdims=True)
+                query_inds = None
+
+            aggregation = None
+
+        can_use_dists = query_inds is not None
+        _, dists = self._get_neighbors(
+            can_use_neighbors=False, can_use_dists=can_use_dists
+        )
+
+        if dists is not None:
+            # Use pre-computed distances
+            dists = dists[query_inds, :]
+        else:
+            keep_inds = self._current_inds
+            index_embeddings = self._embeddings
+            if keep_inds is not None:
+                index_embeddings = index_embeddings[keep_inds]
+
+            dists = skm.pairwise_distances(
+                query, index_embeddings, metric=self.config.metric
+            )
+
+        # Post-aggregation
+        if aggregation is not None:
+            agg_fcn = _AGGREGATIONS[aggregation]
+            dists = agg_fcn(dists, axis=0)
+        else:
+            dists = dists[0, :]
+
+        inds = np.argsort(dists)
+        if reverse:
+            inds = np.flip(inds)
+
+        if k is not None:
+            inds = inds[:k]
+
+        if self.config.patches_field is not None:
+            ids = self.current_label_ids
+        else:
+            ids = self.current_sample_ids
+
+        if return_dists:
+            return ids[inds], dists[inds]
+
+        return ids[inds]
+
+    def _parse_neighbors_query(self, query):
+        # Full index
+        if query is None:
+            return None, None, True, False
+
+        if etau.is_str(query):
+            query_ids = [query]
+            single_query = True
+        else:
+            query = np.asarray(query)
+
+            # Query vector(s)
+            if np.issubdtype(query.dtype, np.number):
+                single_query = query.ndim == 1
+                if single_query:
+                    query = query[np.newaxis, :]
+
+                return query, None, False, single_query
+
+            query_ids = list(query)
+            single_query = False
+
+        # Retrieve indices into active `dists` matrix, if possible
+        ids_to_inds = self._get_ids_to_inds(full=False)
+        query_inds = []
+        for _id in query_ids:
+            _ind = ids_to_inds.get(_id, None)
+            if _ind is not None:
+                query_inds.append(_ind)
+            else:
+                # At least one query ID is not in the active index
+                query_inds = None
+                break
+
+        # Retrieve embeddings
+        ids_to_inds = self._get_ids_to_inds(full=True)
+        inds = []
+        bad_ids = []
+        for _id in query_ids:
+            _ind = ids_to_inds.get(_id, None)
+            if _ind is not None:
+                inds.append(_ind)
+            else:
+                bad_ids.append(_id)
+
+        inds = np.array(inds)
+
+        if bad_ids:
+            raise ValueError(
+                "Query IDs %s do not exist in this index" % bad_ids
+            )
+
+        query = self._embeddings[inds, :]
+
+        if query_inds is not None:
+            query_inds = np.array(query_inds)
+
+        return query, query_inds, False, single_query
+
+    def _get_ids_to_inds(self, full=False):
+        if full:
+            if self._ids_to_inds is None:
+                if self.config.patches_field is not None:
+                    ids = self.label_ids
+                else:
+                    ids = self.sample_ids
+
+                self._ids_to_inds = {_id: i for i, _id in enumerate(ids)}
+
+            return self._ids_to_inds
+
+        if self._curr_ids_to_inds is None:
+            if self.config.patches_field is not None:
+                ids = self.current_label_ids
+            else:
+                ids = self.current_sample_ids
+
+            self._curr_ids_to_inds = {_id: i for i, _id in enumerate(ids)}
+
+        return self._curr_ids_to_inds
+
+    def _get_neighbors(self, can_use_neighbors=True, can_use_dists=True):
+        if self._neighbors_helper is None:
+            self._neighbors_helper = NeighborsHelper(
+                self._embeddings, self.config.metric
+            )
+
+        return self._neighbors_helper.get_neighbors(
+            keep_inds=self._current_inds,
+            can_use_neighbors=can_use_neighbors,
+            can_use_dists=can_use_dists,
+        )
+
+    def _format_output(
+        self, inds, dists, full_index, single_query, return_dists
+    ):
+        if full_index:
+            return (inds, dists) if return_dists else inds
+
+        if self.config.patches_field is not None:
+            index_ids = self.current_label_ids
+        else:
+            index_ids = self.current_sample_ids
+
+        ids = [[index_ids[i] for i in _inds] for _inds in inds]
+        if return_dists:
+            dists = [list(d) for d in dists]
+
+        if single_query:
+            ids = ids[0]
+            if return_dists:
+                dists = dists[0]
+
+        return (ids, dists) if return_dists else ids
+
+    @staticmethod
+    def _parse_data(
+        samples,
+        config,
+        embeddings=None,
+        sample_ids=None,
+        label_ids=None,
+    ):
+        if embeddings is None:
+            embeddings, sample_ids, label_ids = fbu.get_embeddings(
+                samples._dataset,
+                patches_field=config.patches_field,
+                embeddings_field=config.embeddings_field,
+            )
+        elif sample_ids is None:
+            sample_ids, label_ids = fbu.get_ids(
+                samples,
+                patches_field=config.patches_field,
+                data=embeddings,
+                data_type="embeddings",
+            )
+
+        return embeddings, sample_ids, label_ids
+
+    @classmethod
+    def _from_dict(cls, d, samples, config, brain_key):
+        embeddings = d.get("embeddings", None)
+        if embeddings is not None:
+            embeddings = np.array(embeddings)
+
+        sample_ids = d.get("sample_ids", None)
+        if sample_ids is not None:
+            sample_ids = np.array(sample_ids)
+
+        label_ids = d.get("label_ids", None)
+        if label_ids is not None:
+            label_ids = np.array(label_ids)
+
+        return cls(
+            samples,
+            config,
+            brain_key,
+            embeddings=embeddings,
+            sample_ids=sample_ids,
+            label_ids=label_ids,
+        )
+
+
+class NeighborsHelper(object):
+
+    _UNAVAILABLE = "UNAVAILABLE"
+
+    def __init__(self, embeddings, metric):
+        self.embeddings = embeddings
+        self.metric = metric
+
+        self._initialized = False
+        self._full_dists = None
+
+        self._curr_keep_inds = None
+        self._curr_neighbors = None
+        self._curr_dists = None
+
+    def get_neighbors(
+        self,
+        keep_inds=None,
+        can_use_neighbors=True,
+        can_use_dists=True,
+    ):
+        iokay = self._same_keep_inds(keep_inds)
+        nokay = not can_use_neighbors or self._curr_neighbors is not None
+        dokay = not can_use_dists or self._curr_dists is not None
+
+        if iokay and nokay and dokay:
+            neighbors = self._curr_neighbors
+            dists = self._curr_dists
+        else:
+            neighbors, dists = self._build(
+                keep_inds=keep_inds,
+                can_use_neighbors=can_use_neighbors,
+                can_use_dists=can_use_dists,
+            )
+
+            if not iokay:
+                self._curr_keep_inds = keep_inds
+
+            if self._curr_neighbors is None or not iokay:
+                self._curr_neighbors = neighbors
+
+            if self._curr_dists is None or not iokay:
+                self._curr_dists = dists
+
+        if not can_use_neighbors or neighbors is self._UNAVAILABLE:
+            neighbors = None
+
+        if not can_use_dists or dists is self._UNAVAILABLE:
+            dists = None
+
+        return neighbors, dists
+
+    def _same_keep_inds(self, keep_inds):
+        # This handles either argument being None
+        return np.array_equal(keep_inds, self._curr_keep_inds)
+
+    def _build(
+        self, keep_inds=None, can_use_neighbors=True, can_use_dists=True
+    ):
+        if can_use_dists:
+            if (
+                self._full_dists is None
+                and len(self.embeddings) <= _MAX_PRECOMPUTE_DISTS
+            ):
+                self._full_dists = self._build_dists(self.embeddings)
+
+            if self._full_dists is not None:
+                if keep_inds is not None:
+                    dists = self._full_dists[keep_inds, :][:, keep_inds]
+                else:
+                    dists = self._full_dists
+            elif (
+                keep_inds is not None
+                and len(keep_inds) <= _MAX_PRECOMPUTE_DISTS
+            ):
+                dists = self._build_dists(self.embeddings[keep_inds])
+            else:
+                dists = self._UNAVAILABLE
+        else:
+            dists = None
+
+        if can_use_neighbors:
+            if not isinstance(dists, np.ndarray):
+                embeddings = self.embeddings
+                if keep_inds is not None:
+                    embeddings = embeddings[keep_inds]
+
+                neighbors = self._build_neighbors(embeddings)
+            else:
+                neighbors = self._UNAVAILABLE
+        else:
+            neighbors = None
+
+        return neighbors, dists
+
+    def _build_dists(self, embeddings):
+        logger.info("Generating index for %d embeddings...", len(embeddings))
+
+        # Center embeddings
+        embeddings = np.asarray(embeddings)
+        embeddings -= embeddings.mean(axis=0, keepdims=True)
+
+        dists = skm.pairwise_distances(embeddings, metric=self.metric)
+
+        logger.info("Index complete")
+
+        return dists
+
+    def _build_neighbors(self, embeddings):
+        logger.info(
+            "Generating neighbors graph for %d embeddings...",
+            len(embeddings),
+        )
+
+        # Center embeddings
+        embeddings = np.asarray(embeddings)
+        embeddings -= embeddings.mean(axis=0, keepdims=True)
+
+        metric = self.metric
+
+        if metric == "cosine":
+            # Nearest neighbors does not directly support cosine distance, so
+            # we approximate via euclidean distance on unit-norm embeddings
+            cosine_hack = True
+            embeddings = skp.normalize(embeddings, axis=1)
+            metric = "euclidean"
+        else:
+            cosine_hack = False
+
+        neighbors = skn.NearestNeighbors(metric=metric)
+        neighbors.fit(embeddings)
+
+        setattr(neighbors, _COSINE_HACK_ATTR, cosine_hack)
+
+        logger.info("Index complete")
+
+        return neighbors
+
+
+def _get_inds(ids, index_ids, ftype, allow_missing, warn_missing):
+    if etau.is_str(ids):
+        ids = [ids]
+
+    ids_map = {_id: i for i, _id in enumerate(index_ids)}
+
+    inds = []
+    bad_ids = []
+
+    for _id in ids:
+        idx = ids_map.get(_id, None)
+        if idx is not None:
+            inds.append(idx)
+        else:
+            bad_ids.append(_id)
+
+    num_missing = len(bad_ids)
+
+    if num_missing > 0:
+        if not allow_missing:
+            raise ValueError(
+                "Found %d %s IDs (eg '%s') that are not present in the index"
+                % (num_missing, ftype, bad_ids[0])
+            )
+
+        if warn_missing:
+            logger.warning(
+                "Ignoring %d %s IDs that are not present in the index",
+                num_missing,
+                ftype,
+            )
+
+    return np.array(inds)
diff --git a/fiftyone/brain/internal/core/utils.py b/fiftyone/brain/internal/core/utils.py
index b63c0dab..d676e62a 100644
--- a/fiftyone/brain/internal/core/utils.py
+++ b/fiftyone/brain/internal/core/utils.py
@@ -7,11 +7,14 @@
 """
 import itertools
 import logging
+import random
+import string
 
 import numpy as np
 
 import eta.core.utils as etau
 
+import fiftyone.brain as fob
 import fiftyone.core.fields as fof
 import fiftyone.core.labels as fol
 import fiftyone.core.patches as fop
@@ -22,6 +25,154 @@
 logger = logging.getLogger(__name__)
 
 
+def parse_data(
+    samples,
+    patches_field=None,
+    data=None,
+    data_type="embeddings",
+    allow_missing=True,
+    warn_missing=True,
+):
+    if isinstance(data, fob.SimilarityIndex):
+        return get_embeddings_from_index(
+            samples,
+            data,
+            patches_field=None,
+            allow_missing=True,
+            warn_missing=True,
+        )
+
+    if patches_field is None:
+        if isinstance(data, dict):
+            sample_ids, data = zip(*data.items())
+            return np.array(data), np.array(sample_ids), None
+
+        sample_ids, _ = get_ids(samples, data=data, data_type=data_type)
+        return data, sample_ids, None
+
+    if isinstance(data, dict):
+        value = next(iter(data.values()), None)
+        if isinstance(value, np.ndarray) and value.ndim == 1:
+            label_ids, data = zip(*data.items())
+            return _parse_label_data(
+                samples,
+                patches_field,
+                label_ids,
+                data,
+                data_type,
+                allow_missing,
+                warn_missing,
+            )
+
+    sample_ids, label_ids = get_ids(
+        samples,
+        patches_field=patches_field,
+        data=data,
+        data_type=data_type,
+    )
+
+    return data, sample_ids, label_ids
+
+
+def _parse_label_data(
+    samples,
+    patches_field,
+    label_ids,
+    data,
+    data_type,
+    allow_missing,
+    warn_missing,
+):
+    if samples._is_patches:
+        sample_id_path = "sample_id"
+    else:
+        sample_id_path = "id"
+
+    label_type, label_id_path = samples._get_label_field_path(
+        patches_field, "id"
+    )
+    is_list_field = issubclass(label_type, fol._LABEL_LIST_FIELDS)
+
+    ref_sample_ids, ref_label_ids = samples._dataset.values(
+        [sample_id_path, label_id_path]
+    )
+
+    if is_list_field:
+        ids_map = {}
+        for _sample_id, _lids in zip(ref_sample_ids, ref_label_ids):
+            if _lids:
+                for _label_id in _lids:
+                    ids_map[_label_id] = _sample_id
+    else:
+        ids_map = dict(zip(ref_label_ids, ref_sample_ids))
+
+    _data = []
+    _sample_ids = []
+    _label_ids = []
+    _missing_ids = []
+    for _lid, _d in zip(label_ids, data):
+        _sid = ids_map.get(_lid, None)
+        if _sid is not None:
+            _data.append(_d)
+            _sample_ids.append(_sid)
+            _label_ids.append(_lid)
+        else:
+            _missing_ids.append(_lid)
+
+    num_missing = len(_missing_ids)
+    if num_missing > 0:
+        if not allow_missing:
+            raise ValueError(
+                "Unable to retrieve sample IDs for %d label IDs (eg %s)"
+                % (num_missing, _missing_ids[0])
+            )
+
+        if warn_missing:
+            logger.warning(
+                "Ignoring %s for %d label IDs (eg %s) for which sample IDs "
+                "could not be retrieved",
+                data_type,
+                num_missing,
+                _missing_ids[0],
+            )
+
+    return np.array(_data), np.array(_sample_ids), np.array(_label_ids)
+
+
+def get_embeddings_from_index(
+    samples,
+    similarity_index,
+    patches_field=None,
+    allow_missing=True,
+    warn_missing=True,
+):
+    if patches_field is None:
+        if samples._is_patches:
+            sample_id_path = "sample_id"
+        else:
+            sample_id_path = "id"
+
+        sample_ids = samples.values(sample_id_path)
+        label_ids = None
+    else:
+        if samples._is_patches:
+            label_id_path = "id"
+        else:
+            _, label_id_path = samples._get_label_field_path(
+                patches_field, "id"
+            )
+
+        sample_ids = None
+        label_ids = samples.values(label_id_path, unwind=True)
+
+    return similarity_index.get_embeddings(
+        sample_ids=sample_ids,
+        label_ids=label_ids,
+        allow_missing=allow_missing,
+        warn_missing=warn_missing,
+    )
+
+
 def get_ids(
     samples,
     patches_field=None,
@@ -80,6 +231,9 @@ def filter_ids(
         else:
             sample_ids = np.array(samples.values("id"))
 
+        if index_sample_ids is None:
+            return sample_ids, None, None, None
+
         keep_inds, good_inds, bad_ids = _parse_ids(
             sample_ids,
             index_sample_ids,
@@ -95,6 +249,9 @@ def filter_ids(
 
     sample_ids, label_ids = _get_patch_ids(samples, patches_field)
 
+    if index_label_ids is None:
+        return sample_ids, label_ids, None, None
+
     keep_inds, good_inds, bad_ids = _parse_ids(
         label_ids,
         index_label_ids,
@@ -228,6 +385,243 @@ def _parse_ids(ids, index_ids, ftype, allow_missing, warn_missing):
     return keep_inds, good_inds, bad_ids
 
 
+def skip_ids(samples, ids, patches_field=None, warn_existing=False):
+    sample_ids, label_ids = get_ids(samples, patches_field=patches_field)
+
+    if patches_field is not None:
+        exclude_ids = list(set(label_ids) - set(ids))
+        num_existing = len(exclude_ids)
+
+        if num_existing > 0:
+            if warn_existing:
+                logger.warning("Skipping %d existing label IDs", num_existing)
+
+            samples = samples.exclude_labels(
+                ids=exclude_ids, fields=patches_field
+            )
+    else:
+        exclude_ids = list(set(sample_ids) - set(ids))
+        num_existing = len(exclude_ids)
+
+        if num_existing > 0:
+            if warn_existing:
+                logger.warning("Skipping %d existing sample IDs", num_existing)
+
+        samples = samples.exclude(exclude_ids)
+
+    return samples
+
+
+def add_ids(
+    sample_ids,
+    label_ids,
+    index_sample_ids,
+    index_label_ids,
+    patches_field=None,
+    overwrite=True,
+    allow_existing=True,
+    warn_existing=False,
+):
+    if patches_field is not None:
+        ids = label_ids
+        index_ids = index_label_ids
+    else:
+        ids = sample_ids
+        index_ids = index_sample_ids
+
+    ii = []
+    jj = []
+
+    ids_map = {_id: _i for _i, _id in enumerate(index_ids)}
+    new_idx = len(index_ids)
+    for _i, _id in enumerate(ids):
+        _idx = ids_map.get(_id, None)
+        if _idx is None:
+            ii.append(_i)
+            jj.append(new_idx)
+            new_idx += 1
+        elif overwrite:
+            ii.append(_i)
+            jj.append(_idx)
+
+    ii = np.array(ii)
+    jj = np.array(jj)
+
+    n = len(index_sample_ids)
+
+    if not allow_existing:
+        existing_inds = np.nonzero(jj < n)[0]
+        num_existing = existing_inds.size
+
+        if num_existing > 0:
+            if warn_existing:
+                logger.warning(
+                    "Ignoring %d IDs (eg '%s') that are already present in "
+                    "the index",
+                    num_existing,
+                    ids[ii[0]],
+                )
+
+                ii = np.delete(ii, existing_inds)
+                jj = np.delete(jj, existing_inds)
+            else:
+                raise ValueError(
+                    "Found %d IDs (eg '%s') that are already present in the "
+                    "index" % (num_existing, ids[ii[0]])
+                )
+
+    if ii.size > 0:
+        sample_ids = np.array(sample_ids)
+        if patches_field is not None:
+            label_ids = np.array(label_ids)
+
+        m = jj[-1] - n + 1
+
+        if n == 0:
+            index_sample_ids = np.array([], dtype=sample_ids.dtype)
+            if patches_field is not None:
+                index_label_ids = np.array([], dtype=label_ids.dtype)
+
+        if m > 0:
+            index_sample_ids = np.concatenate(
+                (index_sample_ids, np.empty(m, dtype=index_sample_ids.dtype))
+            )
+            if patches_field is not None:
+                index_label_ids = np.concatenate(
+                    (index_label_ids, np.empty(m, dtype=index_label_ids.dtype))
+                )
+
+        index_sample_ids[jj] = sample_ids[ii]
+        if patches_field is not None:
+            index_label_ids[jj] = label_ids[ii]
+
+    return index_sample_ids, index_label_ids, ii, jj
+
+
+def add_embeddings(
+    samples,
+    embeddings,
+    sample_ids,
+    label_ids,
+    embeddings_field,
+    patches_field=None,
+):
+    dataset = samples._dataset
+
+    if patches_field is not None:
+        _, embeddings_path = dataset._get_label_field_path(
+            patches_field, embeddings_field
+        )
+
+        values = dict(zip(label_ids, embeddings))
+        dataset.set_label_values(embeddings_path, values)
+    else:
+        values = dict(zip(sample_ids, embeddings))
+        dataset.set_values(embeddings_field, values, key_field="id")
+
+
+def remove_ids(
+    sample_ids,
+    label_ids,
+    index_sample_ids,
+    index_label_ids,
+    patches_field=None,
+    allow_missing=True,
+    warn_missing=False,
+):
+    rm_inds = []
+
+    if sample_ids is not None:
+        rm_inds.extend(
+            _find_ids(
+                sample_ids,
+                index_sample_ids,
+                allow_missing,
+                warn_missing,
+                "sample",
+            )
+        )
+
+    if label_ids is not None:
+        rm_inds.extend(
+            _find_ids(
+                label_ids,
+                index_label_ids,
+                allow_missing,
+                warn_missing,
+                "label",
+            )
+        )
+
+    rm_inds = np.array(rm_inds)
+
+    if rm_inds.size > 0:
+        index_sample_ids = np.delete(index_sample_ids, rm_inds)
+        if patches_field is not None:
+            index_label_ids = np.delete(index_label_ids, rm_inds)
+
+    return index_sample_ids, index_label_ids, rm_inds
+
+
+def _find_ids(ids, index_ids, allow_missing, warn_missing, ftype):
+    found_inds = []
+    missing_ids = []
+
+    ids_map = {_id: _i for _i, _id in enumerate(index_ids)}
+    for _id in ids:
+        ind = ids_map.get(_id, None)
+        if ind is not None:
+            found_inds.append(ind)
+        elif not allow_missing:
+            missing_ids.append(_id)
+
+    num_missing = len(missing_ids)
+
+    if num_missing > 0:
+        if not allow_missing:
+            raise ValueError(
+                "Found %d %d IDs (eg '%s') that are not present in the index"
+                % (num_missing, ftype, missing_ids[0])
+            )
+
+        if warn_missing:
+            logger.warning(
+                "Ignoring %d %d IDs (eg '%s') that are not present in the "
+                "index",
+                num_missing,
+                ftype,
+                missing_ids[0],
+            )
+
+    return found_inds
+
+
+def remove_embeddings(
+    samples,
+    embeddings_field,
+    sample_ids=None,
+    label_ids=None,
+    patches_field=None,
+):
+    dataset = samples._dataset
+
+    if patches_field is not None:
+        _, embeddings_path = dataset._get_label_field_path(
+            patches_field, embeddings_field
+        )
+
+        if sample_ids is not None and label_ids is None:
+            _, id_path = dataset._get_label_field_path(patches_field, "id")
+            label_ids = dataset.select(sample_ids).values(id_path, unwind=True)
+
+        if label_ids is not None:
+            values = dict(zip(label_ids, itertools.repeat(None)))
+            dataset.set_label_values(embeddings_path, values)
+    elif sample_ids is not None:
+        values = dict(zip(sample_ids, itertools.repeat(None)))
+        dataset.set_values(embeddings_field, values, key_field="id")
+
+
 def filter_values(values, keep_inds, patches_field=None):
     if patches_field:
         _values = list(itertools.chain.from_iterable(values))
@@ -318,12 +712,19 @@ def get_embeddings(
     skip_failures=True,
 ):
     if model is None and embeddings_field is None and embeddings is None:
-        raise ValueError(
-            "One of `model`, `embeddings_field`, or `embeddings` must be "
-            "provided"
+        return _empty_embeddings(patches_field)
+
+    if isinstance(embeddings, fob.SimilarityIndex):
+        allow_missing = handle_missing == "skip"
+        return get_embeddings_from_index(
+            samples,
+            embeddings,
+            patches_field=patches_field,
+            allow_missing=allow_missing,
+            warn_missing=True,
         )
 
-    if model is not None:
+    if embeddings is None and model is not None:
         if etau.is_str(model):
             model = foz.load_zoo_model(model)
 
@@ -350,7 +751,7 @@ def get_embeddings(
                 skip_failures=skip_failures,
             )
 
-    if embeddings_field is not None:
+    if embeddings is None and embeddings_field is not None:
         embeddings, samples = _load_embeddings(
             samples, embeddings_field, patches_field=patches_field
         )
@@ -366,14 +767,7 @@ def get_embeddings(
         )
 
     if not isinstance(embeddings, np.ndarray) and not embeddings:
-        embeddings = np.empty((0, 0), dtype=float)
-        sample_ids = np.array([], dtype="<U24")
-        if patches_field is not None:
-            label_ids = np.array([], dtype="<U24")
-        else:
-            label_ids = None
-
-        return embeddings, sample_ids, label_ids
+        return _empty_embeddings(patches_field)
 
     if patches_field is not None:
         if agg_fcn is not None:
@@ -398,6 +792,36 @@ def get_embeddings(
     return embeddings, sample_ids, label_ids
 
 
+def get_unique_name(name, ref_names):
+    ref_names = set(ref_names)
+
+    if name in ref_names:
+        name += "-" + _get_random_characters(6)
+
+    while name in ref_names:
+        name += _get_random_characters(1)
+
+    return name
+
+
+def _get_random_characters(n):
+    return "".join(
+        random.choice(string.ascii_lowercase + string.digits) for _ in range(n)
+    )
+
+
+def _empty_embeddings(patches_field):
+    embeddings = np.empty((0, 0), dtype=float)
+    sample_ids = np.array([], dtype="<U24")
+
+    if patches_field is not None:
+        label_ids = np.array([], dtype="<U24")
+    else:
+        label_ids = None
+
+    return embeddings, sample_ids, label_ids
+
+
 def _load_embeddings(samples, embeddings_field, patches_field=None):
     if patches_field is not None:
         label_type, embeddings_path = samples._get_label_field_path(
diff --git a/fiftyone/brain/internal/core/visualization.py b/fiftyone/brain/internal/core/visualization.py
index ccf54833..f43ec909 100644
--- a/fiftyone/brain/internal/core/visualization.py
+++ b/fiftyone/brain/internal/core/visualization.py
@@ -70,7 +70,7 @@ def compute_visualization(
         embeddings_field = None
         num_dims = points.shape[1]
     elif model is None and embeddings is None:
-        model = foz.load_zoo_model(_DEFAULT_MODEL)
+        model = _DEFAULT_MODEL
         if batch_size is None:
             batch_size = _DEFAULT_BATCH_SIZE
 
@@ -112,7 +112,7 @@ def compute_visualization(
         logger.info("Generating visualization...")
         points = brain_method.fit(embeddings)
     else:
-        sample_ids, label_ids = fbu.get_ids(
+        points, sample_ids, label_ids = fbu.parse_data(
             samples,
             patches_field=patches_field,
             data=points,
@@ -122,6 +122,7 @@ def compute_visualization(
     results = VisualizationResults(
         samples,
         config,
+        brain_key,
         points,
         sample_ids=sample_ids,
         label_ids=label_ids,
@@ -201,9 +202,6 @@ def _is_expr(arg):
 
 
 class Visualization(fob.BrainMethod):
-    def ensure_requirements(self):
-        pass
-
     def fit(self, embeddings):
         raise NotImplementedError("subclass must implement fit()")
 
@@ -214,9 +212,6 @@ def get_fields(self, samples, brain_key):
 
         return fields
 
-    def cleanup(self, samples, brain_key):
-        pass
-
 
 class UMAPVisualization(Visualization):
     def ensure_requirements(self):
diff --git a/fiftyone/brain/similarity.py b/fiftyone/brain/similarity.py
index f3f40f9b..861793c5 100644
--- a/fiftyone/brain/similarity.py
+++ b/fiftyone/brain/similarity.py
@@ -5,88 +5,310 @@
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
+from collections import defaultdict
+from copy import deepcopy
+import logging
+
+from bson import ObjectId
 import numpy as np
 
 import eta.core.utils as etau
 
+import fiftyone.brain as fb
 import fiftyone.core.brain as fob
+import fiftyone.core.context as foc
+import fiftyone.core.fields as fof
+import fiftyone.core.labels as fol
+import fiftyone.core.patches as fop
+import fiftyone.core.stages as fos
 import fiftyone.core.utils as fou
+import fiftyone.core.validation as fov
+import fiftyone.zoo as foz
+from fiftyone import ViewField as F
 
-fbs = fou.lazy_import("fiftyone.brain.internal.core.similarity")
 fbu = fou.lazy_import("fiftyone.brain.internal.core.utils")
 
 
-class SimilarityResults(fob.BrainResults):
-    """Class storing the results of :meth:`fiftyone.brain.compute_similarity`.
+logger = logging.getLogger(__name__)
+
+_DEFAULT_MODEL = "mobilenet-v2-imagenet-torch"
+_DEFAULT_BATCH_SIZE = None
+
+
+def compute_similarity(
+    samples,
+    patches_field,
+    embeddings,
+    brain_key,
+    model,
+    force_square,
+    alpha,
+    batch_size,
+    num_workers,
+    skip_failures,
+    backend,
+    **kwargs,
+):
+    """See ``fiftyone/brain/__init__.py``."""
+
+    fov.validate_collection(samples)
+
+    if model is None and embeddings is None:
+        model = _DEFAULT_MODEL
+        if batch_size is None:
+            batch_size = _DEFAULT_BATCH_SIZE
+
+    if etau.is_str(embeddings):
+        embeddings_field = fbu.parse_embeddings_field(
+            samples,
+            embeddings,
+            patches_field=patches_field,
+            allow_embedded=model is None,
+        )
+        embeddings = None
+    else:
+        embeddings_field = None
+
+    if etau.is_str(model):
+        _model = foz.load_zoo_model(model)
+        try:
+            supports_prompts = _model.can_embed_prompts
+        except:
+            supports_prompts = None
+    else:
+        _model = model
+        supports_prompts = None
+
+    config = _parse_config(
+        backend,
+        embeddings_field=embeddings_field,
+        patches_field=patches_field,
+        model=model,
+        supports_prompts=supports_prompts,
+        **kwargs,
+    )
+    brain_method = config.build()
+    brain_method.ensure_requirements()
+
+    if brain_key is not None:
+        # Don't allow overwriting an existing run with same key, since we
+        # need the existing run in order to perform workflows like
+        # automatically cleaning up the backend's index
+        brain_method.register_run(samples, brain_key, overwrite=False)
+
+    if embeddings is not False:
+        embeddings, sample_ids, label_ids = fbu.get_embeddings(
+            samples,
+            model=_model,
+            patches_field=patches_field,
+            embeddings=embeddings,
+            embeddings_field=embeddings_field,
+            force_square=force_square,
+            alpha=alpha,
+            batch_size=batch_size,
+            num_workers=num_workers,
+            skip_failures=skip_failures,
+        )
+    else:
+        # Special syntax to allow embeddings to be added later
+        embeddings = None
+
+    results = brain_method.initialize(samples, brain_key)
+
+    if embeddings is not None:
+        results.add_to_index(embeddings, sample_ids, label_ids=label_ids)
+
+    brain_method.save_run_results(samples, brain_key, results)
+
+    return results
+
+
+def _parse_config(name, **kwargs):
+    if name is None:
+        name = fb.brain_config.default_similarity_backend
+
+    backends = fb.brain_config.similarity_backends
+
+    if name not in backends:
+        raise ValueError(
+            "Unsupported backend '%s'. The available backends are %s"
+            % (name, sorted(backends.keys()))
+        )
+
+    params = deepcopy(backends[name])
+
+    config_cls = kwargs.pop("config_cls", None)
+
+    if config_cls is None:
+        config_cls = params.pop("config_cls", None)
+
+    if config_cls is None:
+        raise ValueError("Similarity backend '%s' has no `config_cls`" % name)
+
+    if etau.is_str(config_cls):
+        config_cls = etau.get_class(config_cls)
+
+    params.update(**kwargs)
+    return config_cls(**params)
+
+
+class SimilarityConfig(fob.BrainMethodConfig):
+    """Similarity configuration.
 
     Args:
-        samples: the :class:`fiftyone.core.collections.SampleCollection` used
-        config: the :class:`SimilarityConfig` used
-        embeddings (None): a ``num_embeddings x num_dims`` array of embeddings
-        sample_ids (None): a ``num_embeddings`` array of sample IDs
-        label_ids (None): a ``num_embeddings`` array of label IDs, if
-            applicable
+        embeddings_field (None): the sample field containing the embeddings,
+            if one was provided
+        model (None): the :class:`fiftyone.core.models.Model` or name of the
+            zoo model that was used to compute embeddings, if known
+        patches_field (None): the sample field defining the patches being
+            analyzed, if any
+        supports_prompts (False): whether this run supports prompt queries
     """
 
     def __init__(
-        self, samples, config, embeddings=None, sample_ids=None, label_ids=None
+        self,
+        embeddings_field=None,
+        model=None,
+        patches_field=None,
+        supports_prompts=None,
+        **kwargs,
     ):
-        if embeddings is None:
-            embeddings, sample_ids, label_ids = fbu.get_embeddings(
-                samples._dataset,
-                patches_field=config.patches_field,
-                embeddings_field=config.embeddings_field,
-            )
-        elif sample_ids is None:
-            sample_ids, label_ids = fbu.get_ids(
-                samples,
-                patches_field=config.patches_field,
-                data=embeddings,
-                data_type="embeddings",
-            )
+        if model is not None and not etau.is_str(model):
+            model = None
+
+        self.embeddings_field = embeddings_field
+        self.model = model
+        self.patches_field = patches_field
+        self.supports_prompts = supports_prompts
+        super().__init__(**kwargs)
 
-        self._samples = samples
-        self._config = config
-        self.embeddings = embeddings
-        self.sample_ids = sample_ids
-        self.label_ids = label_ids
+    @property
+    def method(self):
+        """The name of the similarity backend."""
+        raise NotImplementedError("subclass must implement method")
 
-        self._last_view = None
+    @property
+    def max_k(self):
+        """A maximum k value for nearest neighbor queries, or None if there is
+        no limit.
+        """
+        raise NotImplementedError("subclass must implement max_k")
+
+    @property
+    def supports_least_similarity(self):
+        """Whether this backend supports least similarity queries."""
+        raise NotImplementedError(
+            "subclass must implement supports_least_similarity"
+        )
+
+    @property
+    def supported_aggregations(self):
+        """A tuple of supported values for the ``aggregation`` parameter of the
+        backend's
+        :meth:`sort_by_similarity() <SimilarityIndex.sort_by_similarity>` and
+        :meth:`_kneighbors() <SimilarityIndex._kneighbors>` methods.
+        """
+        raise NotImplementedError(
+            "subclass must implement supported_aggregations"
+        )
+
+    def load_credentials(self, **kwargs):
+        self._load_parameters(**kwargs)
+
+    def _load_parameters(self, **kwargs):
+        name = self.method
+        parameters = fb.brain_config.similarity_backends.get(name, {})
+
+        for name, value in kwargs.items():
+            if value is None:
+                value = parameters.get(name, None)
+
+            if value is not None:
+                setattr(self, name, value)
+
+
+class Similarity(fob.BrainMethod):
+    """Base class for similarity factories.
+
+    Args:
+        config: a :class:`SimilarityConfig`
+    """
+
+    def initialize(self, samples, brain_key):
+        """Initializes a similarity index.
+
+        Args:
+            samples: a :class:`fiftyone.core.collections.SampleColllection`
+            brain_key: the brain key
+
+        Returns:
+            a :class:`SimilarityIndex`
+        """
+        raise NotImplementedError("subclass must implement initialize()")
+
+    def get_fields(self, samples, brain_key):
+        fields = []
+        if self.config.patches_field is not None:
+            fields.append(self.config.patches_field)
+
+        if self.config.embeddings_field is not None:
+            fields.append(self.config.embeddings_field)
+
+        return fields
+
+
+class SimilarityIndex(fob.BrainResults):
+    """Base class for similarity indexes.
+
+    Args:
+        samples: the :class:`fiftyone.core.collections.SampleCollection` used
+        config: the :class:`SimilarityConfig` used
+        brain_key: the brain key
+        backend (None): a :class:`Similarity` backend
+    """
+
+    def __init__(self, samples, config, brain_key, backend=None):
+        super().__init__(samples, config, brain_key, backend=backend)
+
+        self._model = None
         self._curr_view = None
         self._curr_sample_ids = None
         self._curr_label_ids = None
         self._curr_keep_inds = None
-        self._curr_good_inds = None
-        self._neighbors_helper = None
-        self._thresh = None
-        self._unique_ids = None
-        self._duplicate_ids = None
-        self._neighbors_map = None
-        self._model = None
+        self._curr_missing_size = None
+        self._last_view = None
+        self._last_views = []
 
         self.use_view(samples)
 
     def __enter__(self):
-        self._last_view = self.view
+        self._last_views.append(self._last_view)
         return self
 
     def __exit__(self, *args):
-        self.use_view(self._last_view)
-        self._last_view = None
+        try:
+            last_view = self._last_views.pop()
+        except:
+            last_view = self._samples
+
+        self.use_view(last_view)
 
     @property
     def config(self):
-        """The :class:`SimilarityConfig` for the results."""
+        """The :class:`SimilarityConfig` for these results."""
         return self._config
 
     @property
-    def index_size(self):
-        """The number of active data points in the index.
+    def sample_ids(self):
+        """The sample IDs of the full index, or ``None`` if not supported."""
+        return None
 
-        If :meth:`use_view` has been called to restrict the index, this
-        property will reflect the size of the active index.
+    @property
+    def label_ids(self):
+        """The label IDs of the full index, or ``None`` if not applicable or
+        not supported.
         """
-        return len(self._curr_sample_ids)
+        return None
 
     @property
     def total_index_size(self):
@@ -95,22 +317,7 @@ def total_index_size(self):
         If :meth:`use_view` has been called to restrict the index, this value
         may be larger than the current :meth:`index_size`.
         """
-        return len(self.embeddings)
-
-    @property
-    def missing_size(self):
-        """The total number of data points in :meth:`view` that are missing
-        from this index.
-
-        This property is only applicable when :meth:`use_view` has been called,
-        and it will be ``None`` if no data points are missing.
-        """
-        good = self._curr_good_inds
-
-        if good is None:
-            return None
-
-        return good.size - np.count_nonzero(good)
+        raise NotImplementedError("subclass must implement total_index_size")
 
     @property
     def view(self):
@@ -122,13 +329,6 @@ def view(self):
         """
         return self._curr_view
 
-    @property
-    def thresh(self):
-        """The threshold used by the last call to :meth:`find_duplicates` or
-        :meth:`find_unique`.
-        """
-        return self._thresh
-
     @property
     def current_sample_ids(self):
         """The sample IDs of the currently active data points in the index.
@@ -149,29 +349,113 @@ def current_label_ids(self):
         return self._curr_label_ids
 
     @property
-    def unique_ids(self):
-        """A list of unique IDs from the last call to :meth:`find_duplicates`
-        or :meth:`find_unique`.
+    def _current_inds(self):
+        """The indices of :meth:`current_sample_ids` in :meth:`sample_ids`, or
+        ``None`` if not supported or if the full index is currently being used.
         """
-        return self._unique_ids
+        return self._curr_keep_inds
 
     @property
-    def duplicate_ids(self):
-        """A list of duplicate IDs from the last call to
-        :meth:`find_duplicates` or :meth:`find_unique`.
+    def index_size(self):
+        """The number of active data points in the index.
+
+        If :meth:`use_view` has been called to restrict the index, this
+        property will reflect the size of the active index.
         """
-        return self._duplicate_ids
+        return len(self._curr_sample_ids)
 
     @property
-    def neighbors_map(self):
-        """A dictionary mapping IDs to lists of ``(dup_id, dist)`` tuples from
-        the last call to :meth:`find_duplicates`.
+    def missing_size(self):
+        """The total number of data points in :meth:`view` that are missing
+        from this index, or ``None`` if unknown.
+
+        This property is only applicable when :meth:`use_view` has been called,
+        and it will be ``None`` if no data points are missing or when the
+        backend does not support it.
         """
-        return self._neighbors_map
+        return self._curr_missing_size
 
-    def use_view(
-        self, sample_collection, allow_missing=True, warn_missing=False
+    def add_to_index(
+        self,
+        embeddings,
+        sample_ids,
+        label_ids=None,
+        overwrite=True,
+        allow_existing=True,
+        warn_existing=False,
+    ):
+        """Adds the given embeddings to the index.
+
+        Args:
+            embeddings: a ``num_embeddings x num_dims`` array of embeddings
+            sample_ids: a ``num_embeddings`` array of sample IDs
+            label_ids (None): a ``num_embeddings`` array of label IDs, if
+                applicable
+            overwrite (True): whether to replace (True) or ignore (False)
+                existing embeddings with the same sample/label IDs
+            allow_existing (True): whether to ignore (True) or raise an error
+                (False) when ``overwrite`` is False and a provided ID already
+                exists in the
+            warn_missing (False): whether to log a warning if an embedding is
+                not added to the index because its ID already exists
+        """
+        raise NotImplementedError("subclass must implement add_to_index()")
+
+    def remove_from_index(
+        self,
+        sample_ids=None,
+        label_ids=None,
+        allow_missing=True,
+        warn_missing=False,
+    ):
+        """Removes the specified embeddings from the index.
+
+        Args:
+            sample_ids (None): an array of sample IDs
+            label_ids (None): an array of label IDs, if applicable
+            allow_missing (True): whether to allow the index to not contain IDs
+                that you provide (True) or whether to raise an error in this
+                case (False)
+            warn_missing (False): whether to log a warning if the index does
+                not contain IDs that you provide
+        """
+        raise NotImplementedError(
+            "subclass must implement remove_from_index()"
+        )
+
+    def get_embeddings(
+        self,
+        sample_ids=None,
+        label_ids=None,
+        allow_missing=True,
+        warn_missing=False,
     ):
+        """Retrieves the embeddings for the given IDs from the index.
+
+        If no IDs are provided, the entire index is returned.
+
+        Args:
+            sample_ids (None): a sample ID or list of sample IDs for which to
+                retrieve embeddings
+            label_ids (None): a label ID or list of label IDs for which to
+                retrieve embeddings
+            allow_missing (True): whether to allow the index to not contain IDs
+                that you provide (True) or whether to raise an error in this
+                case (False)
+            warn_missing (False): whether to log a warning if the index does
+                not contain IDs that you provide
+
+        Returns:
+            a tuple of:
+
+            -   a ``num_embeddings x num_dims`` array of embeddings
+            -   a ``num_embeddings`` array of sample IDs
+            -   a ``num_embeddings`` array of label IDs, if applicable, or else
+                ``None``
+        """
+        raise NotImplementedError("subclass must implement get_embeddings()")
+
+    def use_view(self, samples, allow_missing=True, warn_missing=False):
         """Restricts the index to the provided view.
 
         Subsequent calls to methods on this instance will only contain results
@@ -204,8 +488,7 @@ def use_view(
                 plot.show()
 
         Args:
-            sample_collection: a
-                :class:`fiftyone.core.collections.SampleCollection`
+            samples: a :class:`fiftyone.core.collections.SampleCollection`
             allow_missing (True): whether to allow the provided collection to
                 contain data points that this index does not contain (True) or
                 whether to raise an error in this case (False)
@@ -217,19 +500,25 @@ def use_view(
             self
         """
         sample_ids, label_ids, keep_inds, good_inds = fbu.filter_ids(
-            sample_collection,
+            samples,
             self.sample_ids,
             self.label_ids,
-            patches_field=self._config.patches_field,
+            patches_field=self.config.patches_field,
             allow_missing=allow_missing,
             warn_missing=warn_missing,
         )
 
-        self._curr_view = sample_collection
+        if good_inds is not None:
+            missing_size = good_inds.size - np.count_nonzero(good_inds)
+        else:
+            missing_size = None
+
+        self._last_view = self._curr_view
+        self._curr_view = samples
         self._curr_sample_ids = sample_ids
         self._curr_label_ids = label_ids
         self._curr_keep_inds = keep_inds
-        self._curr_good_inds = good_inds
+        self._curr_missing_size = missing_size
 
         return self
 
@@ -240,6 +529,20 @@ def clear_view(self):
         """
         self.use_view(self._samples)
 
+    def reload(self):
+        """Reloads the index for the current view.
+
+        Subclasses may override this method, but by default this method simply
+        passes the current :meth:`view` back into :meth:`use_view`, which
+        updates the index's current ID set based on any changes to the view
+        since the index was last loaded.
+        """
+        self.use_view(self._curr_view)
+
+    def cleanup(self):
+        """Deletes the similarity index from the backend."""
+        raise NotImplementedError("subclass must implement cleanup()")
+
     def values(self, path_or_expr):
         """Extracts a flat list of values from the given field or expression
         corresponding to the current :meth:`view`.
@@ -259,31 +562,17 @@ def values(self, path_or_expr):
         Returns:
             a list of values
         """
-        return fbs.values(self, path_or_expr)
-
-    def plot_distances(self, bins=100, log=False, backend="plotly", **kwargs):
-        """Plots a histogram of the distance between each example and its
-        nearest neighbor.
-
-        If `:meth:`find_duplicates` or :meth:`find_unique` has been executed,
-        the threshold used is also indicated on the plot.
-
-        Args:
-            bins (100): the number of bins to use
-            log (False): whether to use a log scale y-axis
-            backend ("plotly"): the plotting backend to use. Supported values
-                are ``("plotly", "matplotlib")``
-            **kwargs: keyword arguments for the backend plotting method
+        samples = self.view
+        patches_field = self.config.patches_field
 
-        Returns:
-            one of the following:
+        if patches_field is not None:
+            ids = self.current_label_ids
+        else:
+            ids = self.current_sample_ids
 
-            -   a :class:`fiftyone.core.plots.plotly.PlotlyNotebookPlot`, if
-                you are working in a notebook context and the plotly backend is
-                used
-            -   a plotly or matplotlib figure, otherwise
-        """
-        return fbs.plot_distances(self, bins, log, backend, **kwargs)
+        return fbu.get_values(
+            samples, path_or_expr, ids, patches_field=patches_field
+        )
 
     def sort_by_similarity(
         self,
@@ -309,13 +598,15 @@ def sort_by_similarity(
                     of vectors
                 -   a prompt or iterable of prompts (if supported by the index)
 
-            k (None): the number of matches to return. By default, all
-                samples/labels are included
-            reverse (False): whether to sort by least similarity
-            aggregation ("mean"): the aggregation method to use to compute
-                composite similarities. Only applicable when ``query`` contains
-                multiple queries. Supported values are
-                ``("mean", "min", "max")``
+            k (None): the number of matches to return. Some backends may
+                support ``None``, in which case all samples will be sorted
+            reverse (False): whether to sort by least similarity (True) or
+                greatest similarity (False). Some backends may not support
+                least similarity
+            aggregation ("mean"): the aggregation method to use when multiple
+                queries are provided. The default is ``"mean"``, which means
+                that the query vectors are averaged prior to searching. Some
+                backends may support additional options
             dist_field (None): the name of a float field in which to store the
                 distance of each example to the specified query. The field is
                 created if necessary
@@ -323,8 +614,383 @@ def sort_by_similarity(
         Returns:
             a :class:`fiftyone.core.view.DatasetView`
         """
-        return fbs.sort_by_similarity(
-            self, query, k, reverse, aggregation, dist_field, _mongo
+        samples = self.view
+        patches_field = self.config.patches_field
+
+        selecting_samples = patches_field is None or isinstance(
+            samples, fop.PatchesView
+        )
+
+        kwargs = dict(
+            query=self._parse_query(query),
+            k=k,
+            reverse=reverse,
+            aggregation=aggregation,
+            return_dists=dist_field is not None,
+        )
+
+        if dist_field is not None:
+            ids, dists = self._kneighbors(**kwargs)
+        else:
+            ids = self._kneighbors(**kwargs)
+
+        if not selecting_samples:
+            label_ids = ids
+
+            _ids = set(ids)
+            bools = np.array([_id in _ids for _id in self.current_label_ids])
+            sample_ids = self.current_sample_ids[bools]
+
+        # Store query distances
+        if dist_field is not None:
+            if selecting_samples:
+                values = dict(zip(ids, dists))
+                samples.set_values(dist_field, values, key_field="id")
+            else:
+                label_type, path = samples._get_label_field_path(
+                    patches_field, dist_field
+                )
+                if issubclass(label_type, fol._LABEL_LIST_FIELDS):
+                    samples._set_list_values_by_id(
+                        path,
+                        sample_ids,
+                        label_ids,
+                        dists,
+                        path.rsplit(".", 1)[0],
+                    )
+                else:
+                    values = dict(zip(sample_ids, dists))
+                    samples.set_values(path, values, key_field="id")
+
+        # Construct sorted view
+        stages = []
+
+        if selecting_samples:
+            stage = fos.Select(ids, ordered=True)
+            stages.append(stage)
+        else:
+            # Sorting by object similarity but this is not a patches view, so
+            # arrange the samples in order of their first occuring label
+            result_sample_ids = _unique_no_sort(sample_ids)
+            stage = fos.Select(result_sample_ids, ordered=True)
+            stages.append(stage)
+
+            if k is not None:
+                _ids = [ObjectId(_id) for _id in ids]
+                stage = fos.FilterLabels(patches_field, F("_id").is_in(_ids))
+                stages.append(stage)
+
+        if _mongo:
+            pipeline = []
+            for stage in stages:
+                stage.validate(samples)
+                pipeline.extend(stage.to_mongo(samples))
+
+            return pipeline
+
+        view = samples
+        for stage in stages:
+            view = view.add_stage(stage)
+
+        return view
+
+    def _parse_query(self, query):
+        if query is None:
+            raise ValueError("At least one query must be provided")
+
+        if isinstance(query, np.ndarray):
+            # Query by vector(s)
+            if query.size == 0:
+                raise ValueError("At least one query vector must be provided")
+
+            return query
+
+        if etau.is_str(query):
+            query = [query]
+        else:
+            query = list(query)
+
+        if not query:
+            raise ValueError("At least one query must be provided")
+
+        if etau.is_numeric(query[0]):
+            return np.asarray(query)
+
+        try:
+            ObjectId(query[0])
+            is_prompts = False
+        except:
+            is_prompts = True
+
+        if is_prompts:
+            if not self.config.supports_prompts:
+                raise ValueError(
+                    "Invalid query '%s'; this model does not support prompts"
+                    % query[0]
+                )
+
+            model = self.get_model()
+            with model:
+                return model.embed_prompts(query)
+
+        return query
+
+    def _kneighbors(
+        self,
+        query=None,
+        k=None,
+        reverse=False,
+        aggregation=None,
+        return_dists=False,
+    ):
+        """Returns the k-nearest neighbors for the given query.
+
+        This method should only return results from the current :meth:`view`.
+
+        Args:
+            query (None): the query, which can be any of the following:
+
+                -   an ID or list of IDs for which to return neighbors
+                -   an embedding or ``num_queries x num_dim`` array of
+                    embeddings for which to return neighbors
+                -   Some backends may also support ``None``, in which case the
+                    neighbors for all points in the current :meth:`view are
+                    returned
+
+            k (None): the number of neighbors to return. Some backends may
+                enforce upper bounds on this parameter
+            reverse (False): whether to sort by least similarity (True) or
+                greatest similarity (False). Some backends may not support
+                least similarity
+            aggregation (None): an optional aggregation method to use when
+                multiple queries are provided. All backends must support
+                ``"mean"``, which averages query vectors prior to searching.
+                Backends may support additional options as well
+            return_dists (False): whether to return query-neighbor distances
+
+        Returns:
+            the query result, in one of the following formats:
+
+                -   an ``(ids, dists)`` tuple, when ``return_dists`` is True
+                -   ``ids``, when ``return_dists`` is False
+
+            In the above, ``ids`` contains the IDs of the nearest neighbors, in
+            one of the following formats:
+
+                -   a list of nearest neighbor IDs, when a single query ID or
+                    vector is provided, **or** when an ``aggregation`` is
+                    provided
+                -   a list of lists of nearest neighbor IDs, when multiple
+                    query IDs/vectors and no ``aggregation`` is provided
+                -   a list of arrays of the **integer indexes** (not IDs) of
+                    nearest neighbor points for every vector in the index, when
+                    no query is provided
+
+            and ``dists`` contains the corresponding query-neighbor distances
+            for each result in ``ids``
+        """
+        raise NotImplementedError("subclass must implement _kneighbors()")
+
+    def get_model(self):
+        """Returns the stored model for this index.
+
+        Returns:
+            a :class:`fiftyone.core.models.Model`
+        """
+        if self._model is None:
+            model = self.config.model
+            if model is None:
+                raise ValueError("These results don't have a stored model")
+
+            if etau.is_str(model):
+                model = foz.load_zoo_model(model)
+
+            self._model = model
+
+        return self._model
+
+    def compute_embeddings(
+        self,
+        samples,
+        model=None,
+        batch_size=None,
+        num_workers=None,
+        skip_failures=True,
+        skip_existing=False,
+        warn_existing=False,
+        force_square=False,
+        alpha=None,
+    ):
+        """Computes embeddings for the given samples using this backend's
+        model.
+
+        Args:
+            samples: a :class:`fiftyone.core.collections.SampleCollection`
+            model (None): a :class:`fiftyone.core.models.Model` to apply. If
+                not provided, these results must have been created with a
+                stored model, which will be used by default
+            batch_size (None): an optional batch size to use when computing
+                embeddings. Only applicable when a ``model`` is provided
+            num_workers (None): the number of workers to use when loading
+                images. Only applicable when a Torch-based model is being used
+                to compute embeddings
+            skip_failures (True): whether to gracefully continue without
+                raising an error if embeddings cannot be generated for a sample
+            skip_existing (False): whether to skip generating embeddings for
+                sample/label IDs that are already in the index
+            warn_existing (False): whether to log a warning if any IDs already
+                exist in the index
+            force_square (False): whether to minimally manipulate the patch
+                bounding boxes into squares prior to extraction. Only
+                applicable when a ``model`` and ``patches_field`` are specified
+            alpha (None): an optional expansion/contraction to apply to the
+                patches before extracting them, in ``[-1, inf)``. If provided,
+                the length and width of the box are expanded (or contracted,
+                when ``alpha < 0``) by ``(100 * alpha)%``. For example, set
+                ``alpha = 1.1`` to expand the boxes by 10%, and set
+                ``alpha = 0.9`` to contract the boxes by 10%. Only applicable
+                when a ``model`` and ``patches_field`` are specified
+
+        Returns:
+            a tuple of:
+
+            -   a ``num_embeddings x num_dims`` array of embeddings
+            -   a ``num_embeddings`` array of sample IDs
+            -   a ``num_embeddings`` array of label IDs, if applicable, or else
+                ``None``
+        """
+        if model is None:
+            model = self.get_model()
+
+        if skip_existing:
+            if self.config.patches_field is not None:
+                index_ids = self.label_ids
+            else:
+                index_ids = self.sample_ids
+
+            if index_ids is not None:
+                samples = fbu.skip_ids(
+                    samples,
+                    index_ids,
+                    patches_field=self.config.patches_field,
+                    warn_existing=warn_existing,
+                )
+            else:
+                logger.warning(
+                    "This index does not support skipping existing IDs"
+                )
+
+        return fbu.get_embeddings(
+            samples,
+            model=model,
+            patches_field=self.config.patches_field,
+            embeddings_field=self.config.embeddings_field,
+            force_square=force_square,
+            alpha=alpha,
+            batch_size=batch_size,
+            num_workers=num_workers,
+            skip_failures=skip_failures,
+        )
+
+    @classmethod
+    def _from_dict(cls, d, samples, config, brain_key):
+        """Builds a :class:`SimilarityIndex` from a JSON representation of it.
+
+        Args:
+            d: a JSON dict
+            samples: the :class:`fiftyone.core.collections.SampleCollection`
+                for the run
+            config: the :class:`SimilarityConfig` for the run
+            brain_key: the brain key
+
+        Returns:
+            a :class:`SimilarityIndex`
+        """
+        raise NotImplementedError("subclass must implement _from_dict()")
+
+
+class DuplicatesMixin(object):
+    """Mixin for :class:`SimilarityIndex` instances that support duplicate
+    detection operations.
+
+    Similarity backends can expose this mixin simply by implementing
+    :meth:`_radius_neighbors`.
+    """
+
+    def __init__(self):
+        self._thresh = None
+        self._unique_ids = None
+        self._duplicate_ids = None
+        self._neighbors_map = None
+
+    @property
+    def thresh(self):
+        """The threshold used by the last call to :meth:`find_duplicates` or
+        :meth:`find_unique`.
+        """
+        return self._thresh
+
+    @property
+    def unique_ids(self):
+        """A list of unique IDs from the last call to :meth:`find_duplicates`
+        or :meth:`find_unique`.
+        """
+        return self._unique_ids
+
+    @property
+    def duplicate_ids(self):
+        """A list of duplicate IDs from the last call to
+        :meth:`find_duplicates` or :meth:`find_unique`.
+        """
+        return self._duplicate_ids
+
+    @property
+    def neighbors_map(self):
+        """A dictionary mapping IDs to lists of ``(dup_id, dist)`` tuples from
+        the last call to :meth:`find_duplicates`.
+        """
+        return self._neighbors_map
+
+    def _radius_neighbors(self, query=None, thresh=None, return_dists=False):
+        """Returns the neighbors within the given distance threshold for the
+        given query.
+
+        This method should only return results from the current :meth:`view`.
+
+        Args:
+            query (None): the query, which can be any of the following:
+
+                -   an ID or list of IDs for which to return neighbors
+                -   an embedding or ``num_queries x num_dim`` array of
+                    embeddings for which to return neighbors
+                -   ``None``, in which case the neighbors for all points in the
+                    current :meth:`view are returned
+
+            thresh (None): the distance threshold to use
+            return_dists (False): whether to return query-neighbor distances
+
+        Returns:
+            the query result, in one of the following formats:
+
+                -   an ``(ids, dists)`` tuple, when ``return_dists`` is True
+                -   ``ids``, when ``return_dists`` is False
+
+            In the above, ``ids`` contains the IDs of the nearest neighbors, in
+            one of the following formats:
+
+                -   a list of nearest neighbor IDs, when a single query ID or
+                    vector is provided
+                -   a list of lists of nearest neighbor IDs, when multiple
+                    query IDs/vectors is provided
+                -   a list of arrays of the **integer indexes** (not IDs) of
+                    nearest neighbor points for every vector in the index, when
+                    no query is provided
+
+            and ``dists`` contains the corresponding query-neighbor distances
+            for each result in ``ids``
+        """
+        raise NotImplementedError(
+            "subclass must implement _radius_neighbors()"
         )
 
     def find_duplicates(self, thresh=None, fraction=None):
@@ -348,7 +1014,56 @@ def find_duplicates(self, thresh=None, fraction=None):
                 automatically tuned to achieve the desired fraction of
                 duplicates
         """
-        return fbs.find_duplicates(self, thresh, fraction)
+        if self.config.patches_field is not None:
+            logger.info("Computing duplicate patches...")
+            ids = self.current_label_ids
+        else:
+            logger.info("Computing duplicate samples...")
+            ids = self.current_sample_ids
+
+        # Detect duplicates
+        if fraction is not None:
+            num_keep = int(round(min(max(0, 1.0 - fraction), 1) * len(ids)))
+            unique_ids, thresh = self._remove_duplicates_count(
+                num_keep, ids, init_thresh=thresh
+            )
+        else:
+            unique_ids = self._remove_duplicates_thresh(thresh, ids)
+
+        _unique_ids = set(unique_ids)
+        duplicate_ids = [_id for _id in ids if _id not in _unique_ids]
+
+        # Locate nearest non-duplicate for each duplicate
+        if unique_ids and duplicate_ids:
+            if self.config.patches_field is not None:
+                unique_view = self._samples.select_labels(
+                    ids=unique_ids, fields=self.config.patches_field
+                )
+            else:
+                unique_view = self._samples.select(unique_ids)
+
+            with self.use_view(unique_view):
+                nearest_ids, dists = self._kneighbors(
+                    query=duplicate_ids, k=1, return_dists=True
+                )
+
+            neighbors_map = defaultdict(list)
+            for dup_id, _ids, _dists in zip(duplicate_ids, nearest_ids, dists):
+                neighbors_map[_ids[0]].append((dup_id, _dists[0]))
+
+            neighbors_map = {
+                k: sorted(v, key=lambda t: t[1])
+                for k, v in neighbors_map.items()
+            }
+        else:
+            neighbors_map = {}
+
+        logger.info("Duplicates computation complete")
+
+        self._thresh = thresh
+        self._unique_ids = unique_ids
+        self._duplicate_ids = duplicate_ids
+        self._neighbors_map = neighbors_map
 
     def find_unique(self, count):
         """Queries the index to select a subset of examples of the specified
@@ -364,7 +1079,127 @@ def find_unique(self, count):
         Args:
             count: the desired number of unique examples
         """
-        return fbs.find_unique(self, count)
+        if self.config.patches_field is not None:
+            logger.info("Computing unique patches...")
+            ids = self.current_label_ids
+        else:
+            logger.info("Computing unique samples...")
+            ids = self.current_sample_ids
+
+        unique_ids, thresh = self._remove_duplicates_count(count, ids)
+
+        _unique_ids = set(unique_ids)
+        duplicate_ids = [_id for _id in ids if _id not in _unique_ids]
+
+        logger.info("Uniqueness computation complete")
+
+        self._thresh = thresh
+        self._unique_ids = unique_ids
+        self._duplicate_ids = duplicate_ids
+        self._neighbors_map = None
+
+    def _remove_duplicates_count(self, num_keep, ids, init_thresh=None):
+        if init_thresh is not None:
+            thresh = init_thresh
+        else:
+            thresh = 1
+
+        if num_keep <= 0:
+            logger.info(
+                "threshold: -, kept: %d, target: %d", num_keep, num_keep
+            )
+            return set(), None
+
+        if num_keep >= len(ids):
+            logger.info(
+                "threshold: -, kept: %d, target: %d", num_keep, num_keep
+            )
+            return set(ids), None
+
+        thresh_lims = [0, None]
+        num_target = num_keep
+        num_keep = -1
+
+        while True:
+            keep_ids = self._remove_duplicates_thresh(thresh, ids)
+            num_keep_last = num_keep
+            num_keep = len(keep_ids)
+
+            logger.info(
+                "threshold: %f, kept: %d, target: %d",
+                thresh,
+                num_keep,
+                num_target,
+            )
+
+            if num_keep == num_target or (
+                num_keep == num_keep_last
+                and thresh_lims[1] is not None
+                and thresh_lims[1] - thresh_lims[0] < 1e-6
+            ):
+                break
+
+            if num_keep < num_target:
+                # Need to decrease threshold
+                thresh_lims[1] = thresh
+                thresh = 0.5 * (thresh_lims[0] + thresh)
+            else:
+                # Need to increase threshold
+                thresh_lims[0] = thresh
+                if thresh_lims[1] is not None:
+                    thresh = 0.5 * (thresh + thresh_lims[1])
+                else:
+                    thresh *= 2
+
+        return keep_ids, thresh
+
+    def _remove_duplicates_thresh(self, thresh, ids):
+        nearest_inds = self._radius_neighbors(thresh=thresh)
+
+        n = len(ids)
+        keep = set(range(n))
+        for ind in range(n):
+            if ind in keep:
+                keep -= {i for i in nearest_inds[ind] if i > ind}
+
+        return {ids[i] for i in keep}
+
+    def plot_distances(self, bins=100, log=False, backend="plotly", **kwargs):
+        """Plots a histogram of the distance between each example and its
+        nearest neighbor.
+
+        If `:meth:`find_duplicates` or :meth:`find_unique` has been executed,
+        the threshold used is also indicated on the plot.
+
+        Args:
+            bins (100): the number of bins to use
+            log (False): whether to use a log scale y-axis
+            backend ("plotly"): the plotting backend to use. Supported values
+                are ``("plotly", "matplotlib")``
+            **kwargs: keyword arguments for the backend plotting method
+
+        Returns:
+            one of the following:
+
+            -   a :class:`fiftyone.core.plots.plotly.PlotlyNotebookPlot`, if
+                you are working in a notebook context and the plotly backend is
+                used
+            -   a plotly or matplotlib figure, otherwise
+        """
+        metric = self.config.metric
+        thresh = self.thresh
+
+        _, dists = self._kneighbors(k=1, return_dists=True)
+        dists = np.array([d[0] for d in dists.values()])
+
+        if backend == "matplotlib":
+            return _plot_distances_mpl(
+                dists, metric, thresh, bins, log, **kwargs
+            )
+
+        return _plot_distances_plotly(
+            dists, metric, thresh, bins, log, **kwargs
+        )
 
     def duplicates_view(
         self,
@@ -412,9 +1247,56 @@ def duplicates_view(
                 "You must first call `find_duplicates()` to generate results"
             )
 
-        return fbs.duplicates_view(
-            self, type_field, id_field, dist_field, sort_by, reverse
-        )
+        samples = self.view
+        patches_field = self.config.patches_field
+        neighbors_map = self.neighbors_map
+
+        if patches_field is not None and not isinstance(
+            samples, fop.PatchesView
+        ):
+            samples = samples.to_patches(patches_field)
+
+        if sort_by == "distance":
+            key = lambda kv: min(e[1] for e in kv[1])
+        elif sort_by == "count":
+            key = lambda kv: len(kv[1])
+        else:
+            raise ValueError(
+                "Invalid sort_by='%s'; supported values are %s"
+                % (sort_by, ("distance", "count"))
+            )
+
+        existing_ids = set(samples.values("id"))
+        neighbors = [
+            (k, v) for k, v in neighbors_map.items() if k in existing_ids
+        ]
+
+        ids = []
+        types = {}
+        nearest_ids = {}
+        dists = {}
+        for _id, duplicates in sorted(neighbors, key=key, reverse=reverse):
+            ids.append(_id)
+            types[_id] = "nearest"
+            nearest_ids[_id] = _id
+            dists[_id] = 0.0
+
+            for dup_id, dist in duplicates:
+                ids.append(dup_id)
+                types[dup_id] = "duplicate"
+                nearest_ids[dup_id] = _id
+                dists[dup_id] = dist
+
+        if type_field is not None:
+            samples.set_values(type_field, types, key_field="id")
+
+        if id_field is not None:
+            samples.set_values(id_field, nearest_ids, key_field="id")
+
+        if dist_field is not None:
+            samples.set_values(dist_field, dists, key_field="id")
+
+        return samples.select(ids, ordered=True)
 
     def unique_view(self):
         """Returns a view that contains only the unique examples generated by
@@ -432,20 +1314,24 @@ def unique_view(self):
                 "to generate results"
             )
 
-        return fbs.unique_view(self)
+        samples = self.view
+        patches_field = self.config.patches_field
+        unique_ids = self.unique_ids
 
-    def visualize_duplicates(
-        self, visualization=None, backend="plotly", **kwargs
-    ):
+        if patches_field is not None and not isinstance(
+            samples, fop.PatchesView
+        ):
+            samples = samples.to_patches(patches_field)
+
+        return samples.select(unique_ids)
+
+    def visualize_duplicates(self, visualization, backend="plotly", **kwargs):
         """Generates an interactive scatterplot of the results generated by the
         last call to :meth:`find_duplicates`.
 
-        If provided, the ``visualization`` argument can be any visualization
-        computed on the same dataset (or subset of it) as long as it contains
-        every sample/object in the view whose results you are visualizing. If
-        no ``visualization`` argument is provided and the embeddings
-        have more than 3 dimensions, a 2D representation of the embeddings is
-        computed via :meth:`fiftyone.brain.compute_visualization`.
+        The ``visualization`` argument can be any visualization computed on the
+        same dataset (or subset of it) as long as it contains every
+        sample/object in the view whose results you are visualizing.
 
         The points are colored based on the following partition:
 
@@ -462,7 +1348,7 @@ def visualize_duplicates(
         points in the plot.
 
         Args:
-            visualization (None): a
+            visualization: a
                 :class:`fiftyone.brain.visualization.VisualizationResults`
                 instance to use to visualize the results
             backend ("plotly"): the plotting backend to use. Supported values
@@ -480,18 +1366,50 @@ def visualize_duplicates(
                 "You must first call `find_duplicates()` to generate results"
             )
 
-        return fbs.visualize_duplicates(self, visualization, backend, **kwargs)
+        samples = self.view
+        duplicate_ids = self.duplicate_ids
+        neighbors_map = self.neighbors_map
+        patches_field = self.config.patches_field
+
+        dup_ids = set(duplicate_ids)
+        nearest_ids = set(neighbors_map.keys())
+
+        with visualization.use_view(samples, allow_missing=True):
+            if patches_field is not None:
+                ids = visualization.current_label_ids
+            else:
+                ids = visualization.current_sample_ids
+
+            labels = []
+            for _id in ids:
+                if _id in dup_ids:
+                    label = "duplicate"
+                elif _id in nearest_ids:
+                    label = "nearest"
+                else:
+                    label = "unique"
+
+                labels.append(label)
+
+            if backend == "plotly":
+                kwargs["edges"] = _build_edges(ids, neighbors_map)
+                kwargs["edges_title"] = "neighbors"
+                kwargs["labels_title"] = "type"
+
+            return visualization.visualize(
+                labels=labels,
+                classes=["unique", "nearest", "duplicate"],
+                backend=backend,
+                **kwargs,
+            )
 
-    def visualize_unique(self, visualization=None, backend="plotly", **kwargs):
+    def visualize_unique(self, visualization, backend="plotly", **kwargs):
         """Generates an interactive scatterplot of the results generated by the
         last call to :meth:`find_unique`.
 
-        If provided, the ``visualization`` argument can be any visualization
-        computed on the same dataset (or subset of it) as long as it contains
-        every sample/object in the view whose results you are visualizing. If
-        no ``visualization`` argument is provided and the embeddings
-        have more than 3 dimensions, a 2D representation of the embeddings is
-        computed via :meth:`fiftyone.brain.compute_visualization`.
+        The ``visualization`` argument can be any visualization computed on the
+        same dataset (or subset of it) as long as it contains every
+        sample/object in the view whose results you are visualizing.
 
         The points are colored based on the following partition:
 
@@ -504,7 +1422,7 @@ def visualize_unique(self, visualization=None, backend="plotly", **kwargs):
         points in the plot.
 
         Args:
-            visualization (None): a
+            visualization: a
                 :class:`fiftyone.brain.visualization.VisualizationResults`
                 instance to use to visualize the results
             backend ("plotly"): the plotting backend to use. Supported values
@@ -522,81 +1440,148 @@ def visualize_unique(self, visualization=None, backend="plotly", **kwargs):
                 "You must first call `find_unique()` to generate results"
             )
 
-        return fbs.visualize_unique(self, visualization, backend, **kwargs)
+        samples = self.view
+        unique_ids = self.unique_ids
+        patches_field = self.config.patches_field
 
-    def attributes(self):
-        attrs = super().attributes()
+        unique_ids = set(unique_ids)
 
-        if self.config.embeddings_field is not None:
-            attrs = [
-                attr
-                for attr in attrs
-                if attr not in ("embeddings", "sample_ids", "label_ids")
-            ]
+        with visualization.use_view(samples, allow_missing=True):
+            if patches_field is not None:
+                ids = visualization.current_label_ids
+            else:
+                ids = visualization.current_sample_ids
 
-        return attrs
+            labels = []
+            for _id in ids:
+                if _id in unique_ids:
+                    label = "unique"
+                else:
+                    label = "other"
 
-    @classmethod
-    def _from_dict(cls, d, samples, config):
-        embeddings = d.get("embeddings", None)
-        if embeddings is not None:
-            embeddings = np.array(embeddings)
+                labels.append(label)
 
-        sample_ids = d.get("sample_ids", None)
-        if sample_ids is not None:
-            sample_ids = np.array(sample_ids)
+            return visualization.visualize(
+                labels=labels,
+                classes=["other", "unique"],
+                backend=backend,
+                **kwargs,
+            )
 
-        label_ids = d.get("label_ids", None)
-        if label_ids is not None:
-            label_ids = np.array(label_ids)
 
-        return cls(
-            samples,
-            config,
-            embeddings=embeddings,
-            sample_ids=sample_ids,
-            label_ids=label_ids,
+def _unique_no_sort(values):
+    seen = set()
+    return [v for v in values if v not in seen and not seen.add(v)]
+
+
+def _build_edges(ids, neighbors_map):
+    inds_map = {_id: idx for idx, _id in enumerate(ids)}
+
+    edges = []
+    for nearest_id, duplicates in neighbors_map.items():
+        nearest_ind = inds_map[nearest_id]
+        for dup_id, _ in duplicates:
+            dup_ind = inds_map[dup_id]
+            edges.append((dup_ind, nearest_ind))
+
+    return np.array(edges)
+
+
+def _plot_distances_plotly(dists, metric, thresh, bins, log, **kwargs):
+    import plotly.graph_objects as go
+    import fiftyone.core.plots.plotly as fopl
+
+    counts, edges = np.histogram(dists, bins=bins)
+    left_edges = edges[:-1]
+    widths = edges[1:] - edges[:-1]
+    customdata = np.stack((edges[:-1], edges[1:]), axis=1)
+
+    hover_lines = [
+        "<b>count: %{y}</b>",
+        "distance: [%{customdata[0]:.2f}, %{customdata[1]:.2f}]",
+    ]
+    hovertemplate = "<br>".join(hover_lines) + "<extra></extra>"
+
+    bar = go.Bar(
+        x=left_edges,
+        y=counts,
+        width=widths,
+        customdata=customdata,
+        offset=0,
+        marker_color="#FF6D04",
+        hovertemplate=hovertemplate,
+        showlegend=False,
+    )
+
+    traces = [bar]
+
+    if thresh is not None:
+        line = go.Scatter(
+            x=[thresh, thresh],
+            y=[0, max(counts)],
+            mode="lines",
+            line=dict(color="#17191C", width=3),
+            hovertemplate="<b>thresh: %{x}</b><extra></extra>",
+            showlegend=False,
         )
+        traces.append(line)
 
+    figure = go.Figure(traces)
 
-class SimilarityConfig(fob.BrainMethodConfig):
-    """Similarity configuration.
+    figure.update_layout(
+        xaxis_title="nearest neighbor distance (%s)" % metric,
+        yaxis_title="count",
+        hovermode="x",
+        yaxis_rangemode="tozero",
+    )
 
-    Args:
-        embeddings_field (None): the sample field containing the embeddings,
-            if one was provided
-        model (None): the :class:`fiftyone.core.models.Model` or class name of
-            the model that was used to compute embeddings, if one was provided
-        patches_field (None): the sample field defining the patches being
-            analyzed, if any
-        metric (None): the embedding distance metric used
-        supports_prompts (False): whether this run supports prompt queries
-    """
+    if log:
+        figure.update_layout(yaxis_type="log")
 
-    def __init__(
-        self,
-        embeddings_field=None,
-        model=None,
-        patches_field=None,
-        metric=None,
-        supports_prompts=None,
+    figure.update_layout(**fopl._DEFAULT_LAYOUT)
+    figure.update_layout(**kwargs)
+
+    if foc.is_jupyter_context():
+        figure = fopl.PlotlyNotebookPlot(figure)
+
+    return figure
+
+
+def _plot_distances_mpl(
+    dists, metric, thresh, bins, log, ax=None, figsize=None, **kwargs
+):
+    import matplotlib.pyplot as plt
+
+    if ax is None:
+        fig, ax = plt.subplots()
+    else:
+        fig = ax.figure
+
+    counts, edges = np.histogram(dists, bins=bins)
+    left_edges = edges[:-1]
+    widths = edges[1:] - edges[:-1]
+
+    ax.bar(
+        left_edges,
+        counts,
+        width=widths,
+        align="edge",
+        color="#FF6D04",
         **kwargs,
-    ):
-        if model is not None and not etau.is_str(model):
-            model = etau.get_class_name(model)
+    )
 
-        self.embeddings_field = embeddings_field
-        self.model = model
-        self.patches_field = patches_field
-        self.metric = metric
-        self.supports_prompts = supports_prompts
-        super().__init__(**kwargs)
+    if thresh is not None:
+        ax.vlines(thresh, 0, max(counts), color="#17191C", linewidth=3)
 
-    @property
-    def method(self):
-        return "similarity"
+    if log:
+        ax.set_yscale("log")
 
-    @property
-    def run_cls(self):
-        run_cls_name = self.__class__.__name__[: -len("Config")]
-        return getattr(fbs, run_cls_name)
+    ax.set_xlabel("nearest neighbor distance (%s)" % metric)
+    ax.set_ylabel("count")
+
+    if figsize is not None:
+        fig.set_size_inches(*figsize)
+
+    plt.tight_layout()
+
+    return fig
diff --git a/fiftyone/brain/visualization.py b/fiftyone/brain/visualization.py
index 06cf17c6..cce0a8dc 100644
--- a/fiftyone/brain/visualization.py
+++ b/fiftyone/brain/visualization.py
@@ -23,14 +23,25 @@ class VisualizationResults(fob.BrainResults):
     Args:
         samples: the :class:`fiftyone.core.collections.SampleCollection` used
         config: the :class:`VisualizationConfig` used
+        brain_key: the brain key
         points: a ``num_points x num_dims`` array of visualization points
         sample_ids (None): a ``num_points`` array of sample IDs
         label_ids (None): a ``num_points`` array of label IDs, if applicable
+        backend (None): a :class:`Visualization` backend
     """
 
     def __init__(
-        self, samples, config, points, sample_ids=None, label_ids=None
+        self,
+        samples,
+        config,
+        brain_key,
+        points,
+        sample_ids=None,
+        label_ids=None,
+        backend=None,
     ):
+        super().__init__(samples, config, brain_key, backend=backend)
+
         if sample_ids is None:
             sample_ids, label_ids = fbu.get_ids(
                 samples,
@@ -39,8 +50,6 @@ def __init__(
                 data_type="points",
             )
 
-        self._samples = samples
-        self._config = config
         self.points = points
         self.sample_ids = sample_ids
         self.label_ids = label_ids
@@ -307,7 +316,7 @@ def visualize(
         )
 
     @classmethod
-    def _from_dict(cls, d, samples, config):
+    def _from_dict(cls, d, samples, config, brain_key):
         points = np.array(d["points"])
 
         sample_ids = d.get("sample_ids", None)
@@ -321,6 +330,7 @@ def _from_dict(cls, d, samples, config):
         return cls(
             samples,
             config,
+            brain_key,
             points,
             sample_ids=sample_ids,
             label_ids=label_ids,
@@ -333,8 +343,8 @@ class VisualizationConfig(fob.BrainMethodConfig):
     Args:
         embeddings_field (None): the sample field containing the embeddings,
             if one was provided
-        model (None): the :class:`fiftyone.core.models.Model` or class name of
-            the model that was used to compute embeddings, if one was provided
+        model (None): the :class:`fiftyone.core.models.Model` or name of the
+            zoo model that was used to compute embeddings, if known
         patches_field (None): the sample field defining the patches being
             analyzed, if any
         num_dims (2): the dimension of the visualization space
@@ -349,7 +359,7 @@ def __init__(
         **kwargs,
     ):
         if model is not None and not etau.is_str(model):
-            model = etau.get_class_name(model)
+            model = None
 
         self.embeddings_field = embeddings_field
         self.model = model
@@ -373,8 +383,8 @@ class UMAPVisualizationConfig(VisualizationConfig):
     Args:
         embeddings_field (None): the sample field containing the embeddings,
             if one was provided
-        model (None): the :class:`fiftyone.core.models.Model` or class name of
-            the model that was used to compute embeddings, if one was provided
+        model (None): the :class:`fiftyone.core.models.Model` or name of the
+            zoo model that was used to compute embeddings, if known
         patches_field (None): the sample field defining the patches being
             analyzed, if any
         num_dims (2): the dimension of the visualization space
@@ -435,8 +445,8 @@ class TSNEVisualizationConfig(VisualizationConfig):
     Args:
         embeddings_field (None): the sample field containing the embeddings,
             if one was provided
-        model (None): the :class:`fiftyone.core.models.Model` or class name of
-            the model that was used to compute embeddings, if one was provided
+        model (None): the :class:`fiftyone.core.models.Model` or name of the
+            zoo model that was used to compute embeddings, if known
         patches_field (None): the sample field defining the patches being
             analyzed, if any
         num_dims (2): the dimension of the visualization space
@@ -514,8 +524,8 @@ class PCAVisualizationConfig(VisualizationConfig):
     Args:
         embeddings_field (None): the sample field containing the embeddings,
             if one was provided
-        model (None): the :class:`fiftyone.core.models.Model` or class name of
-            the model that was used to compute embeddings, if one was provided
+        model (None): the :class:`fiftyone.core.models.Model` or name of the
+            zoo model that was used to compute embeddings, if known
         patches_field (None): the sample field defining the patches being
             analyzed, if any
         num_dims (2): the dimension of the visualization space


Test patch:
diff --git a/tests/intensive/test_similarity.py b/tests/intensive/test_similarity.py
index 1c729de2..5de1d281 100644
--- a/tests/intensive/test_similarity.py
+++ b/tests/intensive/test_similarity.py
@@ -1,37 +1,365 @@
 """
 Similarity tests.
 
+Usage::
+
+    pytest tests/intensive/test_similarity.py -s -k test_XXX
+
+Qdrant setup::
+
+    docker pull qdrant/qdrant
+    docker run -p 6333:6333
+
+    pip install qdrant-client
+
+Pinecone setup::
+
+    # Sign up at https://www.pinecone.io
+    # Download API key and environment
+
+    pip install pinecone-client
+
+Brain config setup at `~/.fiftyone/brain_config.json`::
+
+    {
+        "similarity_backends": {
+            "pinecone": {
+                "environment": "us-east-1-aws",
+                "api_key": "XXXXXXXX"
+            },
+            "qdrant": {
+                "url": "http://localhost:6333"
+            }
+        }
+    }
+
 | Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
+import random
 import unittest
 
+import numpy as np
+
 import fiftyone as fo
-import fiftyone.brain as fob
+import fiftyone.brain as fob  # pylint: disable=import-error,no-name-in-module
 import fiftyone.zoo as foz
 from fiftyone import ViewField as F
 
 
+def test_brain_config():
+    similarity_backends = fob.brain_config.similarity_backends
+
+    assert "sklearn" in similarity_backends
+
+    assert "qdrant" in similarity_backends
+    assert "url" in similarity_backends["qdrant"]
+
+    assert "pinecone" in similarity_backends
+    assert "api_key" in similarity_backends["pinecone"]
+    assert "environment" in similarity_backends["pinecone"]
+
+
+def test_image_similarity_backends():
+    dataset = foz.load_zoo_dataset("quickstart")
+
+    index1 = fob.compute_similarity(
+        dataset,
+        model="clip-vit-base32-torch",
+        metric="euclidean",
+        embeddings=False,
+        backend="sklearn",
+        brain_key="clip_sklearn",
+    )
+
+    index2 = fob.compute_similarity(
+        dataset,
+        model="clip-vit-base32-torch",
+        metric="euclidean",
+        embeddings=False,
+        backend="qdrant",
+        brain_key="clip_qdrant",
+    )
+
+    index3 = fob.compute_similarity(
+        dataset,
+        model="clip-vit-base32-torch",
+        metric="euclidean",
+        embeddings=False,
+        backend="pinecone",
+        brain_key="clip_pinecone",
+    )
+
+    embeddings, sample_ids, _ = index1.compute_embeddings(dataset)
+
+    index1.add_to_index(embeddings, sample_ids)
+    index1.save()
+    index1.reload()
+    assert index1.total_index_size == 200
+    assert index1.index_size == 200
+    assert index1.missing_size is None
+
+    index2.add_to_index(embeddings, sample_ids)
+    assert index2.total_index_size == 200
+    assert index2.index_size == 200
+    assert index2.missing_size is None
+
+    index3.add_to_index(embeddings, sample_ids)
+    assert index3.total_index_size == 200
+    assert index3.index_size == 200
+    assert index3.missing_size is None
+
+    prompt = "kites high in the air"
+
+    view1 = dataset.sort_by_similarity(prompt, k=10, brain_key="clip_sklearn")
+    assert len(view1) == 10
+
+    view2 = dataset.sort_by_similarity(prompt, k=10, brain_key="clip_qdrant")
+    assert len(view2) == 10
+
+    view3 = dataset.sort_by_similarity(prompt, k=10, brain_key="clip_pinecone")
+    assert len(view3) == 10
+
+    del index1
+    del index2
+    del index3
+    dataset.clear_cache()
+
+    print(dataset.get_brain_info("clip_sklearn"))
+    print(dataset.get_brain_info("clip_qdrant"))
+    print(dataset.get_brain_info("clip_pinecone"))
+
+    index1 = dataset.load_brain_results("clip_sklearn")
+    index2 = dataset.load_brain_results("clip_qdrant")
+    index3 = dataset.load_brain_results("clip_pinecone")
+
+    assert index1.total_index_size == 200
+    assert index2.total_index_size == 200
+    assert index3.total_index_size == 200
+
+    embeddings1, sample_ids1, _ = index1.get_embeddings()
+    assert embeddings1.shape == (200, 512)
+    assert sample_ids1.shape == (200,)
+
+    embeddings2, sample_ids2, _ = index2.get_embeddings()
+    assert embeddings2.shape == (200, 512)
+    assert sample_ids2.shape == (200,)
+
+    # Pinecone requires IDs
+    # embeddings3, sample_ids3, _ = index3.get_embeddings()
+    # assert embeddings3.shape == (200, 512)
+    # assert sample_ids3.shape == (200,)
+
+    ids = random.sample(list(index1.sample_ids), 100)
+
+    embeddings1, sample_ids1, _ = index1.get_embeddings(sample_ids=ids)
+    embeddings2, sample_ids2, _ = index2.get_embeddings(sample_ids=ids)
+    embeddings3, sample_ids3, _ = index3.get_embeddings(sample_ids=ids)
+
+    embeddings2_dict = dict(zip(sample_ids2, embeddings2))
+    embeddings3_dict = dict(zip(sample_ids3, embeddings3))
+
+    _embeddings2 = np.array([embeddings2_dict[i] for i in sample_ids1])
+    _embeddings3 = np.array([embeddings3_dict[i] for i in sample_ids1])
+
+    assert embeddings1.shape == (100, 512)
+    assert sample_ids1.shape == (100,)
+
+    assert embeddings2.shape == (100, 512)
+    assert sample_ids2.shape == (100,)
+
+    assert embeddings3.shape == (100, 512)
+    assert sample_ids3.shape == (100,)
+
+    assert set(sample_ids1) == set(sample_ids2)
+    assert set(sample_ids1) == set(sample_ids3)
+
+    assert np.allclose(embeddings1, _embeddings2)
+    assert np.allclose(embeddings1, _embeddings3)
+
+    index1.remove_from_index(sample_ids=ids)
+    index2.remove_from_index(sample_ids=ids)
+    index3.remove_from_index(sample_ids=ids)
+
+    assert index1.total_index_size == 100
+    assert index2.total_index_size == 100
+    assert index3.total_index_size == 100
+
+    index1.cleanup()
+    index2.cleanup()
+    index3.cleanup()
+
+    dataset.delete_brain_run("clip_sklearn")
+    dataset.delete_brain_run("clip_qdrant")
+    dataset.delete_brain_run("clip_pinecone")
+    dataset.delete()
+
+
+def test_patch_similarity_backends():
+    dataset = foz.load_zoo_dataset("quickstart")
+
+    index1 = fob.compute_similarity(
+        dataset,
+        patches_field="ground_truth",
+        model="clip-vit-base32-torch",
+        metric="euclidean",
+        embeddings=False,
+        backend="sklearn",
+        brain_key="gt_clip_sklearn",
+    )
+
+    index2 = fob.compute_similarity(
+        dataset,
+        patches_field="ground_truth",
+        model="clip-vit-base32-torch",
+        metric="euclidean",
+        embeddings=False,
+        backend="qdrant",
+        brain_key="gt_clip_qdrant",
+    )
+
+    index3 = fob.compute_similarity(
+        dataset,
+        patches_field="ground_truth",
+        model="clip-vit-base32-torch",
+        metric="euclidean",
+        embeddings=False,
+        backend="pinecone",
+        brain_key="gt_clip_pinecone",
+    )
+
+    embeddings, sample_ids, label_ids = index1.compute_embeddings(dataset)
+
+    index1.add_to_index(embeddings, sample_ids, label_ids=label_ids)
+    index1.save()
+    index1.reload()
+    assert index1.total_index_size == 1232
+    assert index1.index_size == 1232
+    assert index1.missing_size is None
+
+    index2.add_to_index(embeddings, sample_ids, label_ids=label_ids)
+    assert index2.total_index_size == 1232
+    assert index2.index_size == 1232
+    assert index2.missing_size is None
+
+    index3.add_to_index(embeddings, sample_ids, label_ids=label_ids)
+    assert index3.total_index_size == 1232
+    assert index3.index_size == 1232
+    assert index3.missing_size is None
+
+    view = dataset.to_patches("ground_truth")
+
+    prompt = "cute puppies"
+
+    view1 = view.sort_by_similarity(prompt, k=10, brain_key="gt_clip_sklearn")
+    assert len(view1) == 10
+
+    view2 = view.sort_by_similarity(prompt, k=10, brain_key="gt_clip_qdrant")
+    assert len(view2) == 10
+
+    view3 = view.sort_by_similarity(prompt, k=10, brain_key="gt_clip_pinecone")
+    assert len(view3) == 10
+
+    del index1
+    del index2
+    del index3
+    dataset.clear_cache()
+
+    print(dataset.get_brain_info("gt_clip_sklearn"))
+    print(dataset.get_brain_info("gt_clip_qdrant"))
+    print(dataset.get_brain_info("gt_clip_pinecone"))
+
+    index1 = dataset.load_brain_results("gt_clip_sklearn")
+    index2 = dataset.load_brain_results("gt_clip_qdrant")
+    index3 = dataset.load_brain_results("gt_clip_pinecone")
+
+    assert index1.total_index_size == 1232
+    assert index2.total_index_size == 1232
+    assert index3.total_index_size == 1232
+
+    embeddings1, sample_ids1, label_ids1 = index1.get_embeddings()
+    assert embeddings1.shape == (1232, 512)
+    assert sample_ids1.shape == (1232,)
+    assert label_ids1.shape == (1232,)
+
+    embeddings2, sample_ids2, label_ids2 = index2.get_embeddings()
+    assert embeddings2.shape == (1232, 512)
+    assert sample_ids2.shape == (1232,)
+    assert label_ids2.shape == (1232,)
+
+    # Pinecone requires IDs
+    # embeddings3, sample_ids3, label_ids3 = index3.get_embeddings()
+    # assert embeddings3.shape == (1232, 512)
+    # assert sample_ids3.shape == (1232,)
+    # assert label_ids3.shape == (1232,)
+
+    ids = random.sample(list(index1.label_ids), 100)
+
+    embeddings1, sample_ids1, label_ids1 = index1.get_embeddings(label_ids=ids)
+    embeddings2, sample_ids2, label_ids2 = index2.get_embeddings(label_ids=ids)
+    embeddings3, sample_ids3, label_ids3 = index3.get_embeddings(label_ids=ids)
+
+    embeddings2_dict = dict(zip(label_ids2, embeddings2))
+    embeddings3_dict = dict(zip(label_ids3, embeddings3))
+
+    _embeddings2 = np.array([embeddings2_dict[i] for i in label_ids1])
+    _embeddings3 = np.array([embeddings3_dict[i] for i in label_ids1])
+
+    assert embeddings1.shape == (100, 512)
+    assert sample_ids1.shape == (100,)
+    assert label_ids1.shape == (100,)
+
+    assert embeddings2.shape == (100, 512)
+    assert sample_ids2.shape == (100,)
+    assert label_ids2.shape == (100,)
+
+    assert embeddings3.shape == (100, 512)
+    assert sample_ids3.shape == (100,)
+    assert label_ids3.shape == (100,)
+
+    assert set(label_ids1) == set(label_ids2)
+    assert set(label_ids1) == set(label_ids3)
+
+    assert np.allclose(embeddings1, _embeddings2)
+    assert np.allclose(embeddings1, _embeddings3)
+
+    index1.remove_from_index(label_ids=ids)
+    index2.remove_from_index(label_ids=ids)
+    index3.remove_from_index(label_ids=ids)
+
+    assert index1.total_index_size == 1132
+    assert index2.total_index_size == 1132
+    assert index3.total_index_size == 1132
+
+    index1.cleanup()
+    index2.cleanup()
+    index3.cleanup()
+
+    dataset.delete_brain_run("gt_clip_sklearn")
+    dataset.delete_brain_run("gt_clip_qdrant")
+    dataset.delete_brain_run("gt_clip_pinecone")
+
+
 def test_images():
     dataset = _load_images_dataset()
 
-    results = dataset.load_brain_results("img_sim")
+    index = dataset.load_brain_results("img_sim")
 
-    assert results.total_index_size == len(dataset)
-    assert set(dataset.values("id")) == set(results.sample_ids)
+    assert index.total_index_size == len(dataset)
+    assert set(dataset.values("id")) == set(index.sample_ids)
 
 
 def test_images_subset():
     dataset = _load_images_dataset()
 
-    results = dataset.load_brain_results("img_sim")
+    index = dataset.load_brain_results("img_sim")
 
     view = dataset.take(10)
-    results.use_view(view)
+    index.use_view(view)
 
-    assert results.index_size == len(view)
-    assert set(view.values("id")) == set(results.current_sample_ids)
+    assert index.index_size == len(view)
+    assert set(view.values("id")) == set(index.current_sample_ids)
 
 
 def test_images_missing():
@@ -47,13 +375,13 @@ def test_images_missing():
 
     sample_ids = dataset[:4].values("id")
 
-    results = fob.compute_similarity(dataset, batch_size=1)
+    index = fob.compute_similarity(dataset, batch_size=1)
 
-    assert results.total_index_size == 4
-    assert set(sample_ids) == set(results.sample_ids)
+    assert index.total_index_size == 4
+    assert set(sample_ids) == set(index.sample_ids)
 
     model = foz.load_zoo_model("inception-v3-imagenet-torch")
-    results = fob.compute_similarity(
+    index = fob.compute_similarity(
         dataset,
         model=model,
         embeddings="embeddings_missing",
@@ -61,38 +389,38 @@ def test_images_missing():
     )
 
     assert len(dataset.exists("embeddings_missing")) == 4
-    assert results.index_size == 4
-    assert set(sample_ids) == set(results.sample_ids)
+    assert index.index_size == 4
+    assert set(sample_ids) == set(index.sample_ids)
 
 
 def test_patches():
     dataset = _load_patches_dataset()
 
-    results = dataset.load_brain_results("gt_sim")
+    index = dataset.load_brain_results("gt_sim")
 
     label_ids = dataset.values("ground_truth.detections.id", unwind=True)
 
-    assert results.total_index_size == len(label_ids)
-    assert set(label_ids) == set(results.label_ids)
+    assert index.total_index_size == len(label_ids)
+    assert set(label_ids) == set(index.label_ids)
 
 
 def test_patches_subset():
     dataset = _load_patches_dataset()
 
-    results = dataset.load_brain_results("gt_sim")
+    index = dataset.load_brain_results("gt_sim")
 
     label_ids = dataset.values("ground_truth.detections.id", unwind=True)
 
-    assert results.total_index_size == len(label_ids)
-    assert set(label_ids) == set(results.label_ids)
+    assert index.total_index_size == len(label_ids)
+    assert set(label_ids) == set(index.label_ids)
 
     view = dataset.filter_labels("ground_truth", F("label") == "person")
-    results.use_view(view)
+    index.use_view(view)
 
     label_ids = view.values("ground_truth.detections.id", unwind=True)
 
-    assert results.index_size == len(label_ids)
-    assert set(label_ids) == set(results.current_label_ids)
+    assert index.index_size == len(label_ids)
+    assert set(label_ids) == set(index.current_label_ids)
 
 
 def test_patches_missing():
@@ -112,18 +440,18 @@ def test_patches_missing():
         )
         sample.save()
 
-    results = fob.compute_similarity(
+    index = fob.compute_similarity(
         dataset, patches_field="ground_truth", batch_size=1
     )
 
     num_patches = dataset[:4].count("ground_truth.detections")
     label_ids = dataset[:4].values("ground_truth.detections.id", unwind=True)
 
-    assert results.total_index_size == num_patches
-    assert set(label_ids) == set(results.label_ids)
+    assert index.total_index_size == num_patches
+    assert set(label_ids) == set(index.label_ids)
 
     model = foz.load_zoo_model("inception-v3-imagenet-torch")
-    results = fob.compute_similarity(
+    index = fob.compute_similarity(
         dataset,
         model=model,
         patches_field="ground_truth",
@@ -136,8 +464,8 @@ def test_patches_missing():
     )
 
     assert view.count("ground_truth.detections") == num_patches
-    assert results.total_index_size == num_patches
-    assert set(label_ids) == set(results.label_ids)
+    assert index.total_index_size == num_patches
+    assert set(label_ids) == set(index.label_ids)
 
 
 def _load_images_dataset():