Skip to content

Embeddings#

ragbits.core.embeddings.Embeddings #

Embeddings(default_options: OptionsT | None = None)

Bases: ConfigurableComponent[EmbeddingsOptionsT], ABC

Abstract client for communication with embedding models.

Constructs a new ConfigurableComponent instance.

PARAMETER DESCRIPTION
default_options

The default options for the component.

TYPE: OptionsT | None DEFAULT: None

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py
def __init__(self, default_options: OptionsT | None = None) -> None:
    """
    Constructs a new ConfigurableComponent instance.

    Args:
        default_options: The default options for the component.
    """
    self.default_options: OptionsT = default_options or self.options_cls()

default_options instance-attribute #

default_options: OptionsT = default_options or options_cls()

options_cls instance-attribute #

options_cls: type[EmbeddingsOptionsT]

default_module class-attribute instance-attribute #

default_module: ClassVar = embeddings

configuration_key class-attribute instance-attribute #

configuration_key: ClassVar = 'embedder'

subclass_from_config classmethod #

subclass_from_config(config: ObjectContructionConfig) -> Self

Initializes the class with the provided configuration. May return a subclass of the class, if requested by the configuration.

PARAMETER DESCRIPTION
config

A model containing configuration details for the class.

TYPE: ObjectContructionConfig

RETURNS DESCRIPTION
Self

An instance of the class initialized with the provided configuration.

RAISES DESCRIPTION
InvalidConfigError

The class can't be found or is not a subclass of the current class.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py
@classmethod
def subclass_from_config(cls, config: ObjectContructionConfig) -> Self:
    """
    Initializes the class with the provided configuration. May return a subclass of the class,
    if requested by the configuration.

    Args:
        config: A model containing configuration details for the class.

    Returns:
        An instance of the class initialized with the provided configuration.

    Raises:
        InvalidConfigError: The class can't be found or is not a subclass of the current class.
    """
    subclass = import_by_path(config.type, cls.default_module)
    if not issubclass(subclass, cls):
        raise InvalidConfigError(f"{subclass} is not a subclass of {cls}")

    return subclass.from_config(config.config)

subclass_from_factory classmethod #

subclass_from_factory(factory_path: str) -> Self

Creates the class using the provided factory function. May return a subclass of the class, if requested by the factory.

PARAMETER DESCRIPTION
factory_path

A string representing the path to the factory function in the format of "module.submodule:factory_name".

TYPE: str

RETURNS DESCRIPTION
Self

An instance of the class initialized with the provided factory function.

RAISES DESCRIPTION
InvalidConfigError

The factory can't be found or the object returned is not a subclass of the current class.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py
@classmethod
def subclass_from_factory(cls, factory_path: str) -> Self:
    """
    Creates the class using the provided factory function. May return a subclass of the class,
    if requested by the factory.

    Args:
        factory_path: A string representing the path to the factory function
            in the format of "module.submodule:factory_name".

    Returns:
        An instance of the class initialized with the provided factory function.

    Raises:
        InvalidConfigError: The factory can't be found or the object returned
            is not a subclass of the current class.
    """
    factory = import_by_path(factory_path, cls.default_module)
    obj = factory()
    if not isinstance(obj, cls):
        raise InvalidConfigError(f"The object returned by factory {factory_path} is not an instance of {cls}")
    return obj

subclass_from_defaults classmethod #

subclass_from_defaults(defaults: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None) -> Self

Tries to create an instance by looking at default configuration file, and default factory function. Takes optional overrides for both, which takes a higher precedence.

PARAMETER DESCRIPTION
defaults

The CoreConfig instance containing default factory and configuration details.

TYPE: CoreConfig

factory_path_override

A string representing the path to the factory function in the format of "module.submodule:factory_name".

TYPE: str | None DEFAULT: None

yaml_path_override

A string representing the path to the YAML file containing the Ragstack instance configuration.

TYPE: Path | None DEFAULT: None

RAISES DESCRIPTION
InvalidConfigError

