Core structures

Conditioning Parser

genai_monitor.structures.conditioning_parsers.base.BaseConditioningParser

Bases: ABC

Base class for conditioning parsers.

Use it to create custom conditioning parsers, not available natively in the library.

'parse_func_arguments' is the core method that needs to be overwritten by the subclasses.

db_manager instance-attribute

db_manager: DBManager

persistency_manager instance-attribute

persistency_manager: PersistencyManager

parse_conditioning

parse_conditioning(
    method: Callable, *args, **kwargs
) -> Conditioning

Parse the execution parameters of a function into a Conditioning object.

Inspects the signature of the function and passed arguments/keyword arguments.

PARAMETER	DESCRIPTION
method	The method TYPE: Callable
*args	Arguments of the method. DEFAULT: ()
**kwargs	Keyword arguments of the method. DEFAULT: {}

RETURNS	DESCRIPTION
Conditioning	A Conditioning object parsed constructed based on the execution of the method.

RAISES	DESCRIPTION
NotJsonableError	when the parsed arguments cannot be serialized to json.

Source code in src/genai_monitor/structures/conditioning_parsers/base.py

def parse_conditioning(self, method: Callable, *args, **kwargs) -> Conditioning:
    """Parse the execution parameters of a function into a Conditioning object.

    Inspects the signature of the function and passed arguments/keyword arguments.

    Args:
        method: The method
        *args: Arguments of the method.
        **kwargs: Keyword arguments of the method.

    Returns:
        A Conditioning object parsed constructed based on the execution of the method.

    Raises:
        NotJsonableError: when the parsed arguments cannot be serialized to json.
    """
    conditioning_metadata = kwargs.pop(CONDITIONING_METADATA_FIELDNAME, None)
    inference_params = self._get_call_params_with_defaults(method, *args, **kwargs)
    # Get the instance from the self parameter in wrapped_inference_method
    instance = kwargs.get("self")

    if instance is not None:
        inference_params["__instance__"] = instance

    jsonable_value = self.parse_func_arguments(**inference_params)
    if not is_jsonable(jsonable_value):
        raise NotJsonableError(jsonable_value)

    # Extract seeds and add to metadata
    seed_metadata = self._extract_seeds(**inference_params)

    if conditioning_metadata:
        if not is_jsonable(conditioning_metadata):
            raise NotJsonableError(
                f"Conditioning metadata not jsonable: {CONDITIONING_METADATA_FIELDNAME} = {conditioning_metadata}"
            )
        if isinstance(jsonable_value, dict):
            jsonable_value[CONDITIONING_METADATA_FIELDNAME] = conditioning_metadata
        else:
            jsonable_value = {"content": jsonable_value, "metadata": conditioning_metadata}  # type: ignore

    conditioning_hash = get_hash_from_jsonable(jsonable_value)

    return Conditioning(
        value=jsonable_value,
        hash=conditioning_hash,
        value_metadata={
            "seeds": seed_metadata if seed_metadata else None,
        },
    )

parse_func_arguments abstractmethod

parse_func_arguments(*args, **kwargs) -> Jsonable

Core function to be overwritten in subclasses.

Parse func arguments and convert into a jsonable object - the parsing and conversion approach may vary depending on the type of func arguments.

PARAMETER	DESCRIPTION
*args	Arguments of the method. DEFAULT: ()
**kwargs	Keyword arguments of the method. DEFAULT: {}

RETURNS	DESCRIPTION
Jsonable	Parsed parameters that can be serialized to json.

Source code in src/genai_monitor/structures/conditioning_parsers/base.py

@abstractmethod
def parse_func_arguments(self, *args, **kwargs) -> Jsonable:
    """Core function to be overwritten in subclasses.

    Parse func arguments and convert into a jsonable object - the parsing and conversion approach may vary depending
    on the type of func arguments.

    Args:
        *args: Arguments of the method.
        **kwargs: Keyword arguments of the method.

    Returns:
        Parsed parameters that can be serialized to json.
    """

Output Parser

genai_monitor.structures.output_parsers.base.BaseModelOutputParser

Bases: ABC, Generic[T]

Abstract class for converting between sample and model output types.

model_output_to_bytes abstractmethod

model_output_to_bytes(model_output: T) -> bytes

Converts the model output to a byte representation.

PARAMETER	DESCRIPTION
model_output	The model output to convert. TYPE: T

RETURNS	DESCRIPTION
bytes	The byte representation of the model output.

Source code in src/genai_monitor/structures/output_parsers/base.py

