Executor in Action

Fastai

This Executor uses the ResNet18 network for object classification on images provided by fastai.

The encode function of this executor generates a feature vector for each image in each Document of the input DocumentArray. The feature vector generated is the output activations of the neural network (a vector of 1000 components).

Note

The embedding of each text is performed in a joined operation (all embeddings are created for all images in a single function call) to achieve higher performance.

As a result each Document in the input DocumentArray docs will have an embedding after encode() has completed.

  1. import torch
  2. from fastai.vision.models import resnet18
  4. from jina import Executor, requests
  7. class ResnetImageEncoder(Executor):
  9.     def __init__(self, *args, **kwargs):
  10.         super().__init__(*args, **kwargs)
  11.         self.model = resnet18()
  12.         self.model.eval()
  14.     @requests
  15.     def encode(self, docs, **kwargs):
  16.         batch = torch.Tensor(docs.get_attributes('blob'))
  17.         with torch.no_grad():
  18.             batch_embeddings = self.model(batch).detach().numpy()
  20.         for doc, emb in zip(docs, batch_embeddings):
  21.             doc.embedding = emb

Pytorch Lightning

This code snippet uses an autoencoder pretrained from cifar10-resnet18 to build an executor that encodes Document blob( an ndarray that could for example be an image) into embedding . It demonstrates the use of prebuilt model from PyTorch Lightning Bolts to build a Jina encoder.”

  1. from pl_bolts.models.autoencoders import AE
  3. from jina import Executor, requests
  5. import torch
  8. class PLMwuAutoEncoder(Executor):
  9.     def __init__(self, **kwargs):
  10.         super().__init__()
  11.         self.ae = AE(input_height=32).from_pretrained('cifar10-resnet18')
  12.         self.ae.freeze()
  14.     @requests
  15.     def encode(self, docs, **kwargs):
  16.         with torch.no_grad():
  17.             for doc in docs:
  18.                 input_tensor = torch.from_numpy(doc.blob)
  19.                 output_tensor = self.ae(input_tensor)
  20.                 doc.embedding = output_tensor.detach().numpy()

Paddle

The example below uses the PaddlePaddle Ernie model as the encoder. The Executor loads the pre-trained Ernie tokenizer and model, converts Jina Documents’ doc.text into Paddle Tensors and encodes the text as embeddings. As a result, each Document in the DocumentArray will have an embedding after encode() has completed.

  1. import paddle as P  # paddle==2.1.0
  2. import numpy as np
  3. from ernie.modeling_ernie import ErnieModel  # paddle-ernie 0.2.0.dev1
  4. from ernie.tokenizing_ernie import ErnieTokenizer
  6. from jina import Executor, requests
  9. class PaddleErineExecutor(Executor):
  10.     def __init__(self, **kwargs):
  11.         super().__init__()
  12.         self.tokenizer = ErnieTokenizer.from_pretrained('ernie-1.0')
  13.         self.model = ErnieModel.from_pretrained('ernie-1.0')
  14.         self.model.eval()
  16.     @requests
  17.     def encode(self, docs, **kwargs):
  18.         for doc in docs:
  19.             ids, _ = self.tokenizer.encode(doc.text)
  20.             ids = P.to_tensor(np.expand_dims(ids, 0))
  21.             pooled, encoded = self.model(ids)
  22.             doc.embedding = pooled.numpy()

Tensorflow

This Executor uses the MobileNetV2 network for object classification on images.

It extracts the buffer field (which is the actual byte array) from each input Document in the DocumentArray docs , preprocesses the byte array and uses MobileNet to predict the classes (dog/car/…) found in the image. Those predictions are Numpy arrays encoding the probability for each class supported by the model (1000 in this case). The Executor stores those arrays then in the embedding for each Document.

As a result each Document in the input DocumentArray docs will have an embedding after encode() has completed.

  1. import numpy as np
  2. import tensorflow as tf
  3. from keras.applications.mobilenet_v2 import MobileNetV2, preprocess_input
  4. from tensorflow.python.framework.errors_impl import InvalidArgumentError
  6. from jina import Executor, requests
  9. class TfMobileNetEncoder(Executor):
  10.     def __init__(self, **kwargs):
  11.         super().__init__()
  12.         self.image_dim = 224
  13.         self.model = MobileNetV2(pooling='avg', input_shape=(self.image_dim, self.image_dim, 3))
  15.     @requests
  16.     def encode(self, docs, **kwargs):
  17.         buffers, docs = docs.get_attributes_with_docs('buffer')
  19.         tensors = [tf.io.decode_image(contents=b, channels=3) for b in buffers]
  20.         resized_tensors = preprocess_input(np.array(self._resize_images(tensors)))
  22.         embeds = self.model.predict(np.stack(resized_tensors))
  23.         for d, b in zip(docs, embeds):
  24.             d.embedding = b
  26.     def _resize_images(self, tensors):
  27.         resized_tensors = []
  28.         for t in tensors:
  29.             try:
  30.                 resized_tensors.append(tf.keras.preprocessing.image.smart_resize(t, (self.image_dim, self.image_dim)))
  31.             except InvalidArgumentError:
  32.                 # this can happen if you include empty or other malformed images
  33.                 pass
  34.         return resized_tensors

MindSpore

