Skip to main content

okareo

Okareo Objects​

class Okareo()

A class for interacting with Okareo API and for formatting request data.

seed_data_from_list​

@staticmethod
def seed_data_from_list(data_list: List[SeedDataRow]) -> List[SeedData]

Create a list of SeedData objects from a list of dictionaries.

Each dictionary in the input list must have 'input' and 'result' keys.

Arguments:

  • data_list List[SeedDataRow] - A list of dictionaries, where each dictionary contains 'input' and 'result' keys.

Returns:

  • List[SeedData] - A list of SeedData objects created from the input dictionaries.

get_projects​

def get_projects() -> List[ProjectResponse]

Get a list of all Okareo projects available to the user.

Returns:

  • List[ProjectResponse] - A list of ProjectResponse objects accessible to the user.

Raises:

  • TypeError - If the API response is an error.
  • ValueError - If no response is received from the API.

create_project​

def create_project(name: str,
                   tags: Union[Unset, List[str]] = UNSET) -> ProjectResponse

Create a new Okareo project.

Arguments:

  • name str - The name of the new project.
  • tags Union[Unset, List[str]], optional - Optional list of tags to associate with the project.

Returns:

  • ProjectResponse - The created ProjectResponse object.

Raises:

  • TypeError - If the API response is an error.
  • ValueError - If no response is received from the API.

register_model​

def register_model(
        name: str,
        tags: Union[List[str], None] = None,
        project_id: Union[str, None] = None,
        model: Union[None, BaseModel, List[BaseModel]] = None,
        update: bool = False,
        sensitive_fields: Union[List[str], None] = None) -> ModelUnderTest

Register a new Model Under Test (MUT) to use in an Okareo evaluation.

Arguments:

  • name str - The name of the model. Model names must be unique within a project. Using the same name will return or update the existing model.
  • tags Union[List[str], None], optional - Optional list of tags to associate with the model.
  • project_id Union[str, None], optional - The project ID to associate the model with.
  • model Union[None, BaseModel, List[BaseModel]], optional - The model or list of models to register.
  • update bool, optional - Whether to update an existing model with the same name. Defaults to False.
  • sensitive_fields List[str], optional - A list of sensitive fields to mask in the model parameters. Defaults to None.

Returns:

  • ModelUnderTest - The registered ModelUnderTest object.

Raises:

  • TypeError - If the API response is an error.
  • ValueError - If no response is received from the API.

create_scenario_set​

def create_scenario_set(
        create_request: ScenarioSetCreate) -> ScenarioSetResponse

Create a new scenario set to use in an Okareo evaluation or as a seed for synthetic data generation.

Arguments:

  • create_request ScenarioSetCreate - The request object containing scenario set details and seed data. The ScenarioSetCreate object should include:

Returns:

  • ScenarioSetResponse - The created ScenarioSetResponse object.

Raises:

  • ValueError - If the seed data is empty or if no response is received from the API.
  • TypeError - If the API response is an error.

Example:

seed_data = okareo_client.seed_data_from_list([
    {"input": {"animal": "fish", "color": "red"}, "result": "red"},
    {"input": {"animal": "dog", "color": "blue"}, "result": "blue"},
    {"input": {"animal": "cat", "color": "green"}, "result": "green"}
])
create_request = ScenarioSetCreate(name="My Scenario Set", seed_data=seed_data)
okareo_client.create_scenario_set(create_request)

upload_scenario_set​

def upload_scenario_set(
        scenario_name: str,
        file_path: str,
        project_id: Union[Unset, str] = UNSET) -> ScenarioSetResponse

Upload a file as a scenario set to use in an Okareo evaluation or as a seed for synthetic data generation.

Arguments:

  • scenario_name str - The name to assign to the uploaded scenario set.
  • file_path str - The path to the file to upload.
  • project_id Union[Unset, str], optional - The project ID to associate with the scenario set.

Returns:

  • ScenarioSetResponse - The created ScenarioSetResponse object.

Raises:

  • UnexpectedStatus - If the API returns an unexpected status.
  • TypeError - If the API response is an error.
  • ValueError - If no response is received from the API.

Example:

project_id = "your_project_id"  # Optional, can be None
okareo_client.upload_scenario_set(
    scenario_name="My Uploaded Scenario Set",
    file_path="/path/to/scenario_set_file.json",
    project_id=project_id or None,
)

download_scenario_set​

def download_scenario_set(scenario: Union[ScenarioSetResponse, str],
                          file_path: str = "") -> Any

Download a scenario set from Okareo to the client's local filesystem.

Arguments:

  • scenario_set ScenarioSetResponse - The scenario set to download.
  • file_path str, optional - The path where the file will be saved. If not provided, uses scenario set name.

Returns:

  • File - The downloaded file object.

Example:

response_file = okareo_client.download_scenario_set(create_scenario_set)
with open(response_file.name) as scenario_file:
    for line in scenario_file:
        print(line)

generate_scenarios​