@abstractmethod
def model_output_to_bytes(self, model_output: T) -> bytes:
    """Converts the model output to a byte representation.

    Parameters:
        model_output: The model output to convert.

    Returns:
        The byte representation of the model output.
    """

bytes_to_model_output abstractmethod

bytes_to_model_output(databytes: bytes) -> T

Converts a byte representation back into model output.

PARAMETER	DESCRIPTION
databytes	The byte representation of the model output. TYPE: bytes

RETURNS	DESCRIPTION
T	The model output reconstructed from the byte representation.

Source code in src/genai_monitor/structures/output_parsers/base.py

@abstractmethod
def bytes_to_model_output(self, databytes: bytes) -> T:
    """Converts a byte representation back into model output.

    Parameters:
        databytes: The byte representation of the model output.

    Returns:
        The model output reconstructed from the byte representation.
    """

get_model_output_hash

get_model_output_hash(data: T) -> str

Calculates the hash value of the given data.

PARAMETER	DESCRIPTION
data	The data to calculate the hash value for. TYPE: T

RETURNS	DESCRIPTION
str	The hash value of the data.

Source code in src/genai_monitor/structures/output_parsers/base.py

def get_model_output_hash(self, data: T) -> str:
    """Calculates the hash value of the given data.

    Parameters:
        data: The data to calculate the hash value for.

    Returns:
        The hash value of the data.
    """
    try:
        return self.get_base_type_hash(model_output=data)
    except NotImplementedError:
        logger.info(
            "Conversion of model output to base type not implemented, falling back to calculation of the full hash."
        )
    data_bytes = self.model_output_to_bytes(data)
    return hashlib.sha256(data_bytes).hexdigest()

get_base_type_hash

get_base_type_hash(model_output: T) -> str

Calculate the hash of the base data type stored in the model output.

PARAMETER	DESCRIPTION
model_output	The model outuput to extract the base type from. TYPE: T

RETURNS	DESCRIPTION
str	The data in model output as one of the base supported types.

Source code in src/genai_monitor/structures/output_parsers/base.py

def get_base_type_hash(self, model_output: T) -> str:
    """Calculate the hash of the base data type stored in the model output.

    Args:
        model_output: The model outuput to extract the base type from.

    Returns:
        The data in model output as one of the base supported types.
    """
    base_type_data = self.model_output_to_base_type(model_output=model_output)
    return hash_base_type(data=base_type_data)

model_output_to_base_type

model_output_to_base_type(model_output: T) -> BaseType

Get the base type to calculate the hash upon.

PARAMETER	DESCRIPTION
model_output	The output of the model to extract the data from TYPE: T

RETURNS	DESCRIPTION
BaseType	One of the base data types supported in the system

Source code in src/genai_monitor/structures/output_parsers/base.py

def model_output_to_base_type(self, model_output: T) -> BaseType:
    """Get the base type to calculate the hash upon.

    Args:
        model_output: The output of the model to extract the data from

    Returns:
        One of the base data types supported in the system
    """
    raise NotImplementedError("Method to extract base types from model output not specified.")

get_sample_from_model_output

get_sample_from_model_output(model_output: T) -> Sample

Converts the model output to a sample.

PARAMETER	DESCRIPTION
model_output	The model output to convert to a sample. TYPE: T

RETURNS	DESCRIPTION
Sample	The sample created from the model output.

Source code in src/genai_monitor/structures/output_parsers/base.py

def get_sample_from_model_output(self, model_output: T) -> Sample:
    """Converts the model output to a sample.

    Parameters:
        model_output: The model output to convert to a sample.

    Returns:
        The sample created from the model output.
    """
    data_bytes = self.model_output_to_bytes(model_output)
    model_output_hash = self.get_model_output_hash(model_output)

    return Sample(data=data_bytes, hash=model_output_hash)

get_model_output_from_sample

get_model_output_from_sample(sample: Sample) -> T

Converts the sample to model output.

PARAMETER	DESCRIPTION
sample	The sample to convert to model output. TYPE: Sample

RETURNS	DESCRIPTION
T	The model output created from the sample.

RAISES	DESCRIPTION
ValueError	if the data of a sample is empty

Source code in src/genai_monitor/structures/output_parsers/base.py

def get_model_output_from_sample(self, sample: Sample) -> T:
    """Converts the sample to model output.

    Parameters:
        sample: The sample to convert to model output.

    Returns:
        The model output created from the sample.

    Raises:
        ValueError: if the data of a sample is empty
    """
    if sample.data is None:
        raise ValueError("Sample data is empty.")

    return self.bytes_to_model_output(sample.data)

