深度学习系统设计（二）（4）

public void predict(PredictRequest request, .. .. ..) {
  .. .. ..
  String runId = request.getRunId();                        ❶
  if (predictorManager.containsArtifact(runId)) {           ❷
    artifactInfo = predictorManager.getArtifact(runId);
  } else {
    try {
       artifactInfo = msClient.getArtifact(                 ❷
                GetArtifactRequest.newBuilder()
               .setRunId(runId).build());
     } catch (Exception ex) {
       .. .. .. 
    }
  } 
 # Step 4, pick predictor backend client by model algorithm type
  PredictorBackend predictor;
  if (predictorManager.containsPredictor(
        artifactInfo.getAlgorithm())) {
    predictor = predictorManager.getPredictor(              ❸
        artifactInfo.getAlgorithm());
  } else {
   .. .. ..
  }
  # Step 5, use the selected predictor client to download the model files
  predictor.downloadModel(runId, artifactInfo);             ❹
  # Step 6, use the selected predictor client to call
 # its backend predictor for model serving
  String r = predictor.predict(                             ❺
     artifactInfo, request.getDocument());                  ❺
  .. .. ..
}

service PredictionService {
 rpc Predict(PredictRequest) returns (PredictResponse);
}
message PredictRequest {
 string runId = 3;
 string document = 4;
}
message PredictResponse {
 string response = 1;
}

# the algorithm and predictor mapping can be defined in 
# either app config or docker properties
# enable algorithm types
ps.enabledPredictors=intent-classification
# define algorithm and predictors mapping
# predictor.<algorithm_type>.XXX = predictor[host, port, type]
predictors.intent-classification.host= \      ❶
  Intent-classification-predictor             ❶
predictors.intent-classification.port=51001
predictors.intent-classification.techStack=customGrpc

public class PredictorConnectionManager {
  private final Map<String, List<ManagedChannel>> 
    channels = new HashMap<>();
  private final Map<String, PredictorBackend>                ❶
    clients = new HashMap<>();
  private final Map<String, GetArtifactResponse>             ❷
    artifactCache;
  // create predictor backend objects for 
 // the registered algorithm and predictor
  public void registerPredictor(String algorithm, 
       Properties properties) {
    String host = properties.getProperty(                    ❸
       String.format(“predictors.%s.host”, algorithm));
    int port = Integer.parseInt(properties.getProperty(      ❸
       String.format(“predictors.%s.port”, algorithm)));
    String predictorType = properties.getProperty(           ❸
       String.format(“predictors.%s.techStack”, algorithm));
    ManagedChannel channel = ManagedChannelBuilder
       .forAddress(host, port)
       .usePlaintext().build();
    switch (predictorType) {
      .. ..
      case “customGrpc”:                                     ❹
      default:
        channels.put(algorithm, List.of(channel));
        clients.put(algorithm, new CustomGrpcPredictorBackend(
          channel, modelCachePath, minioClient));
      break;
     }
  }
  .. .. ..
}

public interface PredictorBackend {
   void downloadModel(String runId, 
           GetArtifactResponse artifactResponse);
   String predict(GetArtifactResponse artifact, String document);
   void registerModel(GetArtifactResponse artifact);
}

// calling backend predictor for model serving
public String predict(GetArtifactResponse artifact, String document) {
   return stub.predictorPredict(PredictorPredictRequest
       .newBuilder().setDocument(document)    ❶
       .setRunId(artifact.getRunId())         ❷
       .build()).getResponse();
}

service Predictor {
 rpc PredictorPredict(PredictorPredictRequest) returns (PredictorPredictResponse);
}
message PredictorPredictRequest {
 string runId = 1;
 string document = 2;
}
message PredictorPredictResponse {
 string response = 1;
}