If the default factory or configuration can't be found.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py
@classmethod
def subclass_from_defaults(
    cls, defaults: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None
) -> Self:
    """
    Tries to create an instance by looking at default configuration file, and default factory function.
    Takes optional overrides for both, which takes a higher precedence.

    Args:
        defaults: The CoreConfig instance containing default factory and configuration details.
        factory_path_override: A string representing the path to the factory function
            in the format of "module.submodule:factory_name".
        yaml_path_override: A string representing the path to the YAML file containing
            the Ragstack instance configuration.

    Raises:
        InvalidConfigError: If the default factory or configuration can't be found.
    """
    if yaml_path_override:
        config = get_config_from_yaml(yaml_path_override)
        if type_config := config.get(cls.configuration_key):
            return cls.subclass_from_config(ObjectContructionConfig.model_validate(type_config))

    if factory_path_override:
        return cls.subclass_from_factory(factory_path_override)

    if default_factory := defaults.default_factories.get(cls.configuration_key):
        return cls.subclass_from_factory(default_factory)

    if default_config := defaults.default_instances_config.get(cls.configuration_key):
        return cls.subclass_from_config(ObjectContructionConfig.model_validate(default_config))

    raise NoDefaultConfigError(f"Could not find default factory or configuration for {cls.configuration_key}")

from_config classmethod #

from_config(config: dict[str, Any]) -> ConfigurableComponent

Initializes the class with the provided configuration.

PARAMETER DESCRIPTION
config

A dictionary containing configuration details for the class.

TYPE: dict[str, Any]

RETURNS DESCRIPTION
ConfigurableComponent

An instance of the class initialized with the provided configuration.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py
@classmethod
def from_config(cls, config: dict[str, Any]) -> ConfigurableComponent:
    """
    Initializes the class with the provided configuration.

    Args:
        config: A dictionary containing configuration details for the class.

    Returns:
        An instance of the class initialized with the provided configuration.
    """
    default_options = config.pop("default_options", None)
    options = cls.options_cls(**default_options) if default_options else None
    return cls(**config, default_options=options)

embed_text abstractmethod async #

embed_text(data: list[str], options: EmbeddingsOptionsT | None = None) -> list[list[float]]

Creates embeddings for the given strings.

PARAMETER DESCRIPTION
data

List of strings to get embeddings for.

TYPE: list[str]

options

Additional settings used by the Embeddings model.

TYPE: EmbeddingsOptionsT | None DEFAULT: None

RETURNS DESCRIPTION
list[list[float]]

List of embeddings for the given strings.

Source code in packages/ragbits-core/src/ragbits/core/embeddings/base.py
@abstractmethod
async def embed_text(self, data: list[str], options: EmbeddingsOptionsT | None = None) -> list[list[float]]:
    """
    Creates embeddings for the given strings.

    Args:
        data: List of strings to get embeddings for.
        options: Additional settings used by the Embeddings model.

    Returns:
        List of embeddings for the given strings.
    """

image_support #

image_support() -> bool

Check if the model supports image embeddings.

RETURNS DESCRIPTION
bool

True if the model supports image embeddings, False otherwise.

Source code in packages/ragbits-core/src/ragbits/core/embeddings/base.py
def image_support(self) -> bool:  # noqa: PLR6301
    """
    Check if the model supports image embeddings.

    Returns:
        True if the model supports image embeddings, False otherwise.
    """
    return False

embed_image async #

embed_image(images: list[bytes], options: EmbeddingsOptionsT | None = None) -> list[list[float]]

Creates embeddings for the given images.

PARAMETER DESCRIPTION
images

List of images to get embeddings for.

TYPE: list[bytes]

options

Additional settings used by the Embeddings model.

TYPE: EmbeddingsOptionsT | None DEFAULT: None

RETURNS DESCRIPTION
list[list[float]]

List of embeddings for the given images.

Source code in packages/ragbits-core/src/ragbits/core/embeddings/base.py
async def embed_image(self, images: list[bytes], options: EmbeddingsOptionsT | None = None) -> list[list[float]]:
    """
    Creates embeddings for the given images.

    Args:
        images: List of images to get embeddings for.
        options: Additional settings used by the Embeddings model.

    Returns:
        List of embeddings for the given images.
    """
    raise NotImplementedError("Image embeddings are not supported by this model.")

ragbits.core.embeddings.local.LocalEmbeddings #

LocalEmbeddings(model_name: str, api_key: str | None = None, default_options: LocalEmbeddingsOptions | None = None)

Bases: Embeddings[LocalEmbeddingsOptions]

Class for interaction with any encoder available in HuggingFace.

Constructs a new local LLM instance.

PARAMETER DESCRIPTION
model_name

Name of the model to use.

