Skip to main content

okareo

ProjectScopedPayload Objects​

class ProjectScopedPayload(Protocol)

Any generated request body that carries a project_id field.

BaseGenerationSchema Objects​

class BaseGenerationSchema(PydanticBaseModel)

A base schema class for specifying structured outputs to synthetic data generators.

Okareo Objects​

class Okareo()

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

__init__​

def __init__(api_key: str,
             base_path: str = BASE_URL,
             timeout: float = HTTPX_TIME_OUT,
             project: Union[str, UUID, None] = None)

Arguments:

  • api_key - Your Okareo API key.
  • base_path - Okareo API base URL.
  • timeout - HTTP timeout in seconds.
  • project - The Project this client works in — its name or its id. Omit to keep the server's default Project (the pre-Projects behavior). Resolved and validated here, at construction.

set_project​

def set_project(project: Union[str, UUID, None]) -> None

Switch the client-level Project mid-script by name or id (None clears it).

Fetches a fresh Project list to resolve against — freshness over the cost of one extra call on an infrequent operation.

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 the name has leading or trailing whitespace, or if no response is received from the API.

get_project​

def get_project(project_id: Union[str, UUID]) -> ProjectResponse

Get a single project by id.

Arguments:

  • project_id Union[str, UUID] - The ID of the project to fetch.

Returns:

  • ProjectResponse - The requested project.

Raises:

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

update_project​

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

Update a project's name and/or tags. Only the fields you pass are changed.

Arguments:

  • project_id Union[str, UUID] - The ID of the project to update.
  • name Union[Unset, str], optional - New name for the project.
  • tags Union[Unset, List[str]], optional - Replacement tag list.

Returns:

  • ProjectResponse - The updated project.

Raises:

  • ValueError - If the name has leading or trailing whitespace.

archive_project​

def archive_project(project_id: Union[str, UUID]) -> ProjectResponse

Archive a project. Archiving is reversible and restricts nothing — an archived project stays fully usable; it is only hidden from the project picker. The default ("Global") project cannot be archived.

Note: until the generated client is regenerated, the returned ProjectResponse carries the archive state in additional_properties["is_archived"].

unarchive_project​

def unarchive_project(project_id: Union[str, UUID]) -> ProjectResponse

Unarchive a project, restoring it to the project picker.

Note: until the generated client is regenerated, the returned ProjectResponse carries the archive state in additional_properties["is_archived"].

register_model​

def register_model(
        name: str,
        tags: Union[List[str], None] = None,
        project_id: Union[str, UUID, 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.

get_model​

def get_model(name: str, version: str | int = "latest") -> ModelUnderTest

Fetch a model under test based on the name and version.

Arguments:

  • name str - The name of the model to fetch.
  • version str | int, optional - The version of the model to fetch. Defaults to "latest".

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, UUID] = 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, UUID, ScenarioSetResponse],
    name: str,
    number_examples: int,
    project_id: Union[Unset, str, UUID] = 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: Union[str, UUID]) -> 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
) -> List[Union[TestDataPointItem, FullDataPointItem]]

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:

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

Raises:

  • TypeError - If the API response is an error.

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) -> List[DatapointListItem]

Fetch the datapoints specified by a Datapoint Search.

Arguments:

  • datapoint_search DatapointSearch - The search criteria for fetching datapoints.

Returns:

  • List[DatapointListItem] - A list of datapoint items matching the search.

Raises:

  • TypeError - If the API response is an error.

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)

find_datapoints_filter​

def find_datapoints_filter(
        datapoint_search: DatapointFilterSearchPayload
) -> List[DatapointListItem]

Fetch the datapoints specified by a Datapoint Search.

Arguments:

  • datapoint_search DatapointFilterSearchPayload - The search criteria for fetching datapoints.

Returns:

  • List[DatapointListItem] - A list of datapoint items matching the search.

Raises:

  • TypeError - If the API response is an error.

Example:

from okareo_api_client.models.datapoint_filter_search_payload import DatapointFilterSearchPayload

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

### Find datapoints based on filters on datapoints fields
from okareo_api_client.models.datapoint_filter_search_payload import DatapointFilterSearchPayload
from okareo_api_client.models.filter_condition import FilterCondition
from okareo_api_client.models.comparison_operator import ComparisonOperator

