Python SDK reference

Python package providing access to Robust Intelligence.

The RIME SDK provides a programmatic interface to a Robust Intelligence instance, allowing you to create projects, start stress tests, query the backend for test run results, and more from within your Python code. To begin, initialize a client, which acts as the main entry point to SDK functions.

A Client object provides an interface to RIME’s services for creating Projects, starting Stress Test Jobs, and querying the backend for current Stress Test Jobs.

To initialize the Client, provide the address of your RIME instance.

Parameters:

domain – str The base domain/address of the RIME service.
api_key – str The API key used to authenticate to RIME services.
channel_timeout – Optional[float] The amount of time in seconds to wait for responses from the cluster.
disable_tls – Optional[bool] A Boolean that enables TLS when set to FALSE. By default, this value is set to FALSE.
ssl_ca_cert – Optional[Union[Path, str]] Specifies the path to the certificate file used to verify the peer.
cert_file – Optional[Union[Path, str]] Path to the Client certificate file.
key_file – Optional[Union[Path, str]] Path to the Client key file.
assert_hostname – Optional[bool] Enable/disable SSL hostname verification.

Raises:

ValueError – This error is generated when a connection to the RIME cluster cannot be established before the interval specified in timeout expires.

Example

rime_client = Client("my_vpc.rime.com", "api-key")

create_project(name: str, description: str, model_task: str, use_case: str | None = None, ethical_consideration: str | None = None, profiling_config: dict | None = None, general_access_role: str | None = 'ACTOR_ROLE_NONE') → Project

Create a new RIME Project.

Projects enable you to organize Stress Test runs. A natural way to organize Stress Test runs is to create a Project for each specific ML task, such as predicting whether a transaction is fraudulent.

Parameters:

name – str Name of the new Project.
description – str Description of the new Project.
model_task – str Machine Learning task associated with the Project. Must be one of “MODEL_TASK_REGRESSION”, “MODEL_TASK_BINARY_CLASSIFICATION”, “MODEL_TASK_MULTICLASS_CLASSIFICATION”, “MODEL_TASK_NAMED_ENTITY_RECOGNITION”, “MODEL_TASK_RANKING”, “MODEL_TASK_OBJECT_DETECTION”, “MODEL_TASK_NATURAL_LANGUAGE_INFERENCE”, “MODEL_TASK_FILL_MASK”.
use_case – Optional[str] Description of the use case of the Project.
ethical_consideration – Optional[str] Description of ethical considerations for this Project.
profiling_config – Optional[dict] Configuration for the data and model profiling across all test runs.
general_access_role – Optional[str], Project roles assigned to the workspace members. Allowed Values: ACTOR_ROLE_USER, ACTOR_ROLE_VIEWER, ACTOR_ROLE_NONE.

Return type:

Project

Raises:

ValueError – This error is generated when the request to the Project service fails.

Example

project = rime_client.create_project(
    name="foo", description="bar", model_task="Binary Classification"
)

get_project(project_id: str) → Project

Get Project by ID.

Parameters:: project_id – str ID of the Project to return.
Return type:: Project

Example

project = rime_client.get_project("123-456-789")

delete_project(project_id: str, force: bool | None = False) → None

Delete a Project by ID.

Parameters:

project_id – str ID of the Project to delete.
force – Optional[bool] = False When set to True, the Project will be deleted immediately. By default, a confirmation is required.

Example

project = rime_client.delete_project("123-456-789", True)

create_managed_image(name: str, requirements: List[ManagedImagePipRequirement], package_requirements: List[ManagedImagePackageRequirement] | None = None, python_version: str | None = None) → ImageBuilder

Create a new managed Docker image with the desired custom requirements to run RIME on.

These managed Docker images are managed by the RIME cluster and automatically upgrade when the installed version of RIME upgrades. Note: Images take a few minutes to be built.

This method returns an object that can be used to track the progress of the image building job. The new custom image is only available for use in a stress test once it has status READY.

Parameters:

name – str The name of the new Managed Image. This name serves as the unique identifier of the Managed Image. The call fails when an image with the specified name already exists.
requirements – List[ManagedImagePipRequirement] List of additional pip requirements to be installed on the managed image. A ManagedImagePipRequirement can be created with the helper method Client.pip_requirement. The first argument is the name of the library (e.g. tensorflow or xgboost) and the second argument is a valid pip version specifier <https://www.python.org/dev/peps/pep-0440/#version-specifiers> (e.g. >=0.1.2 or ==1.0.2) or an exact version<https://peps.python.org/pep-0440/>, such as 1.1.2, for the library.
package_requirements – Optional[List[ManagedImagePackageRequirement]] [BETA] An optional List of additional operating system package requirements to install on the Managed Image. Currently only Rocky Linux package requirements are supported. Create a ManagedImagePackageRequirement parameter with the Client.os_requirement helper method. The first argument is the name of the package (e.g. texlive or vim) and the second optional argument is a valid yum version specifier (e.g. 0.1.2) for the package.
python_version – Optional[str] An optional version string specifying only the major and minor version for the python interpreter used. The string should be of the format X.Y and be present in the set of supported versions.

Returns:

A ImageBuilder object that provides an interface for monitoring the job in the backend.

Return type:

ImageBuilder

Raises:

ValueError – This error is generated when the request to the ImageRegistry service fails.

Example

reqs = [
     # Fix the version of `xgboost` to `1.0.2`.
     rime_client.pip_requirement("xgboost", "==1.0.2"),
     # We do not care about the installed version of `tensorflow`.
     rime_client.pip_requirement("tensorflow")
 ]

# Start a new image building job
builder_job = rime_client.create_managed_image("xgboost102_tf", reqs)

# Wait until the job has finished and print out status information.
# Once this prints out the `READY` status, your image is available for
# use in Stress Tests.
builder_job.get_status(verbose=True, wait_until_finish=True)

has_managed_image(name: str, check_status: bool = False) → bool

Check whether a Managed Image with the specified name exists.

Parameters:

name – str The unique name of the Managed Image to check. The call returns False when no image with the specified name exists.
check_status – bool Flag that determines whether to check the image status. When this flag is set to True, the call returns True if and only if the image with the specified name exists AND the image is ready to be used.

Returns:

Specifies whether a Managed Image with this name exists.

Return type:

bool

Example

if rime_client.has_managed_image("xgboost102_tensorflow"):
     print("Image exists.")

get_managed_image(name: str) → Dict

Get Managed Image by name.

Parameters:: name – str The unique name of the new Managed Image. The call raises an error when no image exists with this name.
Returns:: A dictionary with information about the Managed Image.
Return type:: Dict

Example

image = rime_client.get_managed_image("xgboost102_tensorflow")

delete_managed_image(name: str) → None

Delete a managed Docker image.

Parameters:: name – str The unique name of the Managed Image.

Example

image = rime_client.delete_managed_image("xgboost102_tensorflow")

static pip_requirement(name: str, version_specifier: str | None = None) → ManagedImagePipRequirement: Construct a PipRequirement object for use in create_managed_image().

static os_requirement(name: str, version_specifier: str | None = None) → ManagedImagePackageRequirement: Construct a PackageRequirement object for create_managed_image().

static pip_library_filter(name: str, fixed_version: str | None = None) → ListImagesRequestPipLibraryFilter: Construct a PipLibraryFilter object for use in list_managed_images().

list_managed_images(pip_library_filters: List[ListImagesRequestPipLibraryFilter] | None = None) → Iterator[Dict]

List all managed Docker images.

Enables searching for images with specific pip libraries installed so that users can reuse Managed Images for Stress Tests.

Parameters:: pip_library_filters – Optional[List[ListImagesRequestPipLibraryFilter]] Optional list of pip libraries to filter by. Construct each ListImagesRequest.PipLibraryFilter object with the pip_library_filter convenience method.
Returns:: An iterator of dictionaries, each dictionary represents a single Managed Image.
Return type:: Iterator[Dict]
Raises:: ValueError – This error is generated when the request to the ImageRegistry service fails or the list of pip library filters is improperly specified.