TYPE: str

api_key

The API key for Hugging Face authentication.

TYPE: str | None DEFAULT: None

default_options

Default options for the embedding model.

TYPE: LocalEmbeddingsOptions | None DEFAULT: None

RAISES DESCRIPTION
ImportError

If the 'local' extra requirements are not installed.

Source code in packages/ragbits-core/src/ragbits/core/embeddings/local.py
def __init__(
    self,
    model_name: str,
    api_key: str | None = None,
    default_options: LocalEmbeddingsOptions | None = None,
) -> None:
    """Constructs a new local LLM instance.

    Args:
        model_name: Name of the model to use.
        api_key: The API key for Hugging Face authentication.
        default_options: Default options for the embedding model.

    Raises:
        ImportError: If the 'local' extra requirements are not installed.
    """
    if not HAS_LOCAL_EMBEDDINGS:
        raise ImportError("You need to install the 'local' extra requirements to use local embeddings models")

    super().__init__(default_options=default_options)

    self.hf_api_key = api_key
    self.model_name = model_name

    self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    self.model = AutoModel.from_pretrained(self.model_name, token=self.hf_api_key).to(self.device)
    self.tokenizer = AutoTokenizer.from_pretrained(self.model_name, token=self.hf_api_key)

default_module class-attribute instance-attribute #

default_module: ClassVar = embeddings

configuration_key class-attribute instance-attribute #

configuration_key: ClassVar = 'embedder'

default_options instance-attribute #

default_options: OptionsT = default_options or options_cls()

options_cls class-attribute instance-attribute #

options_cls = LocalEmbeddingsOptions

hf_api_key instance-attribute #

hf_api_key = api_key

model_name instance-attribute #

model_name = model_name

device instance-attribute #

device = device('cuda' if is_available() else 'cpu')

model instance-attribute #

model = to(device)

tokenizer instance-attribute #

tokenizer = from_pretrained(model_name, token=hf_api_key)

subclass_from_config classmethod #

subclass_from_config(config: ObjectContructionConfig) -> Self

Initializes the class with the provided configuration. May return a subclass of the class, if requested by the configuration.

PARAMETER DESCRIPTION
config

A model containing configuration details for the class.

TYPE: ObjectContructionConfig

RETURNS DESCRIPTION
Self

An instance of the class initialized with the provided configuration.

RAISES DESCRIPTION
InvalidConfigError

The class can't be found or is not a subclass of the current class.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py
@classmethod
def subclass_from_config(cls, config: ObjectContructionConfig) -> Self:
    """
    Initializes the class with the provided configuration. May return a subclass of the class,
    if requested by the configuration.

    Args:
        config: A model containing configuration details for the class.

    Returns:
        An instance of the class initialized with the provided configuration.

    Raises:
        InvalidConfigError: The class can't be found or is not a subclass of the current class.
    """
    subclass = import_by_path(config.type, cls.default_module)
    if not issubclass(subclass, cls):
        raise InvalidConfigError(f"{subclass} is not a subclass of {cls}")

    return subclass.from_config(config.config)

subclass_from_factory classmethod #

subclass_from_factory(factory_path: str) -> Self

Creates the class using the provided factory function. May return a subclass of the class, if requested by the factory.

PARAMETER DESCRIPTION
factory_path

A string representing the path to the factory function in the format of "module.submodule:factory_name".

TYPE: str

RETURNS DESCRIPTION
Self

An instance of the class initialized with the provided factory function.

RAISES DESCRIPTION
InvalidConfigError

The factory can't be found or the object returned is not a subclass of the current class.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py
@classmethod
def subclass_from_factory(cls, factory_path: str) -> Self:
    """
    Creates the class using the provided factory function. May return a subclass of the class,
    if requested by the factory.

    Args:
        factory_path: A string representing the path to the factory function
            in the format of "module.submodule:factory_name".

    Returns:
        An instance of the class initialized with the provided factory function.

    Raises:
        InvalidConfigError: The factory can't be found or the object returned
            is not a subclass of the current class.
    """
    factory = import_by_path(factory_path, cls.default_module)
    obj = factory()
    if not isinstance(obj, cls):
        raise InvalidConfigError(f"The object returned by factory {factory_path} is not an instance of {cls}")
    return obj

subclass_from_defaults classmethod #

subclass_from_defaults(defaults: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None) -> Self

