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

Issue:
Video brain runs touch up
Ensures that all brain runs either support video datasets or raise an informative error if they don't.

```py
import random
import numpy as np

import fiftyone as fo
import fiftyone.brain as fob
import fiftyone.zoo as foz
from fiftyone import ViewField as F

dataset = foz.load_zoo_dataset("quickstart-video", max_samples=2)
view = dataset.match_frames(F("frame_number") <= 2)

for sample in view.iter_samples(autosave=True):
    for frame in sample.frames.values():
        frame["pred_class"] = fo.Classification(
            label=random.choice(["foo", "bar", "spam", "eggs"]),
            confidence=random.random(),
            logits=np.random.random(4),
        )
        frame["pred_dets"] = frame["detections"].copy()
        for detection in frame["pred_dets"].detections:
            detection.confidence = random.random()

#
# These support video datasets
#

fob.compute_exact_duplicates(view)
fob.compute_hardness(view, "frames.pred_class")
fob.compute_mistakenness(view, "frames.pred_dets", "frames.detections")

#
# These methods do not directly support video datasets, but they do support
# frames views
#

fob.compute_similarity(view, brain_key="vid_sim")
# ValueError: This method cannot use image models to compute video embeddings. Try providing precomputed video embeddings or converting to a frames view via `to_frames()` first

fob.compute_similarity(view, patches_field="frames.detections", brain_key="det_sim")
# ValueError: This method does not directly support frame patches for video collections. Try converting to a frames view via `to_frames()` first

fob.compute_visualization(view, brain_key="vid_viz")
# ValueError: This method cannot use image models to compute video embeddings. Try providing precomputed video embeddings or converting to a frames view via `to_frames()` first

results7 = fob.compute_visualization(view, patches_field="frames.detections", brain_key="det_viz")
# ValueError: This method does not directly support frame patches for video collections. Try converting to a frames view via `to_frames()` first

view2 = view.to_frames(sample_frames=True, sparse=True)

fob.compute_similarity(view2, patches_field="detections", brain_key="frames_det_sim")
fob.compute_visualization(view2, patches_field="detections", brain_key="frames_det_viz")
```



Discussion / hints:

Fix patch:
diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index 64dab306..3dd0e53e 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -56,8 +56,8 @@ jobs:
       fail-fast: false
       matrix:
         python:
-          - 3.7
-          - 3.9
+          - 3.8
+          - "3.10"
     steps:
       - name: Clone fiftyone-brain
         uses: actions/checkout@v1
@@ -124,13 +124,8 @@ jobs:
         env:
           RELEASE_TAG: ${{ github.ref }}
         run: |
-          if [[ $RELEASE_TAG =~ ^refs\/tags\/v.*-rc\..*$ ]]; then
-             echo "TWINE_PASSWORD=${{ secrets.FIFTYONE_TEST_PYPI_TOKEN }}" >> $GITHUB_ENV
-             echo "TWINE_REPOSITORY=testpypi" >> $GITHUB_ENV
-          else
-             echo "TWINE_PASSWORD=${{ secrets.FIFTYONE_PYPI_TOKEN }}" >> $GITHUB_ENV
-             echo "TWINE_REPOSITORY=pypi" >> $GITHUB_ENV
-          fi
+          echo "TWINE_PASSWORD=${{ secrets.FIFTYONE_PYPI_TOKEN }}" >> $GITHUB_ENV
+          echo "TWINE_REPOSITORY=pypi" >> $GITHUB_ENV
       - name: Upload to pypi
         env:
           TWINE_USERNAME: __token__
diff --git a/docs/dev_guide.md b/docs/dev_guide.md
index 2f17eb78..f73154d2 100644
--- a/docs/dev_guide.md
+++ b/docs/dev_guide.md
@@ -71,4 +71,4 @@ from .hardness import compute_hardness
 
 ## Copyright
 
-Copyright 2017-2022, Voxel51, Inc.<br> voxel51.com
+Copyright 2017-2023, Voxel51, Inc.<br> voxel51.com
diff --git a/experiments/labelerror/README.md b/experiments/labelerror/README.md
index 2b035d10..a36fea61 100644
--- a/experiments/labelerror/README.md
+++ b/experiments/labelerror/README.md
@@ -12,5 +12,5 @@ This code does not explicitly use eta or theta because it is intended as somethi
 
 ## Copyright
 
-Copyright 2017-2022, Voxel51, Inc.<br>
+Copyright 2017-2023, Voxel51, Inc.<br>
 voxel51.com
diff --git a/experiments/labelerror/plot_run_scalar.py b/experiments/labelerror/plot_run_scalar.py
index e906d923..31ba8eba 100644
--- a/experiments/labelerror/plot_run_scalar.py
+++ b/experiments/labelerror/plot_run_scalar.py
@@ -6,7 +6,7 @@
 It can also plot data-files that have multiple fields of results and select one
 of them to plot per file.
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
diff --git a/experiments/labelerror/resnet_cifar10/labelerror.py b/experiments/labelerror/resnet_cifar10/labelerror.py
index d4b33c32..3126c765 100644
--- a/experiments/labelerror/resnet_cifar10/labelerror.py
+++ b/experiments/labelerror/resnet_cifar10/labelerror.py
@@ -19,7 +19,7 @@
 In addition, simple reporting on the accuracy of the identification of errors
 in the annotations
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
diff --git a/experiments/labelerror/resnet_cifar10/simple_resnet.py b/experiments/labelerror/resnet_cifar10/simple_resnet.py
index 75acfab4..af605571 100644
--- a/experiments/labelerror/resnet_cifar10/simple_resnet.py
+++ b/experiments/labelerror/resnet_cifar10/simple_resnet.py
@@ -4,7 +4,7 @@
 Original Implementation of this is from David Page's work on fast model
 training with resnets.  <https://github.com/davidcpage/cifar10-fast/>
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
diff --git a/experiments/uniqueness/README.md b/experiments/uniqueness/README.md
index de83ab36..34516a0f 100644
--- a/experiments/uniqueness/README.md
+++ b/experiments/uniqueness/README.md
@@ -8,4 +8,4 @@ Problem: Unlabeled
 
 ## Copyright
 
-Copyright 2017-2022, Voxel51, Inc.<br> voxel51.com
+Copyright 2017-2023, Voxel51, Inc.<br> voxel51.com
diff --git a/fiftyone/brain/__init__.py b/fiftyone/brain/__init__.py
index 19a9ad55..154848df 100644
--- a/fiftyone/brain/__init__.py
+++ b/fiftyone/brain/__init__.py
@@ -4,24 +4,30 @@
 
 See https://github.com/voxel51/fiftyone for more information.
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `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.
@@ -56,29 +62,26 @@ def compute_hardness(samples, label_field, hardness_field="hardness"):
 def compute_mistakenness(
     samples,
     pred_field,
-    label_field="ground_truth",
+    label_field,
     mistakenness_field="mistakenness",
     missing_field="possible_missing",
     spurious_field="possible_spurious",
     use_logits=False,
     copy_missing=False,
 ):
-    """Computes the mistakenness of the labels in the specified
-    ``label_field``, scoring the chance that the labels are incorrect.
-
-    Mistakenness is computed based on the predictions in the ``pred_field``,
-    through either their ``confidence`` or ``logits`` attributes. This measure
-    can be used to detect things like annotation errors and unusually hard
-    samples.
+    """Computes the mistakenness (likelihood of being incorrect) of the labels
+    in ``label_field`` based on the predcted labels in ``pred_field``.
 
-    This method supports both classifications and detections/polylines.
+    Mistakenness is measured based on either the ``confidence`` or ``logits``
+    of the predictions in ``pred_field``. This measure can be used to detect
+    things like annotation errors and unusually hard samples.
 
     For classifications, a ``mistakenness_field`` field is populated on each
     sample that quantifies the likelihood that the label in the ``label_field``
     of that sample is incorrect.
 
-    For detections/polylines, the mistakenness of each object in
-    ``label_field`` is computed, using
+    For objects (detections, polylines, keypoints, etc), the mistakenness of
+    each object in ``label_field`` is computed, using
     :meth:`fiftyone.core.collections.SampleCollection.evaluate_detections` to
     locate corresponding objects in ``pred_field``. Three types of mistakes
     are identified:
@@ -86,9 +89,9 @@ def compute_mistakenness(
     -   **(Mistakes)** Objects in ``label_field`` with a match in
         ``pred_field`` are assigned a mistakenness value in their
         ``mistakenness_field`` that captures the likelihood that the class
-        label of the detection in ``label_field`` is a mistake. A
+        label of the object in ``label_field`` is a mistake. A
         ``mistakenness_field + "_loc"`` field is also populated that captures
-        the likelihood that the detection in ``label_field`` is a mistake due
+        the likelihood that the object in ``label_field`` is a mistake due
         to its localization (bounding box).
 
     -   **(Missing)** Objects in ``pred_field`` with no matches in
@@ -101,8 +104,7 @@ def compute_mistakenness(
         ``pred_field`` but which are likely to be incorrect will have their
         ``spurious_field`` attribute set to True.
 
-    In addition, for detections/polylines, the following sample-level fields
-    are populated:
+    In addition, for objects, the following sample-level fields are populated:
 
     -   **(Mistakes)** The ``mistakenness_field`` of each sample is populated
         with the maximum mistakenness of the objects in ``label_field``
@@ -125,23 +127,25 @@ def compute_mistakenness(
         pred_field: the name of the predicted label field to use from each
             sample. Can be of type
             :class:`fiftyone.core.labels.Classification`,
-            :class:`fiftyone.core.labels.Classifications`, or
-            :class:`fiftyone.core.labels.Detections`
-        label_field ("ground_truth"): the name of the "ground truth" label
-            field that you want to test for mistakes with respect to the
-            predictions in ``pred_field``. Must have the same type as
-            ``pred_field``
+            :class:`fiftyone.core.labels.Classifications`,
+            :class:`fiftyone.core.labels.Detections`,
+            :class:`fiftyone.core.labels.Polylines`,
+            :class:`fiftyone.core.labels.Keypoints`, or
+            :class:`fiftyone.core.labels.TemporalDetections`
+        label_field: the name of the "ground truth" label field that you want
+            to test for mistakes with respect to the predictions in
+            ``pred_field``. Must have the same type as ``pred_field``
         mistakenness_field ("mistakenness"): the field name to use to store the
             mistakenness value for each sample
         missing_field ("possible_missing): the field in which to store
-            per-sample counts of potential missing detections/polylines
+            per-sample counts of potential missing objects
         spurious_field ("possible_spurious): the field in which to store
-            per-sample counts of potential spurious detections/polylines
+            per-sample counts of potential spurious objects
         use_logits (False): whether to use logits (True) or confidence (False)
             to compute mistakenness. Logits typically yield better results,
             when they are available
-        copy_missing (False): whether to copy predicted detections/polylines
-            that were deemed to be missing into ``label_field``
+        copy_missing (False): whether to copy predicted objects that were
+            deemed to be missing into ``label_field``
     """
     import fiftyone.brain.internal.core.mistakenness as fbm
 
@@ -192,17 +196,22 @@ def compute_uniqueness(
             :class:`fiftyone.core.labels.Polyline`, or
             :class:`fiftyone.core.labels.Polylines` field defining a region of
             interest within each image to use to compute uniqueness
-        embeddings (None): pre-computed embeddings to use. Can be any of the
-            following:
+        embeddings (None): if no ``model`` is provided, this argument specifies
+            pre-computed embeddings to use, which can be any of the following:
 
             -   a ``num_samples x num_dims`` array of embeddings
             -   if ``roi_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
 
+            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 ``roi_field``
         model (None): a :class:`fiftyone.core.models.Model` or the name of a
             model from the
-            `FiftyOne Model Zoo <https://voxel51.com/docs/fiftyone/user_guide/model_zoo/models.html>`_
+            `FiftyOne Model Zoo <https://docs.voxel51.com/user_guide/model_zoo/models.html>`_
             to use to generate embeddings. The model must expose embeddings
             (``model.has_embeddings = True``)
         force_square (False): whether to minimally manipulate the patch