search = DatapointFilterSearchPayload(
    filters=[FilterCondition(
        field=DatapointField.TEST_RUN_ID,
        operator=ComparisonOperator.EQUAL,
        value="France"
    )]
)
datapoints = okareo_client.find_datapoints_filter(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(all_versions: bool = False) -> List[EvaluatorBriefResponse]

Fetch all available checks.

Arguments:

  • all_versions - If True, return all versions of every check (full version history). Defaults to False (latest version only).

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)

# Include full version history
all_checks = okareo_client.get_all_checks(all_versions=True)

get_check​

def get_check(
        check_id: Union[str, UUID],
        version: Union[str, int, None] = None) -> EvaluatorDetailedResponse

Fetch details for a specific check by UUID or by name.

Arguments:

  • check_id - A check UUID (str or UUID object) or a check name (str). When a name is given the method resolves it to a UUID via the list endpoint.
  • version - Optional version number or the string "latest". Only used when check_id is a name. None and "latest" both resolve to the most recent version.

Returns:

  • EvaluatorDetailedResponse - The detailed response for the specified check.

Raises:

  • ValueError - If no check matches the given name/version.

Example:

# By UUID (existing behaviour)
check = okareo_client.get_check("your_check_uuid")

# By name (latest version)
check = okareo_client.get_check("my_check")
check = okareo_client.get_check("my_check", version="latest")

# By name + pinned version
check = okareo_client.get_check("my_check", version=1)

delete_check​

def delete_check(check_id: Union[str, UUID], 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,
        tags: Optional[List[str]] = None) -> 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.

  • tags - Optional list of string tags to associate with the check.

    A model-based check's prompt_template may only use the current template variables (see ModelBasedCheck). Okareo rejects any other placeholder on save, including the legacy aliases: replace {generation} with {model_output}, {input} with {scenario_input}, and description0 with description1. description2 and description3 have no replacement — remove them, since audio is injected as multimodal content rather than through a placeholder.

Returns:

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

Raises:

  • description5 - If the response is not an instance of EvaluatorDetailedResponse.
  • description6 - If the response validation fails.

Example:

description7

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

create_or_update_driver​

def create_or_update_driver(driver: Driver) -> Driver

Create or update a simulation driver by name.

Arguments:

  • driver - Driver definition to register. If a driver with the same name already exists, it is updated.

Returns:

  • Driver - The created or updated driver.

get_driver_by_name​

def get_driver_by_name(driver_name: str) -> Driver

Retrieve a simulation driver by name.

Arguments:

  • driver_name - The name of the driver to retrieve.

Returns:

  • Driver - The driver with the specified name.

create_or_update_target​

def create_or_update_target(
        target: Target,
        tags: Optional[List[str]] = None,
        project_id: Optional[str] = None,
        sensitive_fields: Union[List[str], None] = None) -> Target

Create or update a simulation target by name.

Arguments:

  • target - Target definition to register. If a target with the same name already exists, it is updated.

Returns:

  • Target - The created or updated target.

get_target_by_name​

def get_target_by_name(target_name: str) -> Target

Retrieve a simulation target by name.

Arguments:

  • target_name - The name of the target to retrieve.

Returns:

  • Target - The target with the specified name.

run_simulation​

def run_simulation(name: str,
                   scenario: Union[ScenarioSetResponse, str],
                   target: str | Target,
                   driver: Optional[str | Driver] = None,
                   checks: Optional[list[str]] = None,
                   stop_check: Union[StopConfig, dict, None] = None,
                   repeats: Optional[int] = 1,
                   max_turns: Optional[int] = 5,
                   first_turn: Optional[str] = "target",
                   checks_at_every_turn: Optional[bool] = False,
                   concurrent_ask_probability: Optional[float] = 0.0,
                   turn_transition_time: Optional[int] = 1000,
                   augmentation: Optional[Union[Augmentation,
                                                dict[str, Any]]] = None,
                   api_key: Optional[str] = None,
                   api_keys: Optional[dict] = None,
                   metrics_kwargs: Optional[dict] = None,
                   calculate_metrics: bool = True,
                   project_id: Optional[str] = None,
                   tags: Optional[list[str]] = None,
                   sensitive_fields: Union[List[str], None] = None,
                   submit: Optional[bool] = False) -> TestRunItem