Example

# Filter for an image with catboost1.0.3 and tensorflow installed.
filters = [
    rime_client.pip_library_filter("catboost", "1.0.3"),
    rime_client.pip_library_filter("tensorflow"),
]

# Query for the images.
images = rime_client.list_managed_images(
    pip_library_filters=filters)

# To get the names of the returned images.
[image["name"] for image in images]

list_agents() → Iterator[Dict]

List all Agents available to the user.

Returns:: An iterator of dictionaries, each dictionary represents a single Agent.
Return type:: Iterator[Dict]
Raises:: ValueError – This error is generated when the request to the AgentManager service fails.

Example

# Query for the images.
agents = rime_client.list_agents()

# To get the names of the returned Agents.
[agent["name"] for agent in agents]

list_projects() → Iterator[Project]

List all Projects.

Returns:: An iterator of Projects.
Return type:: Iterator[Project]
Raises:: ValueError – This error is generated when the request to the Project service fails.

Example

# Query for projects.
projects = rime_client.list_projects()

start_stress_test(test_run_config: dict, project_id: str, **exp_fields: Dict[str, object]) → Job

Start a Stress Testing run.

Parameters:

test_run_config – dict Configuration for the test to be run, which specifies unique ids to locate the model and datasets to be used for the test.
project_id – str Identifier for the Project where the resulting test run will be stored. When not specified, stores the results in the default Project.
exp_fields – Dict[str, object] [BETA] Fields for experimental features.

Returns:

A Job that provides information about the model Stress Test job.

Return type:

Job

Raises:

ValueError – This error is generated when the request to the ModelTesting service fails.

Example

This example assumes that reference and evaluation datasets are registered with identifiers “foo” and “bar” respectively, and that a model with the unique identifier model_uuid is registered.

config = {
    "data_info": {"ref_dataset_id": "foo", "eval_dataset_id": "bar"},
    "model_id": model_uuid,
    "run_name": "My Stress Test Run",
    "model_task": "Binary Classification",
}

Run the job using the specified configuration and the default Docker image in the RIME backend. Store the results in the RIME Project associated with this object.

job = rime_client.start_stress_test_job(
   test_run_config=config, project_id="123-456-789"
)

get_test_run(test_run_id: str) → TestRun

Get a TestRun object with the specified test_run_id.

Checks to see if the test_run_id exists, then returns TestRun object.

Parameters:: test_run_id – str ID of the test run to query for
Returns:: A TestRun object corresponding to the test_run_id
Return type:: TestRun

get_firewall_for_project(project_id: str) → Firewall

Get the active fw for a Project if it exists.

Query the backend for an active Firewall in a specified Project which can be used to perform Firewall operations. If there is no active Firewall for the Project, this call will error.

Parameters:: project_id – ID of the Project which contains a Firewall.
Returns:: A Firewall object.
Return type:: Firewall
Raises:: ValueError – This error is generated when the Firewall does not exist or when the request to the Project service fails.

Example

# Get FW in foo-project if it exists.
firewall = rime_client.get_firewall_for_project("foo-project")

upload_file(file_path: Path | str, upload_path: str | None = None) → str

Upload a file to make it accessible to the RIME cluster.

The uploaded file is stored in the RIME cluster in a blob store using its file name.

Parameters:

file_path – Union[Path, str] Path to the file to be uploaded to RIME’s blob store.
upload_path – Optional[str] = None Name of the directory in the blob store file system. If omitted, a unique random string will be the directory.

Returns:

A reference to the uploaded file’s location in the blob store. This reference can be used to refer to that object when writing RIME configs. Please store this reference for future access to the file.

Return type:

str

Raises:

FileNotFoundError – When the path file_path does not exist.
IOError – When file_path is not a file.
ValueError – When the specified upload_path is an empty string or there was an error in obtaining a blobstore location from the RIME backend or in uploading file_path to RIME’s blob store. When the file upload fails, the incomplete file is NOT automatically deleted.

Example

# Upload the file at location data_path.
client.upload_file(data_path)

upload_local_image_dataset_file(file_path: Path | str, image_features: List[str], upload_path: str | None = None) → Tuple[List[Dict], str]

Upload an image dataset file where image files are stored locally.

The image dataset file is expected to be a list of JSON dictionaries, with an image_features that reference an image (either an absolute path or a relative path to an image file stored locally). Every image within the file is also uploaded to blob store, and the final file is also uploaded. If your image paths already reference an external blob storage, then use upload_file instead to upload the dataset file.

Parameters:

file_path – Union[Path, str] Path to the file to be uploaded to RIME’s blob store.
image_features – List[str] Keys to image file paths.
upload_path – Optional[str] Name of the directory in the blob store file system. If omitted, a unique random string will be the directory.

Returns:

The list of dicts contains the updated dataset file with image paths replaced by s3 paths. The string contains a reference to the uploaded file’s location in the blob store. This reference can be used to refer to that object when writing RIME configs. Please store this reference for future access to the file.

Return type:

Tuple[List[Dict], str]

Raises:

FileNotFoundError – When the path file_path does not exist.
IOError – When file_path is not a file.
ValueError – When there was an error in obtaining a blobstore location from the RIME backend or in uploading file_path to RIME’s blob store. In the scenario the file fails to upload, the incomplete file will NOT automatically be deleted.

upload_data_frame(data_frame: DataFrame, name: str | None = None, upload_path: str | None = None) → str

Upload a pandas DataFrame to make it accessible to the RIME cluster.

The uploaded file is stored in the RIME cluster in a blob store using its file name.

Parameters:

data_frame – pd.DataFrame Path to the file to be uploaded to RIME’s blob store.
name – Optional[str] = None Name of the file in the blob store file system. If omitted, a unique random string will be assigned as the file name.
upload_path – Optional[str] = None Name of the directory in the blob store file system. If omitted, a unique random string will be the directory.

Returns:

A reference to the uploaded file’s location in the blob store. This reference can be used to refer to that object when writing RIME configs. Please store this reference for future access to the file.

Return type:

str

Raises:

ValueError – When the specified upload_path is an empty string or there was an error in obtaining a blobstore location from the RIME backend or in uploading file_path to RIME’s blob store. When the file upload fails, the incomplete file is NOT automatically deleted.

Example

# Upload pandas data frame.
client.upload_data_frame(df)

upload_directory(dir_path: Path | str, upload_hidden: bool = False, upload_path: str | None = None) → str

Upload a model directory to make it accessible to the RIME cluster.

The uploaded directory is stored in the RIME cluster in a blob store. All files contained within dir_path and its subdirectories are uploaded according to their relative paths within dir_path. When upload_hidden is set to False, all hidden files and subdirectories that begin with a ‘.’ are not uploaded.

Parameters:

dir_path – Union[Path, str] Path to the directory to be uploaded to RIME’s blob store.
upload_hidden – bool = False Whether to upload hidden files or subdirectories (i.e. those beginning with a ‘.’).
upload_path – Optional[str] = None Name of the directory in the blob store file system. If omitted, a unique random string will be the directory.

Returns:

A reference to the uploaded directory’s location in the blob store. This reference can be used to refer to that object when writing RIME configs. Please store this reference for future access to the directory.

Return type:

str

Raises:

FileNotFoundError – When the directory dir_path does not exist.
IOError – When dir_path is not a directory or contains no files.
ValueError – When the specified upload_path is an empty string or there was an error in obtaining a blobstore location from the RIME backend or in uploading dir_path to RIME’s blob store. In the scenario the directory fails to upload, files will NOT automatically be deleted.

list_uploaded_file_urls() → Iterator[str]

Return an iterator of file paths that have been uploaded using client.upload_file.

Returns:: An iterator of file path strings.
Return type:: Iterator[str]

