Skip to content

Module: base_inference_workflow

Module implementing a base inference workflow.

This class is not meant to be subclassed directly; instead, subclass one of [TGIClientInferenceWorkflow, CSSInferenceWorkflow, BaseClassicInferenceWorkflow]

BaseInferenceWorkflow

Base class for an inference workflow

Source code in src/infernet_ml/workflows/inference/base_inference_workflow.py 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
class BaseInferenceWorkflow(metaclass=abc.ABCMeta):
    """
    Base class for an inference workflow
    """

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        """
        Constructor. keeps track of arguments passed in.
        """
        super().__init__()
        self.args: list[Any] = list(args)
        self.kwargs: dict[Any, Any] = kwargs

        self.is_setup = False
        self.__inference_count: int = 0
        self.__proof_count: int = 0

    def setup(self) -> Any:
        """
        calls setup and keeps track of whether or not setup was called
        """
        self.is_setup = True
        return self.do_setup()

    @abc.abstractmethod
    def do_setup(self) -> Any:
        """set up your workflow here.
        For LLMs, this may be parameters like top_k, temperature
        for classical LLM, this may be model hyperparams.

        Returns: Any
        """

    def stream(self, input_data: Any) -> Iterator[Any]:
        """
        Stream data for inference. Subclasses should implement do_stream.
        """
        logging.info("preprocessing input_data %s", input_data)
        preprocessed_data = self.do_preprocessing(input_data)
        yield from self.do_stream(preprocessed_data)

    def inference(self, input_data: Any, log_preprocessed_data: bool = True) -> Any:
        """performs inference. Checks that model is set up before
        performing inference.
        Subclasses should implement do_inference.

        Args:
            input_data (typing.Any): input from user
            log_preprocessed_data (bool, optional): If True, logs the
            preprocessed input data before running the inference. Defaults to True.

        Raises:
            ValueError: if setup not called beforehand

        Returns:
            Any: result of inference
        """
        if not self.is_setup:
            raise ValueError("setup not called before inference")

        logging.info("preprocessing input_data %s", input_data)
        preprocessed_data = self.do_preprocessing(input_data)
        if log_preprocessed_data:
            logging.info("querying model with %s", preprocessed_data)
        model_output = self.do_run_model(preprocessed_data)

        logging.info("postprocessing model_output %s", model_output)
        self.__inference_count += 1
        return self.do_postprocessing(input_data, model_output)

    @abc.abstractmethod
    def do_run_model(self, preprocessed_data: Any) -> Any:
        """run model here. preprocessed_data type is
            left generic since depending on model type
        Args:
            preprocessed_data (typing.Any): preprocessed input into model

        Returns:
            typing.Any: result of running model
        """
        pass

    def do_preprocessing(self, input_data: Any) -> Any:
        """
        Implement any preprocessing of the raw user input. For example, you may need to
        apply feature engineering on the input before it is suitable for model inference.
        By default, this method returns the input data as is.

        Args:
            input_data (Any): raw input from user

        Returns:
            Any: preprocessed input
        """
        return input_data

    @abc.abstractmethod
    def do_stream(self, preprocessed_input: Any) -> Iterator[Any]:
        """
        Implement any streaming logic here. For example, you may
        want to stream data from OpenAI or any LLM. This method
        should return the data to be streamed.

        Returns:
            typing.Any: data to be streamed
        """

    def do_postprocessing(self, input_data: Any, output_data: Any) -> Any:
        """
        Implement any postprocessing of the model output. By default, this method
        returns the output data as is.

        Args:
            input_data (Any): raw input from user
            output_data (Any): model output

        Returns:
            Any: postprocessed output
        """
        return output_data

    @final
    def generate_proof(self) -> None:
        """
        Generates proof. checks that setup performed before hand.
        """
        if self.__inference_count <= self.__proof_count:
            logging.warning(
                "generated %s inferences only but "
                + "already generated %s. Possibly duplicate proof.",
                self.__inference_count,
                self.__proof_count,
            )

        self.do_generate_proof()
        self.__proof_count += 1

    def do_generate_proof(self) -> Any:
        """
        Generates proof, which may vary based on proving system. We currently
        only support EZKL based proving, which does not require proof
        generation to be defined as part of the inference workflow if
        the inference from the circuit directly used for proving. Indeed,
        that is the case for the classic infernet_ml proof service. However, this
        may change with the usage of optimistic and other eager or lazy
        proof systems, which we intend to support in the future. By default,
        will raise NotImplementedError. Override in subclass as needed.
        """
        raise NotImplementedError

