mstar.api_server.request_types

mstar.api_server.request_types#

Classes

APIServerMessage(message_type[, body])

Envelope for messages received by the API server.

PreprocessInput(request_id, text, ...)

RequestComplete(request_id, final_outputs)

Signals that a request has finished processing.

ResultChunk(request_id, modality, data[, ...])

One chunk of generated output for a request.

ResultTensors(request_id, modality, ...)

class mstar.api_server.request_types.APIServerMessage(message_type, body=None)[source]#

Bases: object

Envelope for messages received by the API server.

Parameters:
body: ResultTensors | RequestComplete | None = None#
message_type: str#
class mstar.api_server.request_types.PreprocessInput(request_id: str, text: str | None, file_paths: dict[str, list[str]] | None, input_modalities: list[str], output_modalities: list[str], model_kwargs: dict)[source]#

Bases: object

Parameters:
file_paths: dict[str, list[str]] | None#
input_modalities: list[str]#
model_kwargs: dict#
output_modalities: list[str]#
request_id: str#
text: str | None#
class mstar.api_server.request_types.RequestComplete(request_id, final_outputs)[source]#

Bases: object

Signals that a request has finished processing.

Parameters:
final_outputs: dict[str, NestedLoopIndices]#
request_id: str#
class mstar.api_server.request_types.ResultChunk(request_id, modality, data, metadata=<factory>)[source]#

Bases: object

One chunk of generated output for a request.

Parameters:
data: bytes#
metadata: dict#
modality: str#
request_id: str#
class mstar.api_server.request_types.ResultTensors(request_id: str, modality: str, graph_edge: mstar.graph.base.GraphEdge, loop_indices: mstar.graph.loop_indices.NestedLoopIndices, metadata: dict = <factory>)[source]#

Bases: object

Parameters:
graph_edge: GraphEdge#
loop_indices: NestedLoopIndices#
metadata: dict#
modality: str#
request_id: str#