Example

# List all file URLs
urls = rime_client.list_uploaded_file_urls()

delete_uploaded_file_url(upload_url: str) → None

Delete the file at the specified upload url in the RIME blob store.

Parameters:: upload_url – str Url to the file to be deleted in the RIME blob store.
Returns:: None

Example

# Delete a file URL returned by list_uploaded_file_urls
urls = rime_client.list_uploaded_file_urls()
first_url = next(urls)
rime_client.delete_uploaded_file_url(first_url)

get_job(job_id: str) → BaseJob

Get job by ID.

Parameters:: job_id – ID of the Job to return.
Returns:: A Job object.
Return type:: Job
Raises:: ValueError – This error is generated when no Job with the specified ID exists.

Example

# Get Job with ID if it exists.
job = rime_client.get_job("123-456-789")

start_file_scan(model_file_info: dict, custom_image: RuntimeinfoCustomImage | None = None, rime_managed_image: str | None = None, ram_request_megabytes: int | None = None, cpu_request_millicores: int | None = None, agent_id: str | None = None) → FileScanJob

Start a File Scan job.

Parameters:

model_file_info – dict Configuration for the ML file scan, which specifies the model file or repository.
custom_image – Optional[RuntimeinfoCustomImage] Specification of a customized container image to use running the model test. The image must have all dependencies required by your model. The image must specify a name for the image and optionally a pull secret (of type RuntimeinfoCustomImagePullSecret) with the name of the kubernetes pull secret used to access the given image.
rime_managed_image – Optional[str] Name of a Managed Image to use when running the model test. The image must have all dependencies required by your model. To create new Managed Images with your desired dependencies, use the client’s create_managed_image() method.
ram_request_megabytes – Optional[int] Megabytes of RAM requested for the Stress Test Job. The limit is equal to megabytes requested.
cpu_request_millicores – Optional[int] Millicores of CPU requested for the Stress Test Job. The limit is equal to millicores requested.
agent_id – Optional[str] ID of the Agent that runs the File Scan job. When unspecified, the workspace’s default Agent is used.

Returns:

An ML File Scan Job object.

Return type:

FileScanJob

Raises:

ValueError – This error is generated when the request to the FileScanning service fails.

Example

This example shows how to scan a Huggingface model file.

 model_file_info = {
     "scan_type": "huggingface",
     "scan_path": "https://huggingface.co/transformers/v2.11.0",
 }

job = rime_client.start_file_scan(model_file_info)

list_file_scan_results() → Iterator[dict]

List all ML file scan results.

These contain the security reports for the scanned files or repositories.

Returns:: An iterator of dictionaries, each dictionary represents a single ML File Scan result.
Return type:: Iterator[dict]
Raises:: ValueError – This error is generated when the request to the FileScanning service fails.

Example

# List all ML file scan results.
results = rime_client.list_file_scan_results()

class rime_sdk.Project(api_client: ApiClient, project_id: str)

An interface to a Project object.

The Project object is used for editing, updating, and deleting Projects.

property info: ProjectInfo: Return description, use case and ethical consideration of the Project.

get_link() → str

Return the web app URL to the Project.

This link directs to your organization’s deployment of RIME. You can view more detailed information in the web app, including information on your Test Runs, comparisons of those results, and monitored models.

property name: str: Return the name of this Project.

property description: str: Return the description of this Project.

list_stress_testing_jobs(status_filters: List[str] | None = None) → Iterator[Job]

Get list of Stress Testing Jobs for the Project filtered by status.

Note that this only returns jobs from the last two days, because the time-to-live of job objects in the cluster is set at two days.

Parameters:: status_filters – Optional[List[str]] = None Filter for selecting jobs by a union of statuses. The following list enumerates all acceptable values. [‘pending’, ‘running’, ‘failed’, ‘succeeded’] If omitted, jobs will not be filtered by status.
Returns:: An iterator of Job objects. These are not guaranteed to be in any sorted order.
Return type:: Iterator[Job]
Raises:: ValueError – This error is generated when the request to the JobReader service fails or when the provided status_filters array has invalid values.

Example

# Get all running and succeeded jobs for a Project.
jobs = project.list_stress_testing_jobs(
    status_filters=['pending', 'succeeded'],
)
# To get the names of all jobs.
[job["name"] for job in jobs]

list_test_runs() → Iterator[TestRun]

List the Stress Test Runs associated with this Project.

Returns:: An iterator of TestRun objects.
Return type:: Iterator[TestRun]
Raises:: ValueError – This error is generated when the request to the ResultsReader Service fails.

Example

# List all test runs in the Project.
test_runs = project.list_test_runs()
# Get the IDs of all the test runs.
[test_run.test_run_id for test_run in test_runs]

create_firewall(model_id: str, ref_data_id: str, bin_size: timedelta, scheduled_ct_eval_data_integration_id: str | None = None, scheduled_ct_eval_data_info: dict | None = None, scheduled_ct_eval_pred_integration_id: str | None = None, scheduled_ct_eval_pred_info: dict | None = None, scheduled_ct_rolling_window_size: timedelta | None = None) → Firewall

Create a Firewall in the current Project.

Parameters:

model_id – str The model ID that this Firewall is testing. Model IDs are created by the Registry.
ref_data_id – str The ID of the reference dataset that this Firewall compares against during testing. Dataset IDs are created by the Registry.
bin_size – timedelta The length of each time bin to test over as a timedelta object. Must have a minimum value of 1 hour.
scheduled_ct_eval_data_integration_id – Optional[str] The integration ID will be used to fetch eval data.
scheduled_ct_eval_data_info – Optional[dict] The data info needed to fetch eval data.
scheduled_ct_eval_pred_integration_id – Optional[str] The integration ID will be used to fetch eval predictions.
scheduled_ct_eval_pred_info – Optional[dict] The predcition info needed to fetch eval predictions.
scheduled_ct_rolling_window_size – Optional[timedelta] The size of the rolling window to be used for Scheduled CT evaluation.

Returns:

A Firewall object that is used to monitor the model.

Return type:

Firewall

Raises:

ValueError – This error is generated when the request to the Firewall Service fails.

Example

from datetime import timedelta
# Create Firewall using previously registered model and dataset IDs.
fw = project.create_firewall(model_id, ref_data_id, timedelta(days=2))

get_firewall() → Firewall

Return the active Firewall for a Project if one exists.

Query the backend for an active Firewall in this Project which can be used to perform Firewall operations. If there is no active Firewall for the project, this call will return an error.

Returns:: A Firewall object.
Return type:: Firewall
Raises:: ValueError – This error is generated when the Firewall does not exist or when the request to the Firewall Service fails.

Example

# Get Firewall for this Project.
firewall = project.get_firewall()

has_firewall() → bool: Check whether a Project has a Firewall.

delete_firewall(force: bool | None = False) → None

Delete the Firewall for this Project if one exists.

Parameters:: force – Optional[bool] = False When set to True, the Firewall will be deleted immediately. By default, a confirmation is required.

get_notification_settings() → Dict

List all Notification settings for the Project.

Queries the backend to get a list of Notifications settings added to the Project. The Notifications are grouped by the type of Notification: each type contains a list of emails and webhooks which are added to the Notification setting.

Returns:: A Dictionary of Notification type and corresponding emails and webhooks added for that Notification type.
Return type:: Dict

Example

notification_settings = project.get_notification_settings()

add_email(email: str, notif_type_str: str) → None

Add an email to the Notification settings for the specified Notification type.

The valid Notification types are: [“Job_Action”, “Monitoring”, “Daily_Digest”]

Example

notification_settings = project.add_email("<email>", "Monitoring")

remove_email(email: str, notif_type_str: str) → None

Remove an email from the Notification settings for the specified Notification type.

The valid Notification types are: [“Job_Action”, “Monitoring”, “Daily_Digest”]

Example

notification_settings = project.remove_email("<email>", "Monitoring")

