Dependency injection

Dependency Container

genai_monitor.injectors.containers.DependencyContainer

Bases: DeclarativeContainer

Dependency container for the application.

config class-attribute instance-attribute

config = Configuration()

wrapper_factory class-attribute instance-attribute

wrapper_factory = Singleton(provides=WrapperFactory)

wrapper_registry class-attribute instance-attribute

wrapper_registry = Singleton(
    provides=WrapperRegistry,
    wrapper_factory=wrapper_factory,
)

session_manager class-attribute instance-attribute

session_manager = Factory(
    provides=SessionManager, database_url=url
)

db_manager class-attribute instance-attribute

db_manager = Factory(
    provides=DBManager, session_manager=session_manager
)

persistency_manager class-attribute instance-attribute

persistency_manager = Singleton(
    provides=PersistencyManager, path=path, enabled=enabled
)

runtime_manager class-attribute instance-attribute

runtime_manager = Singleton(provides=RuntimeManager)

artifact_wrapper_factory class-attribute instance-attribute

artifact_wrapper_factory = Singleton(
    provides=ArtifactWrapperFactory
)

artifact_registry class-attribute instance-attribute

artifact_registry = Singleton(
    provides=ArtifactRegistry,
    artifact_wrapper_factory=artifact_wrapper_factory,
)

default_hashing_function class-attribute instance-attribute

default_hashing_function = Object(
    provides=default_model_hashing_function
)

transformers_output_parser class-attribute instance-attribute

transformers_output_parser = Factory(
    provides=TransformersTextGenerationParser
)

transformers_conditioning_parser class-attribute instance-attribute

transformers_conditioning_parser = Factory(
    TransformersTextGenerationConditioningParser
)

transformers_hashing_function class-attribute instance-attribute

transformers_hashing_function = Object(
    provides=get_transformers_model_hash
)

transformers_json_path class-attribute instance-attribute

transformers_json_path = Factory(
    get_absolute_path,
    _TRANSFORMERS_JSON_PATH,
    relative_to=abspath(__file__),
)

transformers_class_def_collection class-attribute instance-attribute

transformers_class_def_collection = Factory(
    from_json, json_path=transformers_json_path
)

register_transformers class-attribute instance-attribute

register_transformers = Callable(
    register_class_collection,
    definition_collection=transformers_class_def_collection,
    registry=wrapper_registry,
    db_manager=db_manager,
    persistency_manager=persistency_manager,
    runtime_manager=runtime_manager,
    output_parser=transformers_output_parser,
    conditioning_parser=transformers_conditioning_parser,
    hashing_function=transformers_hashing_function,
)

openai_output_parser class-attribute instance-attribute

openai_output_parser = Factory(
    provides=OpenAIChatOutputParser
)

openai_conditioning_parser class-attribute instance-attribute

openai_conditioning_parser = Factory(
    OpenAIConditioningParser
)

empty_model_hash class-attribute instance-attribute

empty_model_hash = Object(get_empty_model_hash)

providers_json_path class-attribute instance-attribute

providers_json_path = Factory(
    get_absolute_path,
    _PROVIDERS_JSON_PATH,
    relative_to=abspath(__file__),
)

providers_class_def_collection class-attribute instance-attribute

providers_class_def_collection = Factory(
    from_json, json_path=providers_json_path
)

register_providers class-attribute instance-attribute

register_providers = Callable(
    register_class_collection,
    definition_collection=providers_class_def_collection,
    registry=wrapper_registry,
    db_manager=db_manager,
    persistency_manager=persistency_manager,
    runtime_manager=runtime_manager,
    output_parser=openai_output_parser,
    conditioning_parser=openai_conditioning_parser,
    hashing_function=empty_model_hash,
)

diffusers_output_parser class-attribute instance-attribute

diffusers_output_parser = Factory(
    provides=StableDiffusionOutputParser
)

diffusers_conditioning_parser class-attribute instance-attribute

diffusers_conditioning_parser = Factory(
    StableDiffusionConditioningParser
)

diffusers_hashing_function class-attribute instance-attribute

diffusers_hashing_function = Object(
    provides=get_diffusers_model_hash
)