@@ -259,12 +268,16 @@ 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.
 
-    If no ``embeddings`` or ``model`` is provided, a default model is used to
-    generate embeddings.
+    If no ``embeddings`` or ``model`` is provided, the following default model
+    is used to generate embeddings::
+
+        import fiftyone.zoo as foz
+
+        model = foz.load_zoo_model("mobilenet-v2-imagenet-torch")
 
     You can use the ``method`` parameter to select the dimensionality-reduction
     method to use, and you can optionally customize the method by passing
@@ -287,19 +300,35 @@ def compute_visualization(
             :class:`fiftyone.core.labels.Detections`,
             :class:`fiftyone.core.labels.Polyline`, or
             :class:`fiftyone.core.labels.Polylines`
-        embeddings (None): pre-computed embeddings to use. Can be any of the
-            following:
+        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,
+            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``
         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::
@@ -315,7 +344,7 @@ def compute_visualization(
             values are ``("umap", "tsne", "pca", "manual")``
         model (None): a :class:`fiftyone.core.models.Model` or the name of a
             model from the
-            `FiftyOne Model Zoo <https://voxel51.com/docs/fiftyone/user_guide/model_zoo/index.html>`_
+            `FiftyOne Model Zoo <https://docs.voxel51.com/user_guide/model_zoo/index.html>`_
             to use to generate embeddings. The model must expose embeddings
             (``model.has_embeddings = True``)
         force_square (False): whether to minimally manipulate the patch
@@ -367,36 +396,43 @@ 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
 
-    If no ``embeddings`` or ``model`` is provided, a default model is used to
-    generate embeddings.
+    If no ``embeddings`` or ``model`` is provided, the following default model
+    is used to generate embeddings::
+
+        import fiftyone.zoo as foz
+
+        model = foz.load_zoo_model("mobilenet-v2-imagenet-torch")
 
     Args:
         samples: a :class:`fiftyone.core.collections.SampleCollection`
@@ -406,21 +442,34 @@ def compute_similarity(
             :class:`fiftyone.core.labels.Detections`,
             :class:`fiftyone.core.labels.Polyline`, or
             :class:`fiftyone.core.labels.Polylines`
-        embeddings (None): 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 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://voxel51.com/docs/fiftyone/user_guide/model_zoo/index.html>`_
+            `FiftyOne Model Zoo <https://docs.voxel51.com/user_guide/model_zoo/index.html>`_
             to use to generate embeddings. The model must expose embeddings
             (``model.has_embeddings = True``)
         force_square (False): whether to minimally manipulate the patch
@@ -440,24 +489,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..a64ab5a6
--- /dev/null
+++ b/fiftyone/brain/config.py
@@ -0,0 +1,151 @@
+"""
+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.
+
+    Returns:
+        the path to the :class:`BrainConfig` on disk
+    """
+    if "FIFTYONE_BRAIN_CONFIG_PATH" not in os.environ:
+        return os.path.join(
+            os.path.expanduser("~"), ".fiftyone", "brain_config.json"
+        )
+
+    return os.environ["FIFTYONE_BRAIN_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/__init__.py b/fiftyone/brain/internal/__init__.py
index fff6c747..9ba90b68 100644
--- a/fiftyone/brain/internal/__init__.py
+++ b/fiftyone/brain/internal/__init__.py
@@ -3,7 +3,7 @@
 
 Contains all non-public code powering the ``fiftyone.brain`` public namespace.
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
diff --git a/fiftyone/brain/internal/core/__init__.py b/fiftyone/brain/internal/core/__init__.py
index c8dc42a4..8d756b31 100644
--- a/fiftyone/brain/internal/core/__init__.py
+++ b/fiftyone/brain/internal/core/__init__.py
@@ -1,5 +1,5 @@
 """
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
diff --git a/fiftyone/brain/internal/core/duplicates.py b/fiftyone/brain/internal/core/duplicates.py
index a356c490..ffed4005 100644
--- a/fiftyone/brain/internal/core/duplicates.py
+++ b/fiftyone/brain/internal/core/duplicates.py
@@ -1,7 +1,7 @@
 """
 Duplicates methods.
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
diff --git a/fiftyone/brain/internal/core/hardness.py b/fiftyone/brain/internal/core/hardness.py
index cde3609d..e4d5312a 100644
--- a/fiftyone/brain/internal/core/hardness.py
+++ b/fiftyone/brain/internal/core/hardness.py
@@ -1,7 +1,7 @@
 """
 Hardness methods.
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
@@ -13,6 +13,7 @@
 
 import fiftyone.core.brain as fob
 import fiftyone.core.labels as fol
+import fiftyone.core.media as fom
 import fiftyone.core.utils as fou
 import fiftyone.core.validation as fov
 
@@ -35,6 +36,9 @@ def compute_hardness(samples, label_field, hardness_field):
     fov.validate_collection(samples)
     fov.validate_collection_label_fields(samples, label_field, _ALLOWED_TYPES)
 
+    if samples.media_type == fom.VIDEO:
+        hardness_field, _ = samples._handle_frame_field(hardness_field)
+
     config = HardnessConfig(label_field, hardness_field)
     brain_key = hardness_field
     brain_method = config.build()
diff --git a/fiftyone/brain/internal/core/mistakenness.py b/fiftyone/brain/internal/core/mistakenness.py
index 84a4ed69..fb2535fe 100644
--- a/fiftyone/brain/internal/core/mistakenness.py
+++ b/fiftyone/brain/internal/core/mistakenness.py
@@ -1,7 +1,7 @@
 """
 Mistakenness methods.
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
@@ -15,6 +15,7 @@
 from fiftyone import ViewField as F
 import fiftyone.core.brain as fob
 import fiftyone.core.labels as fol
+import fiftyone.core.media as fom
 import fiftyone.core.utils as fou
 import fiftyone.core.validation as fov
 
@@ -27,6 +28,8 @@
     fol.Classifications,
     fol.Detections,
     fol.Polylines,
+    fol.Keypoints,
+    fol.TemporalDetections,
 )
 _MISSED_CONFIDENCE_THRESHOLD = 0.95
 _DETECTION_IOU = 0.5
@@ -63,16 +66,19 @@ def compute_mistakenness(
     # mistakenness, and low confidence predictions result in middling
     # mistakenness.
     #
-    # See the docstring above for additional handling of missing and spurious
-    # detections/polylines.
-    #
 
     fov.validate_collection_label_fields(
         samples, (pred_field, label_field), _ALLOWED_TYPES, same_type=True
     )
 
+    if samples.media_type == fom.VIDEO:
+        mistakenness_field, _ = samples._handle_frame_field(mistakenness_field)
+        missing_field, _ = samples._handle_frame_field(missing_field)
+        spurious_field, _ = samples._handle_frame_field(spurious_field)
+
     is_objects = samples._is_label_field(
-        pred_field, (fol.Detections, fol.Polylines)
+        pred_field,
+        (fol.Detections, fol.Polylines, fol.Keypoints, fol.TemporalDetections),
     )
     if is_objects:
         eval_key = _make_eval_key(samples, mistakenness_field)
diff --git a/fiftyone/brain/internal/core/pinecone.py b/fiftyone/brain/internal/core/pinecone.py
new file mode 100644
index 00000000..f5c3f4b1
--- /dev/null
+++ b/fiftyone/brain/internal/core/pinecone.py
@@ -0,0 +1,589 @@
+"""
+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,
+        reload=True,
+        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,
+            )
+
+        if reload:
+            self.reload()
+
+    def remove_from_index(
+        self,
+        sample_ids=None,
+        label_ids=None,
+        allow_missing=True,
+        warn_missing=False,
+        reload=True,
+    ):
+        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)
+
+        if reload:
+            self.reload()
+
+    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..ecf58eb3
--- /dev/null
+++ b/fiftyone/brain/internal/core/qdrant.py
@@ -0,0 +1,620 @@
+"""
+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,
+        reload=True,
+        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,
+                ),
+            )
+
+        if reload:
+            self.reload()
+
+    def remove_from_index(
+        self,
+        sample_ids=None,
+        label_ids=None,
+        allow_missing=True,
+        warn_missing=False,
+        reload=True,
+    ):
+        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),
+        )
+
+        if reload:
+            self.reload()
+
+    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 8e353cff..00000000
--- a/fiftyone/brain/internal/core/similarity.py
+++ /dev/null
@@ -1,911 +0,0 @@
-"""
-Similarity methods.
-
-| Copyright 2017-2022, 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 = foz.load_zoo_model(_DEFAULT_MODEL)
-        if batch_size is None:
-            batch_size = _DEFAULT_BATCH_SIZE
-
-    if etau.is_str(embeddings):
-        embeddings_field = embeddings
-        embeddings = None
-    else:
-        embeddings_field = None
-
-    config = SimilarityConfig(
-        embeddings_field=embeddings_field,
-        model=model,
-        patches_field=patches_field,
-        metric=metric,
-    )
-    brain_method = config.build()
-    brain_method.ensure_requirements()
-
-    if brain_key is not None:
-        brain_method.register_run(samples, brain_key)
-
-    embeddings = 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)
-    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_ids, k, reverse, aggregation, dist_field, mongo
-):
-    _ensure_neighbors(results)
-
-    samples = results.view
-    sample_ids = results._curr_sample_ids
-    label_ids = results._curr_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 etau.is_str(query_ids):
-        query_ids = [query_ids]
-
-    if not query_ids:
-        raise ValueError("At least one query ID must be provided")
-
-    if aggregation not in _AGGREGATIONS:
-        raise ValueError(
-            "Unsupported aggregation method '%s'. Supported values are %s"
-            % (aggregation, tuple(_AGGREGATIONS.keys()))
-        )
-
-    #
-    # Parse query (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 sorting
-    #
-
-    dists = results._neighbors_helper.get_distances()
-
-    if dists is not None:
-        if keep_inds is not None:
-            dists = dists[keep_inds, :]
-
-        dists = dists[:, query_inds]
-    else:
-        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]
-
-    #
-    # Store query distances
-    #
-
-    if dist_field is not None:
-        if selecting_samples:
-            values = {sample_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(sample_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 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._curr_label_ids
-        logger.info("Computing duplicate patches...")
-    else:
-        ids = results._curr_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._curr_label_ids
-        logger.info("Computing unique patches...")
-    else:
-        ids = results._curr_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 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
-
-    if patches_field is not None:
-        _, id_path = samples._get_label_field_path(patches_field, "id")
-        ids = samples.values(id_path, unwind=True)
-    else:
-        ids = samples.values("id")
-
-    dup_ids = set(duplicate_ids)
-    nearest_ids = set(neighbors_map.keys())
-
-    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"
-
-    with visualization.use_view(samples):
-        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
-
-    if patches_field is not None:
-        _, id_path = samples._get_label_field_path(patches_field, "id")
-        ids = samples.values(id_path, unwind=True)
-    else:
-        ids = samples.values("id")
-
-    _unique_ids = set(unique_ids)
-
-    labels = []
-    for _id in ids:
-        if _id in _unique_ids:
-            label = "unique"
-        else:
-            label = "other"
-
-        labels.append(label)
-
-    with visualization.use_view(samples):
-        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..6fc529a3
--- /dev/null
+++ b/fiftyone/brain/internal/core/sklearn.py
@@ -0,0 +1,847 @@
+"""
+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,
+        reload=True,
+    ):
+        _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
+
+        if reload:
+            self.reload()
+
+    def remove_from_index(
+        self,
+        sample_ids=None,
+        label_ids=None,
+        allow_missing=True,
+        warn_missing=False,
+        reload=True,
+    ):
+        _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
+
+        if reload:
+            self.reload()
+
+    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/uniqueness.py b/fiftyone/brain/internal/core/uniqueness.py
index 6d85632a..864cbe5a 100644
--- a/fiftyone/brain/internal/core/uniqueness.py
+++ b/fiftyone/brain/internal/core/uniqueness.py
@@ -1,7 +1,7 @@
 """
 Uniqueness methods.
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
@@ -25,7 +25,6 @@
 
 logger = logging.getLogger(__name__)
 
-
 _ALLOWED_ROI_FIELD_TYPES = (
     fol.Detection,
     fol.Detections,
@@ -65,23 +64,25 @@ def compute_uniqueness(
     # to dense clusters of related samples.
     #
 
-    fov.validate_collection(samples)
+    fov.validate_image_collection(samples)
 
     if roi_field is not None:
         fov.validate_collection_label_fields(
             samples, roi_field, _ALLOWED_ROI_FIELD_TYPES
         )
 
-    if samples.media_type == fom.VIDEO:
-        raise ValueError("Uniqueness does not yet support video collections")
-
     if model is None and embeddings is None:
         model = fbm.load_model(_DEFAULT_MODEL)
         if batch_size is None:
             batch_size = _DEFAULT_BATCH_SIZE
 
     if etau.is_str(embeddings):
-        embeddings_field = embeddings
+        embeddings_field = fbu.parse_embeddings_field(
+            samples,
+            embeddings,
+            patches_field=roi_field,
+            allow_embedded=model is None,
+        )
         embeddings = None
     else:
         embeddings_field = None
@@ -103,7 +104,7 @@ def compute_uniqueness(
     else:
         agg_fcn = None
 
-    embeddings = fbu.get_embeddings(
+    embeddings, sample_ids, _ = fbu.get_embeddings(
         samples,
         model=model,
         patches_field=roi_field,
@@ -121,8 +122,12 @@ def compute_uniqueness(
     logger.info("Computing uniqueness...")
     uniqueness = _compute_uniqueness(embeddings)
 
+    # Ensure field exists, even if `uniqueness` is empty
     samples._dataset.add_sample_field(uniqueness_field, fof.FloatField)
-    samples.set_values(uniqueness_field, uniqueness)
+
+    uniqueness = {_id: u for _id, u in zip(sample_ids, uniqueness)}
+    if uniqueness:
+        samples.set_values(uniqueness_field, uniqueness, key_field="id")
 
     brain_method.save_run_results(samples, brain_key, None)
 
@@ -130,11 +135,12 @@ def compute_uniqueness(
 
 
 def _compute_uniqueness(embeddings, metric="euclidean"):
-    # @todo convert to a parameter with a default, for tuning
     K = 3
 
     num_embeddings = len(embeddings)
-    if num_embeddings <= _MAX_PRECOMPUTE_DISTS:
+    if num_embeddings <= K:
+        return [1] * num_embeddings
+    elif num_embeddings <= _MAX_PRECOMPUTE_DISTS:
         embeddings = skm.pairwise_distances(embeddings, metric=metric)
         metric = "precomputed"
     else:
diff --git a/fiftyone/brain/internal/core/utils.py b/fiftyone/brain/internal/core/utils.py
index af5a68f1..486aedef 100644
--- a/fiftyone/brain/internal/core/utils.py
+++ b/fiftyone/brain/internal/core/utils.py
@@ -1,16 +1,24 @@
 """
 Utilities.
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
+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.models as fom
+import fiftyone.core.media as fomm
 import fiftyone.core.patches as fop
 import fiftyone.zoo as foz
 from fiftyone import ViewField as F