add_webhook(webhook: str, notif_type_str: str) → None

Add a webhook to the Notification settings for the specified Notification type.

The valid Notification types are: [“Job_Action”, “Monitoring”, “Daily_Digest”]

Example

notification_settings = project.add_webhook("<webhook>", "Monitoring")

remove_webhook(webhook: str, notif_type_str: str) → None

Remove a webhook from the Notification settings for the specified Notification type.

The valid Notification types are: [“Job_Action”, “Monitoring”, “Daily_Digest”]

Example

notification_settings = project.remove_webhook("<webhook>", "Monitoring")

Register and validate a new dataset in this Project.

Parameters:

name – str The chosen name of the dataset.
data_config – dict A dictionary that contains the data configuration. The data configuration must match the API specification of the data_info field in the RegisterDataset request.
integration_id – Optional[str] = None, Provide the integration ID for datasets that require an integration.
tags – Optional[List[str]] = None, An optional list of tags to associate with the dataset.
metadata – Optional[dict] = None, An optional dictionary of metadata to associate with the dataset.
ct_info – Optional[dict] = None, An optional dictionary that contains the CT info. The CT info must match the API specification of the ct_info field in the RegisterDataset request.
skip_validation – Optional[bool] = False, An optional boolean that indicates whether to skip validation. Validation ensures that the dataset is valid for Robust Intelligence’s systems.
agent_id – Optional[str] = None, Agent for running validation. If omitted the workspace’s default agent will be used.

Returns:

The ID of the newly registered dataset.

Return type:

str

Raises:

ValueError – This error is generated when the request to the Registry service fails.
DatasetValidationError – This error is generated when the dataset is invalid.

Example

dataset_id = project.register_dataset(
    name=DATASET_NAME,
    data_config={
        "connection_info": {"data_file": {"path": FILE_PATH}},
        "data_params": {"label_col": LABEL_COL},
    },
    integration_id=INTEGRATION_ID,
    agent_id=AGENT_ID,
)

Register and validate a new dataset in this Project.

Parameters:

name – str The chosen name of the dataset.
remote_path – str The path to the dataset artifact.
tags – Optional[List[str]] = None, An optional list of tags to associate with the dataset.
metadata – Optional[dict] = None, An optional dictionary of metadata to associate with the dataset.
skip_validation – Optional[bool] = False, An optional boolean that indicates whether to skip validation. Validation ensures that the dataset is valid for Robust Intelligence’s systems.
agent_id – Optional[str] = None, Agent for running validation. If omitted the workspace’s default agent will be used.

Returns:

The ID of the newly registered dataset.

Return type:

str

Raises:

ValueError – This error is generated when the request to the Registry service fails.
DatasetValidationError – This error is generated when the dataset is invalid.

Register and validate a new model in this Project.

Parameters:

name – str The chosen name of the model.
model_config – Optional[dict] = None, A dictionary that contains the model configuration. Any model configuration that is provided must match the API specification for the model_info field of the RegisterModel request.
tags – Optional[List[str]] = None, An optional list of tags to associate with the model.
metadata – Optional[dict] = None, An optional dictionary of metadata to associate with the model.
external_id – Optional[str] = None, An optional external ID that can be used to identify the model.
integration_id – Optional[str] = None, Provide the integration ID for models that require an integration for access.
skip_validation – Optional[bool] = False, An optional boolean that indicates whether to skip validation. Validation ensures that the model is valid for Robust Intelligence’s systems.
agent_id – Optional[str] = None, Agent for running validation. If omitted the workspace’s default agent will be used.

Returns:

The ID of the newly registered model.

Return type:

str

Raises:

ValueError – This error is generated when the request to the Registry service fails.
ModelValidationError – This error is generated when the model is invalid.

Example

model_id = project.register_model(
    name=MODEL_NAME,
    model_config={
        "hugging_face": {
            "model_uri": URI,
            "kwargs": {
                "tokenizer_uri": TOKENIZER_URI,
                "class_map": MAP,
                "ignore_class_names": True,
            },
        }
    },
    tags=[MODEL_TAG],
    metadata={KEY: VALUE},
    external_id=EXTERNAL_ID,
    agent_id=AGENT_ID,
)

Register and validate a new model in this Project.

Parameters:

name – str The chosen name of the model.
remote_path – str The path to the model artifact.
tags – Optional[List[str]] = None, An optional list of tags to associate with the model.
metadata – Optional[dict] = None, An optional dictionary of metadata to associate with the model.
external_id – Optional[str] = None, An optional external ID that can be used to identify the model.
integration_id – Optional[str] = None, Provide the integration ID for models that require an integration for access.
skip_validation – Optional[bool] = False, An optional boolean that indicates whether to skip validation. Validation ensures that the model is valid for Robust Intelligence’s systems.
agent_id – Optional[str] = None, Agent for running validation. If omitted the workspace’s default agent will be used.

Returns:

The ID of the newly registered model.

Return type:

str

Raises:

ValueError – This error is generated when the request to the Registry service fails.
ModelValidationError – This error is generated when the model is invalid.

Example

model_id = project.register_model_path(
    name=MODEL_NAME,
    remote_path=MODEL_PATH,
)

register_predictions(dataset_id: str, model_id: str, pred_config: dict, integration_id: str | None = None, tags: List[str] | None = None, metadata: dict | None = None, skip_validation: bool | None = False, agent_id: str | None = None) → None

Register and validate a new prediction corresponding to a model and a dataset.

Parameters:

dataset_id – str, The ID of the dataset used to generate the predictions.
model_id – str, The ID of the model used to generate the predictions.
pred_config – dict, A dictionary that contains the prediction configuration. The prediction configuration must match the API specification for the pred_info field of the RegisterPredictions request.
integration_id – Optional[str] = None, Provide the integration ID for predictions that require an integration to use.
tags – Optional[List[str]] = None, An optional list of tags to associate with the predictions.
metadata – Optional[dict] = None, An optional dictionary of metadata to associate with the predictions.
skip_validation – Optional[bool] = False, An optional boolean that indicates whether to skip validation. Validation ensures that the predictions are valid for Robust Intelligence’s systems.
agent_id – Optional[str] = None, Agent for running validation. If omitted the workspace’s default agent will be used.

Returns:

None

Raises:

ValueError – This error is generated when the request to the Registry service fails.
PredictionsValidationError – This error is generated when the predictions are invalid.

Example

project.register_predictions(
    dataset_id=DATASET_ID,
    model_id=MODEL_ID,
    pred_config={
        "connection_info": {
            "databricks": {
                # Unix timestamp equivalent to 02/08/2023
                "start_time": 1675922943,
                # Unix timestamp equivalent to 03/08/2023
                "end_time": 1678342145,
                "table_name": TABLE_NAME,
                "time_col": TIME_COL,
            },
        },
        "pred_params": {"pred_col": PREDS},
    },
    tags=[TAG],
    metadata={KEY: VALUE},
    agent_id=AGENT_ID,
)

register_predictions_from_file(dataset_id: str, model_id: str, remote_path: str, pred_params: dict | None = None, integration_id: str | None = None, tags: List[str] | None = None, metadata: dict | None = None, skip_validation: bool | None = False, agent_id: str | None = None) → None

Register and validate a new set of predictions for a model on a dataset.

Parameters:

dataset_id – str, The ID of the dataset used to generate the predictions.
model_id – str, The ID of the model used to generate the predictions.
remote_path – str, The path to the prediction artifact.
pred_params – Optional[dict] = None, A dictionary that contains the prediction parameters.
integration_id – Optional[str] = None, Provide the integration ID for predictions that require an integration to use.
tags – Optional[List[str]] = None, An optional list of tags to associate with the predictions.
metadata – Optional[dict] = None, An optional dictionary of metadata to associate with the predictions.
skip_validation – Optional[bool] = False, An optional boolean that indicates whether to skip validation. Validation ensures that the predictions are valid for Robust Intelligence’s systems.
agent_id – Optional[str] = None, Agent for running validation. If omitted the workspace’s default agent will be used.