diffusers_json_path class-attribute instance-attribute

diffusers_json_path = Factory(
    get_absolute_path,
    _DIFFUSERS_JSON_PATH,
    relative_to=abspath(__file__),
)

diffusers_class_def_collection class-attribute instance-attribute

diffusers_class_def_collection = Factory(
    from_json, json_path=diffusers_json_path
)

register_diffusers class-attribute instance-attribute

register_diffusers = Callable(
    register_class_collection,
    definition_collection=diffusers_class_def_collection,
    registry=wrapper_registry,
    db_manager=db_manager,
    persistency_manager=persistency_manager,
    runtime_manager=runtime_manager,
    output_parser=diffusers_output_parser,
    conditioning_parser=diffusers_conditioning_parser,
    hashing_function=diffusers_hashing_function,
)

litellm_output_parser class-attribute instance-attribute

litellm_output_parser = Factory(
    provides=LiteLLMCompletionOutputParser
)

litellm_conditioning_parser class-attribute instance-attribute

litellm_conditioning_parser = Factory(
    provides=LiteLLMCompletionConditioningParser
)

litellm_json_path class-attribute instance-attribute

litellm_json_path = Factory(
    get_absolute_path,
    _LITELLM_JSON_PATH,
    relative_to=abspath(__file__),
)

litellm_def_collection class-attribute instance-attribute

litellm_def_collection = Factory(
    from_json, json_path=litellm_json_path
)

register_litellm class-attribute instance-attribute

register_litellm = Callable(
    register_function_collection,
    definition_collection=litellm_def_collection,
    registry=wrapper_registry,
    db_manager=db_manager,
    persistency_manager=persistency_manager,
    runtime_manager=runtime_manager,
    output_parser=litellm_output_parser,
    conditioning_parser=litellm_conditioning_parser,
    hashing_function=get_function_full_path,
)

genai_monitor.injectors.containers.get_container

get_container() -> DependencyContainer

Get the dependency container.

RETURNS	DESCRIPTION
DependencyContainer	The dependency container.

Source code in src/genai_monitor/injectors/containers.py

def get_container() -> DependencyContainer:
    """Get the dependency container.

    Returns:
        The dependency container.
    """
    global _container  # noqa: PLW0603
    if _container is None:
        _container = DependencyContainer()
    return _container

Registry

genai_monitor.injectors.registry.WrapperRegistry

Registry for function wrappers.

This class is responsible for registering and unregistering wrappers.

ATTRIBUTE	DESCRIPTION
_wrapper_factory	Factory for creating wrappers. TYPE: WrapperFactory
_registry	Dictionary of registered wrappers. TYPE: Dict[str, Union[FunctionWrapper, MethodWrapper]]

register

register(
    func: Callable,
    db_manager: DBManager,
    persistency_manager: PersistencyManager,
    runtime_manager: RuntimeManager,
    output_parser: BaseModelOutputParser,
    conditioning_parser: BaseConditioningParser,
    hashing_function: Callable[[Any], str],
    max_unique_instances: int = 1,
)

Register a function with the registry.

PARAMETER	DESCRIPTION
func	The function or method to register. TYPE: Callable
db_manager	The database manager. TYPE: DBManager
persistency_manager	The persistency manager. TYPE: PersistencyManager
runtime_manager	The runtime manager. TYPE: RuntimeManager
output_parser	The output parser. TYPE: BaseModelOutputParser
conditioning_parser	The conditioning parser. TYPE: BaseConditioningParser
hashing_function	The hashing function. TYPE: Callable[[Any], str]
max_unique_instances	The maximum number of unique sample instances for each conditioning. TYPE: int DEFAULT: 1

Source code in src/genai_monitor/injectors/registry.py