@@ -19,88 +27,682 @@
 logger = logging.getLogger(__name__)
 
 
-def get_ids(samples, patches_field=None):
+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,
+        )
+
+    _validate_args(samples, patches_field=patches_field)
+
     if patches_field is None:
-        sample_ids = np.array(samples.values("id"))
-        return sample_ids, 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 = []
-    for l in samples._get_selected_labels(fields=patches_field):
-        sample_ids.append(l["sample_id"])
-        label_ids.append(l["label_id"])
+    sample_ids, label_ids = get_ids(
+        samples,
+        patches_field=patches_field,
+        data=data,
+        data_type=data_type,
+    )
 
-    return np.array(sample_ids), np.array(label_ids)
+    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],
+            )
 
-def filter_ids(view, samples, sample_ids, label_ids, patches_field=None):
-    # No filtering required
-    if view == samples or view.view() == samples.view():
-        return view, sample_ids, label_ids, None
+    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:
-        _sample_ids = view.values("id")
-        keep_inds = _get_keep_inds(_sample_ids, sample_ids)
-        return view, np.array(_sample_ids), None, keep_inds
+        if samples._is_patches:
+            sample_id_path = "sample_id"
+        else:
+            sample_id_path = "id"
 
-    # Filter labels in patches view
+        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"
+            )
 
-    if (
-        isinstance(view, fop.PatchesView)
-        and patches_field != view.patches_field
-    ):
+        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,
+    data=None,
+    data_type="embeddings",
+    handle_missing="skip",
+    ref_sample_ids=None,
+):
+    _validate_args(samples, patches_field=patches_field)
+
+    if patches_field is None:
+        if ref_sample_ids is not None:
+            sample_ids = ref_sample_ids
+        else:
+            sample_ids = samples.values("id")
+
+        if data is not None and len(sample_ids) != len(data):
+            raise ValueError(
+                "The number of %s (%d) in these results no longer matches the "
+                "number of samples (%d) in the collection. You must "
+                "regenerate the results"
+                % (data_type, len(data), len(sample_ids))
+            )
+
+        return np.array(sample_ids), None
+
+    sample_ids, label_ids = _get_patch_ids(
+        samples,
+        patches_field,
+        handle_missing=handle_missing,
+        ref_sample_ids=ref_sample_ids,
+    )
+
+    if data is not None and len(sample_ids) != len(data):
         raise ValueError(
-            "This patches view contains labels from field '%s', not "
-            "'%s'" % (view.patches_field, patches_field)
+            "The number of %s (%d) in these results no longer matches the "
+            "number of labels (%d) in the '%s' field of the collection. You "
+            "must regenerate the results"
+            % (data_type, len(data), len(sample_ids), patches_field)
         )
 
-    if isinstance(view, fop.EvaluationPatchesView) and patches_field not in (
-        view.gt_field,
-        view.pred_field,
-    ):
-        raise ValueError(
-            "This evaluation patches view contains patches from "
-            "fields '%s' and '%s', not '%s'"
-            % (view.gt_field, view.pred_field, patches_field)
+    return np.array(sample_ids), np.array(label_ids)
+
+
+def filter_ids(
+    samples,
+    index_sample_ids,
+    index_label_ids,
+    patches_field=None,
+    allow_missing=True,
+    warn_missing=False,
+):
+    _validate_args(samples, patches_field=patches_field)
+
+    if patches_field is None:
+        if samples._is_patches:
+            sample_ids = np.array(samples.values("sample_id"))
+        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,
+            "samples",
+            allow_missing,
+            warn_missing,
+        )
+
+        if bad_ids is not None:
+            sample_ids = sample_ids[good_inds]
+
+        return sample_ids, None, keep_inds, good_inds
+
+    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,
+        "labels",
+        allow_missing,
+        warn_missing,
+    )
+
+    if bad_ids is not None:
+        sample_ids = sample_ids[good_inds]
+        label_ids = label_ids[good_inds]
+
+    return sample_ids, label_ids, keep_inds, good_inds
+
+
+def _get_patch_ids(
+    samples, patches_field, handle_missing="skip", ref_sample_ids=None
+):
+    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)
+
+    sample_ids, label_ids = samples.values([sample_id_path, label_id_path])
+
+    if ref_sample_ids is not None:
+        sample_ids, label_ids = _apply_ref_sample_ids(
+            sample_ids, label_ids, ref_sample_ids
         )
 
-    labels = view._get_selected_labels(fields=patches_field)
-    _sample_ids = [l["sample_id"] for l in labels]
-    _label_ids = [l["label_id"] for l in labels]
-    keep_inds = _get_keep_inds(_label_ids, label_ids)
-    return view, np.array(_sample_ids), np.array(_label_ids), keep_inds
+    if is_list_field:
+        sample_ids, label_ids = _flatten_list_ids(
+            sample_ids, label_ids, handle_missing
+        )
+
+    return np.array(sample_ids), np.array(label_ids)
+
+
+def _apply_ref_sample_ids(sample_ids, label_ids, ref_sample_ids):
+    ref_label_ids = [None] * len(ref_sample_ids)
+    inds_map = {_id: i for i, _id in enumerate(ref_sample_ids)}
+    for _id, _lid in zip(sample_ids, label_ids):
+        idx = inds_map.get(_id, None)
+        if idx is not None:
+            ref_label_ids[idx] = _lid
+
+    return ref_sample_ids, ref_label_ids
+
+
+def _flatten_list_ids(sample_ids, label_ids, handle_missing):
+    _sample_ids = []
+    _label_ids = []
+    _add_missing = handle_missing == "image"
+
+    for _id, _lids in zip(sample_ids, label_ids):
+        if _lids:
+            for _lid in _lids:
+                _sample_ids.append(_id)
+                _label_ids.append(_lid)
+        elif _add_missing:
+            _sample_ids.append(_id)
+            _label_ids.append(None)
 
+    return _sample_ids, _label_ids
 
-def _get_keep_inds(ids, ref_ids):
-    if len(ids) == len(ref_ids) and list(ids) == list(ref_ids):
-        return None
 
-    inds_map = {_id: idx for idx, _id in enumerate(ref_ids)}
+def _parse_ids(ids, index_ids, ftype, allow_missing, warn_missing):
+    if np.array_equal(ids, index_ids):
+        return None, None, None
+
+    inds_map = {_id: idx for idx, _id in enumerate(index_ids)}
 
     keep_inds = []
+    bad_inds = []
     bad_ids = []
-    for _id in ids:
+    for _idx, _id in enumerate(ids):
         ind = inds_map.get(_id, None)
         if ind is not None:
             keep_inds.append(ind)
         else:
+            bad_inds.append(_idx)
             bad_ids.append(_id)
 
