Client API
rait_connector.client.RAITClient
Main client for interacting with RAIT API and performing evaluations.
This client handles: - Authentication with RAIT API - Fetching enabled metrics configuration - Running metric evaluations (sequential or parallel) - Encrypting and posting results to API
Example
client = RAITClient() result = client.evaluate( ... prompt_id="123", ... prompt_url="https://example.com/123", ... timestamp="2025-01-01T00:00:00Z", ... model_name="gpt-4", ... model_version="1.0", ... query="What is AI?", ... response="AI is artificial intelligence...", ... environment="production", ... purpose="monitoring" ... )
__init__(rait_api_url=None, rait_client_id=None, rait_client_secret=None, azure_client_id=None, azure_tenant_id=None, azure_client_secret=None, azure_openai_endpoint=None, azure_openai_api_key=None, azure_openai_deployment=None, azure_openai_api_version=None, azure_subscription_id=None, azure_resource_group=None, azure_project_name=None, azure_account_name=None, azure_ai_project_url=None, azure_log_analytics_workspace_id=None, rait_ingest_url=None)
Initialize RAIT client.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
rait_api_url
|
Optional[str]
|
RAIT API endpoint URL |
None
|
rait_client_id
|
Optional[str]
|
RAIT client ID |
None
|
rait_client_secret
|
Optional[str]
|
RAIT client secret |
None
|
azure_client_id
|
Optional[str]
|
Azure AD client ID |
None
|
azure_tenant_id
|
Optional[str]
|
Azure AD tenant ID |
None
|
azure_client_secret
|
Optional[str]
|
Azure AD client secret |
None
|
azure_openai_endpoint
|
Optional[str]
|
Azure OpenAI endpoint |
None
|
azure_openai_api_key
|
Optional[str]
|
Azure OpenAI API key |
None
|
azure_openai_deployment
|
Optional[str]
|
Azure OpenAI deployment name |
None
|
azure_openai_api_version
|
Optional[str]
|
Azure OpenAI API version |
None
|
azure_subscription_id
|
Optional[str]
|
Azure subscription ID |
None
|
azure_resource_group
|
Optional[str]
|
Azure resource group name |
None
|
azure_project_name
|
Optional[str]
|
Azure AI project name |
None
|
azure_account_name
|
Optional[str]
|
Azure account name |
None
|
azure_ai_project_url
|
Optional[str]
|
Azure AI project URL |
None
|
azure_log_analytics_workspace_id
|
Optional[str]
|
Azure Log Analytics workspace ID for telemetry |
None
|
rait_ingest_url
|
Optional[str]
|
Telemetry pipeline API Gateway endpoint URL |
None
|
evaluate(prompt_id, prompt_url, timestamp, model_name, model_version, query, response, environment, purpose, ground_truth='', context='', prompt_response_id='', calibration_run_id='', parallel=True, max_workers=5, fail_fast=False, connector_logs='', for_calibration=False, custom_fields=None)
Evaluate a single prompt and automatically post results to API.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
prompt_id
|
str
|
Unique identifier for the prompt |
required |
prompt_url
|
str
|
URL reference to the prompt |
required |
timestamp
|
str
|
ISO 8601 timestamp |
required |
model_name
|
str
|
Name of the LLM model |
required |
model_version
|
str
|
Version of the LLM model |
required |
query
|
str
|
User's input query |
required |
response
|
str
|
LLM's generated response |
required |
environment
|
str
|
Execution environment |
required |
purpose
|
str
|
Evaluation purpose |
required |
ground_truth
|
str
|
Expected response (optional) |
''
|
context
|
str
|
Additional context (optional) |
''
|
prompt_response_id
|
str
|
Prompt response ID from calibration (optional) |
''
|
calibration_run_id
|
str
|
Calibration run ID from calibration (optional) |
''
|
parallel
|
bool
|
Whether to run evaluations in parallel |
True
|
max_workers
|
int
|
Maximum parallel workers |
5
|
fail_fast
|
bool
|
Whether to stop on first error |
False
|
connector_logs
|
str
|
Optional connector logs to include |
''
|
for_calibration
|
bool
|
Whether this evaluation is for calibration |
False
|
custom_fields
|
Optional[Dict[str, Any]]
|
Optional dict of additional fields to include in model_data_logs |
None
|
Returns:
| Type | Description |
|---|---|
Dict[str, Any]
|
Dict containing prompt data and post response |
Raises:
| Type | Description |
|---|---|
EvaluationError
|
If evaluation fails |
MetricsError
|
If posting fails |
evaluate_batch(prompts, parallel=True, max_workers=5, fail_fast=False, connector_logs='', on_complete=None, for_calibration=False)
Evaluate multiple prompts.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
prompts
|
List[Union[Dict[str, Any], EvaluationInput]]
|
List of prompt data (dicts or EvaluationInput objects). Each prompt can include a "custom_fields" dict for per-prompt additional fields in model_data_logs. |
required |
parallel
|
bool
|
Whether to run evaluations in parallel |
True
|
max_workers
|
int
|
Maximum parallel workers |
5
|
fail_fast
|
bool
|
Whether to stop on first error |
False
|
connector_logs
|
str
|
Optional connector logs to include |
''
|
on_complete
|
Optional[callable]
|
Optional callback function called after all evaluations complete. Receives a dict with results, errors, and summary statistics. |
None
|
for_calibration
|
bool
|
Whether this evaluation is for calibration |
False
|
Returns:
| Type | Description |
|---|---|
Dict[str, Any]
|
Dict with results, errors, and summary statistics |
Example
def my_callback(summary): ... print(f"Completed: {summary['successful']}/{summary['total']}") ... for result in summary['results']: ... print(f" - {result['prompt_id']}")
client.evaluate_batch(prompts, on_complete=my_callback)
fetch_telemetry(tables=None, timespan=timedelta(days=1), custom_queries=None, limit=None)
Fetch telemetry data from Azure Monitor.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tables
|
Optional[List[str]]
|
List of tables to fetch. Defaults to ["AppDependencies", "AppExceptions", "AppAvailabilityResults"] |
None
|
timespan
|
timedelta
|
Time range to query |
timedelta(days=1)
|
custom_queries
|
Optional[Dict[str, str]]
|
Optional dict mapping table names to custom KQL queries |
None
|
limit
|
Optional[int]
|
Maximum rows per table. Defaults to None (all rows). |
None
|
Returns:
| Type | Description |
|---|---|
Dict[str, List[Dict[str, Any]]]
|
Dictionary mapping table names to lists of row dictionaries |
Raises:
| Type | Description |
|---|---|
TelemetryError
|
If fetch fails |
Example
client = RAITClient(azure_log_analytics_workspace_id="...") data = client.fetch_telemetry(timespan=timedelta(days=2)) print(data.keys()) dict_keys(['AppDependencies', 'AppExceptions', 'AppAvailabilityResults'])
get_calibration_prompts(model_name='', model_version='', model_environment='')
Fetch calibration run prompts from the calibrator API.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model_name
|
str
|
Name of the LLM model |
''
|
model_version
|
str
|
Version of the LLM model |
''
|
model_environment
|
str
|
Environment (e.g. "staging", "production") |
''
|
Returns:
| Type | Description |
|---|---|
Dict[str, Any]
|
Dict with calibration_run_id and list of prompts |
Raises:
| Type | Description |
|---|---|
CalibrationError
|
If API call fails |
get_enabled_metrics(model_name='', model_version='', model_environment='', force_refresh=False)
Fetch enabled metrics configuration from API.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
force_refresh
|
bool
|
If True, bypass cache and fetch from API |
False
|
Returns:
| Type | Description |
|---|---|
List[Dict[str, Any]]
|
List of ethical dimensions with their metrics |
Raises:
| Type | Description |
|---|---|
MetricsError
|
If API call fails |
get_model_calibration_prompts(model_name='', model_version='', model_environment='')
Fetch calibration prompts from the model registry API.
This endpoint returns prompts without responses. Use with a model callback to generate responses for calibration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model_name
|
str
|
Name of the LLM model |
''
|
model_version
|
str
|
Version of the LLM model |
''
|
model_environment
|
str
|
Environment (e.g. "staging", "production") |
''
|
Returns:
| Type | Description |
|---|---|
List[Dict[str, Any]]
|
List of dicts containing prompt_id and prompt_text |
Raises:
| Type | Description |
|---|---|
CalibrationError
|
If API call fails |
get_prompts_response(model_name='', model_version='', model_environment='')
Fetch prompts needing model responses from the calibrator API.
Used by the calibration scheduler to collect prompts that require
model responses. Pairs with :meth:update_prompts_response to complete
the calibrator response collection flow.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model_name
|
str
|
Name of the LLM model |
''
|
model_version
|
str
|
Version of the LLM model |
''
|
model_environment
|
str
|
Environment (e.g. "staging", "production") |
''
|
Returns:
| Type | Description |
|---|---|
List[Dict[str, Any]]
|
List of dicts with prompt_response_id, prompt_text, and model_response |
Raises:
| Type | Description |
|---|---|
CalibrationError
|
If API call fails |
post_calibration_responses(model_name, model_version, model_environment, model_purpose, calibration_responses, connector_logs='')
Post calibration responses to the RAIT ingest URL.
Called by the calibration scheduler after collecting model responses
for calibration prompts fetched via :meth:get_model_calibration_prompts.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model_name
|
str
|
Name of the LLM model. |
required |
model_version
|
str
|
Version of the LLM model. |
required |
model_environment
|
str
|
Execution environment (e.g. |
required |
model_purpose
|
str
|
Evaluation purpose label. |
required |
calibration_responses
|
List[Dict[str, Any]]
|
List of dicts with |
required |
connector_logs
|
str
|
Optional connector log string. |
''
|
Returns:
| Type | Description |
|---|---|
Dict[str, Any]
|
Dict with |
Raises:
| Type | Description |
|---|---|
MetricsError
|
If the post fails. |
post_telemetry(model_name, model_version, model_environment, model_purpose, telemetry_data, connector_logs='')
Post telemetry data to the RAIT ingest URL.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model_name
|
str
|
Name of the LLM model. |
required |
model_version
|
str
|
Version of the LLM model. |
required |
model_environment
|
str
|
Execution environment (e.g. |
required |
model_purpose
|
str
|
Evaluation purpose label. |
required |
telemetry_data
|
Dict[str, List[Dict[str, Any]]]
|
Azure Monitor data as returned by :meth: |
required |
connector_logs
|
str
|
Optional connector log string. |
''
|
Returns:
| Type | Description |
|---|---|
Dict[str, Any]
|
Dict with |
Raises:
| Type | Description |
|---|---|
MetricsError
|
If the post fails. |
update_prompts_response(responses, model_name='', model_version='', model_environment='')
Submit model-generated responses to the calibrator API.
Used by the calibration scheduler after collecting responses via
:meth:get_prompts_response. Each entry in responses must contain
prompt_response_id, prompt_text, and model_response.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
responses
|
List[Dict[str, str]]
|
List of dicts with keys prompt_response_id, prompt_text, and model_response |
required |
model_name
|
str
|
Name of the LLM model |
''
|
model_version
|
str
|
Version of the LLM model |
''
|
model_environment
|
str
|
Environment (e.g. "staging", "production") |
''
|
Returns:
| Type | Description |
|---|---|
Dict[str, Any]
|
Dict with status_code and response text |
Raises:
| Type | Description |
|---|---|
CalibrationError
|
If API call fails |