Returns:

None

Raises:

ValueError – This error is generated when the request to the Registry service fails.
PredictionsValidationError – This error is generated when the predictions are invalid.

Example

project.register_predictions_from_file(
    dataset_id=DATASET_ID,
    model_id=MODEL_ID,
    remote_path=PREDICTIONS_PATH,
    agent_id=AGENT_ID,
)

list_datasets() → Iterator[Dict]

Return a list of datasets registered in this Project.

Returns:: Iterator of dictionaries: each dictionary represents a dataset.
Return type:: Iterator[Dict]
Raises:: ValueError – This error is generated when the request to the Registry service fails.

list_models() → Iterator[Dict]

Return a list of models.

Returns:: Iterator of dictionaries: each dictionary represents a model.
Return type:: Iterator[Dict]
Raises:: ValueError – This error is generated when the request to the Registry service fails.

get_data_collector() → DataCollector

Get Data Collector for Project.

When Project has no existing Data Collector, this method creates and returns a new Data Collector.

Returns:: Data Collector object for Project
Return type:: DataCollector
Raises:: ValueError – This error is generated when the request to the Data Collector service fails.

get_general_access_roles() → Dict[RimeActorRole, str]

Get the roles of workspace members for the project.

Returns:

Dict[RimeActorRole, str]: Returns a map of Actor Roles for workspace and their roles for the project.

list_predictions(model_id: str | None = None, dataset_id: str | None = None) → Iterator[Dict]

Return a list of predictions.

Parameters:

model_id – Optional[str] = None The ID of the model to which the prediction sets belong.
dataset_id – Optional[str] = None The ID of the dataset to which the prediction sets belong.

Returns:

Iterator of dictionaries: each dictionary represents a prediction.

Return type:

Iterator[Dict]

Raises:

ValueError – This error is generated when the request to the Registry service fails.

get_dataset(dataset_id: str | None = None, dataset_name: str | None = None) → Dict

Return a registered dataset.

Parameters:

dataset_id – Optional[str] = None, The ID of the dataset to retrieve.
dataset_name – Optional[str] = None, The name of the dataset to retrieve.

Returns:

A dictionary representing the dataset.

Return type:

Dict

Raises:

ValueError – This error is generated when the request to the Registry service fails.

has_dataset(dataset_id: str | None = None, dataset_name: str | None = None) → bool

Return a boolean on whether the dataset is present.

Parameters:

dataset_id – Optional[str] = None The ID of the dataset to check for.
dataset_name – Optional[str] = None The name of the dataset to check for.

Returns:

A boolean on whether the dataset is present.

Return type:

bool

Raises:

ValueError – This error is generated any error other than HTTPStatus.NOT_FOUND is returned from the Registry service.

get_model(model_id: str | None = None, model_name: str | None = None) → Dict

Return a registered model.

Parameters:

model_id – Optional[str] = None, The ID of the model to retrieve.
model_name – Optional[str] = None, The name of the model to retrieve.

Returns:

A dictionary representing the model.

Return type:

Dict

Raises:

ValueError – This error is generated when the request to the Registry service fails.

get_predictions(model_id: str, dataset_id: str) → Dict

Get predictions for a model and dataset.

Parameters:

model_id – str, The ID of the model used to generate the predictions.
dataset_id – str, The ID of the dataset used to generate the predictions.

Returns:

A dictionary representing the predictions.

Return type:

Dict

Raises:

ValueError – This error is generated when the request to the Registry service fails.

delete_dataset(dataset_id: str) → None

Delete a dataset.

Parameters:: dataset_id – str, The ID of the dataset to delete.
Returns:: None
Raises:: ValueError – This error is generated when the request to the Registry service fails.

delete_model(model_id: str) → None

Delete a model.

Parameters:: model_id – str, The ID of the model to delete.
Returns:: None
Raises:: ValueError – This error is generated when the request to the Registry service fails.

delete_predictions(model_id: str, dataset_id: str) → None

Delete predictions for a model and dataset.

Parameters:

model_id – str, The ID of the model used to generate the predictions.
dataset_id – str, The ID of the dataset used to generate the predictions.

Returns:

None

Raises:

ValueError – This error is generated when the request to the Registry service fails.

update_stress_test_categories(categories: List[str]) → None

Update the project’s stress test categories.

Parameters:: categories – List[str] The list of stress test categories to update.
Returns:: None
Raises:: ValueError – This error is generated when the request to the Project service fails.

update_continuous_test_categories(categories: List[str]) → None

Update the project’s continuous test categories.

Parameters:: categories – List[str] The list of continuous test categories to update.
Returns:: None
Raises:: ValueError – This error is generated when the request to the Project service fails.

update_model_profiling_config(model_profiling_config: dict) → None

Update the project’s model profiling configuration.

Parameters:: model_profiling_config – dict Model profiling configuration with which to update the project.
Returns:: None
Raises:: ValueError – This error is generated when the request to the Project service fails.

update_data_profiling_config(data_profiling_config: dict) → None

Update the project’s data profiling configuration.

Parameters:: data_profiling_config – dict Data profiling configuration with which to update the project.
Returns:: None
Raises:: ValueError – This error is generated when the request to the Project service fails.

update_test_suite_config(test_suite_config: dict) → None

Update the project’s test suite config.

Parameters:: test_suite_config – dict Test suite configuration with which to update the project.
Returns:: None
Raises:: ValueError – This error is generated when the request to the Project service fails.

class rime_sdk.Job(api_client: ApiClient, job_id: str)

This object provides an interface for monitoring a Stress Test Job in the RIME backend.

get_test_run_id() → str

Get the Test Run ID corresponding to a successful Job.

Raises:: ValueError – This error is generated when the job does not have state ‘SUCCEEDED’ or when the request to the Job Reader service fails.

get_test_run() → TestRun

Get the Test Run object corresponding to a successful Job.

Raises:: ValueError – This error is generated when the job does not have state ‘SUCCEEDED’ or if the request to the Job Reader service fails.

class rime_sdk.TestRun(api_client: ApiClient, test_run_id: str)

An interface for a RIME Test Run object.

get_link() → str

Return the web app URL for the Firewall Continuous Tests page.

This page contains results for all Test Runs. To jump to the view which shows results for this specific Test Run, click on the corresponding time bin in the UI.

get_result_df() → DataFrame

Return high level summary information for a complete stress Test Run in a single-row dataframe.

This dataframe includes information such as model metrics on the reference and evaluation datasets, overall RIME results such as severity across tests, and high level metadata such as the project ID and model task.

Place these rows together to build a table of test to build a table of test run results for comparison. This only works for stress test jobs that have succeeded.

Returns:: A pandas.DataFrame object containing the Test Run result. Use the .columns method on the returned dataframe to see what columns represent. Generally, these columns have information about the model and datasets as well as summary statistics such as the number of failing Test Cases or number of high severity Test Cases.
Return type:: pd.DataFrame

Example

test_run = client.get_test_run(some_test_run_id)
test_run_result_df = test_run.get_result_df()

get_test_cases_df(show_test_case_metrics: bool = False) → DataFrame

Return all the Test Cases for a completed stress Test Run in a dataframe.

This enables you to perform granular queries on Test Cases. For example, if you only care about subset performance tests and want to see the results on each feature, you can fetch all the Test Cases in a dataframe, then query on that dataframe by test type. This only works on stress test jobs that have succeeded.

Parameters:

show_test_case_metrics – bool = False Whether to show Test Case specific metrics. This could result in a sparse dataframe that is returned, since Test Cases return different metrics. Defaults to False.

Returns:

A pandas.DataFrame object containing the Test Case results. Here is a selected list of columns in the output:

test_run_id: ID of the parent Test Run.