-    num_bad = len(bad_ids)
+    num_missing_index = len(index_ids) - len(keep_inds)
+    if num_missing_index > 0:
+        if not allow_missing:
+            raise ValueError(
+                "The index contains %d %s that are not present in the "
+                "provided collection" % (num_missing_index, ftype)
+            )
+
+        if warn_missing:
+            logger.warning(
+                "Ignoring %d %s from the index that are not present in the "
+                "provided collection",
+                num_missing_index,
+                ftype,
+            )
+
+    num_missing_collection = len(bad_ids)
+    if num_missing_collection > 0:
+        if not allow_missing:
+            raise ValueError(
+                "The provided collection contains %d %s not present in the "
+                "index" % (num_missing_collection, ftype)
+            )
+
+        if warn_missing:
+            logger.warning(
+                "Ignoring %d %s from the provided collection that are not "
+                "present in the index",
+                num_missing_collection,
+                ftype,
+            )
+
+        bad_inds = np.array(bad_inds, dtype=np.int64)
+
+        good_inds = np.full(ids.shape, True)
+        good_inds[bad_inds] = False
+    else:
+        good_inds = None
+        bad_ids = None
+
+    keep_inds = np.array(keep_inds, dtype=np.int64)
+
+    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")
+
 
-    if num_bad == 1:
+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))
+    else:
+        _values = values
+
+    _values = np.asarray(_values)
+
+    if _values.size == keep_inds.size:
+        _values = _values[keep_inds]
+    else:
+        num_expected = np.count_nonzero(keep_inds)
+        if _values.size != num_expected:
+            raise ValueError(
+                "Expected %d raw values or %d pre-filtered values; found %d "
+                "values" % (keep_inds.size, num_expected, values.size)
+            )
+
+    # @todo we might need to re-ravel patch values here in the future
+    # We currently do not do this because all downstream users of this data
+    # will gracefully handle either flat or nested list data
+
+    return _values
+
+
+def get_values(samples, path_or_expr, ids, patches_field=None):
+    _validate_args(
+        samples, patches_field=patches_field, path_or_expr=path_or_expr
+    )
+    return samples._get_values_by_id(
+        path_or_expr, ids, link_field=patches_field
+    )
+
+
+def parse_embeddings_field(
+    samples, embeddings_field, patches_field=None, allow_embedded=True
+):
+    if not etau.is_str(embeddings_field):
         raise ValueError(
-            "The provided view contains ID '%s' not present in the index"
-            % bad_ids[0]
+            "Invalid embeddings_field=%s; expected a string field name"
+            % embeddings_field
         )
 
-    if num_bad > 1:
+    if patches_field is None:
+        _embeddings_field, is_frame_field = samples._handle_frame_field(
+            embeddings_field
+        )
+
+        if not allow_embedded and "." in _embeddings_field:
+            ftype = "frame" if is_frame_field else "sample"
+            raise ValueError(
+                "Invalid embeddings_field=%s; expected a top-level %s field "
+                "name that contains no '.'" % (_embeddings_field, ftype)
+            )
+
+        return embeddings_field
+
+    if embeddings_field.startswith(patches_field + "."):
+        _, root = samples._get_label_field_path(patches_field) + "."
+        if not embeddings_field.startswith(root):
+            raise ValueError(
+                "Invalid embeddings_field=%s for patches_field=%s"
+                % (embeddings_field, patches_field)
+            )
+
+        embeddings_field = embeddings_field[len(root) + 1]
+
+    if not allow_embedded and "." in embeddings_field:
         raise ValueError(
-            "The provided view contains %d IDs (eg '%s') not present in the "
-            "index" % (num_bad, bad_ids[0])
+            "Invalid embeddings_field=%s for patches_field=%s; expected a "
+            "label attribute name that contains no '.'"
+            % (embeddings_field, patches_field)
         )
 
