IBM watsonx.governance¶

pip install pineflow-monitors-watsonx

Monitors¶

class WatsonxPromptMonitor¶

Provides functionality to interact with IBM watsonx.governance for monitoring IBM watsonx.ai LLMs.

Note

One of the following parameters is required to create a prompt monitor: project_id or space_id, but not both.

Parameters:

api_key (str) – The API key for IBM watsonx.governance.
space_id (str, optional) – The space ID in watsonx.governance.
project_id (str, optional) – The project ID in watsonx.governance.
region (str, optional) – The region where watsonx.governance is hosted when using IBM Cloud. Defaults to us-south.
cpd_creds (CloudPakforDataCredentials, optional) – The Cloud Pak for Data environment credentials.

Example

from pineflow.monitors.watsonx import (
    WatsonxPromptMonitor,
    CloudPakforDataCredentials,
)

# watsonx.governance (IBM Cloud)
wxgov_client = WatsonxPromptMonitor(api_key="API_KEY", space_id="SPACE_ID")

# watsonx.governance (CP4D)
cpd_creds = CloudPakforDataCredentials(
    url="CPD_URL",
    username="USERNAME",
    password="PASSWORD",
    version="5.0",
    instance_id="openshift",
)

wxgov_client = WatsonxPromptMonitor(
    space_id="SPACE_ID", cpd_creds=cpd_creds
)

add_prompt_monitor(name, model_id, task_id, description='', model_parameters=None, prompt_variables=None, input_text=None, context_fields=None, question_field=None)¶

Creates an IBM Prompt Template Asset and sets up monitors for the given prompt template asset.

Parameters:

name (str) – The name of the Prompt Template Asset.
model_id (str) – The ID of the model associated with the prompt.
task_id (str) – The task identifier. Currently supports “retrieval_augmented_generation” and “summarization” tasks.
description (str, optional) – A description of the Prompt Template Asset.
model_parameters (dict, optional) – A dictionary of model parameters and their respective values.
prompt_variables (List[str], optional) – A list of values for prompt input variables.
input_text (str, optional) – The input text for the prompt.
context_fields (List[str], optional) – A list of fields that will provide context to the prompt. Applicable only for the retrieval_augmented_generation task type.
question_field (str, optional) – The field containing the question to be answered. Applicable only for the retrieval_augmented_generation task type.

Example

wxgov_client.create_prompt_monitor(
    name="IBM prompt template",
    model_id="ibm/granite-3-2b-instruct",
    task_id="retrieval_augmented_generation",
    prompt_variables=["context1", "context2", "input_query"],
    input_text="Prompt text to be given",
    context_fields=["context1", "context2"],
    question_field="input_query",
)

store_payload_records(records_request, subscription_id)¶

Stores records to the payload logging system.

Parameters:

records_request (List[dict]) – A list of records to be logged. Each record is represented as a dictionary.
subscription_id (str) – The subscription ID associated with the records being logged.

Example

wxgov_client.store_payload_records(
    records_request=[
        {
            "context1": "value_context1",
            "context2": "value_context1",
            "input_query": "What's Pineflow?",
            "input_token_count": 25,
            "generated_token_count": 150,
        }
    ],
    subscription_id="5d62977c-a53d-4b6d-bda1-7b79b3b9d1a0",
)

class WatsonxExternalPromptMonitor¶

Provides functionality to interact with IBM watsonx.governance for monitoring external LLMs.

Note

One of the following parameters is required to create a prompt monitor: project_id or space_id, but not both.

Parameters:

api_key (str) – The API key for IBM watsonx.governance.
space_id (str, optional) – The space ID in watsonx.governance.
project_id (str, optional) – The project ID in watsonx.governance.
region (str, optional) – The region where watsonx.governance is hosted when using IBM Cloud. Defaults to us-south.
cpd_creds (CloudPakforDataCredentials, optional) – The Cloud Pak for Data environment credentials.