Persistency Manager

genai_monitor.structures.persistency_manager.PersistencyManager

PersistencyManager(path: Union[str, Path], enabled: bool)

Manager for saving and loading model outputs in binary format to disk.

Source code in src/genai_monitor/structures/persistency_manager.py

def __init__(self, path: Union[str, Path], enabled: bool):  # noqa: D107, ANN204
    self.enabled = enabled
    path = path if path is not None else DEFAULT_PERSISTENCY_PATH

    if self.enabled:
        self.path = Path(path) if isinstance(path, str) else path
        self.path.mkdir(parents=True, exist_ok=True)

    self._configured = True
    status = "enabled" if self.enabled else "disabled"
    logger.info(f"PersistencyManager configured in {status} mode.{f' Path: {self.path}' if self.enabled else ''}")

path instance-attribute

path: Path

enabled class-attribute instance-attribute

enabled: bool = enabled

configure

configure(config: PersistencyManagerConfig)

Configures the persistency manager.

PARAMETER	DESCRIPTION
config	The configuration object specifying the parameters of PersistencyManager. TYPE: PersistencyManagerConfig

RAISES	DESCRIPTION
ValueError	If the configuration file cannot be loaded.

Source code in src/genai_monitor/structures/persistency_manager.py

def configure(self, config: PersistencyManagerConfig):
    """Configures the persistency manager.

    Args:
        config: The configuration object specifying the parameters of PersistencyManager.

    Raises:
        ValueError: If the configuration file cannot be loaded.
    """
    self.enabled = config.enabled
    path = config.path if config.path is not None else DEFAULT_PERSISTENCY_PATH

    if self.enabled:
        self.path = Path(path) if isinstance(path, str) else path
        self.path.mkdir(parents=True, exist_ok=True)

    self._configured = True
    logger.info(
        f"PersistencyManager configured in {'enabled' if self.enabled else 'disabled'} mode with path: {self.path}."
    )

save_sample

save_sample(sample: Sample)

Saves the sample data to a binary file.

PARAMETER	DESCRIPTION
sample	The sample to save. TYPE: Sample

RAISES	DESCRIPTION
ValueError	If the sample data is None.
ValueError	If the sample ID is None.
ValueError	If the PersistencyManager is disabled.

Source code in src/genai_monitor/structures/persistency_manager.py

def save_sample(self, sample: Sample):
    """Saves the sample data to a binary file.

    Args:
        sample: The sample to save.

    Raises:
        ValueError: If the sample data is None.
        ValueError: If the sample ID is None.
        ValueError: If the PersistencyManager is disabled.
    """
    if not self.is_configured():
        raise ValueError("PersistencyManager has not been configured.")

    if self.enabled:
        if sample.data is None:
            raise ValueError("Sample data is None.")

        if sample.id is None:
            raise ValueError("Sample has to be written to the database before saving.")

        self.save_bytes_to_disk(sample.data, f"samples/{sample.id}.bin")
        logger.success(f"Saved sample #{sample.id} to disk.")

load_sample

load_sample(sample: Sample) -> bytes

Loads the binary file for the given sample.

PARAMETER	DESCRIPTION
sample	The sample to load. TYPE: Sample

RETURNS	DESCRIPTION
bytes	The binary data from the file.

RAISES	DESCRIPTION
ValueError	If the PersistencyManager is disabled.
ValueError	If the sample ID is None.
FileNotFoundError	If the binary file for the sample ID is not found.
ValueError	If the PersistencyManager has not been configured.

Source code in src/genai_monitor/structures/persistency_manager.py

def load_sample(self, sample: Sample) -> bytes:
    """Loads the binary file for the given sample.

    Args:
        sample: The sample to load.

    Returns:
        The binary data from the file.

    Raises:
        ValueError: If the PersistencyManager is disabled.
        ValueError: If the sample ID is None.
        FileNotFoundError: If the binary file for the sample ID is not found.
        ValueError: If the PersistencyManager has not been configured.
    """
    if not self.is_configured():
        raise ValueError("PersistencyManager has not been configured.")

    if not self.enabled:
        raise ValueError("Cannot load binary file while PersistencyManager is disabled.")

    if sample.id is None:
        raise ValueError("Sample id is None.")

    bytesdata = self.load_bytes_from_disk(f"samples/{sample.id}.bin")
    logger.success(f"Loaded sample #{sample.id} from disk.")
    return bytesdata

