aiq.profiler.forecasting.models.random_forest_regressor#

Attributes#

logger

Classes#

RandomForestModel

A random forest regressor that predicts n_step token usage and call latency.

Module Contents#

logger#
class RandomForestModel#

Bases: aiq.profiler.forecasting.models.forecasting_base_model.ForecastingBaseModel

A random forest regressor that predicts n_step token usage and call latency.

model#
matrix_length = None#
fit(
raw_stats: list[list[aiq.data_models.intermediate_step.IntermediateStep]],
)#

X: shape (N, M) # M = matrix_length * 4 y: shape (N, 4)

predict(
raw_stats: list[list[aiq.data_models.intermediate_step.IntermediateStep]],
) → numpy.ndarray#

Predict using the fitted linear model. Returns shape (N, 4)

_prep_single(raw_stats) → numpy.ndarray#
_prep_for_model_training(raw_stats)#
_preprocess_for_forecasting(
arrays: list[numpy.ndarray],
n: int = 3,
k: int = 3,
) → list[tuple[numpy.ndarray, numpy.ndarray]]#

Preprocess a list of arrays where each array has shape (T, 3), with columns:

0: seconds_since_last_llm_call 1: input_prompt_tokens 2: output_prompt_tokens

For each row ‘i’ in each array, produce:
X: shape (n, 3)
-> The previous n calls up to row i (padded if needed).

For row i itself, set output_prompt_tokens=0 (simulate unknown current output).

Y: shape (k, 3)

-> The next k calls after row i (padded if needed).

Parameters#

arrayslist of np.ndarray

Each array is shape (T, 3).

nint

Number of past calls to include for the input context (window size).

kint

Number of future calls to include in the label (forecast horizon).

Returns#

sampleslist of (X, Y) tuples

Each X has shape (n, 3), each Y has shape (k, 3).

_extract_token_usage_meta(
all_requests_data: list[list[aiq.data_models.intermediate_step.IntermediateStep]],
)#
_flatten_features(x_list, y_list)#

X_list: list of arrays, each of shape (matrix_length, 4) y_list: list of arrays, each of shape (1, 4)

Returns:

X_flat: np.array of shape (N, matrix_length*4) y_flat: np.array of shape (N, 4)