Example

from pineflow.monitors.watsonx import (
    WatsonxExternalPromptMonitor,
    CloudPakforDataCredentials,
)

# watsonx.governance (IBM Cloud)
wxgov_client = WatsonxExternalPromptMonitor(
    api_key="API_KEY", space_id="SPACE_ID"
)

# watsonx.governance (CP4D)
cpd_creds = CloudPakforDataCredentials(
    url="CPD_URL",
    username="USERNAME",
    password="PASSWORD",
    version="5.0",
    instance_id="openshift",
)

wxgov_client = WatsonxExternalPromptMonitor(
    space_id="SPACE_ID", cpd_creds=cpd_creds
)

add_prompt_monitor(name, model_id, task_id, detached_model_provider, description='', model_parameters=None, detached_model_name=None, detached_model_url=None, detached_prompt_url=None, detached_prompt_additional_info=None, prompt_variables=None, input_text=None, context_fields=None, question_field=None)¶

Creates a Detached/External Prompt Template Asset and sets up monitors for a given prompt template asset.

Parameters:

name (str) – The name of the External Prompt Template Asset.
model_id (str) – The ID of the model associated with the prompt.
task_id (str) – The task identifier. Currently supports “retrieval_augmented_generation” and “summarization” tasks.
detached_model_provider (str) – The external model provider.
description (str, optional) – A description of the External Prompt Template Asset.
model_parameters (dict, optional) – Model parameters and their respective values.
detached_model_name (str, optional) – The name of the external model.
detached_model_url (str, optional) – The URL of the external model.
detached_prompt_url (str, optional) – The URL of the external prompt.
detached_prompt_additional_info (dict, optional) – Additional information related to the external prompt.
prompt_variables (List[str], optional) – Values for the prompt variables.
input_text (str, optional) – The input text for the prompt.
context_fields (List[str], optional) – A list of fields that will provide context to the prompt. Applicable only for “retrieval_augmented_generation” task type.
question_field (str, optional) – The field containing the question to be answered. Applicable only for “retrieval_augmented_generation” task type.

Example

wxgov_client.create_prompt_monitor(
    name="Detached prompt (model AWS Anthropic)",
    model_id="anthropic.claude-v2",
    task_id="retrieval_augmented_generation",
    detached_model_provider="AWS Bedrock",
    detached_model_name="Anthropic Claude 2.0",
    detached_model_url="https://docs.aws.amazon.com/bedrock/latest/userguide/model-parameters-claude.html",
    prompt_variables=["context1", "context2", "input_query"],
    input_text="Prompt text to be given",
    context_fields=["context1", "context2"],
    question_field="input_query",
)

store_payload_records(records_request, subscription_id)¶

Stores records to the payload logging system.

Parameters:

records_request (List[dict]) – A list of records to be logged, where each record is represented as a dictionary.
subscription_id (str) – The subscription ID associated with the records being logged.

Example

wxgov_client.store_payload_records(
    records_request=[
        {
            "context1": "value_context1",
            "context2": "value_context1",
            "input_query": "What's Pineflow?",
            "input_token_count": 25,
            "generated_token_count": 150,
        }
    ],
    subscription_id="5d62977c-a53d-4b6d-bda1-7b79b3b9d1a0",
)

Custom Metrics¶

class WatsonxCustomMetric¶

Provides functionality to set up a custom metric to measure your model’s performance with IBM watsonx.governance.

Parameters:

api_key (str) – The API key for IBM watsonx.governance.
region (str, optional) – The region where IBM watsonx.governance is hosted when using IBM Cloud. Defaults to us-south.
cpd_creds (CloudPakforDataCredentials, optional) – IBM Cloud Pak for Data environment credentials.

Example

from pineflow.monitors.watsonx import (
    WatsonxCustomMetric,
    CloudPakforDataCredentials,
)

# watsonx.governance (IBM Cloud)
wxgov_client = WatsonxCustomMetric(api_key="API_KEY")