def register(  # noqa: PLR0913
    self,
    func: Callable,
    db_manager: DBManager,
    persistency_manager: PersistencyManager,
    runtime_manager: RuntimeManager,
    output_parser: BaseModelOutputParser,
    conditioning_parser: BaseConditioningParser,
    hashing_function: Callable[[Any], str],
    max_unique_instances: int = 1,
):
    """Register a function with the registry.

    Args:
        func: The function or method to register.
        db_manager: The database manager.
        persistency_manager: The persistency manager.
        runtime_manager: The runtime manager.
        output_parser: The output parser.
        conditioning_parser: The conditioning parser.
        hashing_function: The hashing function.
        max_unique_instances: The maximum number of unique sample instances for each conditioning.
    """
    func_name = func.__qualname__

    if func_name in self._registry:
        logger.warning(f"Overriding function {func_name}, which is already registered.")

    wrapper = self._wrapper_factory.create(
        func=func,
        db_manager=db_manager,
        persistency_manager=persistency_manager,
        runtime_manager=runtime_manager,
        output_parser=output_parser,
        conditioning_parser=conditioning_parser,
        hashing_function=hashing_function,
        max_unique_instances=max_unique_instances,
    )
    self._registry[func_name] = wrapper
    func_wrapped = wrapper.wrap(func=func)
    override_func_in_module(func=func, func_override=func_wrapped)
    override_func_in_imported_modules(func=func, func_override=func_wrapped)

unregister

unregister(func: Callable)

Unregister a function from the registry.

PARAMETER	DESCRIPTION
func	The function to unregister. TYPE: Callable

Source code in src/genai_monitor/injectors/registry.py

def unregister(self, func: Callable):
    """Unregister a function from the registry.

    Args:
        func: The function to unregister.
    """
    func_name = func.__qualname__
    wrapper = self._registry.pop(func_name, None)
    if wrapper is None:
        logger.warning(f"Function {func_name} can't be unregistered as it doesn't exist in the registry.")
        return
    override_func_in_module(func=func, func_override=wrapper.func)
    override_func_in_imported_modules(func=func, func_override=wrapper.func)

get_registered_list

get_registered_list() -> List[str]

Get a list of registered functions.

RETURNS	DESCRIPTION
List[str]	A list of registered functions.

Source code in src/genai_monitor/injectors/registry.py

def get_registered_list(self) -> List[str]:
    """Get a list of registered functions.

    Returns:
        A list of registered functions.
    """
    return list(self._registry.keys())

genai_monitor.injectors.registry.ArtifactRegistry

Artifact registry.

ATTRIBUTE	DESCRIPTION
_artifact_wrapper_factory	Factory for creating artifact wrappers. TYPE: ArtifactWrapperFactory
_registry	Dictionary of registered artifacts (both forward and backward tracking). TYPE: Dict[str, Dict[str, Union[ArtifactFunctionWrapper, ArtifactMethodWrapper]]]

register

register(
    func: Callable,
    direction: str,
    db_manager: DBManager,
    persistency_manager: PersistencyManager,
    runtime_manager: RuntimeManager,
)

Registers an artifact function or method.

PARAMETER	DESCRIPTION
func	The function or method to register. TYPE: Callable
direction	The direction of the artifact (either 'forward' or 'backward'). TYPE: str
db_manager	The database manager. TYPE: DBManager
persistency_manager	The persistency manager. TYPE: PersistencyManager
runtime_manager	The runtime manager. TYPE: RuntimeManager

RAISES	DESCRIPTION
ValueError	If the direction is not 'forward' or 'backward'.

Source code in src/genai_monitor/injectors/registry.py

def register(
    self,
    func: Callable,
    direction: str,
    db_manager: DBManager,
    persistency_manager: PersistencyManager,
    runtime_manager: RuntimeManager,
):
    """Registers an artifact function or method.

    Args:
        func: The function or method to register.
        direction: The direction of the artifact (either 'forward' or 'backward').
        db_manager: The database manager.
        persistency_manager: The persistency manager.
        runtime_manager: The runtime manager.

    Raises:
        ValueError: If the direction is not 'forward' or 'backward'.
    """
    func_name = func.__qualname__

    if func_name in self._registry[direction]:
        logger.warning(f"Overriding artifact {func_name}, which is already registered.")

    wrapper = self._artifact_wrapper_factory.create(
        func=func, db_manager=db_manager, persistency_manager=persistency_manager, runtime_manager=runtime_manager
    )
    if direction == "backward":
        func_wrapped = wrapper.wrap_backward(func=func)

    elif direction == "forward":
        func_wrapped = wrapper.wrap_forward(func=func)

    else:
        raise ValueError(f"Invalid direction: {direction} (must be 'forward' or 'backward').")

    self._registry[direction][func_name] = wrapper

    override_func_in_module(func=func, func_override=func_wrapped)
    override_func_in_imported_modules(func=func, func_override=func_wrapped)

