Sweep: Utility Function for Vector Similarity Search (✓ Sandbox Passed)

API/database.py

@@ -22,3 +22,31 @@ def find_one_and_delete(self, collection, query):
 
     def update_one(self, collection, query, update):
         return self.db[collection].update_one(query, update)
+    def find_similar_vectors(self, collection, embedding_vector, n):
+        """
+        Find the top n most similar vectors in the database to the given embedding_vector.
+        This method uses the Euclidean distance for similarity measure.
+
+        :param collection: The MongoDB collection to search within.
+        :param embedding_vector: The embedding vector to find similar vectors for.
+        :param n: The number of top similar vectors to return.
+        :return: The top n most similar vectors from the MongoDB database.
+        """
+        pipeline = [
+            {
+                "$addFields": {
+                    "distance": {
+                        "$sqrt": {
+                            "$reduce": {
+                                "input": {"$zip": {"inputs": ["$vector", embedding_vector]}},
+                                "initialValue": 0,
+                                "in": {"$add": ["$$value", {"$pow": [{"$subtract": ["$$this.0", "$$this.1"]}, 2]}]}
+                            }
+                        }
+                    }
+                }
+            },
+            {"$sort": {"distance": 1}},
+            {"$limit": n}
+        ]
+        return list(self.db[collection].aggregate(pipeline))

API/route.py

@@ -267,3 +267,23 @@ async def delete_employees(EmployeeCode: int):
     client.find_one_and_delete(collection, {"EmployeeCode": EmployeeCode})
 
     return {"Message": "Successfully Deleted"}
+@router.post("/recognise_face")
+async def recognise_face(embedding: List[float], n: int):
+    """
+    Recognise a face by finding the most similar face embeddings in the database.
+
+    Args:
+        embedding (List[float]): The embedding vector of the face to be recognised.
+        n (int): The number of top similar vectors to return.
+
+    Returns:
+        dict: A dictionary containing the top n most similar face embeddings.
+
+    """
+    logging.info("Recognising face")
+    try:
+        similar_faces = client.find_similar_vectors(collection, embedding, n)
+        return {"similar_faces": similar_faces}
+    except Exception as e:
+        logging.error(f"Error recognising face: {str(e)}")
+        raise HTTPException(status_code=500, detail="Internal server error")