-    return np.array(keep_inds, dtype=np.int64)
+    return embeddings_field
 
 
 def get_embeddings(
@@ -117,10 +719,30 @@ def get_embeddings(
     num_workers=None,
     skip_failures=True,
 ):
-    if model is not None:
+    _validate_args(samples, patches_field=patches_field)
+
+    if model is None and embeddings_field is None and embeddings is None:
+        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 embeddings is None and model is not None:
         if etau.is_str(model):
             model = foz.load_zoo_model(model)
 
+        if not isinstance(model, fom.Model):
+            raise ValueError(
+                "Model must be a %s; found %s" % (fom.Model, type(model))
+            )
+
         if patches_field is not None:
             logger.info("Computing patch embeddings...")
             embeddings = samples.compute_patch_embeddings(
@@ -135,6 +757,16 @@ def get_embeddings(
                 skip_failures=skip_failures,
             )
         else:
+            if (
+                samples.media_type == fomm.VIDEO
+                and model.media_type == fomm.IMAGE
+            ):
+                raise ValueError(
+                    "This method cannot use image models to compute video "
+                    "embeddings. Try providing precomputed video embeddings "
+                    "or converting to a frames view via `to_frames()` first"
+                )
+
             logger.info("Computing embeddings...")
             embeddings = samples.compute_embeddings(
                 model,
@@ -143,85 +775,181 @@ def get_embeddings(
                 num_workers=num_workers,
                 skip_failures=skip_failures,
             )
-    elif embeddings_field is not None:
-        embeddings = samples.values(embeddings_field)
 
-    if embeddings is None:
-        raise ValueError(
-            "One of `model`, `embeddings_field`, or `embeddings` must be "
-            "provided"
+    if embeddings is None and embeddings_field is not None:
+        embeddings, samples = _load_embeddings(
+            samples, embeddings_field, patches_field=patches_field
+        )
+        ref_sample_ids = None
+    else:
+        if isinstance(embeddings, dict):
+            embeddings = [
+                embeddings.get(_id, None) for _id in samples.values("id")
+            ]
+
+        embeddings, ref_sample_ids = _handle_missing_embeddings(
+            embeddings, samples
         )
 
-    if isinstance(embeddings, dict):
-        embeddings = [
-            embeddings.get(_id, None) for _id in samples.values("id")
-        ]
+    if not isinstance(embeddings, np.ndarray) and not embeddings:
+        return _empty_embeddings(patches_field)
 
     if patches_field is not None:
-        _handle_missing_patch_embeddings(embeddings, samples, patches_field)
-
         if agg_fcn is not None:
-            embeddings = [agg_fcn(e) for e in embeddings]
-            embeddings = np.stack(embeddings)
+            embeddings = np.stack([agg_fcn(e) for e in embeddings])
         else:
             embeddings = np.concatenate(embeddings, axis=0)
+    elif not isinstance(embeddings, np.ndarray):
+        embeddings = np.stack(embeddings)
+
+    if agg_fcn is not None:
+        patches_field = None
+
+    sample_ids, label_ids = get_ids(
+        samples,
+        patches_field=patches_field,
+        data=embeddings,
+        data_type="embeddings",
+        handle_missing=handle_missing,
+        ref_sample_ids=ref_sample_ids,
+    )
+
+    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:
-        _handle_missing_embeddings(embeddings)
+        label_ids = None
 
-        if agg_fcn is not None:
-            embeddings = [agg_fcn(e) for e in embeddings]
+    return embeddings, sample_ids, label_ids
 
-        embeddings = np.stack(embeddings)
 
-    return embeddings
+def _load_embeddings(samples, embeddings_field, patches_field=None):
+    if patches_field is not None:
+        label_type, embeddings_path = samples._get_label_field_path(
+            patches_field, embeddings_field
+        )
+        is_list_field = issubclass(label_type, fol._LABEL_LIST_FIELDS)
+    else:
+        embeddings_path = embeddings_field
+        is_list_field = False
 
+    if is_list_field:
+        samples = samples.filter_labels(
+            patches_field, F(embeddings_field) != None
+        )
+    else:
+        samples = samples.match(F(embeddings_path) != None)
 
-def _handle_missing_embeddings(embeddings):
-    if isinstance(embeddings, np.ndarray):
-        return
+    if samples.has_field(embeddings_path):
+        _field = None
+    else:
+        _field = fof.VectorField()
 
-    missing_inds = []
-    num_dims = None
-    for idx, embedding in enumerate(embeddings):
-        if embedding is None:
-            missing_inds.append(idx)
-        elif num_dims is None:
-            num_dims = embedding.size
+    embeddings = samples.values(embeddings_path, _field=_field)
 
-    if not missing_inds:
+    if is_list_field:
+        embeddings = [np.stack(e) for e in embeddings if e]
+
+    return embeddings, samples
+
+
+def _validate_args(samples, patches_field=None, path_or_expr=None):
+    if patches_field is not None:
+        _validate_patches_args(
+            samples, patches_field, path_or_expr=path_or_expr
+        )
+    else:
+        _validate_samples_args(samples, path_or_expr=path_or_expr)
+
+
+def _validate_samples_args(samples, path_or_expr=None):
+    if not etau.is_str(path_or_expr):
         return
 
-    missing_embedding = np.zeros(num_dims or 16)
-    for idx in missing_inds:
-        embeddings[idx] = missing_embedding.copy()
+    path, _, list_fields, _, _ = samples._parse_field_name(path_or_expr)
+
+    if list_fields:
+        raise ValueError(
+            "Values path '%s' contains invalid list field '%s'"
+            % (path, list_fields[0])
+        )
 
-    logger.warning("Using zeros for %d missing embeddings", len(missing_inds))
 
+def _validate_patches_args(samples, patches_field, path_or_expr=None):
+    if samples.media_type == fomm.VIDEO:
+        raise ValueError(
+            "This method does not directly support frame patches for video "
+            "collections. Try converting to a frames view via `to_frames()` "
+            "first"
+        )
+
+    if etau.is_str(path_or_expr) and not path_or_expr.startswith(
+        patches_field + "."
+    ):
+        raise ValueError(
+            "Values path '%s' must start with patches field '%s'"
+            % (path_or_expr, patches_field)
+        )
+
+    if (
+        isinstance(samples, fop.PatchesView)
+        and patches_field != samples.patches_field
+    ):
+        raise ValueError(
+            "This patches view contains labels from field '%s', not "
+            "'%s'" % (samples.patches_field, patches_field)
+        )
+
+    if isinstance(
+        samples, fop.EvaluationPatchesView
+    ) and patches_field not in (
+        samples.gt_field,
+        samples.pred_field,
+    ):
+        raise ValueError(
+            "This evaluation patches view contains patches from "
+            "fields '%s' and '%s', not '%s'"
+            % (samples.gt_field, samples.pred_field, patches_field)
+        )
+
+
+def _handle_missing_embeddings(embeddings, samples):
+    if isinstance(embeddings, np.ndarray):
+        return embeddings, None
 
-def _handle_missing_patch_embeddings(embeddings, samples, patches_field):
     missing_inds = []
-    num_dims = None
     for idx, embedding in enumerate(embeddings):
         if embedding is None:
             missing_inds.append(idx)
-        elif num_dims is None:
-            num_dims = embedding.shape[1]
 
     if not missing_inds:
-        return
+        return embeddings, None
 
-    missing_embedding = np.zeros(num_dims or 16)
+    embeddings = [e for e in embeddings if e is not None]
+    ref_sample_ids = list(np.delete(samples.values("id"), missing_inds))
 
-    _, labels_path = samples._get_label_field_path(patches_field)
-    patch_counts = samples.values(F(labels_path).length())
-
-    num_missing = 0
-    for idx in missing_inds:
-        count = patch_counts[idx]
-        embeddings[idx] = np.tile(missing_embedding, (count, 1))
-        num_missing += count
-
-    if num_missing > 0:
-        logger.warning(
-            "Using zeros for %d missing patch embeddings", num_missing
-        )
+    return embeddings, ref_sample_ids
diff --git a/fiftyone/brain/internal/core/visualization.py b/fiftyone/brain/internal/core/visualization.py
index d972caf8..e26a8b2d 100644
--- a/fiftyone/brain/internal/core/visualization.py
+++ b/fiftyone/brain/internal/core/visualization.py
@@ -1,7 +1,7 @@
 """
 Visualization methods.
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
@@ -14,6 +14,8 @@
 import eta.core.utils as etau
 
 import fiftyone.core.brain as fob
+import fiftyone.core.expressions as foe
+import fiftyone.core.plots as fop
 import fiftyone.core.utils as fou
 import fiftyone.core.validation as fov
 import fiftyone.zoo as foz
@@ -32,7 +34,6 @@
 
 logger = logging.getLogger(__name__)
 
-
 _DEFAULT_MODEL = "mobilenet-v2-imagenet-torch"
 _DEFAULT_BATCH_SIZE = None
 
@@ -69,12 +70,17 @@ 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
 
     if etau.is_str(embeddings):
-        embeddings_field = embeddings
+        embeddings_field = fbu.parse_embeddings_field(
+            samples,
+            embeddings,
+            patches_field=patches_field,
+            allow_embedded=model is None,
+        )
         embeddings = None
     else:
         embeddings_field = None
@@ -90,7 +96,7 @@ def compute_visualization(
         brain_method.register_run(samples, brain_key)
 
     if points is None:
-        embeddings = fbu.get_embeddings(
+        embeddings, sample_ids, label_ids = fbu.get_embeddings(
             samples,
             model=model,
             patches_field=patches_field,
@@ -105,17 +111,97 @@ def compute_visualization(
 
         logger.info("Generating visualization...")
         points = brain_method.fit(embeddings)
+    else:
+        points, sample_ids, label_ids = fbu.parse_data(
+            samples,
+            patches_field=patches_field,
+            data=points,
+            data_type="points",
+        )
+
+    results = VisualizationResults(
+        samples,
+        config,
+        brain_key,
+        points,
+        sample_ids=sample_ids,
+        label_ids=label_ids,
+    )
 
-    results = VisualizationResults(samples, config, points)
     brain_method.save_run_results(samples, brain_key, results)
 
     return results
 
 
-class Visualization(fob.BrainMethod):
-    def ensure_requirements(self):
-        pass
+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(
+    results,
+    labels=None,
+    sizes=None,
+    classes=None,
+    backend="plotly",
+    **kwargs,
+):
+    points = results.current_points
+    samples = results.view
+    patches_field = results.config.patches_field
+    good_inds = results._curr_good_inds
+    if patches_field is not None:
+        ids = results.current_label_ids
+    else:
+        ids = results.current_sample_ids
+
+    if good_inds is not None:
+        if etau.is_container(labels) and not _is_expr(labels):
+            labels = fbu.filter_values(
+                labels, good_inds, patches_field=patches_field
+            )
+
+        if etau.is_container(sizes) and not _is_expr(sizes):
+            sizes = fbu.filter_values(
+                sizes, good_inds, patches_field=patches_field
+            )
+
+    if labels is not None and _is_expr(labels):
+        labels = fbu.get_values(
+            samples, labels, ids, patches_field=patches_field
+        )
+
+    if sizes is not None and _is_expr(sizes):
+        sizes = fbu.get_values(
+            samples, sizes, ids, patches_field=patches_field
+        )
+
+    return fop.scatterplot(
+        points,
+        samples=samples,
+        ids=ids,
+        link_field=patches_field,
+        labels=labels,
+        sizes=sizes,
+        classes=classes,
+        backend=backend,
+        **kwargs,
+    )
+
 
+def _is_expr(arg):
+    return isinstance(arg, (foe.ViewExpression, dict))
+
+
+class Visualization(fob.BrainMethod):
     def fit(self, embeddings):
         raise NotImplementedError("subclass must implement fit()")
 
@@ -126,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/internal/models/__init__.py b/fiftyone/brain/internal/models/__init__.py
index 4b9f39e5..93b5040f 100644
--- a/fiftyone/brain/internal/models/__init__.py
+++ b/fiftyone/brain/internal/models/__init__.py
@@ -1,7 +1,7 @@
 """
 Brain models.
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
diff --git a/fiftyone/brain/internal/models/simple_resnet.py b/fiftyone/brain/internal/models/simple_resnet.py
index 5317c06a..8ed489f5 100644
--- a/fiftyone/brain/internal/models/simple_resnet.py
+++ b/fiftyone/brain/internal/models/simple_resnet.py
@@ -4,7 +4,7 @@
 The original implementation of this is from David Page's work on fast model
 training with resnets at https://github.com/davidcpage/cifar10-fast.
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
diff --git a/fiftyone/brain/internal/models/torch.py b/fiftyone/brain/internal/models/torch.py
index 049d7610..2b7c7abe 100644
--- a/fiftyone/brain/internal/models/torch.py
+++ b/fiftyone/brain/internal/models/torch.py
@@ -1,7 +1,7 @@
 """
 PyTorch utilities.
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
diff --git a/fiftyone/brain/similarity.py b/fiftyone/brain/similarity.py
index be4b2705..dbe50a63 100644
--- a/fiftyone/brain/similarity.py
+++ b/fiftyone/brain/similarity.py
@@ -1,125 +1,468 @@
 """
 Similarity interface.
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `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)
+
+    results = brain_method.initialize(samples, brain_key)
+
+    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
+
+    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: a ``num_embeddings x num_dims`` array of embeddings
+        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):
-        sample_ids, label_ids = fbu.get_ids(
-            samples, patches_field=config.patches_field
+    def __init__(
+        self,
+        embeddings_field=None,
+        model=None,
+        patches_field=None,
+        supports_prompts=None,
+        **kwargs,
+    ):
+        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)
+
+    @property
+    def method(self):
+        """The name of the similarity backend."""
+        raise NotImplementedError("subclass must implement method")
+
+    @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"
         )
 
-        if len(sample_ids) != len(embeddings):
-            ptype = "label" if config.patches_field is not None else "sample"
-            raise ValueError(
-                "Number of %s IDs (%d) does not match number of embeddings "
-                "(%d). You may have missing data/labels that you need to omit "
-                "from your view" % (ptype, len(sample_ids), len(embeddings))
-            )
+    @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"
+        )
 
-        self.embeddings = embeddings
+    def load_credentials(self, **kwargs):
+        self._load_parameters(**kwargs)
 
-        self._samples = samples
-        self._config = config
-        self._sample_ids = sample_ids
-        self._label_ids = label_ids
-        self._last_view = None
+    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._neighbors_helper = None
-        self._thresh = None
-        self._unique_ids = None
-        self._duplicate_ids = None
-        self._neighbors_map = 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 examples 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):
+        """The total number of data points in the index.
+
+        If :meth:`use_view` has been called to restrict the index, this value
+        may be larger than the current :meth:`index_size`.
+        """
+        raise NotImplementedError("subclass must implement total_index_size")
 
     @property
     def view(self):
         """The :class:`fiftyone.core.collections.SampleCollection` against
         which results are currently being generated.
 
-        If :meth:`use_view` has been called, this view may be a subset of the
-        collection on which the full index was generated.
+        If :meth:`use_view` has been called, this view may be different than
+        the collection on which the full index was generated.
         """
         return self._curr_view
 
     @property
-    def thresh(self):
-        """The threshold used by the last call to :meth:`find_duplicates` or
-        :meth:`find_unique`.
+    def current_sample_ids(self):
+        """The sample IDs of the currently active data points in the index.
+
+        If :meth:`use_view` has been called, this may be a subset of the full
+        index.
         """
-        return self._thresh
+        return self._curr_sample_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_label_ids(self):
+        """The label IDs of the currently active data points in the index, or
+        ``None`` if not applicable.
+
+        If :meth:`use_view` has been called, this may be a subset of the full
+        index.
         """
-        return self._unique_ids
+        return self._curr_label_ids
 
     @property
-    def duplicate_ids(self):
-        """A list of duplicate 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._duplicate_ids
+        return self._curr_keep_inds
 
     @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 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._neighbors_map
+        return len(self._curr_sample_ids)
+
+    @property
+    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._curr_missing_size
 
-    def use_view(self, sample_collection):
-        """Restricts the index to the provided view, which must be a subset of
-        the full index's collection.
+    def add_to_index(
+        self,
+        embeddings,
+        sample_ids,
+        label_ids=None,
+        overwrite=True,
+        allow_existing=True,
+        warn_existing=False,
+        reload=True,
+    ):
+        """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
+            reload (True): whether to call :meth:`reload` to refresh the
+                current view after the update
+        """
+        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,
+        reload=True,
+    ):
+        """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
+            reload (True): whether to call :meth:`reload` to refresh the
+                current view after the update
+        """
+        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
         from the specified view rather than the full index.
@@ -151,25 +494,37 @@ def use_view(self, sample_collection):
                 plot.show()
 
         Args:
-            sample_collection: a
-                :class:`fiftyone.core.collections.SampleCollection` defining a
-                subset of this index to use
+            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)
+            warn_missing (False): whether to log a warning if the provided
+                collection contains data points that this index does not
+                contain
 
         Returns:
             self
         """
-        view, sample_ids, label_ids, keep_inds = fbu.filter_ids(
-            sample_collection,
-            self._samples,
-            self._sample_ids,
-            self._label_ids,
-            patches_field=self._config.patches_field,
+        sample_ids, label_ids, keep_inds, good_inds = fbu.filter_ids(
+            samples,
+            self.sample_ids,
+            self.label_ids,
+            patches_field=self.config.patches_field,
+            allow_missing=allow_missing,
+            warn_missing=warn_missing,
         )
 
-        self._curr_view = view
+        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_missing_size = missing_size
 
         return self
 
@@ -180,33 +535,54 @@ def clear_view(self):
         """
         self.use_view(self._samples)
 
-    def plot_distances(self, bins=100, log=False, backend="plotly", **kwargs):
-        """Plots a histogram of the distance between each example and its
-        nearest neighbor.
+    def reload(self):
+        """Reloads the index for the current view.
 
-        If `:meth:`find_duplicates` or :meth:`find_unique` has been executed,
-        the threshold used is also indicated on the plot.
+        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`.
+
+        This method always returns values in the same order as
+        :meth:`current_sample_ids` and :meth:`current_label_ids`.
 
         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
+            path_or_expr: the values to extract, which can be:
 
-        Returns:
-            one of the following:
+                -   the name of a sample field or ``embedded.field.name`` from
+                    which to extract numeric or string values
+                -   a :class:`fiftyone.core.expressions.ViewExpression`
+                    defining numeric or string values to compute via
+                    :meth:`fiftyone.core.collections.SampleCollection.values`
 
-            -   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
+        Returns:
+            a list of values
         """
-        return fbs.plot_distances(self, bins, log, backend, **kwargs)
+        samples = self.view
+        patches_field = self.config.patches_field
+
+        if patches_field is not None:
+            ids = self.current_label_ids
+        else:
+            ids = self.current_sample_ids
+
+        return fbu.get_values(
+            samples, path_or_expr, ids, patches_field=patches_field
+        )
 
     def sort_by_similarity(
         self,
-        query_ids,
+        query,
         k=None,
         reverse=False,
         aggregation="mean",
@@ -214,20 +590,29 @@ def sort_by_similarity(
         _mongo=False,
     ):
         """Returns a view that sorts the samples/labels in :meth:`view` by
-        visual similarity to the specified query.
+        similarity to the specified query.
 
-        The query IDs can be any IDs in the full index of this instance, even
-        if the current :meth:`view` contains a subset of the full index.
+        When querying by IDs, the query can be any ID(s) in the full index of
+        this instance, even if the current :meth:`view` contains a subset of
+        the full index.
 
         Args:
-            query_ids: an ID or iterable of query IDs
-            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_ids``
-                contains multiple IDs. Supported values are
-                ``("mean", "min", "max")``
+            query: the query, which can be any of the following:
+
+                -   an ID or iterable of IDs
+                -   a ``num_dims`` vector or ``num_queries x num_dims`` array
+                    of vectors
+                -   a prompt or iterable of prompts (if supported by the index)
+
+            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
@@ -235,8 +620,383 @@ def sort_by_similarity(
         Returns:
             a :class:`fiftyone.core.view.DatasetView`
         """
-        return fbs.sort_by_similarity(
-            self, query_ids, 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):
@@ -260,7 +1020,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
@@ -276,7 +1085,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,
@@ -324,9 +1253,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
@@ -344,20 +1320,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:
 
@@ -374,7 +1354,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
@@ -392,18 +1372,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:
 
@@ -416,7 +1428,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
@@ -434,49 +1446,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
 
-    @classmethod
-    def _from_dict(cls, d, samples, config):
-        embeddings = np.array(d["embeddings"])
-        return cls(samples, config, embeddings)
+        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
 
-class SimilarityConfig(fob.BrainMethodConfig):
-    """Similarity configuration.
+            labels = []
+            for _id in ids:
+                if _id in unique_ids:
+                    label = "unique"
+                else:
+                    label = "other"
 
-    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
-    """
+                labels.append(label)
 
-    def __init__(
-        self,
-        embeddings_field=None,
-        model=None,
-        patches_field=None,
-        metric=None,
+            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 _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)
+
+    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 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
-        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 1da6c476..cce0a8dc 100644
--- a/fiftyone/brain/visualization.py
+++ b/fiftyone/brain/visualization.py
@@ -1,7 +1,7 @@
 """
 Visualization interface.
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
@@ -10,7 +10,6 @@
 import eta.core.utils as etau
 
 import fiftyone.core.brain as fob
-import fiftyone.core.plots as fop
 import fiftyone.core.utils as fou
 
 fbu = fou.lazy_import("fiftyone.brain.internal.core.utils")
@@ -24,34 +23,44 @@ 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, label_ids = fbu.get_ids(
-            samples, patches_field=config.patches_field
-        )
-
-        if len(sample_ids) != len(points):
-            ptype = "label" if config.patches_field is not None else "sample"
-            raise ValueError(
-                "Number of %s IDs (%d) does not match number of points (%d). "
-                "You may have missing data/labels that you need to omit from "
-                "your view" % (ptype, len(sample_ids), len(points))
+    def __init__(
+        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,
+                patches_field=config.patches_field,
+                data=points,
+                data_type="points",
             )
 
         self.points = points
+        self.sample_ids = sample_ids
+        self.label_ids = label_ids
 
-        self._samples = samples
-        self._config = config
-        self._sample_ids = sample_ids
-        self._label_ids = label_ids
         self._last_view = None
         self._curr_view = None
+        self._curr_points = None
         self._curr_sample_ids = None
         self._curr_label_ids = None
         self._curr_keep_inds = None
-        self._curr_points = None
+        self._curr_good_inds = None
 
         self.use_view(samples)
 
@@ -70,26 +79,79 @@ def config(self):
 
     @property
     def index_size(self):
-        """The number of examples in the index.
+        """The number of active 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 len(self._curr_sample_ids)
 
+    @property
+    def total_index_size(self):
+        """The total number of data points in the index.
+
+        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.points)
+
+    @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)
+
+    @property
+    def current_points(self):
+        """The currently active points in the index.
+
+        If :meth:`use_view` has been called, this may be a subset of the full
+        index.
+        """
+        return self._curr_points
+
+    @property
+    def current_sample_ids(self):
+        """The sample IDs of the currently active points in the index.
+
+        If :meth:`use_view` has been called, this may be a subset of the full
+        index.
+        """
+        return self._curr_sample_ids
+
+    @property
+    def current_label_ids(self):
+        """The label IDs of the currently active points in the index, or
+        ``None`` if not applicable.
+
+        If :meth:`use_view` has been called, this may be a subset of the full
+        index.
+        """
+        return self._curr_label_ids
+
     @property
     def view(self):
         """The :class:`fiftyone.core.collections.SampleCollection` against
         which results are currently being generated.
 
-        If :meth:`use_view` has been called, this view may be a subset of the
-        collection on which the full index was generated.
+        If :meth:`use_view` has been called, this view may be different than
+        the collection on which the full index was generated.
         """
         return self._curr_view
 
-    def use_view(self, sample_collection):
-        """Restricts the index to the provided view, which must be a subset of
-        the full index's collection.
+    def use_view(
+        self, sample_collection, allow_missing=True, warn_missing=False
+    ):
+        """Restricts the index to the provided view.
 
         Subsequent calls to methods on this instance will only contain results
         from the specified view rather than the full index.
@@ -119,18 +181,24 @@ def use_view(self, sample_collection):
 
         Args:
             sample_collection: a
-                :class:`fiftyone.core.collections.SampleCollection` defining a
-                subset of this index to use
+                :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)
+            warn_missing (False): whether to log a warning if the provided
+                collection contains data points that this index does not
+                contain
 
         Returns:
             self
         """
-        view, sample_ids, label_ids, keep_inds = fbu.filter_ids(
+        sample_ids, label_ids, keep_inds, good_inds = fbu.filter_ids(
             sample_collection,
-            self._samples,
-            self._sample_ids,
-            self._label_ids,
+            self.sample_ids,
+            self.label_ids,
             patches_field=self._config.patches_field,
+            allow_missing=allow_missing,
+            warn_missing=warn_missing,
         )
 
         if keep_inds is not None:
@@ -138,11 +206,12 @@ def use_view(self, sample_collection):
         else:
             points = self.points
 
-        self._curr_view = view
+        self._curr_view = sample_collection
+        self._curr_points = points
         self._curr_sample_ids = sample_ids
         self._curr_label_ids = label_ids
         self._curr_keep_inds = keep_inds
-        self._curr_points = points
+        self._curr_good_inds = good_inds
 
         return self
 
@@ -153,6 +222,28 @@ def clear_view(self):
         """
         self.use_view(self._samples)
 
+    def values(self, path_or_expr):
+        """Extracts a flat list of values from the given field or expression
+        corresponding to the current :meth:`view`.
+
+        This method always returns values in the same order as
+        :meth:`current_points`, :meth:`current_sample_ids`, and
+        :meth:`current_label_ids`.
+
+        Args:
+            path_or_expr: the values to extract, which can be:
+
+                -   the name of a sample field or ``embedded.field.name`` from
+                    which to extract numeric or string values
+                -   a :class:`fiftyone.core.expressions.ViewExpression`
+                    defining numeric or string values to compute via
+                    :meth:`fiftyone.core.collections.SampleCollection.values`
+
+        Returns:
+            a list of values
+        """
+        return fbv.values(self, path_or_expr)
+
     def visualize(
         self,
         labels=None,
@@ -161,7 +252,8 @@ def visualize(
         backend="plotly",
         **kwargs,
     ):
-        """Generates an interactive scatterplot of the visualization results.
+        """Generates an interactive scatterplot of the visualization results
+        for the current :meth:`view`.
 
         This method supports 2D or 3D visualizations, but interactive point
         selection is only available in 2D.
@@ -214,10 +306,8 @@ def visualize(
         Returns:
             an :class:`fiftyone.core.plots.base.InteractivePlot`
         """
-        return fop.scatterplot(
-            self._curr_points,
-            samples=self._curr_view,
-            link_field=self._config.patches_field,
+        return fbv.visualize(
+            self,
             labels=labels,
             sizes=sizes,
             classes=classes,
@@ -226,9 +316,25 @@ def visualize(
         )
 
     @classmethod
-    def _from_dict(cls, d, samples, config):
+    def _from_dict(cls, d, samples, config, brain_key):
         points = np.array(d["points"])
-        return cls(samples, config, points)
+
+        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,
+            points,
+            sample_ids=sample_ids,
+            label_ids=label_ids,
+        )
 
 
 class VisualizationConfig(fob.BrainMethodConfig):
@@ -237,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
@@ -253,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
@@ -277,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
@@ -339,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
@@ -418,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
diff --git a/install.bash b/install.bash
index 924cd28f..b0d47194 100644
--- a/install.bash
+++ b/install.bash
@@ -4,7 +4,7 @@
 # Usage:
 #   bash install.bash
 #
-# Copyright 2017-2022, Voxel51, Inc.
+# Copyright 2017-2023, Voxel51, Inc.
 # voxel51.com
 #
 
diff --git a/production/README.md b/production/README.md
index 47c01def..f3df7cc1 100644
--- a/production/README.md
+++ b/production/README.md
@@ -40,4 +40,4 @@ To add a model to the Brain, following these steps:
 
 ## Copyright
 
-Copyright 2017-2022, Voxel51, Inc.<br> voxel51.com
+Copyright 2017-2023, Voxel51, Inc.<br> voxel51.com
diff --git a/production/marketing/README.md b/production/marketing/README.md
index 958b5ada..e1759db9 100644
--- a/production/marketing/README.md
+++ b/production/marketing/README.md
@@ -18,4 +18,4 @@ experiment details with proprietary information that only exists in the brain.
 
 ## Copyright
 
-Copyright 2017-2022, Voxel51, Inc.<br> voxel51.com
+Copyright 2017-2023, Voxel51, Inc.<br> voxel51.com
diff --git a/production/marketing/shot_bdd_detections.py b/production/marketing/shot_bdd_detections.py
index e562fb28..52f1dd10 100644
--- a/production/marketing/shot_bdd_detections.py
+++ b/production/marketing/shot_bdd_detections.py
@@ -12,7 +12,7 @@
     # From inside IPython
     run shot_bdd_detections.py
 
-Copyright 2017-2022, Voxel51, Inc.
+Copyright 2017-2023, Voxel51, Inc.
 voxel51.com
 """
 import os
diff --git a/production/marketing/shot_neardups.py b/production/marketing/shot_neardups.py
index fe3fc9de..2560880a 100644
--- a/production/marketing/shot_neardups.py
+++ b/production/marketing/shot_neardups.py
@@ -13,7 +13,7 @@
     # From inside IPython
     run shot_neardups.py
 
-Copyright 2017-2022, Voxel51, Inc.
+Copyright 2017-2023, Voxel51, Inc.
 voxel51.com
 """
 import os
diff --git a/production/marketing/use_cases_code.md b/production/marketing/use_cases_code.md
index 076e5e22..fabb960d 100644
--- a/production/marketing/use_cases_code.md
+++ b/production/marketing/use_cases_code.md
@@ -115,4 +115,4 @@ fo.launch_dashboard(view=mistakes_view)
 
 ## Copyright
 
-Copyright 2017-2022, Voxel51, Inc.<br> voxel51.com
+Copyright 2017-2023, Voxel51, Inc.<br> voxel51.com
diff --git a/production/models/simple_resnet/README.md b/production/models/simple_resnet/README.md
index e3d9240b..5a2638f0 100644
--- a/production/models/simple_resnet/README.md
+++ b/production/models/simple_resnet/README.md
@@ -27,4 +27,4 @@ Steps to train and deploy the model:
 
 ## Copyright
 
-Copyright 2017-2022, Voxel51, Inc.<br> voxel51.com
+Copyright 2017-2023, Voxel51, Inc.<br> voxel51.com
diff --git a/production/models/simple_resnet/config.py b/production/models/simple_resnet/config.py
index 0c9614d3..d31abf25 100644
--- a/production/models/simple_resnet/config.py
+++ b/production/models/simple_resnet/config.py
@@ -1,7 +1,7 @@
 """
 Simple configuration setup for these experiments
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
diff --git a/production/models/simple_resnet/datasets.py b/production/models/simple_resnet/datasets.py
index d42f40d4..854ced54 100644
--- a/production/models/simple_resnet/datasets.py
+++ b/production/models/simple_resnet/datasets.py
@@ -1,7 +1,7 @@
 """
 Implementation of datasets for the experiments
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
diff --git a/production/models/simple_resnet/preprocess.py b/production/models/simple_resnet/preprocess.py
index 64a7441b..1ba697fb 100644
--- a/production/models/simple_resnet/preprocess.py
+++ b/production/models/simple_resnet/preprocess.py
@@ -1,7 +1,7 @@
 """
 Preprocessing functions
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
diff --git a/production/models/simple_resnet/train_classifier.py b/production/models/simple_resnet/train_classifier.py
index 23009364..23c3a260 100644
--- a/production/models/simple_resnet/train_classifier.py
+++ b/production/models/simple_resnet/train_classifier.py
@@ -18,7 +18,7 @@
 
     run train_classifier.py -t 2000 -e 12 -b 64 --n_rounds 1 --p_initial 1.0 -m /tmp/foo.pth
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
diff --git a/production/models/simple_resnet/training.py b/production/models/simple_resnet/training.py
index 5da83525..75ad5fcb 100644
--- a/production/models/simple_resnet/training.py
+++ b/production/models/simple_resnet/training.py
@@ -1,7 +1,7 @@
 """
 Training functions
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
diff --git a/production/models/simple_resnet/utils.py b/production/models/simple_resnet/utils.py
index 537fd10f..fef716b9 100644
--- a/production/models/simple_resnet/utils.py
+++ b/production/models/simple_resnet/utils.py
@@ -1,7 +1,7 @@
 """
 Utilities for experiments.
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
diff --git a/requirements/dev.txt b/requirements/dev.txt
index 7b44b253..d73bf272 100644
--- a/requirements/dev.txt
+++ b/requirements/dev.txt
@@ -6,5 +6,5 @@ ipython>=7.16.1
 pandas==1.1.5
 pre-commit==2.0.1
 pylint==2.3.1
-pytest==5.4.3
+pytest==7.3.1
 twine>=3
diff --git a/setup.py b/setup.py
index 38549f7f..b05596a8 100644
--- a/setup.py
+++ b/setup.py
@@ -2,7 +2,7 @@
 """
 Installs `fiftyone-brain`.
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
@@ -17,7 +17,7 @@
     long_description += "\n## License\n\n" + fh.read()
 
 
-VERSION = "0.9.1"
+VERSION = "0.12.0"
 
 
 def get_version():


Test patch:
diff --git a/experiments/labelerror/resnet_cifar10/le_test2.py b/experiments/labelerror/resnet_cifar10/le_test2.py
index 2924c5ec..e86499a5 100644
--- a/experiments/labelerror/resnet_cifar10/le_test2.py
+++ b/experiments/labelerror/resnet_cifar10/le_test2.py
@@ -2,7 +2,7 @@
 le_test2 --> initial tests for using fiftyone with this code
 Loads the data, assigns ground-truth labels, then trains a clean model
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
diff --git a/experiments/labelerror/resnet_cifar10/le_test3.py b/experiments/labelerror/resnet_cifar10/le_test3.py
index 45cb7c86..11d31535 100644
--- a/experiments/labelerror/resnet_cifar10/le_test3.py
+++ b/experiments/labelerror/resnet_cifar10/le_test3.py
@@ -3,7 +3,7 @@
 Loads the data, assigns ground-truth labels, loads a model
 Runs prediction and associates predictions with the fiftyone dataset
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
diff --git a/production/models/simple_resnet/test_classifier.py b/production/models/simple_resnet/test_classifier.py
index 35eed6cc..ea361614 100644
--- a/production/models/simple_resnet/test_classifier.py
+++ b/production/models/simple_resnet/test_classifier.py
@@ -11,7 +11,7 @@
 Uses
     -t number
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
diff --git a/tests/README.md b/tests/README.md
index a2420f2a..c484b204 100644
--- a/tests/README.md
+++ b/tests/README.md
@@ -34,4 +34,4 @@ pytest <file>.py -s -k <test_function_name>
 
 ## Copyright
 
-Copyright 2017-2022, Voxel51, Inc.<br> voxel51.com
+Copyright 2017-2023, Voxel51, Inc.<br> voxel51.com
diff --git a/tests/intensive/test_interface.py b/tests/intensive/test_interface.py
index 713606af..6ab78a24 100644
--- a/tests/intensive/test_interface.py
+++ b/tests/intensive/test_interface.py
@@ -1,7 +1,7 @@
 """
 Brain interface tests.
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
diff --git a/tests/intensive/test_similarity.py b/tests/intensive/test_similarity.py
new file mode 100644
index 00000000..5de1d281
--- /dev/null
+++ b/tests/intensive/test_similarity.py
@@ -0,0 +1,536 @@
+"""
+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  # 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()
+
+    index = dataset.load_brain_results("img_sim")
+
+    assert index.total_index_size == len(dataset)
+    assert set(dataset.values("id")) == set(index.sample_ids)
+
+
+def test_images_subset():
+    dataset = _load_images_dataset()
+
+    index = dataset.load_brain_results("img_sim")
+
+    view = dataset.take(10)
+    index.use_view(view)
+
+    assert index.index_size == len(view)
+    assert set(view.values("id")) == set(index.current_sample_ids)
+
+
+def test_images_missing():
+    dataset = _load_images_dataset().limit(4).clone()
+    dataset.add_samples(
+        [
+            fo.Sample(filepath="non-existent1.png"),
+            fo.Sample(filepath="non-existent2.png"),
+            fo.Sample(filepath="non-existent3.png"),
+            fo.Sample(filepath="non-existent4.png"),
+        ]
+    )
+
+    sample_ids = dataset[:4].values("id")
+
+    index = fob.compute_similarity(dataset, batch_size=1)
+
+    assert index.total_index_size == 4
+    assert set(sample_ids) == set(index.sample_ids)
+
+    model = foz.load_zoo_model("inception-v3-imagenet-torch")
+    index = fob.compute_similarity(
+        dataset,
+        model=model,
+        embeddings="embeddings_missing",
+        batch_size=1,
+    )
+
+    assert len(dataset.exists("embeddings_missing")) == 4
+    assert index.index_size == 4
+    assert set(sample_ids) == set(index.sample_ids)
+
+
+def test_patches():
+    dataset = _load_patches_dataset()
+
+    index = dataset.load_brain_results("gt_sim")
+
+    label_ids = dataset.values("ground_truth.detections.id", unwind=True)
+
+    assert index.total_index_size == len(label_ids)
+    assert set(label_ids) == set(index.label_ids)
+
+
+def test_patches_subset():
+    dataset = _load_patches_dataset()
+
+    index = dataset.load_brain_results("gt_sim")
+
+    label_ids = dataset.values("ground_truth.detections.id", unwind=True)
+
+    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")
+    index.use_view(view)
+
+    label_ids = view.values("ground_truth.detections.id", unwind=True)
+
+    assert index.index_size == len(label_ids)
+    assert set(label_ids) == set(index.current_label_ids)
+
+
+def test_patches_missing():
+    dataset = _load_patches_dataset().limit(4).clone()
+    dataset.add_samples(
+        [
+            fo.Sample(filepath="non-existent1.png"),
+            fo.Sample(filepath="non-existent2.png"),
+            fo.Sample(filepath="non-existent3.png"),
+            fo.Sample(filepath="non-existent4.png"),
+        ]
+    )
+
+    for sample in dataset[4:]:
+        sample["ground_truth"] = fo.Detections(
+            detections=[fo.Detection(bounding_box=[0.1, 0.1, 0.8, 0.8])]
+        )
+        sample.save()
+
+    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 index.total_index_size == num_patches
+    assert set(label_ids) == set(index.label_ids)
+
+    model = foz.load_zoo_model("inception-v3-imagenet-torch")
+    index = fob.compute_similarity(
+        dataset,
+        model=model,
+        patches_field="ground_truth",
+        embeddings="embeddings_missing",
+        batch_size=1,
+    )
+
+    view = dataset.filter_labels(
+        "ground_truth", F("embeddings_missing") != None
+    )
+
+    assert view.count("ground_truth.detections") == num_patches
+    assert index.total_index_size == num_patches
+    assert set(label_ids) == set(index.label_ids)
+
+
+def _load_images_dataset():
+    name = "test-similarity-images"
+
+    if fo.dataset_exists(name):
+        return fo.load_dataset(name)
+
+    return _make_images_dataset(name)
+
+
+def _load_patches_dataset():
+    name = "test-similarity-patches"
+
+    if fo.dataset_exists(name):
+        return fo.load_dataset(name)
+
+    return _make_patches_dataset(name)
+
+
+def _make_images_dataset(name):
+    dataset = foz.load_zoo_dataset(
+        "quickstart", max_samples=20, dataset_name=name
+    )
+    model = foz.load_zoo_model("inception-v3-imagenet-torch")
+
+    # Embed images
+    dataset.compute_embeddings(
+        model, embeddings_field="embeddings", batch_size=8
+    )
+
+    # Image similarity
+    fob.compute_similarity(
+        dataset, embeddings="embeddings", brain_key="img_sim"
+    )
+
+    return dataset
+
+
+def _make_patches_dataset(name):
+    dataset = foz.load_zoo_dataset(
+        "quickstart", max_samples=20, dataset_name=name
+    )
+    model = foz.load_zoo_model("inception-v3-imagenet-torch")
+
+    # Embed ground truth patches
+    dataset.compute_patch_embeddings(
+        model,
+        "ground_truth",
+        embeddings_field="embeddings",
+        batch_size=8,
+        force_square=True,
+    )
+
+    # Patch similarity
+    fob.compute_similarity(
+        dataset,
+        patches_field="ground_truth",
+        embeddings="embeddings",
+        brain_key="gt_sim",
+    )
+
+    return dataset
+
+
+if __name__ == "__main__":
+    fo.config.show_progress_bars = True
+    unittest.main(verbosity=2)
diff --git a/tests/intensive/test_uniqueness.py b/tests/intensive/test_uniqueness.py
index 646b6e3e..5eea6f59 100644
--- a/tests/intensive/test_uniqueness.py
+++ b/tests/intensive/test_uniqueness.py
@@ -1,16 +1,12 @@
 """
 Uniqueness tests.
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
-import os
 import unittest
 
-import eta.core.storage as etas
-import eta.core.utils as etau
-
 import fiftyone as fo
 import fiftyone.brain as fob
 import fiftyone.zoo as foz
@@ -20,30 +16,74 @@ def test_uniqueness():
     _run_uniqueness()
 
 
-def test_uniqueness_torch_model():
+def test_uniqueness_torch():
     model = foz.load_zoo_model("inception-v3-imagenet-torch")
     _run_uniqueness(model=model, batch_size=16)
 
 
-def test_uniqueness_tf_model():
+def test_uniqueness_tf():
     model = foz.load_zoo_model("resnet-v2-50-imagenet-tf1")
     _run_uniqueness(model=model, batch_size=16)
 
 
+def test_uniqueness_missing():
+    dataset = fo.Dataset()
+    dataset.add_samples(
+        [
+            fo.Sample(filepath="non-existent1.png"),
+            fo.Sample(filepath="non-existent2.png"),
+            fo.Sample(filepath="non-existent3.png"),
+            fo.Sample(filepath="non-existent4.png"),
+        ]
+    )
+
+    fob.compute_uniqueness(dataset, batch_size=1)
+
+    view = dataset.exists("uniqueness")
+
+    assert dataset.has_field("uniqueness")
+    assert len(view) == 0
+
+
 def test_roi_uniqueness():
     _run_uniqueness(roi_field="ground_truth")
 
 
-def test_roi_uniqueness_torch_model():
+def test_roi_uniqueness_torch():
     model = foz.load_zoo_model("inception-v3-imagenet-torch")
     _run_uniqueness(roi_field="ground_truth", model=model, batch_size=16)
 
 
-def test_roi_uniqueness_tf_model():
+def test_roi_uniqueness_tf():
     model = foz.load_zoo_model("resnet-v2-50-imagenet-tf1")
     _run_uniqueness(roi_field="ground_truth", model=model, batch_size=16)
 
 
+def test_roi_uniqueness_missing():
+    dataset = fo.Dataset()
+    dataset.add_samples(
+        [
+            fo.Sample(filepath="non-existent1.png"),
+            fo.Sample(filepath="non-existent2.png"),
+            fo.Sample(filepath="non-existent3.png"),
+            fo.Sample(filepath="non-existent4.png"),
+        ]
+    )
+
+    for sample in dataset:
+        sample["ground_truth"] = fo.Detections(
+            detections=[fo.Detection(bounding_box=[0.1, 0.1, 0.8, 0.8])]
+        )
+        sample.save()
+
+    fob.compute_uniqueness(dataset, roi_field="ground_truth", batch_size=1)
+
+    view = dataset.exists("uniqueness")
+
+    assert dataset.has_field("uniqueness")
+    assert len(view) == 0
+
+
 def _run_uniqueness(roi_field=None, model=None, batch_size=None):
     dataset = foz.load_zoo_dataset(
         "quickstart", dataset_name=fo.get_default_dataset_name()
@@ -53,7 +93,6 @@ def _run_uniqueness(roi_field=None, model=None, batch_size=None):
     view = dataset.take(50)
     num_samples = len(view)
 
-    # Custom Torch model
     fob.compute_uniqueness(
         view, roi_field=roi_field, model=model, batch_size=batch_size
     )
diff --git a/tests/intensive/test_visualization.py b/tests/intensive/test_visualization.py
index a80b8226..c7818e6c 100644
--- a/tests/intensive/test_visualization.py
+++ b/tests/intensive/test_visualization.py
@@ -1,7 +1,11 @@
 """
 Visualization tests.
 
-| Copyright 2017-2022, Voxel51, Inc.
+All of these tests are designed to be run manually via::
+
+    pytest tests/intensive/test_visualization.py -s -k test_<name>
+
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
@@ -13,9 +17,7 @@
 import fiftyone as fo
 import fiftyone.brain as fob
 import fiftyone.zoo as foz
-
-
-_DATASET_NAME = "visualization-test"
+from fiftyone import ViewField as F
 
 
 def test_mnist():
@@ -30,7 +32,11 @@ def test_mnist():
     )
 
     results = fob.compute_visualization(
-        dataset, embeddings=embeddings, num_dims=2, verbose=True, seed=51
+        dataset,
+        embeddings=embeddings,
+        num_dims=2,
+        verbose=True,
+        seed=51,
     )
 
     plot = results.visualize(labels="ground_truth.label")
@@ -40,11 +46,12 @@ def test_mnist():
 
 
 def test_images():
-    dataset = _load_dataset()
+    dataset = _load_images_dataset()
 
-    results = fob.compute_visualization(
-        dataset, embeddings="embeddings", num_dims=2, verbose=True, seed=51
-    )
+    results = dataset.load_brain_results("img_viz")
+
+    assert results.total_index_size == len(dataset)
+    assert set(dataset.values("id")) == set(results.sample_ids)
 
     plot = results.visualize(labels="uniqueness")
     plot.show()
@@ -52,17 +59,90 @@ def test_images():
     input("Press enter to continue...")
 
 
-def test_objects():
-    dataset = _load_dataset()
+def test_images_subset():
+    dataset = _load_images_dataset()
+
+    results = dataset.load_brain_results("img_viz")
+
+    view = dataset.take(10)
+    results.use_view(view)
+
+    assert results.index_size == len(view)
+    assert set(view.values("id")) == set(results.current_sample_ids)
+
+    plot = results.visualize(labels="uniqueness")
+    plot.show()
+
+    input("Press enter to continue...")
+
+
+def test_images_missing():
+    dataset = _load_images_dataset().limit(4).clone()
+    dataset.add_samples(
+        [
+            fo.Sample(filepath="non-existent1.png"),
+            fo.Sample(filepath="non-existent2.png"),
+            fo.Sample(filepath="non-existent3.png"),
+            fo.Sample(filepath="non-existent4.png"),
+        ]
+    )
+
+    sample_ids = dataset[:4].values("id")
 
+    results = fob.compute_visualization(dataset, batch_size=1)
+
+    assert results.total_index_size == 4
+    assert set(sample_ids) == set(results.sample_ids)
+
+    model = foz.load_zoo_model("inception-v3-imagenet-torch")
     results = fob.compute_visualization(
         dataset,
-        patches_field="ground_truth",
-        embeddings="gt_embeddings",
-        num_dims=2,
-        verbose=True,
-        seed=51,
+        model=model,
+        embeddings="embeddings_missing",
+        batch_size=1,
+    )
+
+    assert len(dataset.exists("embeddings_missing")) == 4
+    assert results.total_index_size == 4
+    assert set(sample_ids) == set(results.sample_ids)
+
+
+def test_patches():
+    dataset = _load_patches_dataset()
+
+    results = dataset.load_brain_results("gt_viz")
+
+    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)
+
+    plot = results.visualize(labels="ground_truth.detections.label")
+    plot.show()
+
+    input("Press enter to continue...")
+
+
+def test_patches_subset():
+    dataset = _load_patches_dataset()
+
+    results = dataset.load_brain_results("gt_viz")
+
+    plot = results.visualize(
+        labels="ground_truth.detections.label",
+        classes=["person"],
     )
+    plot.show()
+
+    input("Press enter to continue...")
+
+    view = dataset.filter_labels("ground_truth", F("label") == "person")
+    results.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)
 
     plot = results.visualize(labels="ground_truth.detections.label")
     plot.show()
@@ -70,52 +150,119 @@ def test_objects():
     input("Press enter to continue...")
 
 
-def test_objects_subset():
-    dataset = _load_dataset()
+def test_patches_missing():
+    dataset = _load_patches_dataset().limit(4).clone()
+    dataset.add_samples(
+        [
+            fo.Sample(filepath="non-existent1.png"),
+            fo.Sample(filepath="non-existent2.png"),
+            fo.Sample(filepath="non-existent3.png"),
+            fo.Sample(filepath="non-existent4.png"),
+        ]
+    )
+
+    for sample in dataset[4:]:
+        sample["ground_truth"] = fo.Detections(
+            detections=[fo.Detection(bounding_box=[0.1, 0.1, 0.8, 0.8])]
+        )
+        sample.save()
 
+    results = fob.compute_visualization(
+        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)
+
+    model = foz.load_zoo_model("inception-v3-imagenet-torch")
     results = fob.compute_visualization(
         dataset,
+        model=model,
         patches_field="ground_truth",
-        embeddings="gt_embeddings",
-        num_dims=2,
-        verbose=True,
-        seed=51,
+        embeddings="embeddings_missing",
+        batch_size=1,
     )
 
-    counts = dataset.count_values("ground_truth.detections.label")
-    classes = sorted(counts, key=counts.get, reverse=True)[:5]
-
-    plot = results.visualize(
-        labels="ground_truth.detections.label", classes=classes
+    view = dataset.filter_labels(
+        "ground_truth", F("embeddings_missing") != None
     )
-    plot.show()
 
-    input("Press enter to continue...")
+    assert view.count("ground_truth.detections") == num_patches
+    assert results.total_index_size == num_patches
+    assert set(label_ids) == set(results.label_ids)
+
+
+def _load_images_dataset():
+    name = "test-visualization-images"
 
+    if fo.dataset_exists(name):
+        return fo.load_dataset(name)
 
-def _load_dataset():
-    if fo.dataset_exists(_DATASET_NAME):
-        return fo.load_dataset(_DATASET_NAME)
+    return _make_images_dataset(name)
 
-    dataset = foz.load_zoo_dataset("quickstart", dataset_name=_DATASET_NAME)
-    dataset.persistent = True
 
+def _load_patches_dataset():
+    name = "test-visualization-patches"
+
+    if fo.dataset_exists(name):
+        return fo.load_dataset(name)
+
+    return _make_patches_dataset(name)
+
+
+def _make_images_dataset(name):
+    dataset = foz.load_zoo_dataset(
+        "quickstart", max_samples=20, dataset_name=name
+    )
     model = foz.load_zoo_model("inception-v3-imagenet-torch")
 
     # Embed images
     dataset.compute_embeddings(
-        model, embeddings_field="embeddings", batch_size=16
+        model, embeddings_field="embeddings", batch_size=8
+    )
+
+    # Image visualization
+    fob.compute_visualization(
+        dataset,
+        embeddings="embeddings",
+        num_dims=2,
+        verbose=True,
+        seed=51,
+        brain_key="img_viz",
+    )
+
+    return dataset
+
+
+def _make_patches_dataset(name):
+    dataset = foz.load_zoo_dataset(
+        "quickstart", max_samples=20, dataset_name=name
     )
+    model = foz.load_zoo_model("inception-v3-imagenet-torch")
 
     # Embed ground truth patches
     dataset.compute_patch_embeddings(
         model,
         "ground_truth",
-        embeddings_field="gt_embeddings",
-        batch_size=16,
+        embeddings_field="embeddings",
+        batch_size=8,
         force_square=True,
     )
 
+    # Patch visualization
+    fob.compute_visualization(
+        dataset,
+        patches_field="ground_truth",
+        embeddings="embeddings",
+        num_dims=2,
+        verbose=True,
+        seed=51,
+        brain_key="gt_viz",
+    )
+
     return dataset
 
 
diff --git a/tests/models/test_simple_resnet.py b/tests/models/test_simple_resnet.py
index 42c78d4c..f65bd176 100644
--- a/tests/models/test_simple_resnet.py
+++ b/tests/models/test_simple_resnet.py
@@ -1,7 +1,7 @@
 """
 Tests for :mod:`fiftyone.brain.internal.models.simple_resnet`.
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """
diff --git a/tests/test_uniqueness.py b/tests/test_uniqueness.py
index a121a284..fc6e3bf8 100644
--- a/tests/test_uniqueness.py
+++ b/tests/test_uniqueness.py
@@ -1,7 +1,7 @@
 """
 Uniqueness tests.
 
-| Copyright 2017-2022, Voxel51, Inc.
+| Copyright 2017-2023, Voxel51, Inc.
 | `voxel51.com <https://voxel51.com/>`_
 |
 """