features: List of features that the Test Case ran on.

test_batch_type: Type of test that was run (e.g. Subset AUC).

status: Status of the Test Case (e.g. Pass, Fail, etc.).

severity: Denotes the severity of the failure of the test.

Return type:

pd.DataFrame

Example

# Wait until the job has finished, since this method only works on
# SUCCEEDED jobs.
job.get_status(verbose=True, wait_until_finish=True)
# Get the Test Run result.
test_run = job.get_test_run()
# Dump the Test Cases in dataframe ``df``.
df = test_run.get_test_cases_df()

get_test_batch(test_type: str) → TestBatch

Return the Test Batch object for the specified test type on this Test Run.

A TestBatch object allows a user to query the results for the corresponding test. For example, the TestBatch object representing subset_performance:subset_accuracy allows a user to understand the results of the subset_performance:subset_accuracy test to varying levels of granularity.

Parameters:: test_type – str Name of the test. Structured as test_type:test_name, e.g. subset_performance:subset_accuracy.
Returns:: A TestBatch representing test_type.
Return type:: TestBatch

Example

batch = test_run.get_test_batch("unseen_categorical")

get_test_batches() → Iterator[TestBatch]

Return all TestBatch objects for the Test Run.

Returns:: An iterator of TestBatch objects.
Return type:: Iterator[TestBatch]

get_summary_tests_df(show_summary_test_metrics: bool = False) → DataFrame

Get summary tests for a completed stress Test Run in a dataframe.

This gives you the ability to perform granular queries on summary tests. This only works on stress test jobs that have succeeded.

Returns:: A pandas.DataFrame object containing the Test Case results. Here is a selected list of columns in the output: 1. id: ID of the parent Test Run. 2. name: name of the summary test. 3. severity: Denotes the severity of the failure of the test. 4. test_batch_types: List of tests that this summary tests uses. 5. failing_test_types: List of tests in this summary test that fail. 6. num_none_severity: Count of tests with NONE severity. 7. num_low_severity: Count of tests with LOW severity. 9. num_high_severity: Count of tests with HIGH severity.
Return type:: pd.DataFrame

Example

# Wait until the job has finished, since this method only works on
# SUCCEEDED jobs.
job.get_status(verbose=True, wait_until_finish=True)
# Get the Test Run result.
test_run = job.get_test_run()
# Dump the Test Cases in dataframe ``df``.
df = test_run.get_summary_tests_df()

class rime_sdk.TestBatch(api_client: ApiClient, test_run_id: str, test_type: str)

An interface for a Test Batch object in a RIME Test Run.

summary(show_batch_metrics: bool = False) → Series

Return the Test Batch summary as a Pandas Series.

The summary contains high level information about a Test Batch. For example, the name of the Test Batch, the category, and the severity of the Test Batch as a whole.

Returns:

A Pandas Series with the following columns (and optional additional columns for batch-level metrics):

test_run_id

test_type

test_name

category

duration_in_millis

severity

failing_features

description

summary_counts.total

summary_counts.pass

summary_counts.fail

summary_counts.warning

summary_counts.skip

Return type:

pd.Series

get_test_cases_df() → DataFrame

Return all Test Cases in the Test Batch as a Pandas DataFrame.

Different tests will have different columns/information. For example, some tests may have a column representing the number of failing rows.

Returns:: A Pandas Dataframe where each row represents a Test Case.
Return type:: pd.DataFrame

Example

# Wait until the job has finished, since this method only works on
# SUCCEEDED jobs.
job.get_status(verbose=True, wait_until_finish=True)
# Get the Test Run result.
test_run = job.get_test_run()
# Get the "subset accuracy" Test Batch from this Test Run.
test_batch = test_run.get_test_batch("subset_performance:subset_recall")
# Return the Test Cases from this Test Batch in a dataframe ``df``.
df = test_batch.get_test_cases_df()

class rime_sdk.Firewall(api_client: ApiClient, firewall_id: str)

An interface to a RIME Firewall.

update_firewall(model_id: str | None = None, ref_data_id: str | None = None, scheduled_ct_eval_data_integration_id: str | None = None, scheduled_ct_eval_data_info: dict | None = None, scheduled_ct_eval_pred_integration_id: str | None = None, scheduled_ct_eval_pred_info: dict | None = None, update_rolling_window: bool = False, scheduled_ct_rolling_window_size: timedelta | None = None, disable_scheduled_ct: bool | None = False) → RimeUpdateFirewallResponse

Update the Firewall with specified model and reference data.

Parameters:

model_id – Optional[str] The ID of the model to use for the Firewall.
ref_data_id – Optional[str] The ID of the reference data to use for the Firewall.
scheduled_ct_eval_data_integration_id – Optional[str] The integration id of the evaluation data for scheduled CT.
scheduled_ct_eval_data_info – Optional[Dict] The data info of the evaluation data for scheduled CT.
scheduled_ct_eval_pred_integration_id – Optional[str] The integration id of the evaluation prediction for scheduled CT.
scheduled_ct_eval_pred_info – Optional[Dict] The data info of the evaluation prediction for scheduled CT.
update_rolling_window – bool Specifies whether to update rolling window
scheduled_ct_rolling_window_size – Optional[timedelta] If update_rolling_window is True, the new rolling window size will be set to this value. The rolling window for the firewall is used to determine the reference set for scheduled CT if it is not None.
disable_scheduled_ct – Optional[bool] Specifies whether to disable continuous testing

Returns:

Dictionary representation of updated Firewall object.

Return type:

Dict

Raises:

ValueError – This error is generated when no fields are submitted to be updated or when the request to the Firewall service fails.

Example

response = fw.update_firewall(ref_data_id="New reference data ID")

activate_ct_scheduling(data_info: dict, data_integration_id: str | None = None, pred_integration_id: str | None = None, pred_info: dict | None = None, rolling_window_size: timedelta | None = None) → None

Activate scheduled CT.

Parameters:

data_info – dict The data info of the evaluation data for scheduled CT.
data_integration_id – Optional[str] The integration id of the evaluation data for scheduled CT.
pred_integration_id – Optional[str] The integration id of the evaluation prediction for scheduled CT.
pred_info – Optional[dict] The prediction info of the evaluation data for scheduled CT.
rolling_window_size – Optional[timedelta] If rolling_window_size is not None, it is used to determine the reference set for scheduled CT.

update_scheduled_ct_info(data_integration_id: str | None = None, data_info: dict | None = None, pred_integration_id: str | None = None, pred_info: dict | None = None) → None

Update scheduled CT.

Parameters:

data_integration_id – Optional[str] If data_integration_id is not None, the Firewall will use it for scheduled CT.
data_info – Optional[dict] If data_info is not None, the Firewall will use it for scheduled CT.
pred_integration_id – Optional[str] If pred_integration_id is not None, the Firewall will use it for scheduled CT.
pred_info – Optional[dict] If pred_info is not None, the Firewall will use it for scheduled CT.

deactivate_ct_scheduling() → None: Deactivate scheduled CT.

get_link() → str: Return a URL to the Firewall.

get_bin_size() → timedelta: Return the bin size of this Firewall.

get_ref_data_id() → str: Return the ID of the Firewall’s current reference set.

get_model_id() → str: Return the ID of the Firewall’s current model.

get_scheduled_ct_info() → FirewallScheduledCTInfo: Return the scheduled continuous testing info of this Firewall.

is_scheduled_ct_enabled() → bool: Return whether scheduled continuous testing is enabled for this Firewall.

get_events_df() → DataFrame

Get a dataframe of Detected Events for the given Firewall.

Monitors detect Events when degradations occur. For example, a Monitor for the metric “Accuracy” will detect an Event when the value of the model performance metric drops below a threshold.

list_monitors(monitor_types: List[str] | None = None, risk_category_types: List[str] | None = None) → Iterator[Monitor]

List Monitors for the given Firewall.