unregister

unregister(func: Callable, direction: str)

Removes an artifact function or method from the registry.

PARAMETER	DESCRIPTION
func	The function or method to unregister. TYPE: Callable
direction	The direction of the artifact (either 'forward' or 'backward'). TYPE: str

Source code in src/genai_monitor/injectors/registry.py

def unregister(self, func: Callable, direction: str):
    """Removes an artifact function or method from the registry.

    Args:
        func: The function or method to unregister.
        direction: The direction of the artifact (either 'forward' or 'backward').
    """
    func_name = func.__qualname__
    wrapper = self._registry[direction].pop(func_name, None)

    if wrapper is None:
        logger.warning(f"Artifact {func_name} can't be unregistered as it doesn't exist in the registry.")
        return
    override_func_in_module(func=func, func_override=wrapper.func)
    override_func_in_imported_modules(func=func, func_override=wrapper.func)

Wrappers

Inference

genai_monitor.injectors.wrapper.MethodWrapper

Bases: Wrapper

Wrapper for class methods.

db_manager instance-attribute

db_manager: DBManager

persistency_manager instance-attribute

persistency_manager: PersistencyManager

runtime_manager instance-attribute

runtime_manager: RuntimeManager

output_parser instance-attribute

output_parser: BaseModelOutputParser

conditioning_parser instance-attribute

conditioning_parser: BaseConditioningParser

max_unique_instances instance-attribute

max_unique_instances: int

hashing_function instance-attribute

hashing_function: Callable[[Any], str]

func class-attribute instance-attribute

func: Callable = field(init=False)

wrap

wrap(func: Callable) -> Callable

Source code in src/genai_monitor/injectors/wrapper.py

def wrap(self, func: Callable) -> Callable:
    self.func = deepcopy(func)

    @wraps(func)
    def wrapped(obj_self, *args, **kwargs) -> Any:
        model_hash = self.hashing_function(obj_self)
        model_cls_name = obj_self.__class__.__name__
        conditioning = self.conditioning_parser.parse_conditioning(func, *args, **kwargs)
        kwargs.pop(CONDITIONING_METADATA_FIELDNAME, None)

        conditioning = self._resolve_conditioning(conditioning)
        generator, existing_generations = self._get_generations(model_hash, conditioning, model_cls_name)

        max_unique_instances = generator.model_metadata.get("max_unique_instances", 1)
        next_instance = (conditioning.value_metadata.get("latest_instance", -1) + 1) % max_unique_instances

        if not self.persistency_manager.enabled:
            logger.info("PersistencyManager is disabled. Output will be generated.")
            model_output = func(obj_self, *args, **kwargs)

            if not existing_generations:
                sample = self._create_sample_placeholder(conditioning=conditioning, generator=generator)
                self._finish_sample_generation(
                    sample=sample,
                    model_output=model_output,
                    conditioning=conditioning,
                    generator=generator,
                )

            return model_output

        if len(existing_generations) >= max_unique_instances:
            logger.info(f"Max instances ({max_unique_instances}) reached. Returning existing generation")

            model_output = self._return_existing_generation(
                existing_generations=existing_generations,
                generator=generator,
                generation_id=next_instance,
            )

            self.update_generation_id(conditioning, next_instance)

            return model_output

        sample_placeholder = self._create_sample_placeholder(
            conditioning=conditioning, generator=generator, generation_id=next_instance
        )

        try:
            model_output = func(obj_self, *args, **kwargs)

        except Exception as e:
            self.db_manager.update(
                model=SampleTable,
                filters={"id": sample_placeholder.id},
                values={"status": SampleStatus.FAILED.value},
            )
            logger.error(f"Could not generate sample: {e}")
            raise e

        self._finish_sample_generation(
            sample=sample_placeholder,
            model_output=model_output,
            conditioning=conditioning,
            generator=generator,
            generation_id=next_instance,
        )
        self.update_generation_id(conditioning, next_instance)
        return model_output

    return wrapped