The code snippet below takes docs as input and perform matrix multiplication with self.encoding_matrix. It leverages Mindspore Tensor conversion and operation. Finally, the embedding of each document will be set as the multiplied result as np.ndarray.

  1. import numpy as np
  2. from mindspore import Tensor  # mindspore 1.2.0
  3. import mindspore.ops as ops
  4. import mindspore.context as context
  6. from jina import Executor, requests
  9. class MindsporeMwuExecutor(Executor):
  10.     def __init__(self, **kwargs):
  11.         super().__init__()
  12.         context.set_context(mode=context.PYNATIVE_MODE, device_target='CPU')
  13.         self.encoding_mat = Tensor(np.random.rand(5, 5))
  15.     @requests
  16.     def encode(self, docs, **kwargs):
  17.         matmul = ops.MatMul()
  18.         for doc in docs:
  19.             input_tensor = Tensor(doc.blob)  # convert the ``ndarray`` of the doc to ``Tensor``
  20.             output_tensor = matmul(self.encoding_mat, input_tensor)  # multiply the input with the encoding matrix.
  21.             doc.embedding = output_tensor.asnumpy()  # assign the encoding results to ``embedding``

Scikit-learn

This Executor uses a TF-IDF feature vector to generate sparse embeddings for text search.

The class TFIDFTextEncoder extracts stores a tfidf_vectorizer object that it is fitted with a dataset already present in sklearn. The executor provides an encode method that receives a DocumentArray and updates each document in the DocumentArray with an embedding attribute that is the tf-idf representation of the text found in the document.

Note

The embedding of each text is perfomed in a joined operation (all embeddings are creted for all texts in a single function call) to achieve higher performance.

As a result, each Document in the DocumentArray will have an embedding after encode() has completed.

  1. from sklearn.datasets import fetch_20newsgroups
  2. from sklearn.feature_extraction.text import TfidfVectorizer
  4. from jina import Executor, requests, DocumentArray
  7. class TFIDFTextEncoder(Executor):
  9.     def __init__(self, *args, **kwargs):
  10.         super().__init__(*args, **kwargs)
  12.         from sklearn import datasets
  14.         dataset = fetch_20newsgroups()
  15.         tfidf_vectorizer = TfidfVectorizer()
  16.         tfidf_vectorizer.fit(dataset.data)
  17.         self.ttfidf_vectorizer = tfidf_vectorizer
  19.     @requests
  20.     def encode(self, docs: DocumentArray, *args, **kwargs):
  21.         iterable_of_texts = docs.get_attributes('text')
  22.         embedding_matrix = self.tfidf_vectorizer.transform(iterable_of_texts)
  24.         for i, doc in enumerate(docs):
  25.             doc.embedding = embedding_matrix[i]

PyTorch

The code snippet below takes docs as input and perform feature extraction with modelnet v2. It leverages Pytorch’s Tensor conversion and operation. Finally, the embedding of each document will be set as the extracted features.

  1. import torch  # 1.8.1
  2. import torchvision.models as models  # 0.9.1
  3. from jina import Executor, requests
  6. class PytorchMobilNetExecutor(Executor):
  7.     def __init__(self, **kwargs):
  8.         super().__init__()
  9.         self.model = models.quantization.mobilenet_v2(pretrained=True, quantize=True)
  10.         self.model.eval()
  12.     @requests
  13.     def encode(self, docs, **kwargs):
  14.         blobs = torch.Tensor(docs.get_attributes('blob'))
  15.         with torch.no_grad():
  16.             embeds = self.model(blobs).detach().numpy()
  17.             for doc, embed in zip(docs, embeds):
  18.                 doc.embedding = embed

ONNX

The code snippet bellow converts a Pytorch model to the ONNX and leverage onnxruntime to run inference tasks on models from hugging-face transformers.

  1. from pathlib import Path
  3. import numpy as np
  4. import onnxruntime
  5. from jina import Executor, requests
  6. from transformers import BertTokenizerFast, convert_graph_to_onnx
  9. class ONNXBertExecutor(Executor):
  10.     def __init__(self, **kwargs):
  11.         super().__init__()
  13.         # export your huggingface model to onnx
  14.         convert_graph_to_onnx.convert(
  15.             framework="pt",
  16.             model="bert-base-cased",
  17.             output=Path("onnx/bert-base-cased.onnx"),
  18.             opset=11,
  19.         )
  21.         # create the tokenizer
  22.         self.tokenizer = BertTokenizerFast.from_pretrained("bert-base-cased")
  24.         # create the inference session
  25.         options = onnxruntime.SessionOptions()
  26.         options.intra_op_num_threads = 1  # have an impact on performances
  27.         options.graph_optimization_level = (
  28.             onnxruntime.GraphOptimizationLevel.ORT_ENABLE_ALL
  29.         )
  31.         # Load the model as a graph and prepare the CPU backend
  32.         self.session = onnxruntime.InferenceSession(
  33.             "onnx/bert-base-cased.onnx", options
  34.         )
  35.         self.session.disable_fallback()
  37.     @requests
  38.     def encode(self, docs, **kwargs):
  39.         for doc in docs:
  40.             tokens = self.tokenizer.encode_plus(doc.text)
  41.             inputs = {name: np.atleast_2d(value) for name, value in tokens.items()}
  43.             output, pooled = self.session.run(None, inputs)
  44.             # assign the encoding results to ``embedding``
  45.             doc.embedding = pooled[0]