save_bytes_to_disk

save_bytes_to_disk(data: bytes, filename: str)

Saves the bytes to a binary file.

PARAMETER	DESCRIPTION
data	The bytes to save. TYPE: bytes
filename	The name of the file to save the bytes to. TYPE: str

RAISES	DESCRIPTION
ValueError	If the PersistencyManager is disabled.

Source code in src/genai_monitor/structures/persistency_manager.py

def save_bytes_to_disk(self, data: bytes, filename: str):
    """Saves the bytes to a binary file.

    Args:
        data: The bytes to save.
        filename: The name of the file to save the bytes to.

    Raises:
        ValueError: If the PersistencyManager is disabled.
    """
    if not self.is_configured():
        raise ValueError("PersistencyManager has not been configured.")

    if self.enabled:
        filepath = self.path / filename
        filepath.parent.mkdir(parents=True, exist_ok=True)
        with open(filepath, "wb") as file:
            file.write(data)

load_bytes_from_disk

load_bytes_from_disk(filename: str) -> bytes

Loads the bytes from a binary file.

PARAMETER	DESCRIPTION
filename	The name of the file to load the bytes from. TYPE: str

RETURNS	DESCRIPTION
bytes	The bytes from the file.

RAISES	DESCRIPTION
FileNotFoundError	If the file is not found.

Source code in src/genai_monitor/structures/persistency_manager.py

def load_bytes_from_disk(self, filename: str) -> bytes:
    """Loads the bytes from a binary file.

    Args:
        filename: The name of the file to load the bytes from.

    Returns:
        The bytes from the file.

    Raises:
        FileNotFoundError: If the file is not found.
    """
    with open(self.path / filename, "rb") as file:
        bytesdata = file.read()
    return bytesdata

is_configured

is_configured() -> bool

Checks if the PersistencyManager has been configured.

RETURNS	DESCRIPTION
bool	Boolean indicating if the PersistencyManager has been configured.

Source code in src/genai_monitor/structures/persistency_manager.py

def is_configured(self) -> bool:
    """Checks if the PersistencyManager has been configured.

    Returns:
        Boolean indicating if the PersistencyManager has been configured.
    """
    return self._configured

save_artifact

save_artifact(artifact: Artifact)

Saves the artifact data to a binary file.

PARAMETER	DESCRIPTION
artifact	The artifact to save. TYPE: Artifact

RAISES	DESCRIPTION
ValueError	If the artifact ID is None.
ValueError	If the PersistencyManager is disabled.

Source code in src/genai_monitor/structures/persistency_manager.py

def save_artifact(self, artifact: Artifact):
    """Saves the artifact data to a binary file.

    Args:
        artifact: The artifact to save.

    Raises:
        ValueError: If the artifact ID is None.
        ValueError: If the PersistencyManager is disabled.
    """
    if not self.is_configured():
        raise ValueError("PersistencyManager has not been configured.")

    if self.enabled:
        if artifact.id is None:
            raise ValueError("Artifact has to be written to the database before saving.")

        self.save_bytes_to_disk(json.dumps(artifact.value).encode("utf-8"), f"artifacts/{artifact.id}.bin")
        logger.success(f"Saved artifact #{artifact.id} to disk.")

load_artifact

load_artifact(artifact: Artifact) -> bytes

Loads the binary file for the given artifact.

PARAMETER	DESCRIPTION
artifact	The artifact to load. TYPE: Artifact

RETURNS	DESCRIPTION
bytes	The binary data from the file.

RAISES	DESCRIPTION
ValueError	If the PersistencyManager is disabled.
ValueError	If the artifact ID is None.
FileNotFoundError	If the binary file for the artifact ID is not found.
ValueError	If the PersistencyManager has not been configured.

Source code in src/genai_monitor/structures/persistency_manager.py

def load_artifact(self, artifact: Artifact) -> bytes:
    """Loads the binary file for the given artifact.

    Args:
        artifact: The artifact to load.

    Returns:
        The binary data from the file.

    Raises:
        ValueError: If the PersistencyManager is disabled.
        ValueError: If the artifact ID is None.
        FileNotFoundError: If the binary file for the artifact ID is not found.
        ValueError: If the PersistencyManager has not been configured.
    """
    if not self.is_configured():
        raise ValueError("PersistencyManager has not been configured.")

    if not self.enabled:
        raise ValueError("Cannot load binary file while PersistencyManager is disabled.")

    if artifact.id is None:
        raise ValueError("Artifact id is None.")

    bytesdata = self.load_bytes_from_disk(f"artifacts/{artifact.id}.bin")
    logger.success(f"Loaded artifact #{artifact.id} from disk.")
    return json.loads(bytesdata.decode("utf-8"))