attach_artifacts_to_sample

attach_artifacts_to_sample(sample: Sample) -> None

Attach artifacts to a sample.

PARAMETER	DESCRIPTION
sample	The sample to attach artifacts to. TYPE: Sample

Source code in src/genai_monitor/injectors/wrapper.py

def attach_artifacts_to_sample(self, sample: Sample) -> None:
    """Attach artifacts to a sample.

    Args:
        sample: The sample to attach artifacts to.
    """
    while self.runtime_manager.artifacts_for_next_sample:
        artifact = self.runtime_manager.artifacts_for_next_sample.popleft()

        existing_artifacts = self.db_manager.search(
            ArtifactTable, {"name": artifact.name, "hash": artifact.hash, "sample_id": sample.id}
        )

        if existing_artifacts:
            logger.info(f"Artifact {artifact.name} already exists in the database.")

        else:  # noqa: PLR5501
            if self.persistency_manager.enabled:
                value = artifact.value
                artifact.value = None
                artifact.sample_id = sample.id
                artifact.hash = get_hash_from_jsonable(value)
                artifact = Artifact.from_orm(self.db_manager.save(artifact.to_orm()))
                artifact.value = value
                self.persistency_manager.save_artifact(artifact)
                logger.info(f"Saved artifact #{artifact.id} (sample #{sample.id}) to database and disk.")
            else:
                artifact = Artifact.from_orm(self.db_manager.save(artifact.to_orm()))
                logger.info(f"Saved artifact #{artifact.id} (sample #{sample.id}) to database.")

update_generation_id

update_generation_id(
    conditioning: Conditioning, generation_id: int
)

Update the generation_id of the most recently used generation.

PARAMETER	DESCRIPTION
conditioning	The conditioning. TYPE: Conditioning
generation_id	The generation id. TYPE: int

Source code in src/genai_monitor/injectors/wrapper.py

def update_generation_id(self, conditioning: Conditioning, generation_id: int):
    """Update the generation_id of the most recently used generation.

    Args:
        conditioning: The conditioning.
        generation_id: The generation id.
    """
    conditioning_metadata = conditioning.value_metadata
    conditioning_metadata["latest_instance"] = generation_id

    self.db_manager.update(
        model=ConditioningTable, filters={"id": conditioning.id}, values={"value_metadata": conditioning_metadata}
    )

genai_monitor.injectors.wrapper.FunctionWrapper

Bases: Wrapper

Wrapper for functions.

db_manager instance-attribute

db_manager: DBManager

persistency_manager instance-attribute

persistency_manager: PersistencyManager

runtime_manager instance-attribute

runtime_manager: RuntimeManager

output_parser instance-attribute

output_parser: BaseModelOutputParser

conditioning_parser instance-attribute

conditioning_parser: BaseConditioningParser

max_unique_instances instance-attribute

max_unique_instances: int

hashing_function instance-attribute

hashing_function: Callable[[Any], str]

func class-attribute instance-attribute

func: Callable = field(init=False)

wrap

wrap(func: Callable) -> Callable

Wraps the function.

PARAMETER	DESCRIPTION
func	The function to wrap. TYPE: Callable

RETURNS	DESCRIPTION
Callable	The wrapped function. TYPE: Callable

Source code in src/genai_monitor/injectors/wrapper.py

