caikit.runtime.service_generation.rpcs

This package has classes that will serialize a python interface to a protocol buffer interface.

Typically used for `caikit.core.module`s that expose .train and .run functions.

Attributes

`log`
`INDENT`

Classes

`CaikitRPCBase`	Helper class that provides a standard way to create an ABC using
`ModuleClassTrainRPC`	Helper class to create a unique RPC corresponding with the train function
`TaskPredictRPC`	Helper class to create a unique RPC for the aggregate set of Modules that
`TaskPredictionJobRPC`	Helper class to create a unique RPC for prediction tasks. This reuses a
`TaskPredictionManagementRPC`	Helper class to create RPCs for each of the job management types
`TaskPredictionStatusRPC`	Helper class to create RPCs for each of the job management types
`TaskPredictionCancelRPC`	Helper class to create RPCs for each of the job management types
`TaskPredictionResultRPC`	Helper class to create RPCs for each of the job management types
`_RequestMessage`	Helper class to create the input request message that wraps up the inputs

Module Contents

caikit.runtime.service_generation.rpcs.log[source]

caikit.runtime.service_generation.rpcs.INDENT = ' '

class caikit.runtime.service_generation.rpcs.CaikitRPCBase[source]

Bases: abc.ABC

Helper class that provides a standard way to create an ABC using inheritance.

property module_list: List[Type[caikit.core.ModuleBase]]

Abstractmethod:

Return the list of caikit.core Modules that can be invoked using this RPC

property request: _RequestMessage

Abstractmethod:

Return the internal representation of the request message type for this RPC

property name: str

create_request_data_model(package_name: str) → Type[caikit.core.data_model.base.DataBase][source]: Dynamically create data model for this RPC’s request message

create_rpc_json(package_name: str) → Dict[source]: Return json snippet for the service definition of this RPC

class caikit.runtime.service_generation.rpcs.ModuleClassTrainRPC(method_signature: caikit.core.signature_parsing.CaikitMethodSignature)[source]

Bases: CaikitRPCBase

Helper class to create a unique RPC corresponding with the train function for a given module class

clz: Type[caikit.core.ModuleBase]

_method

_name = ''

return_type

_inner_request

_req

create_request_data_model(package_name: str)[source]: Partial override of CaikitRPCBase.create_request_data_model to take care of the inner request data model

property module_list: List[Type[caikit.core.ModuleBase]]: Returns a list containing the single caikit.core.module type that this RPC is for

property request: _RequestMessage: Return the internal representation of the request message type for this RPC

static module_class_to_rpc_name(module_class: Type[caikit.core.ModuleBase]) → str[source]: Helper function to convert from the name of a module to the name of the request RPC function

static module_class_to_req_name(module_class: Type[caikit.core.ModuleBase]) → str[source]

Helper function to convert from the name of a module to the name of the request RPC message

Example: self.clz._module__ = sample_lib.modules.sample_task.sample_implementation

return: SampleTaskSampleModuleTrainRequest

static module_class_to_inner_request_name(module_class: Type[caikit.core.ModuleBase]) → str[source]

Helper function to convert from a module to the name of the inner message containing all the training parameters

Example: self.clz._module__ = sample_lib.modules.sample_task.sample_implementation

return: SampleTaskSampleModuleTrainParameters

static _mutate_method_signature_for_training(signature: caikit.core.signature_parsing.CaikitMethodSignature) → caikit.core.signature_parsing.CaikitMethodSignature | None[source]

class caikit.runtime.service_generation.rpcs.TaskPredictRPC(task: Type[caikit.core.TaskBase], method_signatures: List[caikit.core.signature_parsing.CaikitMethodSignature], input_streaming: bool = False, output_streaming: bool = False)[source]

Bases: CaikitRPCBase

Helper class to create a unique RPC for the aggregate set of Modules that implement the same task

task

_method_signatures

_input_streaming = False

_output_streaming = False

_req

return_type

_name = ''

property module_list: List[Type[caikit.core.ModuleBase]]: Returns the list of all caikit.core.modules that this RPC will be for. These should all be of the same ai-problem, e.g. my_caikit_library.modules.classification