save_conditioning

save_conditioning(conditioning: Conditioning)

Saves the conditioning data to a binary file.

PARAMETER	DESCRIPTION
conditioning	The conditioning to save. TYPE: Conditioning

RAISES	DESCRIPTION
ValueError	If the conditioning ID is None.
ValueError	If the PersistencyManager is disabled.

Source code in src/genai_monitor/structures/persistency_manager.py

def save_conditioning(self, conditioning: Conditioning):
    """Saves the conditioning data to a binary file.

    Args:
        conditioning: The conditioning to save.

    Raises:
        ValueError: If the conditioning ID is None.
        ValueError: If the PersistencyManager is disabled.
    """
    if not self.is_configured():
        raise ValueError("PersistencyManager has not been configured.")

    if self.enabled:
        if conditioning.id is None:
            raise ValueError("Conditioning has to be written to the database before saving.")

        self.save_bytes_to_disk(
            json.dumps(conditioning.value).encode("utf-8"), f"conditionings/{conditioning.id}.bin"
        )
        logger.success(f"Saved conditioning #{conditioning.id} to disk.")

load_conditioning

load_conditioning(conditioning: Conditioning) -> Dict

Loads the binary file for the given conditioning.

PARAMETER	DESCRIPTION
conditioning	The conditioning to load. TYPE: Conditioning

RETURNS	DESCRIPTION
Dict	The binary data from the file.

RAISES	DESCRIPTION
ValueError	If the PersistencyManager is disabled.
ValueError	If the conditioning ID is None.
FileNotFoundError	If the binary file for the conditioning ID is not found.
ValueError	If the PersistencyManager has not been configured.

Source code in src/genai_monitor/structures/persistency_manager.py

def load_conditioning(self, conditioning: Conditioning) -> Dict:
    """Loads the binary file for the given conditioning.

    Args:
        conditioning: The conditioning to load.

    Returns:
        The binary data from the file.

    Raises:
        ValueError: If the PersistencyManager is disabled.
        ValueError: If the conditioning ID is None.
        FileNotFoundError: If the binary file for the conditioning ID is not found.
        ValueError: If the PersistencyManager has not been configured.
    """
    if not self.is_configured():
        raise ValueError("PersistencyManager has not been configured.")

    if not self.enabled:
        raise ValueError("Cannot load binary file while PersistencyManager is disabled.")

    if conditioning.id is None:
        raise ValueError("Conditioning id is None.")

    bytesdata = self.load_bytes_from_disk(f"conditionings/{conditioning.id}.bin")
    logger.success(f"Loaded conditioning #{conditioning.id} from disk.")
    return json.loads(bytesdata.decode("utf-8"))

Runtime Manager

genai_monitor.structures.runtime_manager.RuntimeManager

Class for managing runtime data.

ATTRIBUTE	DESCRIPTION
user_id	The user ID. TYPE: Optional[int]
version	The runtime version identifier. TYPE: Optional[str]
latest_sample	The most recently created sample. TYPE: Optional[Sample]
artifacts_for_next_sample	Collection of artifacts to be associated with the next sample. TYPE: Deque[Artifact]

user_id class-attribute instance-attribute

user_id: Optional[int] = None

version class-attribute instance-attribute

version: Optional[str] = None

latest_sample class-attribute instance-attribute

latest_sample: Optional[Sample] = None

artifacts_for_next_sample class-attribute instance-attribute

artifacts_for_next_sample: Deque[Artifact] = field(
    factory=deque
)

set_user_id

set_user_id(user_id: int) -> None

Set the user ID.

PARAMETER	DESCRIPTION
user_id	The user ID. TYPE: int

Source code in src/genai_monitor/structures/runtime_manager.py

def set_user_id(self, user_id: int) -> None:
    """Set the user ID.

    Args:
        user_id: The user ID.
    """
    self.user_id = user_id

set_runtime_version

set_runtime_version(version: str) -> None

Set the version.

PARAMETER	DESCRIPTION
version	The version. TYPE: str

Source code in src/genai_monitor/structures/runtime_manager.py

def set_runtime_version(self, version: str) -> None:
    """Set the version.

    Args:
        version: The version.
    """
    self.version = version