__init__(*args, **kwargs)

Constructor. keeps track of arguments passed in.

Source code in src/infernet_ml/workflows/inference/base_inference_workflow.py 
21
22
23
24
25
26
27
28
29
30
31
def __init__(self, *args: Any, **kwargs: Any) -> None:
    """
    Constructor. keeps track of arguments passed in.
    """
    super().__init__()
    self.args: list[Any] = list(args)
    self.kwargs: dict[Any, Any] = kwargs

    self.is_setup = False
    self.__inference_count: int = 0
    self.__proof_count: int = 0

do_generate_proof()

Generates proof, which may vary based on proving system. We currently only support EZKL based proving, which does not require proof generation to be defined as part of the inference workflow if the inference from the circuit directly used for proving. Indeed, that is the case for the classic infernet_ml proof service. However, this may change with the usage of optimistic and other eager or lazy proof systems, which we intend to support in the future. By default, will raise NotImplementedError. Override in subclass as needed.

Source code in src/infernet_ml/workflows/inference/base_inference_workflow.py 
153
154
155
156
157
158
159
160
161
162
163
164
def do_generate_proof(self) -> Any:
    """
    Generates proof, which may vary based on proving system. We currently
    only support EZKL based proving, which does not require proof
    generation to be defined as part of the inference workflow if
    the inference from the circuit directly used for proving. Indeed,
    that is the case for the classic infernet_ml proof service. However, this
    may change with the usage of optimistic and other eager or lazy
    proof systems, which we intend to support in the future. By default,
    will raise NotImplementedError. Override in subclass as needed.
    """
    raise NotImplementedError

do_postprocessing(input_data, output_data)

Implement any postprocessing of the model output. By default, this method returns the output data as is.

Parameters:

Name Type Description Default
input_data Any

raw input from user

 required
output_data Any

model output

 required

Returns:

Name Type Description
Any Any

postprocessed output
Source code in src/infernet_ml/workflows/inference/base_inference_workflow.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
def do_postprocessing(self, input_data: Any, output_data: Any) -> Any:
    """
    Implement any postprocessing of the model output. By default, this method
    returns the output data as is.

    Args:
        input_data (Any): raw input from user
        output_data (Any): model output

    Returns:
        Any: postprocessed output
    """
    return output_data

do_preprocessing(input_data)

Implement any preprocessing of the raw user input. For example, you may need to apply feature engineering on the input before it is suitable for model inference. By default, this method returns the input data as is.

Parameters:

Name Type Description Default
input_data Any

raw input from user

 required

Returns:

Name Type Description
Any Any

preprocessed input
Source code in src/infernet_ml/workflows/inference/base_inference_workflow.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def do_preprocessing(self, input_data: Any) -> Any:
    """
    Implement any preprocessing of the raw user input. For example, you may need to
    apply feature engineering on the input before it is suitable for model inference.
    By default, this method returns the input data as is.

    Args:
        input_data (Any): raw input from user

    Returns:
        Any: preprocessed input
    """
    return input_data

do_run_model(preprocessed_data) abstractmethod

run model here. preprocessed_data type is left generic since depending on model type Args: preprocessed_data (typing.Any): preprocessed input into model

Returns:

Type Description
Any

typing.Any: result of running model
Source code in src/infernet_ml/workflows/inference/base_inference_workflow.py 
86
87
88
89
90
91
92
93
94
95
96
@abc.abstractmethod
def do_run_model(self, preprocessed_data: Any) -> Any:
    """run model here. preprocessed_data type is
        left generic since depending on model type
    Args:
        preprocessed_data (typing.Any): preprocessed input into model

    Returns:
        typing.Any: result of running model
    """
    pass

