Deployment API

System Requirements and Setup

See the Toolchain Setup Guide in the toolchain documentation.

Model/Pipelines Deployment and Compile

In this guide, we will outline the application development workflow using NeuReality’s Deployment API.

In the following sections, we will elucidate key concepts by guiding developers through the process of adapting a sample Roberta QA pipeline for performing inference on the NR1 module platform.

Initialization - Original Application Code

Use your chosen pipeline (it is taken from Hugging Face in this sample).

from transformers import pipeline

pipe = pipeline("text-classification", model="fabriceyhc/bert-base-uncased-imdb")
question="why are distributed systems better?"
context="Distributed systems are taking the world by storm, allowing for more natural scaling and less bottlenecks"
result = pipe(question=question, context=context)
print("Answer:", result["answer"]) # Answer: more natural scaling and less bottlenecks

Refactor the pipeline

To improve code modularity and convert your code into a graphable format, you should follow these steps to split the pipeline into its parts β€” tokenizer, model, and postprocessing. This approach will help make your code more maintainable, testable, and reusable. To achieve this, start from defining the Modules: Tokenizer Module responsible for preprocessing the input data into a format suitable for the model; Model Module that handles the core processing logic, which could include running a machine learning model or performing ML computations; Postprocessing Module that takes the output from the model and converts it into a final usable format.

# Load model directly
from transformers import AutoTokenizer, AutoModelForQuestionAnswering

tokenizer = AutoTokenizer.from_pretrained("deepset/roberta-base-squad2")
model = AutoModelForQuestionAnswering.from_pretrained("deepset/roberta-base-squad2")

# Tokenize inputs
inputs = tokenizer.encode_plus(question, context, return_tensors="pt", max_length=128, truncation=True)

# Perform inference
with torch.no_grad():
    outputs = model(**inputs)

# Decode and print the answer
answer_start_scores = outputs.start_logits
answer_end_scores = outputs.end_logits
answer_start = torch.argmax(answer_start_scores)
answer_end = torch.argmax(answer_end_scores) + 1
answer = tokenizer.decode(inputs["input_ids"][0][answer_start:answer_end])
print("Answer:", answer) # Answer: more natural scaling and less bottlenecks