├── manifest.json                          ❶
├── model.pth                              ❷
└── vocab.pth                              ❸
// A sample manifest.json file 
{
  "Algorithm": "intent-classification",    ❹
  "Framework": "Pytorch",
  "FrameworkVersion": "1.9.0",
  "ModelName": "intent",
  "CodeVersion": "80bf0da",
  "ModelVersion": "1.0",
  "classes": {                             ❺
    "0": "cancel",
    "1": "ingredients_list",
    "2": "nutrition_info",
    "3": "greeting",
     .. .. ..
}

class TextClassificationModel(nn.Module):
  def __init__(self, vocab_size, embed_dim,   ❶
      fc_size, num_class):                    ❶
    super(TextClassificationModel, self).__init__()
    self.embedding = nn.EmbeddingBag(vocab_size, embed_dim, sparse=True)
    self.fc1 = nn.Linear(embed_dim, fc_size)
    self.fc2 = nn.Linear(fc_size, num_class)
    self.init_weights()
  def forward(self, text, offsets):
    embedded = self.embedding(text, offsets)
    return self.fc2(self.fc1(embedded))

class ModelManager:
  def __init__(self, config, tokenizer, device):
    self.model_dir = config.MODEL_DIR
    self.models = {}                                      ❶
  # load model file and initialize model
  def load_model(self, model_id):
    if model_id in self.models:
      return
    # load model files, including vocabulary, prediction class mapping.
    vacab_path = os.path.join(self.model_dir, model_id, "vocab.pth")
    manifest_path = os.path.join(self.model_dir, model_id, "manifest.json")
    model_path = os.path.join(self.model_dir, model_id, "model.pth")
    vocab = torch.load(vacab_path)
    with open(manifest_path, 'r') as f:
    manifest = json.loads(f.read())
    classes = manifest['classes']
    # initialize model graph and load model weights
    num_class, vocab_size, emsize = len(classes), len(vocab), 64
    model = TextClassificationModel(vocab_size, emsize, 
       self.config.FC_SIZE, num_class).to(self.device)
    model.load_state_dict(torch.load(model_path))
    model.eval()
    self.models[self.model_key(model_id)] = model        ❷
self.models[self.model_vocab_key(model_id)]              ❷
  ➥ = vocab                                             ❷
self.models[self.model_classes(model_id)]                ❷
  ➥ = classes                                           ❷
  # run model to make prediction
  def predict(self, model_id, document):
    # fetch model graph, dependency and 
    # classes from cache by model id 
    model = self.models[self.model_key(model_id)]
    vocab = self.models[self.model_vocab_key(model_id)]
    classes = self.models[self.model_classes(model_id)]
    def text_pipeline(x):
      return vocab(self.tokenizer(x))
    # transform user input data (text string) 
 # to model graph’s input
    processed_text = torch.tensor(text_pipeline(document), dtype=torch.int64)
    offsets = [0, processed_text.size(0)]
    offsets = torch.tensor(offsets[:-1]).cumsum(dim=0)
    val = model(processed_text, offsets)                ❸
    # convert prediction result from an integer to 
 # a text string (class)
    res_index = val.argmax(1).item()
    res = classes[str(res_index)]
    print("label is {}, {}".format(res_index, res))
    return res

def serve():
  .. .. ..
  model_manager = ModelManager(config, 
    tokenizer=get_tokenizer('basic_english'), device="cpu")
  server = grpc.server(futures.
    ThreadPoolExecutor(max_workers=10))                         ❶
  prediction_service_pb2_grpc.add_PredictorServicer_to_server(
    PredictorServicer(model_manager), server)                   ❷
  .. .. ..
class PredictorServicer(prediction_service_pb2_grpc.PredictorServicer):
  def __init__(self, model_manager):
    self.model_manager = model_manager
 # Serving logic 
  def PredictorPredict(self, request, context: grpc.ServicerContext):
 # load model 
    self.model_manager.load_model(model_id=request.runId)
    class_name = self.model_manager.                            ❸
      predict(request.runId, request.document)
    return PredictorPredictResponse(response=json.dumps({'res': class_name}))

# step 1: start torchserve backend
docker run --name intent-classification-torch-predictor\
 --network orca3 --rm -d -p "${ICP_TORCH_PORT}":7070 \
 -p "${ICP_TORCH_MGMT_PORT}":7071 \
 -v "${MODEL_CACHE_DIR}":/models \                     ❶
 -v "$(pwd)/config/torch_server_config.properties": \
     /home/model-server/config.properties \
 pytorch/torchserve:0.5.2-cpu torchserve \             ❷
 --start --model-store /models                         ❸
# step 2: start the prediction service (the web frontend)
docker build -t orca3/services:latest -f services.dockerfile .
docker run --name prediction-service --network orca3 \
  --rm -d -p "${PS_PORT}":51001 \
  -v "${MODEL_CACHE_DIR}":/tmp/modelCache \            ❹
 orca3/services:latest  \
 prediction-service.jar
# step 3: make a prediction request, ask intent for “merry christmas”
grpcurl -plaintext 
  -d "{
    "runId": "${MODEL_ID}",
    "document": "merry christmas"
 }" 
 localhost:"${PS_PORT}" prediction.PredictionService/Predict

public void registerModel(GetArtifactResponse artifact) {
  String modelUrl = String.format(MODEL_FILE_NAME_TEMPLATE,
        artifact.getRunId());
  String torchModelName = String.format(TORCH_MODEL_NAME_TEMPLATE,
            artifact.getName(), artifact.getVersion());
  ManagementResponse r = managementStub.registerModel(     ❶
           RegisterModelRequest.newBuilder()
             .setUrl(modelUrl)
             .setModelName(torchModelName)
             .build());
  # Assign resource (TorchServe worker) for this model
  managementStub.scaleWorker(ScaleWorkerRequest.newBuilder()
             .setModelName(torchModelName)
             .setMinWorker(1)
             .build());
}

# prediction with single input on model resnet-18
curl http:/ /localhost:8080/predictions/resnet-18 \
 -F "data=@kitten_small.jpg"
# prediction with multiple inputs on model squeezenet1_1
curl http:/ /localhost:8080/predictions/squeezenet1_1 \
 -F 'data=@docs/images/dogs-before.jpg' \
 -F 'data=@docs/images/kitten_small.jpg'

// call TorchServe gRPC prediction api
public String predict(GetArtifactResponse artifact, String document) {
  return stub.predictions(PredictionsRequest.newBuilder()
           .setModelName(String.format(TORCH_MODEL_NAME_TEMPLATE,
              artifact.getName(), artifact.getVersion()))
           .putAllInput(ImmutableMap.of("data",                         ❶
              ByteString.copyFrom(document, StandardCharsets.UTF_8)))
                .build()).getPrediction()
           .toString(StandardCharsets.UTF_8);
}