Tries to create an instance by looking at default configuration file, and default factory function. Takes optional overrides for both, which takes a higher precedence.

PARAMETER DESCRIPTION
defaults

The CoreConfig instance containing default factory and configuration details.

TYPE: CoreConfig

factory_path_override

A string representing the path to the factory function in the format of "module.submodule:factory_name".

TYPE: str | None DEFAULT: None

yaml_path_override

A string representing the path to the YAML file containing the Ragstack instance configuration.

TYPE: Path | None DEFAULT: None

RAISES DESCRIPTION
InvalidConfigError

If the default factory or configuration can't be found.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py
@classmethod
def subclass_from_defaults(
    cls, defaults: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None
) -> Self:
    """
    Tries to create an instance by looking at default configuration file, and default factory function.
    Takes optional overrides for both, which takes a higher precedence.

    Args:
        defaults: The CoreConfig instance containing default factory and configuration details.
        factory_path_override: A string representing the path to the factory function
            in the format of "module.submodule:factory_name".
        yaml_path_override: A string representing the path to the YAML file containing
            the Ragstack instance configuration.

    Raises:
        InvalidConfigError: If the default factory or configuration can't be found.
    """
    if yaml_path_override:
        config = get_config_from_yaml(yaml_path_override)
        if type_config := config.get(cls.configuration_key):
            return cls.subclass_from_config(ObjectContructionConfig.model_validate(type_config))

    if factory_path_override:
        return cls.subclass_from_factory(factory_path_override)

    if default_factory := defaults.default_factories.get(cls.configuration_key):
        return cls.subclass_from_factory(default_factory)

    if default_config := defaults.default_instances_config.get(cls.configuration_key):
        return cls.subclass_from_config(ObjectContructionConfig.model_validate(default_config))

    raise NoDefaultConfigError(f"Could not find default factory or configuration for {cls.configuration_key}")

from_config classmethod #

from_config(config: dict[str, Any]) -> ConfigurableComponent

Initializes the class with the provided configuration.

PARAMETER DESCRIPTION
config

A dictionary containing configuration details for the class.

TYPE: dict[str, Any]

RETURNS DESCRIPTION
ConfigurableComponent

An instance of the class initialized with the provided configuration.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py
@classmethod
def from_config(cls, config: dict[str, Any]) -> ConfigurableComponent:
    """
    Initializes the class with the provided configuration.

    Args:
        config: A dictionary containing configuration details for the class.

    Returns:
        An instance of the class initialized with the provided configuration.
    """
    default_options = config.pop("default_options", None)
    options = cls.options_cls(**default_options) if default_options else None
    return cls(**config, default_options=options)

image_support #

image_support() -> bool

Check if the model supports image embeddings.

RETURNS DESCRIPTION
bool

True if the model supports image embeddings, False otherwise.

Source code in packages/ragbits-core/src/ragbits/core/embeddings/base.py
def image_support(self) -> bool:  # noqa: PLR6301
    """
    Check if the model supports image embeddings.

    Returns:
        True if the model supports image embeddings, False otherwise.
    """
    return False

embed_image async #

embed_image(images: list[bytes], options: EmbeddingsOptionsT | None = None) -> list[list[float]]

Creates embeddings for the given images.

PARAMETER DESCRIPTION
images

List of images to get embeddings for.

TYPE: list[bytes]

options

Additional settings used by the Embeddings model.

TYPE: EmbeddingsOptionsT | None DEFAULT: None

RETURNS DESCRIPTION
list[list[float]]

List of embeddings for the given images.

Source code in packages/ragbits-core/src/ragbits/core/embeddings/base.py
async def embed_image(self, images: list[bytes], options: EmbeddingsOptionsT | None = None) -> list[list[float]]:
    """
    Creates embeddings for the given images.

    Args:
        images: List of images to get embeddings for.
        options: Additional settings used by the Embeddings model.

    Returns:
        List of embeddings for the given images.
    """
    raise NotImplementedError("Image embeddings are not supported by this model.")

embed_text async #

embed_text(data: list[str], options: LocalEmbeddingsOptions | None = None) -> list[list[float]]

Calls the appropriate encoder endpoint with the given data and options.

PARAMETER DESCRIPTION
data

List of strings to get embeddings for.

TYPE: list[str]

options

Additional options to pass to the embedding model.

TYPE: LocalEmbeddingsOptions | None DEFAULT: None