Moving to NeuReality Programming Model

  • NeuReality programming model supports the following constructs represented by the corresponding python classes:

    QualcommProgamContainerMOdel, PythonModel, EnsembleModel, and accelerated pre/post-processing operators located in neureality.ops package.

    • General python code executed on CPU is wrapped by PythonModel.

    • The model itself should be compiled with Qualcomm Cloud AI100 SDK tools and then wrapped with QualcommProgramContainerModel. Qualcomm Cloud AI100 SDK supports PyTorch and ONNX model compilation.

    • If accelerated pre/post processing operators are used, the original calls to these operators should be replaced with the calls from neureality.ops package.

    • If the AI pipeline represents an ensemble of modules, it can be implemented by wrapping all the calls with EnsembleModel. _Note: β€œFree code” (in contrast to calling the known operators) can not be written inside the EnsembleModel. If β€œfree code” is needed, wrap it with PythonModel instead.

    For more information, refer to the Deployment API reference.

  • Split the logic of the code into one or more preprocessing, model, and postprocessing stages (classes) as needed. Each part should inherit from the relevant class from NeuReality programming model.

  • Once using Qualcomm Cloud AI100 accelerator, you should convert your model into a qpc - Qualcomm Cloud AI100 execution blob, by using Qualcomm Cloud AI100 SDK. This process includes generating qpc files includes converting the PyTorch model to onnx and the ONNX model to qpc using β€˜/opt/qti-aic/exec/qaic-exec’ tool. Once the qpc is created, assume it is saved under (or copy it from where it saved so it would be here). ```sh ./programqpc.bin

Wrap it with QualcommProgramContainerModel. This class requires a qpc path, which is a path to the binary file representing the model (programqpc.bin) which you have created.

For more information, refer to the official Qualcomm Cloud AI100 SDK documentation and model compilation flow guide.

  • Each class should define init and forward functions. The init function is called during the class initialization, and the forward function is called during runtime.

Back to the example, there are three main phases:

  • Preprocessing (tokenizer)

  • The model

  • Postprocessing (get answers spans, extract the enswers and detokenize)

After reviewing the flow of each phase, we can infer that the tokenizer should be wrapped with PythonModel, the model should be wrapped with QualcommProgamContainerMOdel, and the postprocessing will be implemented with NeuReality operators.

Preprocess

import neureality as nr
from transformers import RobertaTokenizer
import numpy as np

# This code has been rearranged based on the findings from the review of the tokenizer's source code.
class Tokenizer(nr.PythonModel):

    """Question answering tokenizer which handles the conversion of text (strings) into token IDs (the numerical representations that the model can understand)

    Attributes:
        tokenizer: Roberta tokenizer
        pad_token_idx: the token index for padding
        sentence_sep_token_idx: the token index for sentence separator
        batch_size: number of sequences in a batch

    """

    def __init__(self, max_length=128, batch_size=5):

        """Question answering tokenizer

        Args:
            max_length (int, optional):  maximum number of tokens in a sentence, sequences will be padded to this number. Defaults to 128.
            batch_size (int, optional): number of sequences in a batch. Defaults to 5.
        """
        super().__init__()
        self.tokenizer = RobertaTokenizer.from_pretrained("deepset/roberta-base-squad2")
        self.tokenizer.model_max_length = max_length
        self.pad_token_idx = 1
        self.sentence_sep_token_idx = 2
        self.batch_size = batch_size

    def generate_non_answer_mask(self, input_tokens):

        """Generates a mask for all items that cannot be the answer, i.e. all special tokens such as question mark, <bos>, <sep>"""

        mask_or_pad = np.where(input_tokens == self.pad_token_idx, 1, 0) | np.where(
            input_tokens == self.sentence_sep_token_idx, 1, 0
        )
        sep_indices = [tokens.tolist().index(self.sentence_sep_token_idx) for tokens in input_tokens]
        question = np.zeros(input_tokens.shape, dtype=np.int64)
        for sentence_idx, sep_idx in enumerate(sep_indices):
            question[sentence_idx, :sep_idx] = 1

        SOFTMAX_MASK = -10000
        non_answer_mask = mask_or_pad | question
        non_answer_mask = non_answer_mask * SOFTMAX_MASK
        return non_answer_mask.astype(np.float16)

    def forward(self, questions_uint_seqs, contexts_uint_seqs):

        """
        Args:
            questions_uint_seqs: numpy array of the question sequence
            contexts_uint_seqs: numpy array of the context sequence
        Returns:
            input_tokens: Indices of input sequence tokens in the vocabulary
            attention_mask: Mask to avoid performing attention on padding token indices. Mask values selected in [0, 1]:
                            1 for tokens that are not masked,
                            0 for tokens that are masked.
            non_answer_mask: Mask for everything that cannot be the answer, i.e. all question, <bos>, <sep> tokens
        """

        questions_strs = nr.AppUtils.get_strings_from_np_uint_arr(questions_uint_seqs)
        contexts_strs = nr.AppUtils.get_strings_from_np_uint_arr(contexts_uint_seqs)
        tokenizer_output = self.tokenizer(questions_strs, contexts_strs, return_tensors="np", padding="max_length")

        input_tokens = tokenizer_output["input_ids"].astype(np.int32)
        attention_mask = tokenizer_output["attention_mask"].astype(np.float16)
        non_answer_mask = self.generate_non_answer_mask(input_tokens)
        return input_tokens, attention_mask, non_answer_mask

Model

Given the qpc model generated with Qualcomm Cloud AI100 SDK (assumed it is under     ./programqpc.bin    as we said before), use QualcommProgramContainerModel to wrap it.

from neureality.api.qpc_model import QualcommProgramContainerModel

qpc_path="./programqpc.bin"
QualcommProgramContainerModel(qpc_path)

Postprocess

Get Answer Spans
from neureality import ops

class GetAnswerSpans(nr.EnsembleModel):
    """Post process model predictions to generate answer spans within the input context sequences.
    Attributes:
        max_answer_len: maximum number of input tokens that can be used for an answer
        batch_size: number of sequences in a batch
    """

    def __init__(self, max_answer_tokens=15, batch_size=5):
        """Post process model predictions to generate answer spans within the input context sequences
        Args:
            max_answer_tokens: maximum number of input tokens that can be used for an answer
            batch_size: number of sequences in a batch
        """
        super().__init__()
        self.max_answer_len = max_answer_tokens
        self.batch_size = batch_size

    def forward(self, probs_outer_product):
        """
        Args:
            probs_outer_product: model predictions. a tensor of shape (N, max_seq_len, max_seq_len)
                                of start_idx,end_idx outer product probabilities for each input token
        Returns:
            answers_spans: tensor of shape (N, 2), start and end indices for every answer
            answers_scores: tensor of shape (N), the score of the answer
        """

        # only answers with start > end are legal
        # we keep only the (0,0) diagonal element for impossible answer
        legal_answers_probs = ops.trilu(probs_outer_product, upper=1, keep_first_element=1)
        # only answers <= max_answer_len are considered
        fitting_answers_probs = ops.trilu(legal_answers_probs, upper=0, k=self.max_answer_len - 1)
        max_idx = ops.argmax(fitting_answers_probs, k=1)
        answers_spans = ops.unravel_index(
            max_idx,
            np.tile(
                np.asarray([fitting_answers_probs.shape[1], fitting_answers_probs.shape[2]]).reshape(1, -1),
                (self.batch_size, 1),
            ),
        )

        answers_scores = ops.max(fitting_answers_probs, axis=(1, 2))
        return answers_spans, answers_scores
Extract Answers and Detokenize
class ExtractAnswerAndDetokenize(nr.PythonModel):
    """Extract the answers from input context tokens using the predicted spans.
    Then detokenize the answers
     Args:
         buffer_size_bytes: maximum answer string size in bytes
    """

    def __init__(self, buffer_size_bytes=1000):
        super().__init__()
        self.buffer_size = buffer_size_bytes
        self.tokenizer = RobertaTokenizer.from_pretrained("deepset/roberta-base-squad2")

    def forward(self, answers_spans, input_tokens):
        """Args:
        answer_spans: tensor of shape (N, 2), start and end indices for every answer
        input_tokens: tensor of shape (N, max_seq_len), input sequece
        """
        answers_strs = []
        for answer_idx, span in enumerate(answers_spans):
            start_idx, end_idx = span
            answer_tokens = input_tokens[answer_idx, start_idx : end_idx + 1]
            answer = self.tokenizer.decode(answer_tokens)
            answers_strs.append(answer)

        return nr.AppUtils.get_np_uint_arr_from_strings(answers_strs, buffer_size=self.buffer_size)

Full Pipeline

# one place to define all parameters regarding the qpc
@dataclass
class QPCRobertaQAConfig:
    batch_size: int = 5
    max_length: int = 128


class RobertaQAEnsemble(nr.EnsembleModel):
    """
    The Roberta QA ensemble model takes a question and context as inputs and returns the answer and the predictions. All tensors except for predictions are passed in uint8 format, predictions have float16 format.
    """

    def __init__(self, qpc_path: str):
        super().__init__()
        self.config = QPCRobertaQAConfig()
        # initializing the preprocess we defined above
        self.tokenizer = Tokenizer(max_length=self.config.max_length, batch_size=self.config.batch_size)
        # using the model we initilaized above
        self.roberta = QualcommProgramContainerModel(qpc_path)
        # initializing the postprocess parts we defined above
        self.get_answer_spans = GetAnswerSpans(batch_size=self.config.batch_size)
        self.extract_and_detokenize = ExtractAnswerAndDetokenize()

    def forward(self, questions_uint_seqs, contexts_uint_seqs):
        """Question answering pipeline forward
        Args:
            questions_uint_seqs: np uint array with question ascii coded strings padded to max length (batch_size, 1000)
            contexts_uint_seqs: np uint array with context ascii coded strings padded to max length (batch_size, 1000)
        Returns:
            answers: the string answer array as numpy, shape (batch_szie, max_length=1000), dtype uint8
        """

        input_tokens, attention_mask, non_answer_mask = self.tokenizer(questions_uint_seqs, contexts_uint_seqs)
        probs_outer_product = self.roberta(input_tokens, attention_mask, non_answer_mask)
        answers_spans, answers_scores = self.get_answer_spans(probs_outer_product)
        answers_strs = self.extract_and_detokenize(answers_spans, input_tokens)
        return answers_strs

Pre-compilation (run the reference code on X86)

  • Intialize the pipeline.

  • Create the inputs. The inputs should be as numpy arrays.

  • Run on X86 and get reference results.

# intialize the defined pipeline
roberta_ensemble = RobertaQAEnsemble(qpc_path)
config = QPCRobertaQAConfig()
batch_size = config.batch_size
# create the inputs
question = [question] if isinstance(question, str) else question
context = [context] if isinstance(context, str) else context

# create batched_input if it is not already batched
if len(question) != config.batch_size:
    question = question * config.batch_size
if len(context) != config.batch_size:
    context = context * config.batch_size

# create numpy array from the question and context strings
question_np = nr.AppUtils.get_np_uint_arr_from_strings(question)
context_np = nr.AppUtils.get_np_uint_arr_from_strings(context)

# run the pipeline
cpu_uint_strs = roberta_ensemble(question_np, context_np)
cpu_strs = nr.AppUtils.get_strings_from_np_uint_arr(cpu_uint_strs)[0]
print("Answer: ", cpu_strs)  # Answer: more natural scaling and less bottlenecks

Compile, Simulate, Validate and Upload

Compile

  • Compile the pipeline for either the device or the simulator. The compilation is done by calling the β€œbuild” function.

  • Compile to device: Compile and obtain artifacts (compute graph (CG) descriptor and binary directories). Using this CG, we can run the pipeline on NR1β„’ AI Inference Appliance and get results.

  • Compile to simulator: This process also generates the CG and binaries. However, the artifacts differ between the two modes. For example, target operating system will be x86 for simulator binaries and ARM in a case of on-device binaries.

  • Runtime parameters can also be defined before the build, like setting the number of activations, this is probably related for building to device since only there we care about performance. Choose the numbers according to the Qualcomm AI100 accelerator you are using (for example Qualcomm Cloud AI100 Pro accelerator has 16 cores).

# Create an input sample for tracing the application
# Input samples are nr.Tensor which are numpy wrappers.
tracing_sample = [nr.Tensor(question_np), nr.Tensor(context_np)]
build_path="./roberta_artifacts"

# Runtime parameters - see the next commented line (uncomment it if want to compile to device):
# roberta_ensemble.roberta.set_build_settings({"ai100_num_activations": "16"})

# Call build function will compile the pipeline
nr_model = roberta_ensemble.build(
                build_path=build_path,
                input_sample=tracing_sample,
                python_requirements=["transformers==4.28.1"],
                compilation_mode=compilation_mode,
            )

For more information, see the build parameters in the Deployment API documentation.

A brief reminder of the arguments for the build method is below (the full explanation can be found in the link above):

  • build_path - the location where the artifacts will be saved

  • input_sample - a tensor with a specific shape and dtype, used for tracing

  • python_requirements - the specific dependencies that need to be installed for the pipeline

  • compilation_mode - simulator/device. Use NR_CONFIG_MODE_SIM_TURBO_OFF for simulator compilation and NR_CONFIG_MODE_DEVICE for device compilation.

As mentioned above, the compilation outcomes include a CG file which is stored in the directory roberta_artifacts, specified by the build_path argument. The other artifacts are directories representing various compute nodes in the CG. Each directory includes its own CG file (sub-CG of the the full CG for the entire pipeline). These directories are also stored in the directory roberta_artifacts, specified by the build_path argument. For the operators which are running on the DSP additional directories are generated in the same location. They include a DSP descriptor binary and txt files. If Python models are included in the pipeline, the corresponding docker files are generated for each model and stored in dcr.int.neureality.ai registry. A requirements.txt file is also generated according to the python_requirements argument.

The 'nr_model' we get from building has a SimulatorExecutor type. You can use it with a simulator tool (see more info in the next session). For more information, refer to SimulatorExecutor in the Deployment API documentation.

Simulate, Validate and Upload

Simulate (only for simulator mode)

  • Simulate the CG using the simulator tool and compare the results to the reference results.

  • The input provided for the simulation should be the same as the input used for the pre-compilation execution.

Note:
For more information about the simulation and profiling tools see Deployment API section.

# run NR1β„’ simulation using compiled binaries
bold_print("\nSimulating NR1 binaries ...\n")
sim_uint_strs = nr_model([question_np, context_np])
sim_strs = nr.AppUtils.get_strings_from_np_uint_arr(sim_uint_strs)[0]
print("Answer: ", sim_strs)  # Answer: more natural scaling and less bottlenecks

Validate (only for building to simulator mode)

# compare the results with reference model
bold_print("\nComparing simulation with reference model ...")
assert cpu_strs == sim_strs, f"Roberta ensemble simulator returns wrong results \n{cpu_strs=}\n{sim_strs=}"
bold_print("\nSuccess!")

If β€œSuccess!” is displayed on the screen, it indicates that the pipeline has been successfully written according to NeuReality programming model, compiled, simulated, and validated.

You can see two flows here, the left is the flow which is done trough Qualcomm Cloud AI100 accelerator tools, and the right is what we have discussed in this session:

My Image

Upload the artifacts to CG store (for both building to device and to simulator)

In order to provision the artifacts to the NR1β„’ module, you need to upload them to S3 storage or to an NFS server.

Upload to S3 storage

There is a utility function to upload the artifacts to S3.

Let’s assume that you have a bucket named β€œcg-store-dev” and you want to upload the compiled compute graph to β€œexample/key” inside this bucket:

bucket = "cg-store-dev"
bucket_obj_key= "example/key"
nr.upload_artifacts_to_s3(build_path, bucket, bucket_obj_key)

CG Provisioning and Inference on NR1β„’ module

Please refer to the following documentation to learn more about NR1 configuration and compute graph provisioning, as well as serving requests:

Profiling

Profiling is an essential part of AI pipeline optimization process. By using the profiler the user can understand what are the main bottlenecks regarding computation and processing time and can improve the performance by adjusting pipeline structure and parameters.

This guide assumes that the user is already running a deployment docker container provided by NeuReality. All commands in this guide are run from within that container.

Build Example Pipeline

Build the example pipeline, by running:

python3 pipelines/demos/resnet50/deploy/model_as_qpc/app.py --input_image_path pipelines/demos/resnet50/sample_inputs/img_dog.jpg --compile_to_nr1

Here’s the structure of the artifacts folder:

pipelines/demos/resnet50/deploy/resnet_artifacts
β”œβ”€β”€ cg_descriptor.gv
β”œβ”€β”€ cg_descriptor.json
β”œβ”€β”€ ensemble_trace.gv
β”œβ”€β”€ Normalize
β”‚ β”œβ”€β”€ cg_descriptor.gv
β”‚ └── cg_descriptor.json
β”œβ”€β”€ QualcommProgramContainerModel
β”‚ β”œβ”€β”€ c_funcs
β”‚ β”œβ”€β”€ cg_descriptor.gv
β”‚ └── cg_descriptor.json
└── vision_dsp_main_0_summary_Normalize__dsp_normalize
β”œβ”€β”€ dsp_desc_0.bin
β”œβ”€β”€ dsp_desc_debug_info.txt
β”œβ”€β”€ optimized_relay_debug_info.txt
β”œβ”€β”€ param0.bin
β”œβ”€β”€ param1.bin
└── relay_debug_info.txt

Set AWS credentials

Profiling tool requires that AWS credentials are set via environment variables AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY and AWS_SESSION_TOKEN. The values for these variables can be obtained by looking at the section β€œAccess Keys” of the AWS account used to connect to the S3 storage.

For more informations, see AWS cli configuration via env variables.

Upload Artifacts to CG storage

Assuming that bucket=cg-store-dev and bucket_obj_key=example/key, run the following command:

python3 pipelines/demos/resnet50/deploy/model_as_qpc/upload_to_cg_store.py --build_path pipelines/demos/resnet50/deploy/resnet_artifacts --bucket cg-store-dev --bucket_obj_key example/key

Run Profiler

In the container run the following command:

python3 python/neureality/samples/generic_profiler/main.py --server_address <NR1_server_address> --cg_path pipelines/demos/resnet50/deploy/resnet_artifacts

The profiler will output a report:

Profiler report for CG 'ResNet50EnsemblePipeline', Latency = 63029.1[us]
|-----------------------------------------------------|---------------|---------------------|-------------------|-------------|
|                       CG-Node                       |    CE (id)    | start-timestamp[us] | end-timestamp[us] | latency[us] |
|:---------------------------------------------------:|:-------------:|:-------------------:|:-----------------:|:-----------:|
|                      data-read                      |    nif (0)    |         0.0         |      11644.0      |   11644.0   |
|         1/ResNet50EnsemblePipeline/control          |    jpu (0)    |       11663.9       |      11694.9      |    31.0     |
|          Normalize_n987004__dsp_normalize           | audio-dsp (2) |       11694.9       |      12022.4      |    327.5    |
|             4/Normalize_n987004/control             |    jpu (0)    |       12024.2       |      12062.7      |    38.5     |
| QualcommProgramContainerModel__NrExternalTargetCall |   gpcpu (0)   |       12164.2       |      62934.2      |   50770.0   |
|       4/QualcommProgramContainerModel/control       |    jpu (0)    |       62953.6       |      62983.4      |    29.8     |
|                       12/EXIT                       |   gpcpu (0)   |       62993.9       |      63000.9      |     7.0     |
|         13/ResNet50EnsemblePipeline/control         |    jpu (0)    |       63008.8       |      63029.1      |    20.3     |

Each line in the report corresponds to a single compute node in compute graph (CG).

The report also specifies the actual placement assignment of each compute node on NR1 accelerating compute engines. The following assignments are reported:

  1. NIF – network interface.

  2. JPU – Job processing unit, handles control flows between the compute nodes.

  3. GP-DSP - Operations that run on the DSP (audio DSP or video DSP).

  4. GP-CPU - Operations handled by the main processor on the chip, which might or might not wrap an operation that is sent to an external DLA.

Start/end execution times and latency are also reported for each compute node.

In the example above, the pipeline spent some time bringing the data onto the chip - 11,644 us. Then, the Job Processing Unit handled its input.

Before performing the main RESNET operation a preprocessing step in the audio-dsp took 327 us.

Most of the processing time was running the RESNET model – 50,770 us.

Deployment API Reference

Open the in-depth Deployment API documentation to view the API reference.

NeuReality Operators Reference

Refer to the detailed NeuReality Operators API to learn how to utilize the operators.

Workload samples

NR Samples github repository provides a collection of typical workload samples along with exemplary code to generate NR1 artifacts. This repository serves as an excellent starting point for utilizing the Deployment API guide. Please refer to the documentation associated with the samples for specific instructions.