Run a multiturn simulation against a target.

This method resolves or creates the referenced driver and target, builds Simulation parameters (including optional augmentation), and executes either ModelUnderTest.run_test (blocking) or ModelUnderTest.submit_test (async server-side) using TestRunType.MULTI_TURN.

Parameter summary:

  • name: Name of the resulting test run.
  • scenario: Scenario set object or scenario set ID.
  • target / driver: Registered names or object definitions.
  • checks: Optional checks to execute.
  • stop_check, repeats, max_turns, first_turn: Turn-flow controls.
  • checks_at_every_turn, concurrent_ask_probability, turn_transition_time: Runtime simulation controls.
  • augmentation: Optional augmentation settings.
  • api_key / api_keys: Provider credentials for target calls.
  • metrics_kwargs / calculate_metrics: Metrics configuration.
  • project_id / sensitive_fields: Registration controls.
  • tags: Tags applied to the test run this call creates — always, regardless of how target is supplied. They are persisted on the run and can be filtered on with Okareo.find_test_runs(tags=[...]). Additionally, when target is a Target object (which this call registers or updates), the same tags are applied to that Target. When target is a string name, the existing Target is looked up and its tags are left untouched — only the test run is tagged.
  • submit: If True, submit asynchronously via ModelUnderTest.submit_test.

Returns a TestRunItem representing the created simulation test run.

run_load_test​

def run_load_test(name: str,
                  target: Union[str, Target],
                  load_concurrent: int,
                  load_duration_s: float,
                  per_call_max_duration_s: Optional[float] = None,
                  seed_data: Optional[List[dict]] = None,
                  driver: Optional[Union[str, Driver]] = None,
                  checks: Optional[List[str]] = None,
                  api_key: Optional[str] = None,
                  api_keys: Optional[dict] = None,
                  project_id: Optional[str] = None,
                  calculate_metrics: bool = True) -> TestRunItem

Run a closed-loop VOICE LOAD TEST: hold load_concurrent conversations for load_duration_s seconds.