RETURNS DESCRIPTION
list[list[float]]

List of embeddings for the given strings.

Source code in packages/ragbits-core/src/ragbits/core/embeddings/local.py
async def embed_text(self, data: list[str], options: LocalEmbeddingsOptions | None = None) -> list[list[float]]:
    """Calls the appropriate encoder endpoint with the given data and options.

    Args:
        data: List of strings to get embeddings for.
        options: Additional options to pass to the embedding model.

    Returns:
        List of embeddings for the given strings.
    """
    merged_options = (self.default_options | options) if options else self.default_options

    embeddings = []
    for batch in self._batch(data, merged_options.batch_size):
        batch_dict = self.tokenizer(
            batch,
            max_length=self.tokenizer.model_max_length,
            padding=True,
            truncation=True,
            return_tensors="pt",
        ).to(self.device)
        with torch.no_grad():
            outputs = self.model(**batch_dict)
            batch_embeddings = self._average_pool(outputs.last_hidden_state, batch_dict["attention_mask"])
            batch_embeddings = F.normalize(batch_embeddings, p=2, dim=1)
        embeddings.extend(batch_embeddings.to("cpu").tolist())

    torch.cuda.empty_cache()
    return embeddings

ragbits.core.embeddings.litellm.LiteLLMEmbeddings #

LiteLLMEmbeddings(model: str = 'text-embedding-3-small', default_options: LiteLLMEmbeddingsOptions | None = None, api_base: str | None = None, api_key: str | None = None, api_version: str | None = None)

Bases: Embeddings[LiteLLMEmbeddingsOptions]

Client for creating text embeddings using LiteLLM API.

Constructs the LiteLLMEmbeddingClient.

PARAMETER DESCRIPTION
model

Name of the LiteLLM supported model to be used. Default is "text-embedding-3-small".

TYPE: str DEFAULT: 'text-embedding-3-small'

default_options

Defualt options to pass to the LiteLLM API.

TYPE: LiteLLMEmbeddingsOptions | None DEFAULT: None

api_base

The API endpoint you want to call the model with.

TYPE: str | None DEFAULT: None

api_key

API key to be used. API key to be used. If not specified, an environment variable will be used, for more information, follow the instructions for your specific vendor in the LiteLLM documentation.

TYPE: str | None DEFAULT: None

api_version

The API version for the call.

TYPE: str | None DEFAULT: None

Source code in packages/ragbits-core/src/ragbits/core/embeddings/litellm.py
def __init__(
    self,
    model: str = "text-embedding-3-small",
    default_options: LiteLLMEmbeddingsOptions | None = None,
    api_base: str | None = None,
    api_key: str | None = None,
    api_version: str | None = None,
) -> None:
    """
    Constructs the LiteLLMEmbeddingClient.

    Args:
        model: Name of the [LiteLLM supported model](https://docs.litellm.ai/docs/embedding/supported_embedding)\
            to be used. Default is "text-embedding-3-small".
        default_options: Defualt options to pass to the LiteLLM API.
        api_base: The API endpoint you want to call the model with.
        api_key: API key to be used. API key to be used. If not specified, an environment variable will be used,
            for more information, follow the instructions for your specific vendor in the\
            [LiteLLM documentation](https://docs.litellm.ai/docs/embedding/supported_embedding).
        api_version: The API version for the call.
    """
    super().__init__(default_options=default_options)
    self.model = model
    self.api_base = api_base
    self.api_key = api_key
    self.api_version = api_version

default_module class-attribute instance-attribute #

default_module: ClassVar = embeddings

configuration_key class-attribute instance-attribute #

configuration_key: ClassVar = 'embedder'

default_options instance-attribute #

default_options: OptionsT = default_options or options_cls()

options_cls class-attribute instance-attribute #

options_cls = LiteLLMEmbeddingsOptions

model instance-attribute #

model = model

api_base instance-attribute #

api_base = api_base

api_key instance-attribute #

api_key = api_key

api_version instance-attribute #

api_version = api_version

subclass_from_config classmethod #

subclass_from_config(config: ObjectContructionConfig) -> Self

Initializes the class with the provided configuration. May return a subclass of the class, if requested by the configuration.

PARAMETER DESCRIPTION
config

A model containing configuration details for the class.

TYPE: ObjectContructionConfig

RETURNS DESCRIPTION
Self