do_setup() abstractmethod

set up your workflow here. For LLMs, this may be parameters like top_k, temperature for classical LLM, this may be model hyperparams.

Returns: Any

Source code in src/infernet_ml/workflows/inference/base_inference_workflow.py 
40
41
42
43
44
45
46
47
@abc.abstractmethod
def do_setup(self) -> Any:
    """set up your workflow here.
    For LLMs, this may be parameters like top_k, temperature
    for classical LLM, this may be model hyperparams.

    Returns: Any
    """

do_stream(preprocessed_input) abstractmethod

Implement any streaming logic here. For example, you may want to stream data from OpenAI or any LLM. This method should return the data to be streamed.

Returns:

Type Description
Iterator[Any]

typing.Any: data to be streamed
Source code in src/infernet_ml/workflows/inference/base_inference_workflow.py 
112
113
114
115
116
117
118
119
120
121
@abc.abstractmethod
def do_stream(self, preprocessed_input: Any) -> Iterator[Any]:
    """
    Implement any streaming logic here. For example, you may
    want to stream data from OpenAI or any LLM. This method
    should return the data to be streamed.

    Returns:
        typing.Any: data to be streamed
    """

generate_proof()

Generates proof. checks that setup performed before hand.

Source code in src/infernet_ml/workflows/inference/base_inference_workflow.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
@final
def generate_proof(self) -> None:
    """
    Generates proof. checks that setup performed before hand.
    """
    if self.__inference_count <= self.__proof_count:
        logging.warning(
            "generated %s inferences only but "
            + "already generated %s. Possibly duplicate proof.",
            self.__inference_count,
            self.__proof_count,
        )

    self.do_generate_proof()
    self.__proof_count += 1

inference(input_data, log_preprocessed_data=True)

performs inference. Checks that model is set up before performing inference. Subclasses should implement do_inference.

Parameters:

Name Type Description Default
input_data Any

input from user

 required
log_preprocessed_data bool

If True, logs the

 True

Raises:

Type Description
ValueError

if setup not called beforehand

Returns:

Name Type Description
Any Any

result of inference
Source code in src/infernet_ml/workflows/inference/base_inference_workflow.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def inference(self, input_data: Any, log_preprocessed_data: bool = True) -> Any:
    """performs inference. Checks that model is set up before
    performing inference.
    Subclasses should implement do_inference.

    Args:
        input_data (typing.Any): input from user
        log_preprocessed_data (bool, optional): If True, logs the
        preprocessed input data before running the inference. Defaults to True.

    Raises:
        ValueError: if setup not called beforehand

    Returns:
        Any: result of inference
    """
    if not self.is_setup:
        raise ValueError("setup not called before inference")

    logging.info("preprocessing input_data %s", input_data)
    preprocessed_data = self.do_preprocessing(input_data)
    if log_preprocessed_data:
        logging.info("querying model with %s", preprocessed_data)
    model_output = self.do_run_model(preprocessed_data)

    logging.info("postprocessing model_output %s", model_output)
    self.__inference_count += 1
    return self.do_postprocessing(input_data, model_output)

setup()

calls setup and keeps track of whether or not setup was called

Source code in src/infernet_ml/workflows/inference/base_inference_workflow.py 
33
34
35
36
37
38
def setup(self) -> Any:
    """
    calls setup and keeps track of whether or not setup was called
    """
    self.is_setup = True
    return self.do_setup()

stream(input_data)

Stream data for inference. Subclasses should implement do_stream.

Source code in src/infernet_ml/workflows/inference/base_inference_workflow.py 
49
50
51
52
53
54
55
def stream(self, input_data: Any) -> Iterator[Any]:
    """
    Stream data for inference. Subclasses should implement do_stream.
    """
    logging.info("preprocessing input_data %s", input_data)
    preprocessed_data = self.do_preprocessing(input_data)
    yield from self.do_stream(preprocessed_data)