def generate_scenarios(
    source_scenario: Union[str, ScenarioSetResponse],
    name: str,
    number_examples: int,
    project_id: Union[Unset, str] = UNSET,
    generation_type: Union[Unset,
                           ScenarioType] = ScenarioType.REPHRASE_INVARIANT
) -> ScenarioSetResponse

Generate a synthetic scenario set based on an existing seed scenario.

Arguments:

  • source_scenario Union[str, ScenarioSetResponse] - The source scenario set or its ID to generate from.
  • name str - The name for the new generated scenario set.
  • number_examples int - The number of synthetic examples to generate per seed scenario row.
  • project_id Union[Unset, str], optional - The project ID to associate with the generated scenario set.
  • generation_type Union[Unset, ScenarioType], optional - The type of scenario generation to use.

Returns:

  • ScenarioSetResponse - The generated synthetic scenario set.

Raises:

  • TypeError - If the API response is an error.
  • ValueError - If no response is received from the API.

Example:

source_scenario = "source_scenario_id"  # or ScenarioSetResponse object
generated_set = okareo_client.generate_scenarios(
    source_scenario=source_scenario,
    name="Generated Scenario Set",
    number_examples=100,
    project_id="your_project_id",
    generation_type=ScenarioType.REPHRASE_INVARIANT
)
print(generated_set.app_link) # Prints the link to the generated scenario set

generate_scenario_set​

def generate_scenario_set(
        create_request: ScenarioSetGenerate) -> ScenarioSetResponse

Generate a synthetic scenario set based on an existing seed scenario and a ScenarioSetGenerate object. Offers more controls than the comparable generate_scenarios method.

Arguments:

  • create_request ScenarioSetGenerate - The request object specifying scenario generation parameters.

Returns:

  • ScenarioSetResponse - The generated synthetic scenario set.

Example:

generate_request = ScenarioSetGenerate(
    source_scenario_id="seed_scenario_id",
    name="My Synthetic Scenario Set",
    number_examples=50,
    project_id="your_project_id",
    generation_type=ScenarioType.REPHRASE_INVARIANT,
)
generated_set = okareo_client.generate_scenario_set(generate_request)
print(generated_set.app_link)  # Prints the link to the generated scenario set

get_scenario_data_points​

def get_scenario_data_points(
        scenario_id: str) -> List[ScenarioDataPoinResponse]

Fetch the scenario data points associated with a scenario set with scenario_id.

Arguments:

  • scenario_id str - The ID of the scenario set to fetch data points for.

Returns:

  • List[ScenarioDataPoinResponse] - A list of scenario data point responses associated with the scenario set.

Example:

okareo_client = Okareo(api_key="your_api_key")
scenario_id = "your_scenario_id"
data_points = okareo_client.get_scenario_data_points(scenario_id)
for dp in data_points:
    print(dp.input_, dp.result)

find_test_data_points​

def find_test_data_points(
    test_data_point_payload: FindTestDataPointPayload
) -> Union[List[Union[TestDataPointItem, FullDataPointItem]], ErrorResponse]

Fetch the test run data points associated as specified in the payload.

Arguments:

  • test_data_point_payload FindTestDataPointPayload - The payload specifying the test data point search criteria.

Returns:

Union[List[Union[TestDataPointItem, FullDataPointItem]], ErrorResponse]: A list of test or full data point items, or an error response.

Example:

from okareo_api_client.models.find_test_data_point_payload import (
    FindTestDataPointPayload,
)

test_run_id = "your_test_run_id"  # Replace with your actual test run ID
payload = FindTestDataPointPayload(
    test_run_id=test_run_id,
)
data_points = okareo_client.find_test_data_points(payload)
for dp in data_points:
    print(dp)

find_datapoints​

def find_datapoints(
    datapoint_search: DatapointSearch
) -> Union[List[DatapointListItem], ErrorResponse]

Fetch the datapoints specified by a Datapoint Search.

Arguments:

  • datapoint_search DatapointSearch - The search criteria for fetching datapoints.

Returns:

Union[List[DatapointListItem], ErrorResponse]: A list of datapoint items matching the search, or an error response.

Example:

from okareo_api_client.models.datapoint_search import DatapointSearch

### Search based on a test run ID
test_run__id = "your_test_run_id"  # Replace with your actual test run ID
search = DatapointSearch(
    test_run_id=test_run__id,
)
datapoints = okareo_client.find_datapoints(search)
for dp in datapoints:
    print(dp)

### Search based on a context token from a logger
logger_config = {
    "api_key": "<API_KEY>",
    "tags": ["logger-test"],
    "context_token": random_string(10),
}
# Use the logger config to log completions from CrewAI or Autogen
...

# Search for the logged datapoints by the context token
search = DatapointSearch(
    context_token=context_token,
)
datapoints = okareo_client.find_datapoints(search)
for dp in datapoints:
    print(dp)

generate_check​

def generate_check(
        create_check: EvaluatorSpecRequest) -> EvaluatorGenerateResponse

Generate the contents of a Check based on an EvaluatorSpecRequest. Can be used to generate a behavioral (model-based) or a deterministic (code-based) check. Check names must be unique within a project.

Arguments:

  • create_check EvaluatorSpecRequest - The specification for the check to generate.