def wrap(self, func: Callable) -> Callable:
    """Wraps the function.

    Args:
        func: The function to wrap.

    Returns:
        Callable: The wrapped function.
    """
    self.func = deepcopy(func)

    @wraps(func)
    def wrapped(*args, **kwargs) -> Any:
        function_hash = self.hashing_function(func)
        function_name = func.__name__

        conditioning = self.conditioning_parser.parse_conditioning(func, *args, **kwargs)
        kwargs.pop(CONDITIONING_METADATA_FIELDNAME, None)

        conditioning = self._resolve_conditioning(conditioning)
        generator, existing_generations = self._get_generations(function_hash, conditioning, function_name)

        max_unique_instances = generator.model_metadata.get("max_unique_instances", 1)
        next_instance = (conditioning.value_metadata.get("latest_instance", -1) + 1) % max_unique_instances

        if not self.persistency_manager.enabled:
            logger.info("PersistencyManager is disabled. Output will be generated.")
            model_output = func(*args, **kwargs)

            if not existing_generations:
                sample = self._create_sample_placeholder(conditioning=conditioning, generator=generator)
                self._finish_sample_generation(
                    sample=sample,
                    model_output=model_output,
                    conditioning=conditioning,
                    generator=generator,
                )

            return model_output

        if len(existing_generations) >= max_unique_instances:
            logger.info(f"Max instances ({max_unique_instances}) reached. Returning existing generation.")
            try:
                model_output = self._return_existing_generation(
                    existing_generations=existing_generations,
                    generator=generator,
                    generation_id=next_instance,
                )

                self.update_generation_id(conditioning, next_instance)

                return model_output

            except Exception as e:
                logger.error(f"Could not return existing generation: {e}")
                logger.info("Generating new sample without sample creation.")
                return func(*args, **kwargs)

        sample_placeholder = self._create_sample_placeholder(
            conditioning=conditioning, generator=generator, generation_id=next_instance
        )

        try:
            model_output = func(*args, **kwargs)

        except Exception as e:
            self.db_manager.update(
                model=SampleTable,
                filters={"id": sample_placeholder.id},
                values={"status": SampleStatus.FAILED.value},
            )
            logger.error(f"Could not generate sample: {e}")
            raise e

        self._finish_sample_generation(
            sample=sample_placeholder,
            model_output=model_output,
            conditioning=conditioning,
            generator=generator,
            generation_id=next_instance,
        )
        self.update_generation_id(conditioning, next_instance)
        return model_output

    return wrapped

attach_artifacts_to_sample

attach_artifacts_to_sample(sample: Sample) -> None

Attach artifacts to a sample.

PARAMETER	DESCRIPTION
sample	The sample to attach artifacts to. TYPE: Sample

Source code in src/genai_monitor/injectors/wrapper.py

def attach_artifacts_to_sample(self, sample: Sample) -> None:
    """Attach artifacts to a sample.

    Args:
        sample: The sample to attach artifacts to.
    """
    while self.runtime_manager.artifacts_for_next_sample:
        artifact = self.runtime_manager.artifacts_for_next_sample.popleft()

        existing_artifacts = self.db_manager.search(
            ArtifactTable, {"name": artifact.name, "hash": artifact.hash, "sample_id": sample.id}
        )

        if existing_artifacts:
            logger.info(f"Artifact {artifact.name} already exists in the database.")

        else:  # noqa: PLR5501
            if self.persistency_manager.enabled:
                value = artifact.value
                artifact.value = None
                artifact.sample_id = sample.id
                artifact.hash = get_hash_from_jsonable(value)
                artifact = Artifact.from_orm(self.db_manager.save(artifact.to_orm()))
                artifact.value = value
                self.persistency_manager.save_artifact(artifact)
                logger.info(f"Saved artifact #{artifact.id} (sample #{sample.id}) to database and disk.")
            else:
                artifact = Artifact.from_orm(self.db_manager.save(artifact.to_orm()))
                logger.info(f"Saved artifact #{artifact.id} (sample #{sample.id}) to database.")

update_generation_id

update_generation_id(
    conditioning: Conditioning, generation_id: int
)

Update the generation_id of the most recently used generation.

PARAMETER	DESCRIPTION
conditioning	The conditioning. TYPE: Conditioning
generation_id	The generation id. TYPE: int

Source code in src/genai_monitor/injectors/wrapper.py

def update_generation_id(self, conditioning: Conditioning, generation_id: int):
    """Update the generation_id of the most recently used generation.

    Args:
        conditioning: The conditioning.
        generation_id: The generation id.
    """
    conditioning_metadata = conditioning.value_metadata
    conditioning_metadata["latest_instance"] = generation_id

    self.db_manager.update(
        model=ConditioningTable, filters={"id": conditioning.id}, values={"value_metadata": conditioning_metadata}
    )

genai_monitor.injectors.wrapper.WrapperFactory

Factory for creating wrappers for functions and methods.