# watsonx.governance (CP4D)
cpd_creds = CloudPakforDataCredentials(
    url="CPD_URL",
    username="USERNAME",
    password="PASSWORD",
    version="5.0",
    instance_id="openshift",
)

wxgov_client = WatsonxCustomMetric(cpd_creds=cpd_creds)

add_local_metric_definition(name, monitor_metrics, subscription_id)¶

Creates a custom metric definition to compute metrics at the local (transaction) level for IBM watsonx.governance.

Parameters:

name (str) – The name of the custom transaction metric group.
monitor_metrics (List[WatsonxLocalMetric]) – A list of metrics to be monitored at the local (transaction) level.
subscription_id (str) – The IBM watsonx.governance subscription ID associated with the metric definition.

Example

from pineflow.monitors.watsonx import WatsonxLocalMetric

wxgov_client.add_local_metric_definition(
    name="Custom LLM Local Metric",
    subscription_id="019674ca-0c38-745f-8e9b-58546e95174e",
    monitor_metrics=[
        WatsonxLocalMetric(name="context_quality", data_type="double")
    ],
)

add_metric_definition(name, monitor_metrics, integrated_system_url, integrated_system_credentials, schedule=False)¶

Creates a custom monitor definition for IBM watsonx.governance.

This must be done before using custom metrics.

Parameters:

name (str) – The name of the custom metric group.
monitor_metrics (List[WatsonxMetric]) – A list of metrics to be measured.
schedule (bool, optional) – Enable or disable the scheduler. Defaults to False.
integrated_system_url (str) – The URL of the external metric provider.
integrated_system_credentials (IntegratedSystemCredentials) – The credentials for the integrated system.

Example

from pineflow.monitors.watsonx import (
    WatsonxMetric,
    IntegratedSystemCredentials,
    WatsonxMetricThreshold,
)

wxgov_client.add_metric_definition(
    name="Custom Metric - Custom LLM Quality",
    monitor_metrics=[
        WatsonxMetric(
            name="context_quality",
            applies_to=[
                "retrieval_augmented_generation",
                "summarization",
            ],
            thresholds=[
                WatsonxMetricThreshold(
                    threshold_type="lower_limit", default_value=0.75
                )
            ],
        )
    ],
    integrated_system_url="IS_URL",  # URL to the endpoint computing the metric
    integrated_system_credentials=IntegratedSystemCredentials(
        auth_type="basic", username="USERNAME", password="PASSWORD"
    ),
)

add_monitor_instance(integrated_system_id, monitor_definition_id, subscription_id)¶

Enables a custom monitor for the specified subscription and monitor definition.

Parameters:

integrated_system_id (str) – The ID of the integrated system.
monitor_definition_id (str) – The ID of the custom metric monitor instance.
subscription_id (str) – The ID of the subscription to associate the monitor with.

Example

wxgov_client.add_monitor_instance(
    integrated_system_id="019667ca-5687-7838-8d29-4ff70c2b36b0",
    monitor_definition_id="custom_llm_quality",
    subscription_id="0195e95d-03a4-7000-b954-b607db10fe9e",
)

list_local_metrics(custom_local_metric_id)¶

Lists records from a custom local metric definition.

Parameters:: custom_local_metric_id (str) – The unique ID of the custom transaction metric.

Example

wxgov_client.list_local_metrics(
    custom_local_metric_id="0196ad47-c505-73c0-9d7b-91c082b697e3"
)

publish_metrics(monitor_instance_id, monitor_run_id, records_request)¶

Publishes custom metrics for a specific monitor instance.

Parameters:

monitor_instance_id (str) – The unique ID of the monitor instance.
monitor_run_id (str) – The ID of the monitor run that generated the metrics.
records_request (dict[str | float | int]) – Dict containing the metrics to be published.

Example

