[docs]classMyIndexer(Executor):""" Executor with basic exact search using cosine distance """def__init__(self,**kwargs):super().__init__(**kwargs)ifos.path.exists(self.workspace+'/indexer'):self._docs=DocumentArray.load(self.workspace+'/indexer')else:self._docs=DocumentArray()
[docs]@requests(on=['/search','/eval'])defsearch(self,docs:'DocumentArray',parameters:Dict,**kwargs):"""Append best matches to each document in docs :param docs: documents that are searched :param parameters: dictionary of pairs (parameter,value) :param kwargs: other keyword arguments """docs.match(self._docs,metric='cosine',normalization=(1,0),limit=int(parameters['top_k']),)
[docs]defclose(self):""" Stores the DocumentArray to disk """self._docs.save(self.workspace+'/indexer')
[docs]classMyEncoder(Executor):""" Encode data using SVD decomposition """def__init__(self,**kwargs):super().__init__(**kwargs)np.random.seed(1337)# generate a random orthogonal matrixH=np.random.rand(784,64)u,s,vh=np.linalg.svd(H,full_matrices=False)self.oth_mat=u@vh
[docs]@requestsdefencode(self,docs:'DocumentArray',**kwargs):"""Encode the data using an SVD decomposition :param docs: input documents to update with an embedding :param kwargs: other keyword arguments """# reduce dimension to 50 by random orthogonal projectioncontent=np.stack(docs[:,'content'])content=content[:,:,:,0].reshape(-1,784)embeds=((content/255)@self.oth_mat).astype(np.float32)fordoc,embed,continzip(docs,embeds,content):doc.embedding=embeddoc.content=contdoc.convert_image_tensor_to_uri()doc.pop('tensor')
[docs]classMyConverter(Executor):""" Convert DocumentArrays removing tensor and reshaping tensor as image """
[docs]@requestsdefconvert(self,docs:'DocumentArray',**kwargs):""" Remove tensor and reshape documents as squared images :param docs: documents to modify :param kwargs: other keyword arguments """fordocindocs:doc.convert_image_tensor_to_uri()doc.pop('tensor')