The only load knobs are load_concurrent (# calls to hold) and load_duration_s (plateau seconds). Everything that paces the ramp and protects the system from a runaway — the provider call-creation rate, the ramp deadline, the plateau-detection fraction, and the stop backstops — is an INTERNAL SAFETY mechanism derived server-side; it is deliberately not exposed here so a caller can't mistune it.

Distinct from run_simulation (which evaluates a scenario set once): here you specify LOAD. seed_data rows are round-robin sampled with replacement to fill the sustained load, and checks score every call's datapoint (quality-under-load). Internally this builds a round-robin-tiled scenario, sets the target's max_parallel_requests to load_concurrent, and submits a MULTI_TURN run whose server-side manager holds the plateau and ends it gracefully (measured from the ACTUAL plateau, and aborting a doomed ramp). Pair with a never-end driver prompt.

Optional call cycling (per_call_max_duration_s): cap each individual call's wall-clock duration. Instead of holding load_concurrent long-lived calls, the plateau is composed of short calls that are graceful-ended at the cap and immediately backfilled, so the guarantee (load_concurrent live for load_duration_s) is unchanged while each call is bounded. Shape/pacing caps (floor, hard per-call kill, dial rate) are enforced server-side; the SDK only forwards the flat knob. Omit it for the classic held-call behavior.

Returns a TestRunItem for the load-test run.

generate_driver_prompt​

def generate_driver_prompt(user_input: str,
                           prior_prompt: Optional[str] = None,
                           language: Optional[str] = None,
                           **driver_kwargs: Any) -> Driver

Generate a structured driver prompt from a one-sentence description.

Arguments:

  • user_input - Natural language description of the caller persona.
  • prior_prompt - Optional existing prompt to refine.
  • language - BCP-47 language code (e.g. "en", "es", "fr-CA").
  • **driver_kwargs - Extra fields forwarded to the returned Driver (e.g. voice_instructions, temperature, voice).

Returns:

A Driver with the AI-generated name and prompt_template.

wait_for_test_run​

def wait_for_test_run(test_run_id: Union[str, UUID],
                      poll_interval: float = 10.0,
                      timeout: Optional[float] = None) -> TestRunItem

Block until a submitted Run is FINISHED, polling with short requests.

The companion to run_simulation(..., submit=True) and submit_test: the submit call returns the accepted Run at once, this waits for it. Each poll is one GET with its own timeout; a failed poll is logged and retried.

Arguments:

  • test_run_id - The id of the submitted Run.
  • poll_interval - Seconds between polls. Defaults to 10.
  • timeout - Give up after this many seconds. None (default) waits as long as the Run runs.

Raises:

  • TestRunError - the Run ended FAILED (with the server's failure message), or timeout passed without a terminal status. On a timeout the Run may still be going, so its listener is left running; call again to keep waiting.

find_test_runs​

def find_test_runs(name: Optional[str] = None,
                   tags: Optional[list] = None,
                   project_id: Optional[str] = None,
                   return_model_metrics: bool = False) -> list

Find existing test runs for simulation/evaluation workflows.

Arguments:

  • name - Filter results to runs with this exact name (client-side).
  • tags - Filter results to runs with these tags (server-side).
  • project_id - Scope to a specific project.
  • return_model_metrics - Include model_metrics in the response.

Returns:

List of test run dicts from the server.

re_evaluate​

def re_evaluate(test_run_id: str,
                checks: list,
                name: Optional[str] = None,
                tags: Optional[list] = None) -> TestRunItem

Re-evaluate an existing test run with different checks.

No new simulation or phone call is made. The existing conversation data is re-scored with the specified checks.

Arguments:

  • test_run_id - ID of the source test run to re-evaluate.
  • checks - List of check names or IDs to apply.
  • name - Optional name for the new re-evaluated run.
  • tags - Optional tags for the new run.

Returns:

The newly created TestRunItem with re-evaluated results.

calibrate_check​

def calibrate_check(test_run_id: str,
                    check: Union[BaseCheck, Dict[str, Any]],
                    name: str = "draft_check",
                    inspect_only: bool = False,
                    timeout: float = CALIBRATE_TIME_OUT) -> Dict[str, Any]

Try out a draft check against a finished test run without saving anything.

Nothing is persisted — no check, no test run, no datapoints, no scores. Each returned row carries the check's verdict for that row plus the arguments that went into it, so you can see whether a variable was even populated before trusting the verdict.

The draft is validated first, under the same rules a check must satisfy to be saved. A prompt using an unrecognized or legacy placeholder is refused with a message naming it, and no judge call is made.

Arguments:

  • test_run_id - ID of a completed test run whose rows to calibrate against.
  • check - A ModelBasedCheck / CodeBasedCheck instance, or a raw check_config dict. The config is sent verbatim and echoed back verbatim, so a saved response says which draft produced which verdicts.
  • name - Name for the draft. Never saved; it only labels the result column.
  • inspect_only - Return the arguments for every row and skip execution entirely — no judge calls, no cost.
  • timeout - HTTP timeout in seconds. Defaults to CALIBRATE_TIME_OUT, which sits just above the server's own wall clock; Okareo.__init__ does not forward its own timeout to the HTTP client, so this is the only one.

Returns:

The parsed calibration response: the echoed check_config, which argument surface was used (check1), row counts with a check2 flag when the server's row cap was applied, and one entry per test datapoint.

Raises:

  • check3 - On any non-2xx response, carrying the status code and the response body — a 422's check4 names the placeholder to fix, and a 504 means the calibration exceeded the server's wall clock.

download_call_recording​

def download_call_recording(call_sid: str) -> bytes

Download a voice call recording by its Twilio CallSid.

Arguments:

  • call_sid - The Twilio CallSid from datapoint metadata (e.g. dp.model_metadata.additional_properties["call_sid"]).

Returns:

Raw WAV audio bytes.

upload_voice​

def upload_voice(file_path: Optional[str] = None,
                 file_bytes: Optional[bytes] = None,
                 project_id: Optional[str] = None) -> VoiceUploadResponse

Upload a voice file for use in voice scenarios/simulations.

Provide either a local file_path or in-memory file_bytes.

Arguments:

  • file_path - Path to a local audio file.
  • file_bytes - Raw audio bytes.
  • project_id - Optional project ID to associate with the uploaded file.

Returns:

  • VoiceUploadResponse - Metadata including a downloadable file_url.

Raises:

  • ValueError - If neither file_path nor file_bytes is provided.

download_voice​

def download_voice(file_url: str) -> bytes

Download a voice file from Okareo.

Files are always stored as MP3 on the server.

Arguments:

Returns:

Raw MP3 audio bytes.

create_scenario_set_with_audio_files​

def create_scenario_set_with_audio_files(
        name: str,
        data_list: List[Dict[str, str]],
        project_id: Optional[Union[str, UUID]] = None) -> ScenarioSetResponse

Upload local audio files and create a scenario set in one call.

Each item's 'input' field should be a local file path to a WAV or MP3 file. The file is uploaded to Okareo (coerced to MP3), and 'input' is replaced with the returned file URL before creating the scenario.

Arguments:

  • name - Scenario set name.
  • data_list - List of dicts with 'input' (local file path) and 'result' (expected transcript string).
  • project_id - Optional project to associate files and scenario with.

Returns:

ScenarioSetResponse from the created scenario set.

ingest_conversations​

def ingest_conversations(
        project_id: Union[str, UUID, None] = None,
        conversations: Optional[List[Dict[str, Any]]] = None,
        mut_id: Union[str, UUID, None] = None) -> Dict[str, Any]

Ingest voice conversations for monitoring.

Accepts one or more conversations from voice platforms (Retell, Twilio, VAPI, etc.) and enqueues them for async processing. Each conversation's turns will become Datapoint rows, and configured monitors will automatically match and run checks.

This is the monitoring path, not the simulation path. No ScenarioSets are created. The mut_id is optional - when omitted, datapoints are created without MUT association and rely entirely on monitor/filter group matching.

Arguments:

  • project_id - Okareo project ID.
  • conversations - List of conversation dictionaries, each containing:
    • source_platform (str): Platform source ('retell', 'twilio', 'vapi', 'elevenlabs', or 'custom')
    • call_id (str): Platform-specific call identifier
    • context_token (str, optional): Context token for correlation (defaults to call_id)
    • audio (dict, optional): Preferred audio shape with one of:
    • {"type": "url", "url": "https://..."}
    • {"type": "voice_file_id", "voice_file_id": "uuid"}
    • {"type": "inline_b64", "inline_b64": "..."}
    • recording_url (str, optional): Legacy compatibility alias for audio URL
    • recording_bytes_b64 (str, optional): Legacy compatibility alias for inline base64 audio
    • transcript (list, optional): Pre-parsed transcript as list of turns with 'role' and 'content'
    • diarization (bool, optional): When transcript is absent, controls whether Okareo runs diarization + ASR. Defaults to True.
    • metadata (dict, optional): Platform-specific metadata
    • tags (list, optional): Tags for monitor matching
    • first_turn (str, optional): For audio-only diarization ('user' or 'assistant' spoke first, defaults to 'assistant')
  • mut_id - Optional model under test ID. If not provided, datapoints are created without MUT association (monitoring path).

Returns:

Dict with 'status' and list of conversation identifiers.

Raises:

  • httpx.HTTPStatusError - If the API returns an error status.

    Example (monitoring-only, no MUT):

okareo.ingest_conversations(
    project_id="your-project-id",
    conversations=[
        {
            "source_platform": "retell",
            "call_id": "call-123",
            "audio": {
                "type": "url",
                "url": "https://retell.ai/recordings/call-123.mp3",
            },
            "tags": ["support", "billing"],
            "metadata": {"customer_id": "cust-456"}
        }
    ]
)

Example (with MUT association):

okareo.ingest_conversations(
    project_id="your-project-id",
    mut_id="your-mut-id",
    conversations=[
        {
            "source_platform": "custom",
            "call_id": "call-456",
            "transcript": [
                {"role": "user", "content": "Hello", "timestamp_ms": 0},
                {"role": "assistant", "content": "Hi, how can I help?", "timestamp_ms": 1000}
            ],
        }
    ]
)