Skip to main content
Version: v1.0

aixplain.v1.modules.model

Model module for aiXplain SDK.

This module provides the Model class and related functionality for working with AI models in the aiXplain platform, including model execution, parameter management, and status tracking.

Copyright 2022 The aiXplain SDK authors

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Author: Duraikrishna Selvaraju, Thiago Castro Ferreira, Shreyas Sharma and Lucas Pavanelli Date: September 1st 2022

Model Objects​

class Model(Asset)

[view_source]

A ready-to-use AI model that can be executed synchronously or asynchronously.

This class represents a deployable AI model in the aiXplain platform. It provides functionality for model execution, parameter management, and status tracking. Models can be run with both synchronous and asynchronous APIs, and some models support streaming responses.

Attributes:

  • id Text - ID of the model.
  • name Text - Name of the model.
  • description Text - Detailed description of the model's functionality.
  • api_key Text - Authentication key for API access.
  • url Text - Endpoint URL for model execution.
  • supplier Union[Dict, Text, Supplier, int] - Provider/creator of the model.
  • version Text - Version identifier of the model.
  • function Function - The AI function this model performs.
  • backend_url str - Base URL for the backend API.
  • cost Dict - Pricing information for model usage.
  • input_params ModelParameters - Parameters accepted by the model.
  • output_params Dict - Description of model outputs.
  • model_params ModelParameters - Configuration parameters for model behavior.
  • supports_streaming bool - Whether the model supports streaming responses.
  • function_type FunctionType - Category of function (AI, UTILITY, etc.).
  • is_subscribed bool - Whether the user has an active subscription.
  • created_at datetime - When the model was created.
  • status AssetStatus - Current status of the model.
  • additional_info dict - Additional model metadata.

__init__​

def __init__(id: Text,
             name: Text = "",
             description: Text = "",
             api_key: Text = config.TEAM_API_KEY,
             supplier: Union[Dict, Text, Supplier, int] = "aiXplain",
             version: Optional[Text] = None,
             function: Optional[Function] = None,
             is_subscribed: bool = False,
             cost: Optional[Dict] = None,
             created_at: Optional[datetime] = None,
             input_params: Optional[Dict] = None,
             output_params: Optional[Dict] = None,
             model_params: Optional[Dict] = None,
             supports_streaming: bool = False,
             status: Optional[AssetStatus] = AssetStatus.ONBOARDED,
             function_type: Optional[FunctionType] = FunctionType.AI,
             **additional_info) -> None

[view_source]

Initialize a new Model instance.

Arguments:

  • id Text - ID of the Model.
  • name Text, optional - Name of the Model. Defaults to "".
  • description Text, optional - Description of the Model. Defaults to "".
  • api_key Text, optional - Authentication key for API access. Defaults to config.TEAM_API_KEY.
  • supplier Union[Dict, Text, Supplier, int], optional - Provider/creator of the model. Defaults to "aiXplain".
  • version Text, optional - Version identifier of the model. Defaults to None.
  • function Function, optional - The AI function this model performs. Defaults to None.
  • is_subscribed bool, optional - Whether the user has an active subscription. Defaults to False.
  • cost Dict, optional - Pricing information for model usage. Defaults to None.
  • created_at Optional[datetime], optional - When the model was created. Defaults to None.
  • input_params Dict, optional - Parameters accepted by the model. Defaults to None.
  • output_params Dict, optional - Description of model outputs. Defaults to None.
  • model_params Dict, optional - Configuration parameters for model behavior. Defaults to None.
  • supports_streaming bool, optional - Whether the model supports streaming responses. Defaults to False.
  • status AssetStatus, optional - Current status of the model. Defaults to AssetStatus.ONBOARDED.
  • function_type FunctionType, optional - Category of function. Defaults to FunctionType.AI.
  • **additional_info - Additional model metadata.

to_dict​

def to_dict() -> Dict

[view_source]

Convert the model instance to a dictionary representation.

Returns:

  • Dict - A dictionary containing the model's configuration with keys:
    • id: Unique identifier
    • name: Model name
    • description: Model description
    • supplier: Model provider
    • additional_info: Extra metadata (excluding None/empty values)
    • input_params: Input parameter configuration
    • output_params: Output parameter configuration
    • model_params: Model behavior parameters
    • function: AI function type
    • status: Current model status

get_parameters​

def get_parameters() -> Optional[ModelParameters]

[view_source]

Get the model's configuration parameters.

Returns:

  • Optional[ModelParameters] - The model's parameter configuration if set, None otherwise.

__repr__​

def __repr__() -> str

[view_source]

Return a string representation of the model.

Returns:

  • str - A string in the format "Model: <name> by <supplier> (id=<id>)".

sync_poll​

def sync_poll(poll_url: Text,
              name: Text = "model_process",
              wait_time: float = 0.5,
              timeout: float = 300) -> ModelResponse

[view_source]

Poll the platform until an asynchronous operation completes or times out.

This method repeatedly checks the status of an asynchronous operation, implementing exponential backoff for the polling interval.