create staticmethod

create(
    func: Callable,
    db_manager: DBManager,
    persistency_manager: PersistencyManager,
    runtime_manager: RuntimeManager,
    output_parser: BaseModelOutputParser,
    conditioning_parser: BaseConditioningParser,
    hashing_function: Callable[[Any], str],
    max_unique_instances: int = 1,
) -> Union[FunctionWrapper, MethodWrapper]

Creates a wrapper for a function or method.

PARAMETER	DESCRIPTION
func	The function or method to wrap. TYPE: Callable
db_manager	The database manager. TYPE: DBManager
persistency_manager	The persistency manager. TYPE: PersistencyManager
runtime_manager	The runtime manager. TYPE: RuntimeManager
output_parser	The output parser. TYPE: BaseModelOutputParser
conditioning_parser	The conditioning parser. TYPE: BaseConditioningParser
hashing_function	The hashing function. TYPE: Callable[[Any], str]
max_unique_instances	The maximum number of unique sample instances for each conditioning. TYPE: int DEFAULT: 1

RETURNS	DESCRIPTION
Union[FunctionWrapper, MethodWrapper]	The wrapper for the function or method.

Source code in src/genai_monitor/injectors/wrapper.py

@staticmethod
def create(
    func: Callable,
    db_manager: DBManager,
    persistency_manager: PersistencyManager,
    runtime_manager: RuntimeManager,
    output_parser: BaseModelOutputParser,
    conditioning_parser: BaseConditioningParser,
    hashing_function: Callable[[Any], str],
    max_unique_instances: int = 1,
) -> Union[FunctionWrapper, MethodWrapper]:
    """Creates a wrapper for a function or method.

    Args:
        func: The function or method to wrap.
        db_manager: The database manager.
        persistency_manager: The persistency manager.
        runtime_manager: The runtime manager.
        output_parser: The output parser.
        conditioning_parser: The conditioning parser.
        hashing_function: The hashing function.
        max_unique_instances: The maximum number of unique sample instances for each conditioning.

    Returns:
        The wrapper for the function or method.
    """
    if is_class_func(func=func):
        return MethodWrapper(
            db_manager=db_manager,
            persistency_manager=persistency_manager,
            runtime_manager=runtime_manager,
            output_parser=output_parser,
            conditioning_parser=conditioning_parser,
            hashing_function=hashing_function,
            max_unique_instances=max_unique_instances,
        )

    return FunctionWrapper(
        db_manager=db_manager,
        persistency_manager=persistency_manager,
        runtime_manager=runtime_manager,
        output_parser=output_parser,
        conditioning_parser=conditioning_parser,
        hashing_function=hashing_function,
        max_unique_instances=max_unique_instances,
    )

Artifacts

genai_monitor.injectors.wrapper.ArtifactMethodWrapper

Bases: ArtifactWrapper

Wrapper for methods for artifact tracking.

db_manager instance-attribute

db_manager: DBManager

persistency_manager instance-attribute

persistency_manager: PersistencyManager

runtime_manager instance-attribute

runtime_manager: RuntimeManager

wrap_forward

wrap_forward(func: Callable) -> Callable

Wraps the function for forward artifact tracking.

PARAMETER	DESCRIPTION
func	The function to wrap. TYPE: Callable

RETURNS	DESCRIPTION
Callable	The wrapped function.

Source code in src/genai_monitor/injectors/wrapper.py

def wrap_forward(self, func: Callable) -> Callable:
    """Wraps the function for forward artifact tracking.

    Args:
        func: The function to wrap.

    Returns:
        The wrapped function.
    """
    self.func = deepcopy(func)

    @wraps(func)
    def wrapped(obj_self, *args, **kwargs) -> Any:
        output = func(*args, **kwargs)
        self._add_artifact_to_queue(output=output, name=f"{obj_self.__class__.__name__}.{func.__name__}")
        return output

    return wrapped

wrap_backward

wrap_backward(func: Callable) -> Callable

Wraps the function for backward artifact tracking.

PARAMETER	DESCRIPTION
func	The function to wrap. TYPE: Callable

RETURNS	DESCRIPTION
Callable	The wrapped function.