Returns:

  • EvaluatorGenerateResponse - The generated check response.

Example:

from okareo_api_client.models.evaluator_spec_request import EvaluatorSpecRequest
from okareo.okareo import OkareoClient, BaseCheck

# Generate a behavioral model-based check
spec = EvaluatorSpecRequest(
    description="Checks if the output contains toxic language.",
    requires_scenario_input=False,
    requires_scenario_result=False,
    output_data_type="bool", # bool, int, float
)
okareo_client = Okareo(api_key="your_api_key")
generated_check = okareo_client.generate_check(spec)

# Inspect the generated check to ensure it meets your requirements
print(generated_check)

# Upload the generated check to Okareo to use in evaluations
toxicity_check = okareo.create_or_update_check(
    name="toxicity_check",
    description=generated_check.description,
    check=ModelBasedCheck(  # type: ignore
        prompt_template=check.generated_prompt,
        check_type=CheckOutputType.PASS_FAIL,
    ),
)
# Inspect the uploaded check
print(toxicity_check)

get_all_checks​

def get_all_checks() -> List[EvaluatorBriefResponse]

Fetch all available checks.

Arguments:

None

Returns:

  • List[EvaluatorBriefResponse] - A list of EvaluatorBriefResponse objects representing all available checks.

Example:

checks = okareo_client.get_all_checks()
for check in checks:
    print(check.name, check.id)

get_check​

def get_check(check_id: str) -> EvaluatorDetailedResponse

Fetch details for a specific check.

Arguments:

  • check_id str - The ID of the check to fetch.

Returns:

  • EvaluatorDetailedResponse - The detailed response for the specified check.

Example:

check_id = "your_check_id"
check_details = okareo_client.get_check(check_id)
print(check_details)

delete_check​

def delete_check(check_id: str, check_name: str) -> str

Deletes a check identified by its ID and name.

Arguments:

  • check_id str - The unique identifier of the check to delete.
  • check_name str - The name of the check to delete.

Returns:

  • str - A message indicating the result of the deletion.

Example:

result = okareo_client.delete_check(check_id="abc123", check_name="MyCheck")
print(result)  # Output: Check deletion was successful

create_or_update_check​

def create_or_update_check(name: str, description: str,
                           check: BaseCheck) -> EvaluatorDetailedResponse

Create or update an existing check. If the check with 'name' already exists, then this method will update the existing check. Otherwise, this method will create a new check.

Arguments:

  • name str - The unique name of the check to create or update.
  • description str - A human-readable description of the check.
  • check BaseCheck - An instance of BaseCheck containing the check configuration.

Returns:

  • EvaluatorDetailedResponse - The detailed response from the evaluator after creating or updating the check.

Raises:

  • AssertionError - If the response is not an instance of EvaluatorDetailedResponse.
  • ValueError - If the response validation fails.

Example:

from okareo.checks import CheckOutputType, ModelBasedCheck

# Define your custom check; here we use a ModelBasedCheck as an example.
# Mustache template is used in the prompt to pipe scenario input and generated output
my_check = ModelBasedCheck(
    prompt_template="Only output the number of words in the following text: {scenario_input} {generation}",
    check_type=CheckOutputType.PASS_FAIL,
)

# Create or update the check
response = okareo_client.create_or_update_check(
    name="my_word_count_check",
    description="Custom check for counting combined total number of words in input and output.",
    check=my_check
)

print(response)

create_trace_eval​

def create_trace_eval(group: Any, context_token: str) -> Any

Create a trace evaluation for a group.

Arguments:

  • group_id str - The ID of the group.
  • context_token str - The context token for the trace.

Returns:

The created trace evaluation details.

Raises:

  • OkareoAPIException - If the API request fails.

evaluate​

def evaluate(name: str,
             test_run_type: TestRunType,
             scenario_id: Union[Unset, str] = UNSET,
             datapoint_ids: Union[Unset, list[str]] = UNSET,
             filter_group_id: Union[Unset, str] = UNSET,
             tags: Union[Unset, list[str]] = UNSET,
             metrics_kwargs: Union[Dict[str, Any], Unset] = UNSET,
             checks: Union[Unset, list[str]] = UNSET) -> TestRunItem

Evaluate datapoints using the specified parameters.

Arguments:

  • scenario_id - ID of the scenario set
  • metrics_kwargs - Dictionary of metrics to be measured
  • name - Name of the test run
  • test_run_type - Type of test run
  • tags - Tags for filtering test runs
  • checks - List of checks to include
  • datapoint_ids - List of datapoint IDs to filter by
  • filter_group_id - ID of the datapoint filter group to apply

Returns:

  • TestRunItem - The evaluation results as a TestRunItem object.

Example:

checks = ["model_refusal"]  # one or more checks to apply in the evaluation
test_run = okareo.evaluate(
    name="My Test Run",
    test_run_type=TestRunType.NL_GENERATION,
    checks=checks,
    datapoint_ids=["datapoint_id_1", "datapoint_id_2"],
)
print(test_run.app_link)  # View link to eval results in Okareo app