Arguments:

  • poll_url Text - URL to poll for operation status.
  • name Text, optional - Identifier for the operation for logging. Defaults to "model_process".
  • wait_time float, optional - Initial wait time in seconds between polls. Will increase exponentially up to 60 seconds. Defaults to 0.5.
  • timeout float, optional - Maximum total time to poll in seconds. Defaults to 300.

Returns:

  • ModelResponse - The final response from the operation. If polling times out or fails, returns a failed response with appropriate error message.

Notes:

The minimum wait time between polls is 0.2 seconds. The wait time increases by 10% after each poll up to a maximum of 60 seconds.

poll​

def poll(poll_url: Text, name: Text = "model_process") -> ModelResponse

[view_source]

Make a single poll request to check operation status.

Arguments:

  • poll_url Text - URL to poll for operation status.
  • name Text, optional - Identifier for the operation for logging. Defaults to "model_process".

Returns:

  • ModelResponse - The current status of the operation. Contains completion status, any results or errors, and usage statistics.

Notes:

This is a low-level method used by sync_poll. Most users should use sync_poll instead for complete operation handling.

run_stream​

def run_stream(data: Union[Text, Dict],
               parameters: Optional[Dict] = None) -> ModelResponseStreamer

[view_source]

Execute the model with streaming response.

Arguments:

  • data Union[Text, Dict] - The input data for the model.
  • parameters Optional[Dict], optional - Additional parameters for model execution. Defaults to None.

Returns:

  • ModelResponseStreamer - A streamer object that yields response chunks.

Raises:

  • AssertionError - If the model doesn't support streaming.

run​

def run(data: Union[Text, Dict],
        name: Text = "model_process",
        timeout: float = 300,
        parameters: Optional[Dict] = None,
        wait_time: float = 0.5,
        stream: bool = False) -> Union[ModelResponse, ModelResponseStreamer]

[view_source]

Execute the model and wait for results.

This method handles both synchronous and streaming execution modes. For asynchronous operations, it polls until completion or timeout.

Arguments:

  • data Union[Text, Dict] - The input data for the model.
  • name Text, optional - Identifier for the operation for logging. Defaults to "model_process".
  • timeout float, optional - Maximum time to wait for completion in seconds. Defaults to 300.
  • parameters Dict, optional - Additional parameters for model execution. Defaults to None.
  • wait_time float, optional - Initial wait time between polls in seconds. Defaults to 0.5.
  • stream bool, optional - Whether to use streaming mode. Requires model support. Defaults to False.

Returns:

Union[ModelResponse, ModelResponseStreamer]: The model's response. For streaming mode, returns a streamer object. For regular mode, returns a response object with results or error information.

Notes:

If the model execution becomes asynchronous, this method will poll for completion using sync_poll with the specified timeout and wait_time.

run_async​

def run_async(data: Union[Text, Dict],
              name: Text = "model_process",
              parameters: Optional[Dict] = None) -> ModelResponse

[view_source]

Start asynchronous model execution.

This method initiates model execution but doesn't wait for completion. Use sync_poll to check the operation status later.

Arguments:

  • data Union[Text, Dict] - The input data for the model.
  • name Text, optional - Identifier for the operation for logging. Defaults to "model_process".
  • parameters Dict, optional - Additional parameters for model execution. Defaults to None.

Returns:

  • ModelResponse - Initial response containing:
    • status: Current operation status
    • url: URL for polling operation status
    • error_message: Any immediate errors
    • other response metadata

check_finetune_status​

def check_finetune_status(after_epoch: Optional[int] = None)

[view_source]

Check the status of the FineTune model.

Arguments:

  • after_epoch Optional[int], optional - status after a given epoch. Defaults to None.

Raises:

  • Exception - If the 'TEAM_API_KEY' is not provided.

Returns:

  • FinetuneStatus - The status of the FineTune model.

delete​

def delete() -> None

[view_source]

Delete this model from the aiXplain platform.

This method attempts to delete the model from the platform. It will fail if the user doesn't have appropriate permissions.

Raises:

  • Exception - If deletion fails or if the user doesn't have permission.

add_additional_info_for_benchmark​

def add_additional_info_for_benchmark(display_name: str,
                                      configuration: Dict) -> None

[view_source]

Add benchmark-specific information to the model.

This method updates the model's additional_info with benchmark-related metadata.

Arguments:

  • display_name str - Name for display in benchmarks.
  • configuration Dict - Model configuration settings for benchmarking.

from_dict​

@classmethod
def from_dict(cls, data: Dict) -> "Model"

[view_source]

Create a Model instance from a dictionary representation.

Arguments:

  • data Dict - Dictionary containing model configuration with keys:
    • id: Model identifier
    • name: Model name
    • description: Model description
    • api_key: API key for authentication
    • supplier: Model provider information
    • version: Model version
    • function: AI function type
    • is_subscribed: Subscription status
    • cost: Pricing information
    • created_at: Creation timestamp (ISO format)
    • input_params: Input parameter configuration
    • output_params: Output parameter configuration
    • model_params: Model behavior parameters
    • additional_info: Extra metadata

Returns:

  • Model - A new Model instance populated with the dictionary data.