Source code in src/genai_monitor/injectors/wrapper.py

def wrap_backward(self, func: Callable) -> Callable:
    """Wraps the function for backward artifact tracking.

    Args:
        func: The function to wrap.

    Returns:
        The wrapped function.
    """
    self.func = deepcopy(func)

    @wraps(func)
    def wrapped(obj_self, *args, **kwargs) -> Any:
        output = func(*args, **kwargs)
        self._save_artifact(output, name=f"{obj_self.__class__.__name__}.{func.__name__}")
        return output

    return wrapped

genai_monitor.injectors.wrapper.ArtifactFunctionWrapper

Bases: ArtifactWrapper

Wrapper for functions for artifact tracking.

db_manager instance-attribute

db_manager: DBManager

persistency_manager instance-attribute

persistency_manager: PersistencyManager

runtime_manager instance-attribute

runtime_manager: RuntimeManager

wrap_forward

wrap_forward(func: Callable) -> Callable

Wraps the function for forward artifact tracking.

PARAMETER	DESCRIPTION
func	The function to wrap. TYPE: Callable

RETURNS	DESCRIPTION
Callable	The wrapped function.

Source code in src/genai_monitor/injectors/wrapper.py

def wrap_forward(self, func: Callable) -> Callable:
    """Wraps the function for forward artifact tracking.

    Args:
        func: The function to wrap.

    Returns:
        The wrapped function.
    """
    self.func = deepcopy(func)

    @wraps(func)
    def wrapped(*args, **kwargs) -> Any:
        output = func(*args, **kwargs)
        self._add_artifact_to_queue(output=output, name=f"{func.__module__}.{func.__name__}")
        return output

    return wrapped

wrap_backward

wrap_backward(func: Callable) -> Callable

Wraps the function for backward artifact tracking.

PARAMETER	DESCRIPTION
func	The function to wrap. TYPE: Callable

RETURNS	DESCRIPTION
Callable	The wrapped function.

Source code in src/genai_monitor/injectors/wrapper.py

def wrap_backward(self, func: Callable) -> Callable:
    """Wraps the function for backward artifact tracking.

    Args:
        func: The function to wrap.

    Returns:
        The wrapped function.
    """
    self.func = deepcopy(func)

    @wraps(func)
    def wrapped(*args, **kwargs) -> Any:
        output = func(*args, **kwargs)
        self._save_artifact(output, name=f"{func.__module__}.{func.__name__}")
        return output

    return wrapped

genai_monitor.injectors.wrapper.ArtifactWrapperFactory

Factory for creating wrappers for artifacts.

create

create(
    func: Callable,
    db_manager: DBManager,
    persistency_manager: PersistencyManager,
    runtime_manager: RuntimeManager,
) -> Union[ArtifactFunctionWrapper, ArtifactMethodWrapper]

Creates a wrapper for an artifact.

PARAMETER	DESCRIPTION
func	The artifact to wrap. TYPE: Callable
db_manager	The database manager. TYPE: DBManager
persistency_manager	The persistency manager. TYPE: PersistencyManager
runtime_manager	The runtime manager. TYPE: RuntimeManager

RETURNS	DESCRIPTION
Union[ArtifactFunctionWrapper, ArtifactMethodWrapper]	The wrapper for the artifact.

Source code in src/genai_monitor/injectors/wrapper.py

def create(
    self,
    func: Callable,
    db_manager: DBManager,
    persistency_manager: PersistencyManager,
    runtime_manager: RuntimeManager,
) -> Union[ArtifactFunctionWrapper, ArtifactMethodWrapper]:
    """Creates a wrapper for an artifact.

    Args:
        func: The artifact to wrap.
        db_manager: The database manager.
        persistency_manager: The persistency manager.
        runtime_manager: The runtime manager.

    Returns:
        The wrapper for the artifact.
    """
    if is_class_func(func=func):
        return ArtifactMethodWrapper(
            db_manager=db_manager, persistency_manager=persistency_manager, runtime_manager=runtime_manager
        )

    return ArtifactFunctionWrapper(
        db_manager=db_manager, persistency_manager=persistency_manager, runtime_manager=runtime_manager
    )