wxgov_client.publish_metrics(
    monitor_instance_id="01966801-f9ee-7248-a706-41de00a8a998",
    monitor_run_id="RUN_ID",
    records_request={"context_quality": 0.914, "sensitivity": 0.85},
)

store_payload_records(custom_local_metric_id, records_request)¶

Stores custom metrics to payload records (transaction/record level).

Parameters:

custom_local_metric_id (str) – The unique ID of the custom transaction metric.
records_request (List[Dict]) – A list of dictionaries containing the records to be stored.

Example

wxgov_client.store_payload_records(
    custom_local_metric_id="0196ad39-1b75-7e77-bddb-cc5393d575c2",
    records_request=[
        {
            "scoring_id": "123-123",
            "run_id": "RUN_ID",
            "computed_on": "payload",
            "context_quality": 0.786,
        }
    ],
)

Credentials¶

class CloudPakforDataCredentials¶

Encapsulates the credentials required for IBM Cloud Pak for Data.

Parameters:

url (str) – The host URL of the Cloud Pak for Data environment.
api_key (str, optional) – The API key for the environment, if IAM is enabled.
username (str, optional) – The username for the environment.
password (str, optional) – The password for the environment.
bedrock_url (str, optional) – The Bedrock URL. Required only when IAM integration is enabled on CP4D 4.0.x clusters.
instance_id (str, optional) – The instance ID.
version (str, optional) – The version of Cloud Pak for Data.
disable_ssl_verification (bool, optional) – Indicates whether to disable SSL certificate verification. Defaults to True.

class IntegratedSystemCredentials¶

Encapsulates the credentials for an Integrated System based on the authentication type.

Depending on the auth_type, only a subset of the properties is required.

Parameters:

auth_type (str) – The type of authentication. Currently supports “basic” and “bearer”.
username (str, optional) – The username for Basic Authentication.
password (str, optional) – The password for Basic Authentication.
token_url (str, optional) – The URL of the authentication endpoint used to request a Bearer token.
token_method (str, optional) – The HTTP method (e.g., “POST”, “GET”) used to request the Bearer token.
token_headers (dict, optional) – Optional headers to include when requesting the Bearer token. Defaults to None.
token_payload (str, optional) – The body or payload to send when requesting the Bearer token. Can be a string (e.g., raw JSON).

Supporting Classes¶

class WatsonxMetric¶

Defines the IBM watsonx.governance global monitor metric.

Parameters:

name (str) – The name of the metric.
applies_to (List[str]) – A list of task types that the metric applies to. Currently supports: “summarization”, “generation”, “question_answering”, “extraction”, and “retrieval_augmented_generation”.
thresholds (List[WatsonxMetricThreshold]) – A list of metric thresholds associated with the metric.

Example

from pineflow.monitors.watsonx import WatsonxMetric, WatsonxMetricThreshold

WatsonxMetric(
    name="context_quality",
    applies_to=["retrieval_augmented_generation", "summarization"],
    thresholds=[
        WatsonxMetricThreshold(
            threshold_type="lower_limit", default_value=0.75
        )
    ],
)

class WatsonxLocalMetric¶

Provides the IBM watsonx.governance local monitor metric definition.

Parameters:

name (str) – The name of the metric.
data_type (str) – The data type of the metric. Currently supports “string”, “integer”, “double”, and “timestamp”.
nullable (bool, optional) – Indicates whether the metric can be null. Defaults to False.

Example

from pineflow.monitors.watsonx import WatsonxLocalMetric

WatsonxLocalMetric(name="context_quality", data_type="double")

class WatsonxMetricThreshold¶

Defines the metric threshold for IBM watsonx.governance.

Parameters:

threshold_type (str) – The threshold type. Can be either lower_limit or upper_limit.
default_value (float) – The metric threshold value.

Example

from pineflow.monitors.watsonx import WatsonxMetricThreshold

WatsonxMetricThreshold(threshold_type="lower_limit", default_value=0.8)