property request: _RequestMessage: Return the internal representation of the request message type for this RPC

property input_streaming: bool

property output_streaming: bool

create_rpc_json(package_name: str) → Dict[source]: Return json snippet for the service definition of this RPC

_handle_task_inputs(method_params: Dict[str, Any]) → Dict[str, Any][source]: Overrides input params with types specified in the Task

_task_to_req_name() → str[source]: Helper function to convert the pair of library name and task name to a request message name

_task_to_rpc_name() → str[source]

Helper function to convert the pair of library name and task name to an RPC name

Example: self.task = (sample_lib, sample_task)

return: SampleTaskPredict

class caikit.runtime.service_generation.rpcs.TaskPredictionJobRPC(task: Type[caikit.core.TaskBase], method_signatures: List[caikit.core.signature_parsing.CaikitMethodSignature])[source]

Bases: TaskPredictRPC

Helper class to create a unique RPC for prediction tasks. This reuses a large amount of the code from the TaskPredictRPC generation

return_type

_task_to_rpc_name() → str[source]

Helper function to convert the pair of library name and task name to an RPC name

Example: self.task = (sample_lib, sample_task)

return: SampleTaskPredict

class caikit.runtime.service_generation.rpcs.TaskPredictionManagementRPC(task: Type[caikit.core.TaskBase], method_signatures: List[caikit.core.signature_parsing.CaikitMethodSignature])[source]

Bases: CaikitRPCBase

Helper class to create RPCs for each of the job management types

task

_method_signatures

_name

_req

property return_type: Type[caikit.core.data_model.base.DataBase]

Abstractmethod:

The return type of the management APIs. Must be implemented by subclasses

property request: _RequestMessage: Return the internal representation of the request message type for this RPC

property module_list: List[Type[caikit.core.ModuleBase]]: Returns the list of all caikit.core.modules that this RPC will be for. These should all be of the same ai-problem, e.g. my_caikit_library.modules.classification

property input_type: Type[caikit.core.data_model.base.DataBase]: The input type for all management requests will be a PredictionJobInfoRequest

create_rpc_json(package_name: str) → Dict[source]: Return json snippet for the service definition of this RPC

abstract get_rpc_name() → str[source]: Helper function to convert the pair of library name and task name to an RPC name. This must be implemented by subclasses

class caikit.runtime.service_generation.rpcs.TaskPredictionStatusRPC(task: Type[caikit.core.TaskBase], method_signatures: List[caikit.core.signature_parsing.CaikitMethodSignature])[source]

Bases: TaskPredictionManagementRPC

Helper class to create RPCs for each of the job management types

return_type: The return type of the management APIs. Must be implemented by subclasses

get_rpc_name() → str[source]: Get the status RPC name

class caikit.runtime.service_generation.rpcs.TaskPredictionCancelRPC(task: Type[caikit.core.TaskBase], method_signatures: List[caikit.core.signature_parsing.CaikitMethodSignature])[source]

Bases: TaskPredictionManagementRPC

Helper class to create RPCs for each of the job management types

return_type: The return type of the management APIs. Must be implemented by subclasses

get_rpc_name() → str[source]: Get the cancel RPC name

class caikit.runtime.service_generation.rpcs.TaskPredictionResultRPC(task: Type[caikit.core.TaskBase], method_signatures: List[caikit.core.signature_parsing.CaikitMethodSignature])[source]

Bases: TaskPredictionManagementRPC

Helper class to create RPCs for each of the job management types

property return_type: The return type of the management APIs. Must be implemented by subclasses

get_rpc_name() → str[source]: Get the result RPC name

class caikit.runtime.service_generation.rpcs._RequestMessage(msg_name: str, params: Dict[str, Type], default_map: Dict[str, Any])[source]

Helper class to create the input request message that wraps up the inputs for a given function. The request Contains N named data-model or primitive objects.

name

triples = []

default_map

create_data_model(package_name: str) → Type[caikit.core.data_model.base.DataBase][source]: Dynamically create a data model for this request message