An instance of the class initialized with the provided configuration.

RAISES DESCRIPTION
InvalidConfigError

The class can't be found or is not a subclass of the current class.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py
@classmethod
def subclass_from_config(cls, config: ObjectContructionConfig) -> Self:
    """
    Initializes the class with the provided configuration. May return a subclass of the class,
    if requested by the configuration.

    Args:
        config: A model containing configuration details for the class.

    Returns:
        An instance of the class initialized with the provided configuration.

    Raises:
        InvalidConfigError: The class can't be found or is not a subclass of the current class.
    """
    subclass = import_by_path(config.type, cls.default_module)
    if not issubclass(subclass, cls):
        raise InvalidConfigError(f"{subclass} is not a subclass of {cls}")

    return subclass.from_config(config.config)

subclass_from_factory classmethod #

subclass_from_factory(factory_path: str) -> Self

Creates the class using the provided factory function. May return a subclass of the class, if requested by the factory.

PARAMETER DESCRIPTION
factory_path

A string representing the path to the factory function in the format of "module.submodule:factory_name".

TYPE: str

RETURNS DESCRIPTION
Self

An instance of the class initialized with the provided factory function.

RAISES DESCRIPTION
InvalidConfigError

The factory can't be found or the object returned is not a subclass of the current class.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py
@classmethod
def subclass_from_factory(cls, factory_path: str) -> Self:
    """
    Creates the class using the provided factory function. May return a subclass of the class,
    if requested by the factory.

    Args:
        factory_path: A string representing the path to the factory function
            in the format of "module.submodule:factory_name".

    Returns:
        An instance of the class initialized with the provided factory function.

    Raises:
        InvalidConfigError: The factory can't be found or the object returned
            is not a subclass of the current class.
    """
    factory = import_by_path(factory_path, cls.default_module)
    obj = factory()
    if not isinstance(obj, cls):
        raise InvalidConfigError(f"The object returned by factory {factory_path} is not an instance of {cls}")
    return obj

subclass_from_defaults classmethod #

subclass_from_defaults(defaults: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None) -> Self

Tries to create an instance by looking at default configuration file, and default factory function. Takes optional overrides for both, which takes a higher precedence.

PARAMETER DESCRIPTION
defaults

The CoreConfig instance containing default factory and configuration details.

TYPE: CoreConfig

factory_path_override

A string representing the path to the factory function in the format of "module.submodule:factory_name".

TYPE: str | None DEFAULT: None

yaml_path_override

A string representing the path to the YAML file containing the Ragstack instance configuration.

TYPE: Path | None DEFAULT: None

RAISES DESCRIPTION
InvalidConfigError

If the default factory or configuration can't be found.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py
@classmethod
def subclass_from_defaults(
    cls, defaults: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None
) -> Self:
    """
    Tries to create an instance by looking at default configuration file, and default factory function.
    Takes optional overrides for both, which takes a higher precedence.

    Args:
        defaults: The CoreConfig instance containing default factory and configuration details.
        factory_path_override: A string representing the path to the factory function
            in the format of "module.submodule:factory_name".
        yaml_path_override: A string representing the path to the YAML file containing
            the Ragstack instance configuration.

    Raises:
        InvalidConfigError: If the default factory or configuration can't be found.
    """
    if yaml_path_override:
        config = get_config_from_yaml(yaml_path_override)
        if type_config := config.get(cls.configuration_key):
            return cls.subclass_from_config(ObjectContructionConfig.model_validate(type_config))

    if factory_path_override:
        return cls.subclass_from_factory(factory_path_override)

    if default_factory := defaults.default_factories.get(cls.configuration_key):
        return cls.subclass_from_factory(default_factory)

    if default_config := defaults.default_instances_config.get(cls.configuration_key):
        return cls.subclass_from_config(ObjectContructionConfig.model_validate(default_config))

    raise NoDefaultConfigError(f"Could not find default factory or configuration for {cls.configuration_key}")

from_config classmethod #

from_config(config: dict[str, Any]) -> ConfigurableComponent

Initializes the class with the provided configuration.

PARAMETER DESCRIPTION
config

A dictionary containing configuration details for the class.

TYPE: dict[str, Any]

RETURNS DESCRIPTION
ConfigurableComponent