Monitors examine time-sequenced data in RIME. Built-in Monitors track model health metrics such as degradations in model performance metrics or attacks on your model. This method can optionally filter by Monitor types or Risk Category types.

Parameters:

monitor_types – Optional[List[str]] Modifies query to return the set of built-in monitors or user-created custom monitors. Accepted values: [“Default”, “Custom”]
risk_category_types – Optional[List[str]] Modifies query to return monitors pertaining to certain categories of AI Risk. For instance, monitors that track model performance help you track down Operational Risk. Accepted values: [“Operational”, “Bias_and_Fairness”, “Security”, “Custom”]

Returns:

A generator of Monitor objects.

Return type:

Iterator[Monitor]

Raises:

ValueError – This error is generated when unrecognized filtering parameters are provided or when the request to the Firewall service fails.

Example

# List all default Monitors on the Firewall
monitors = firewall.list_monitors(monitor_types=["Default"])
# For each Monitor, list all detected Events.
all_events = [monitor.list_detected_events() for monitor in monitors]

start_continuous_test(eval_data_id: str, override_existing_bins: bool = False, agent_id: str | None = None, ram_request_megabytes: int | None = None, cpu_request_millicores: int | None = None, random_seed: int | None = None, rime_managed_image: str | None = None, custom_image: RuntimeinfoCustomImage | None = None, **exp_fields: Dict[str, object]) → ContinuousTestJob

Start a Continuous Testing run.

Runs a Continuous Testing job on a batch of data.

Parameters:

eval_data_id – str ID of the evaluation data.
override_existing_bins – bool Specifies whether to override existing bins.
ram_request_megabytes – Optional[int] Megabytes of RAM set as the Kubernetes pod limit for the Stress Test Job. The default is 4000MB.
cpu_request_millicores – Optional[int] Millicores of CPU set as the Kubernetes pod limit for the Stress Test Job. The default is 1500mi.
agent_id – Optional[str] ID for the Agent where the Continuous Test will be run. Uses the default Agent for the workspace when not specified.
rime_managed_image – Optional[str] Name of a Managed Image to use when running the model test. The image must have all dependencies required by your model. To create new Managed Images with your desired dependencies, use the client’s create_managed_image() method.
custom_image – Optional[RuntimeinfoCustomImage] Specification of a customized container image to use running the model test. The image must have all dependencies required by your model. The image must specify a name for the image and optionally a pull secret (of type RuntimeinfoCustomImagePullSecret) with the name of the kubernetes pull secret used to access the given image.
exp_fields – Dict[str, object] Fields for experimental features.

Returns:

A Job object corresponding to the model Continuous Test Job.

Return type:

ContinuousTestJob

Raises:

ValueError – This error is generated when the request to the ModelTesting service fails.

Example

firewall = project.get_firewall()
eval_data_id = client.register_dataset("example dataset", data_config)
job = firewall.start_continuous_test(
    eval_data_id=eval_data_id,
    ram_request_megabytes=8000,
    cpu_request_millicores=2000,
)

class rime_sdk.ImageBuilder(api_client: ApiClient, name: str, requirements: List[ManagedImagePipRequirement] | None = None, package_requirements: List[ManagedImagePackageRequirement] | None = None, python_version: str | None = None)

An interface to an Image Builder object.

get_status(verbose: bool = True, wait_until_finish: bool = False, poll_rate_sec: float = 5.0) → Dict

Return the status of the Image Build Job.

This method includes a toggle to wait until the Image Build job finishes.

Parameters:

verbose – bool Specifies whether to print diagnostic information such as logs. By default, this value is set to True.
wait_until_finish – bool Specifies whether to block until the Image is READY or FAILED. By default, this value is set to False.
poll_rate_sec – float The frequency with which to poll the Image’s build status. By default, this value is set to 5 seconds.

Returns:

A dictionary representing the Image’s state.

Return type:

Dict

class rime_sdk.DataCollector(api_client: ApiClient, project_id: str)

An interface to a Data Collector.

A Data Collector allows users to log datapoints to be used in RIME at either a datapoint or batch level in a continuous stream.

register_data_stream() → str

Register a data stream with the Data Collector.

A data stream is a location to which data can be uploaded.

Returns:: The ID of the registered data stream.
Return type:: str

Log datapoints in batches.

Parameters:

data_stream_id – str The ID of the data stream to log the datapoints.
inputs – List[Dict] List of inputs to log to the Data Collector. Provide each input should as a dictionary. Feature names are dictionary keys, with their corresponding values.
timestamps – Optional[List[datetime]] List of optional timestamps associated with each input. The default value is the timestamp when the log_datapoints method is called.
labels – Optional[List[Union[Dict, int, float]]] List of optional labels associated with each input.
query_ids – Optional[List[Union[str, float, int]]] List of optional query IDs associated with each input. This parameter is only relevant for ranking use cases.

Returns:

List of the logged datapoint IDs.

Return type:

List[str]

Raises:

ValueError – This error is generated when the length of the inputs, timestamps, labels, or query_ids lists are not equal.

Example: This example registers a data stream and logs two datapoints to the registered data stream.

data_stream_id = data_collector.register_data_stream()
datapoint_ids = data_collector.log_datapoints(
data_stream_id=data_stream_id,
    inputs=[
        {"feature_1": 1, "feature_2": 2},
        {"feature_1": 3, "feature_2": 4},
    ],
    timestamps=[
        datetime(2020, 1, 1, 0, 0, 0),
        datetime(2020, 1, 1, 0, 0, 1),
    ],
    labels=[{"label": "label_1"}, {"label": "label_2"}],
)

log_predictions(model_id: str, datapoint_ids: List[str], predictions: List[Dict]) → None

Log predictions to the Data Collector.

Parameters:

model_id – str The ID of the model to log the predictions.
datapoint_ids – List[str] List of datapoint IDs associated with each prediction.
predictions – List[Dict] List of predictions to log to the Data Collector. Provide each prediction as a dictionary. Feature names are dictionary keys, with their corresponding values.

Raises:

ValueError – This error is generated when the length of the datapoint_ids and predictions lists are not equal.

Example: This example logs two predictions to the Data Collector.

data_collector.log_predictions(
    model_id="model_id",
    datapoint_ids=["datapoint_id_1", "datapoint_id_2"],
    predictions=[
        {"prediction": "prediction_1"}, {"prediction": "prediction_2"}
    ],
)

class rime_sdk.Registry(api_client: ApiClient)

An interface to a RIME Registry.

Register and validate a new dataset in a Project.

Parameters:

project_id – str The ID of the Project in which to register the dataset.
name – str The chosen name of the dataset.
data_config – dict A dictionary that contains the data configuration. The data configuration must match the API specification of the data_info field in the RegisterDataset request.
integration_id – Optional[str] = None, Provide the integration ID for datasets that require an integration.
tags – Optional[List[str]] = None, An optional list of tags to associate with the dataset.
metadata – Optional[dict] = None, An optional dictionary of metadata to associate with the dataset.
ct_info – Optional[dict] = None, An optional dictionary that contains the CT info. The CT info must match the API specification of the ct_info field in the RegisterDataset request.
skip_validation – Optional[bool] = False, An optional boolean that indicates whether to skip validation. Validation ensures that the dataset is valid for Robust Intelligence’s systems.
agent_id – Optional[str] = None, Agent for running validation. If omitted the workspace’s default agent will be used.

Returns:

The ID of the newly registered dataset.

Return type:

str

Raises:

ValueError – This error is generated when the request to the Registry service fails.

Example

dataset_id = registry.register_dataset(
    name=DATASET_NAME,
    data_config={
        "connection_info": {"data_file": {"path": FILE_PATH}},
        "data_params": {"label_col": LABEL_COL},
    },
    integration_id=INTEGRATION_ID,
)

Register and validate a new model in a Project.

Parameters:

project_id – str The ID of the Project in which to register the model.
name – str The chosen name of the model.
model_config – Optional[dict] = None, A dictionary that contains the model configuration. Any model configuration that is provided must match the API specification for the model_info field of the RegisterModel request.
tags – Optional[List[str]] = None, An optional list of tags to associate with the model.
metadata – Optional[dict] = None, An optional dictionary of metadata to associate with the model.
external_id – Optional[str] = None, An optional external ID that can be used to identify the model.
integration_id – Optional[str] = None, Provide the integration ID for models that require an integration for access.
skip_validation – Optional[bool] = False, An optional boolean that indicates whether to skip validation. Validation ensures that the model is valid for Robust Intelligence’s systems.
agent_id – Optional[str] = None, Agent for running validation. If omitted the workspace’s default agent will be used.

Returns:

The ID of the newly registered model.

Return type:

str

Raises:

ValueError – This error is generated when the request to the Registry service fails.

Example

model_id = registry.register_model(
    name=MODEL_NAME,
    model_config={
        "hugging_face": {
            "model_uri": URI,
            "kwargs": {
                "tokenizer_uri": TOKENIZER_URI,
                "class_map": MAP,
                "ignore_class_names": True,
            },
        }
    },
    tags=[MODEL_TAG],
    metadata={KEY: VALUE},
    external_id=EXTERNAL_ID,
    agent_id=AGENT_ID,
)

register_predictions(project_id: str, dataset_id: str, model_id: str, pred_config: dict, integration_id: str | None = None, tags: List[str] | None = None, metadata: dict | None = None, skip_validation: bool | None = False, agent_id: str | None = None) → None

Register and validate a new set of predictions for a model and a dataset.

Parameters:

project_id – str The ID of the Project to which the models belong.
dataset_id – str, The ID of the dataset used to generate the predictions.
model_id – str, The ID of the model used to generate the predictions.
pred_config – dict, A dictionary that contains the prediction configuration. The prediction configuration must match the API specification for the pred_info field of the RegisterPredictions request.
integration_id – Optional[str] = None, Provide the integration ID for predictions that require an integration to use.
tags – Optional[List[str]] = None, An optional list of tags to associate with the predictions.
metadata – Optional[dict] = None, An optional dictionary of metadata to associate with the predictions.
skip_validation – Optional[bool] = False, An optional boolean that indicates whether to skip validation. Validation ensures that the predictions are valid for Robust Intelligence’s systems.
agent_id – Optional[str] = None, Agent for running validation. If omitted the workspace’s default agent will be used.

Returns:

None

Raises:

ValueError – This error is generated when the request to the Registry service fails.

Example

registry.register_predictions(
    dataset_id=DATASET_ID,
    model_id=MODEL_ID,
    pred_config={
        "connection_info": {
            "databricks": {
                # Unix timestamp equivalent to 02/08/2023
                "start_time": 1675922943,
                # Unix timestamp equivalent to 03/08/2023
                "end_time": 1678342145,
                "table_name": TABLE_NAME,
                "time_col": TIME_COL,
            },
        },
        "pred_params": {"pred_col": PREDS},
    },
    tags=[TAG],
    metadata={KEY: VALUE},
)

list_datasets(project_id: str) → Iterator[Dict]

Return a list of datasets.

Parameters:: project_id – str The ID of the Project to which the datasets belong.
Returns:: Iterator of dictionaries: each dictionary represents a dataset.
Return type:: Iterator[Dict]
Raises:: ValueError – This error is generated when the request to the Registry service fails.

list_models(project_id: str) → Iterator[Dict]

Return a list of models.

Parameters:: project_id – str The ID of the Project to which the models belong.
Returns:: Iterator of dictionaries: each dictionary represents a model.
Return type:: Iterator[Dict]
Raises:: ValueError – This error is generated when the request to the Registry service fails.

list_predictions(project_id: str, model_id: str | None = None, dataset_id: str | None = None) → Iterator[Dict]

Return a list of prediction sets.

Parameters:

project_id – str The ID of the Project to which the models belong.
model_id – Optional[str] = None The ID of the model to which the prediction sets belong.
dataset_id – Optional[str] = None The ID of the dataset to which the prediction sets belong.

Returns:

Iterator of dictionaries: each dictionary represents a prediction set.

Return type:

Iterator[Dict]

Raises:

ValueError – This error is generated when the request to the Registry service fails.

get_dataset(dataset_id: str | None = None, dataset_name: str | None = None) → Dict

Return a dataset.

Parameters:

dataset_id – str The ID of the dataset to retrieve.
dataset_name – str The name of the dataset to retrieve.

Returns:

A dictionary representing the dataset.

Return type:

Dict

Raises:

ValueError – This error is generated when the request to the Registry service fails.

has_dataset(dataset_id: str | None = None, dataset_name: str | None = None) → bool

Return a boolean on whether the dataset is present.

Parameters:

dataset_id – Optional[str] = None The ID of the dataset to check for.
dataset_name – Optional[str] = None The name of the dataset to check for.

Returns:

A boolean on whether the dataset is present.

Return type:

bool

Raises:

ValueError – This error is generated any error other than HTTPStatus.NOT_FOUND is returned from the Registry service.

get_model(model_id: str | None = None, model_name: str | None = None) → Dict

Return a model.

Parameters:

model_id – str The ID of the model to retrieve.
model_name – str The name of the model to retrieve.

Returns:

A dictionary representing the model.

Return type:

Dict

Raises:

ValueError – This error is generated when the request to the Registry service fails.

get_predictions(model_id: str, dataset_id: str) → Dict

Get a prediction set.

Parameters:

model_id – str The ID of the model used to generate the predictions.
dataset_id – str The ID of the dataset used to generate the predictions.

Returns:

A dictionary that contains the prediction set.

Return type:

Dict

Raises:

ValueError – This error is generated when the request to the Registry service fails.

delete_dataset(dataset_id: str) → None

Delete a dataset.

Parameters:: dataset_id – str The ID of the dataset to delete.
Raises:: ValueError – This error is generated when the request to the Registry service fails.

delete_model(model_id: str) → None

Delete a model.

Parameters:: model_id – str The ID of the model to delete.
Raises:: ValueError – This error is generated when the request to the Registry service fails.

delete_predictions(model_id: str, dataset_id: str) → None

Delete a prediction set.

Parameters:

model_id – str The ID of the model used to generate the predictions.
dataset_id – str The ID of the dataset used to generate the predictions.

Raises:

ValueError – This error is generated when the request to the Registry service fails.

class rime_sdk.DetectionEvent(api_client: ApiClient, event_dict: dict)

An interface to a Detection Event.

RIME surfaces Detection Events to indicate problems with a model that is in production or during model validation.

to_dict() → dict: Return a dictionary representation of the Event object.

class rime_sdk.Monitor(api_client: ApiClient, monitor_id: str, firewall_id: str, project_id: str)

An interface to a Monitor object.

Monitors track important model events over time including metric degradations or attacks on your model.

update(notify: bool | None = None) → None

Update the settings for the given Monitor in the backend.

Parameters:: notify – Optional[bool] A Boolean that specifies whether to enable Monitoring notifications for a given monitor. When Monitoring notifications are turned on for the same Project and the monitor finds a Detection Event, the system sends an alert.

list_detected_events() → Iterator[DetectionEvent]

List detected Events for the given Monitor.

For each continuous testing bin upload, RIME compares the metric value to the Monitor’s thresholds and creates detection events when a degradation is detected. For a subset of Monitors, we perform Root Cause Analysis to explain the detailed cause of the Event.

Returns:: A generator of dictionary representations of Detection Events. They are sorted in reverse chronological order by the time at which the event occurred.
Return type:: Iterator[DetectionEvent]

Example

# List all default Monitors on the Firewall
monitors = firewall.list_monitors(monitor_types=["Default"])
# For each Monitor, list all detected Events.
all_events = [monitor.list_detected_events() for monitor in monitors]