An instance of the class initialized with the provided configuration.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py
@classmethod
def from_config(cls, config: dict[str, Any]) -> ConfigurableComponent:
    """
    Initializes the class with the provided configuration.

    Args:
        config: A dictionary containing configuration details for the class.

    Returns:
        An instance of the class initialized with the provided configuration.
    """
    default_options = config.pop("default_options", None)
    options = cls.options_cls(**default_options) if default_options else None
    return cls(**config, default_options=options)

image_support #

image_support() -> bool

Check if the model supports image embeddings.

RETURNS DESCRIPTION
bool

True if the model supports image embeddings, False otherwise.

Source code in packages/ragbits-core/src/ragbits/core/embeddings/base.py
def image_support(self) -> bool:  # noqa: PLR6301
    """
    Check if the model supports image embeddings.

    Returns:
        True if the model supports image embeddings, False otherwise.
    """
    return False

embed_image async #

embed_image(images: list[bytes], options: EmbeddingsOptionsT | None = None) -> list[list[float]]

Creates embeddings for the given images.

PARAMETER DESCRIPTION
images

List of images to get embeddings for.

TYPE: list[bytes]

options

Additional settings used by the Embeddings model.

TYPE: EmbeddingsOptionsT | None DEFAULT: None

RETURNS DESCRIPTION
list[list[float]]

List of embeddings for the given images.

Source code in packages/ragbits-core/src/ragbits/core/embeddings/base.py
async def embed_image(self, images: list[bytes], options: EmbeddingsOptionsT | None = None) -> list[list[float]]:
    """
    Creates embeddings for the given images.

    Args:
        images: List of images to get embeddings for.
        options: Additional settings used by the Embeddings model.

    Returns:
        List of embeddings for the given images.
    """
    raise NotImplementedError("Image embeddings are not supported by this model.")

embed_text async #

embed_text(data: list[str], options: LiteLLMEmbeddingsOptions | None = None) -> list[list[float]]

Creates embeddings for the given strings.

PARAMETER DESCRIPTION
data

List of strings to get embeddings for.

TYPE: list[str]

options

Additional options to pass to the Lite LLM API.

TYPE: LiteLLMEmbeddingsOptions | None DEFAULT: None

RETURNS DESCRIPTION
list[list[float]]

List of embeddings for the given strings.

RAISES DESCRIPTION
EmbeddingConnectionError

If there is a connection error with the embedding API.

EmbeddingEmptyResponseError

If the embedding API returns an empty response.

EmbeddingStatusError

If the embedding API returns an error status code.

EmbeddingResponseError

If the embedding API response is invalid.

Source code in packages/ragbits-core/src/ragbits/core/embeddings/litellm.py
async def embed_text(self, data: list[str], options: LiteLLMEmbeddingsOptions | None = None) -> list[list[float]]:
    """
    Creates embeddings for the given strings.

    Args:
        data: List of strings to get embeddings for.
        options: Additional options to pass to the Lite LLM API.

    Returns:
        List of embeddings for the given strings.

    Raises:
        EmbeddingConnectionError: If there is a connection error with the embedding API.
        EmbeddingEmptyResponseError: If the embedding API returns an empty response.
        EmbeddingStatusError: If the embedding API returns an error status code.
        EmbeddingResponseError: If the embedding API response is invalid.
    """
    merged_options = (self.default_options | options) if options else self.default_options

    with trace(
        data=data,
        model=self.model,
        api_base=self.api_base,
        api_version=self.api_version,
        options=merged_options.dict(),
    ) as outputs:
        try:
            response = await litellm.aembedding(
                input=data,
                model=self.model,
                api_base=self.api_base,
                api_key=self.api_key,
                api_version=self.api_version,
                **merged_options.dict(),
            )
        except litellm.openai.APIConnectionError as exc:
            raise EmbeddingConnectionError() from exc
        except litellm.openai.APIStatusError as exc:
            raise EmbeddingStatusError(exc.message, exc.status_code) from exc
        except litellm.openai.APIResponseValidationError as exc:
            raise EmbeddingResponseError() from exc

        if not response.data:
            raise EmbeddingEmptyResponseError()

        outputs.embeddings = [embedding["embedding"] for embedding in response.data]
        if response.usage:
            outputs.completion_tokens = response.usage.completion_tokens
            outputs.prompt_tokens = response.usage.prompt_tokens
            outputs.total_tokens = response.usage.total_tokens

    return outputs.embeddings