Embedders

ragbits.core.embeddings.Embedder

Embedder(default_options: OptionsT | None = None)

Bases: ConfigurableComponent[EmbedderOptionsT], 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[EmbedderOptionsT]

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: ObjectConstructionConfig) -> 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: ObjectConstructionConfig

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: ObjectConstructionConfig) -> 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

preferred_subclass classmethod

preferred_subclass(config: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None) -> Self

Tries to create an instance by looking at project's component preferences, either from YAML or from the factory. Takes optional overrides for both, which takes a higher precedence.

PARAMETER	DESCRIPTION
config	The CoreConfig instance containing preferred 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 preferred_subclass(
    cls, config: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None
) -> Self:
    """
    Tries to create an instance by looking at project's component preferences, either from YAML
    or from the factory. Takes optional overrides for both, which takes a higher precedence.

    Args:
        config: The CoreConfig instance containing preferred 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:
        preferences = get_config_from_yaml(yaml_path_override)
        if type_config := preferences.get(cls.configuration_key):
            return cls.subclass_from_config(ObjectConstructionConfig.model_validate(type_config))

    if factory_path_override:
        return cls.subclass_from_factory(factory_path_override)

    if preferred_factory := config.component_preference_factories.get(cls.configuration_key):
        return cls.subclass_from_factory(preferred_factory)

    if preferred_config := config.preferred_instances_config.get(cls.configuration_key):
        return cls.subclass_from_config(ObjectConstructionConfig.model_validate(preferred_config))

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

from_config classmethod

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

Initializes the class with the provided configuration.

PARAMETER	DESCRIPTION
config	A dictionary containing configuration details for the class. TYPE: dict[str, Any]

RETURNS	DESCRIPTION
Self	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]) -> Self:
    """
    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: EmbedderOptionsT | 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 Embedder model. TYPE: EmbedderOptionsT | 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: EmbedderOptionsT | 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 Embedder 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: EmbedderOptionsT | 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 Embedder model. TYPE: EmbedderOptionsT | 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: EmbedderOptionsT | 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 Embedder model.

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

ragbits.core.embeddings.local.LocalEmbedder

LocalEmbedder(model_name: str, api_key: str | None = None, default_options: LocalEmbedderOptions | None = None)

Bases: Embedder[LocalEmbedderOptions]

Class for interaction with any encoder available in HuggingFace.

Note: Local implementation is not dedicated for production. Use it only in experiments / evaluation

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: LocalEmbedderOptions | 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: LocalEmbedderOptions | 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 = LocalEmbedderOptions

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: ObjectConstructionConfig) -> 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: ObjectConstructionConfig

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: ObjectConstructionConfig) -> 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

preferred_subclass classmethod

preferred_subclass(config: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None) -> Self

Tries to create an instance by looking at project's component preferences, either from YAML or from the factory. Takes optional overrides for both, which takes a higher precedence.

PARAMETER	DESCRIPTION
config	The CoreConfig instance containing preferred 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 preferred_subclass(
    cls, config: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None
) -> Self:
    """
    Tries to create an instance by looking at project's component preferences, either from YAML
    or from the factory. Takes optional overrides for both, which takes a higher precedence.

    Args:
        config: The CoreConfig instance containing preferred 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:
        preferences = get_config_from_yaml(yaml_path_override)
        if type_config := preferences.get(cls.configuration_key):
            return cls.subclass_from_config(ObjectConstructionConfig.model_validate(type_config))

    if factory_path_override:
        return cls.subclass_from_factory(factory_path_override)

    if preferred_factory := config.component_preference_factories.get(cls.configuration_key):
        return cls.subclass_from_factory(preferred_factory)

    if preferred_config := config.preferred_instances_config.get(cls.configuration_key):
        return cls.subclass_from_config(ObjectConstructionConfig.model_validate(preferred_config))

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

from_config classmethod

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

Initializes the class with the provided configuration.

PARAMETER	DESCRIPTION
config	A dictionary containing configuration details for the class. TYPE: dict[str, Any]

RETURNS	DESCRIPTION
Self	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]) -> Self:
    """
    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: EmbedderOptionsT | 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 Embedder model. TYPE: EmbedderOptionsT | 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: EmbedderOptionsT | 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 Embedder 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: LocalEmbedderOptions | 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: LocalEmbedderOptions | 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: LocalEmbedderOptions | 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
    with trace(
        data=data,
        model_name=self.model_name,
        model_obj=repr(self.model),
        tokenizer=repr(self.tokenizer),
        device=self.device,
        options=merged_options.dict(),
    ) as outputs:
        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():
                model_outputs = self.model(**batch_dict)
                batch_embeddings = self._average_pool(model_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()
        outputs.embeddings = embeddings
    return embeddings

ragbits.core.embeddings.litellm.LiteLLMEmbedder

LiteLLMEmbedder(model_name: str = 'text-embedding-3-small', default_options: LiteLLMEmbedderOptions | None = None, *, api_base: str | None = None, base_url: str | None = None, api_key: str | None = None, api_version: str | None = None, router: Router | None = None)

Bases: Embedder[LiteLLMEmbedderOptions]

Client for creating text embeddings using LiteLLM API.

Constructs the LiteLLMEmbeddingClient.

PARAMETER	DESCRIPTION
model_name	Name of the LiteLLM supported model to be used. Default is "text-embedding-3-small". TYPE: str DEFAULT: 'text-embedding-3-small'
default_options	Default options to pass to the LiteLLM API. TYPE: LiteLLMEmbedderOptions | None DEFAULT: None
api_base	The API endpoint you want to call the model with. TYPE: str | None DEFAULT: None
base_url	Alias for api_base. If both are provided, api_base takes precedence. TYPE: str | None DEFAULT: None
api_key	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
router	Router to be used to route requests to different models. TYPE: Router | None DEFAULT: None

Source code in packages/ragbits-core/src/ragbits/core/embeddings/litellm.py

def __init__(
    self,
    model_name: str = "text-embedding-3-small",
    default_options: LiteLLMEmbedderOptions | None = None,
    *,
    api_base: str | None = None,
    base_url: str | None = None,  # Alias for api_base
    api_key: str | None = None,
    api_version: str | None = None,
    router: litellm.Router | None = None,
) -> None:
    """
    Constructs the LiteLLMEmbeddingClient.

    Args:
        model_name: 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: Default options to pass to the LiteLLM API.
        api_base: The API endpoint you want to call the model with.
        base_url: Alias for api_base. If both are provided, api_base takes precedence.
        api_key: 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.
        router: Router to be used to [route requests](https://docs.litellm.ai/docs/routing) to different models.
    """
    super().__init__(default_options=default_options)

    self.model_name = model_name
    self.api_base = api_base or base_url
    self.api_key = api_key
    self.api_version = api_version
    self.router = router

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 = LiteLLMEmbedderOptions

model_name instance-attribute

model_name = model_name

api_base instance-attribute

api_base = api_base or base_url

api_key instance-attribute

api_key = api_key

api_version instance-attribute

api_version = api_version

router instance-attribute

router = router

subclass_from_config classmethod

subclass_from_config(config: ObjectConstructionConfig) -> 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: ObjectConstructionConfig

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: ObjectConstructionConfig) -> 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

preferred_subclass classmethod

preferred_subclass(config: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None) -> Self

Tries to create an instance by looking at project's component preferences, either from YAML or from the factory. Takes optional overrides for both, which takes a higher precedence.

PARAMETER	DESCRIPTION
config	The CoreConfig instance containing preferred 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 preferred_subclass(
    cls, config: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None
) -> Self:
    """
    Tries to create an instance by looking at project's component preferences, either from YAML
    or from the factory. Takes optional overrides for both, which takes a higher precedence.

    Args:
        config: The CoreConfig instance containing preferred 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:
        preferences = get_config_from_yaml(yaml_path_override)
        if type_config := preferences.get(cls.configuration_key):
            return cls.subclass_from_config(ObjectConstructionConfig.model_validate(type_config))

    if factory_path_override:
        return cls.subclass_from_factory(factory_path_override)

    if preferred_factory := config.component_preference_factories.get(cls.configuration_key):
        return cls.subclass_from_factory(preferred_factory)

    if preferred_config := config.preferred_instances_config.get(cls.configuration_key):
        return cls.subclass_from_config(ObjectConstructionConfig.model_validate(preferred_config))

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

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: EmbedderOptionsT | 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 Embedder model. TYPE: EmbedderOptionsT | 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: EmbedderOptionsT | 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 Embedder 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: LiteLLMEmbedderOptions | 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: LiteLLMEmbedderOptions | 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: LiteLLMEmbedderOptions | 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_name,
        api_base=self.api_base,
        api_version=self.api_version,
        options=merged_options.dict(),
    ) as outputs:
        try:
            entrypoint = self.router or litellm
            response = await entrypoint.aembedding(
                input=data,
                model=self.model_name,
                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

from_config classmethod

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

Creates and returns a LiteLLMEmbedder instance.

PARAMETER	DESCRIPTION
config	A configuration object containing the configuration for initializing the LiteLLMEmbedder instance. TYPE: dict[str, Any]

RETURNS	DESCRIPTION
LiteLLMEmbedder	An initialized LiteLLMEmbedder instance. TYPE: Self

Source code in packages/ragbits-core/src/ragbits/core/embeddings/litellm.py

@classmethod
def from_config(cls, config: dict[str, Any]) -> Self:
    """
    Creates and returns a LiteLLMEmbedder instance.

    Args:
        config: A configuration object containing the configuration for initializing the LiteLLMEmbedder instance.

    Returns:
        LiteLLMEmbedder: An initialized LiteLLMEmbedder instance.
    """
    if "router" in config:
        router = litellm.router.Router(model_list=config["router"])
        config["router"] = router

    # Map base_url to api_base if present
    if "base_url" in config and "api_base" not in config:
        config["api_base"] = config.pop("base_url")

    return super().from_config(config)

ragbits.core.embeddings.fastembed.FastEmbedEmbedder

FastEmbedEmbedder(model_name: str, use_gpu: bool = False, default_options: FastEmbedOptions | None = None)

Bases: Embedder[FastEmbedOptions]

Class for creating dense text embeddings using FastEmbed library. For more information, see the FastEmbed GitHub.

Source code in packages/ragbits-core/src/ragbits/core/embeddings/fastembed.py

def __init__(self, model_name: str, use_gpu: bool = False, default_options: FastEmbedOptions | None = None):
    super().__init__(default_options=default_options)
    self.model_name = model_name
    self.use_gpu = use_gpu
    if use_gpu:
        self._model = TextEmbedding(model_name=model_name, providers=["CUDAExecutionProvider"])
    else:
        self._model = TextEmbedding(model_name=model_name)

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 = FastEmbedOptions

model_name instance-attribute

model_name = model_name

use_gpu instance-attribute

use_gpu = use_gpu

subclass_from_config classmethod

subclass_from_config(config: ObjectConstructionConfig) -> 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: ObjectConstructionConfig

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: ObjectConstructionConfig) -> 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

preferred_subclass classmethod

preferred_subclass(config: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None) -> Self

Tries to create an instance by looking at project's component preferences, either from YAML or from the factory. Takes optional overrides for both, which takes a higher precedence.

PARAMETER	DESCRIPTION
config	The CoreConfig instance containing preferred 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 preferred_subclass(
    cls, config: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None
) -> Self:
    """
    Tries to create an instance by looking at project's component preferences, either from YAML
    or from the factory. Takes optional overrides for both, which takes a higher precedence.

    Args:
        config: The CoreConfig instance containing preferred 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:
        preferences = get_config_from_yaml(yaml_path_override)
        if type_config := preferences.get(cls.configuration_key):
            return cls.subclass_from_config(ObjectConstructionConfig.model_validate(type_config))

    if factory_path_override:
        return cls.subclass_from_factory(factory_path_override)

    if preferred_factory := config.component_preference_factories.get(cls.configuration_key):
        return cls.subclass_from_factory(preferred_factory)

    if preferred_config := config.preferred_instances_config.get(cls.configuration_key):
        return cls.subclass_from_config(ObjectConstructionConfig.model_validate(preferred_config))

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

from_config classmethod

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

Initializes the class with the provided configuration.

PARAMETER	DESCRIPTION
config	A dictionary containing configuration details for the class. TYPE: dict[str, Any]

RETURNS	DESCRIPTION
Self	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]) -> Self:
    """
    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: EmbedderOptionsT | 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 Embedder model. TYPE: EmbedderOptionsT | 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: EmbedderOptionsT | 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 Embedder 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: EmbedderOptionsT | None = None) -> list[list[float]]

Embeds a list of strings into a list of embeddings.

PARAMETER	DESCRIPTION
data	List of strings to get embeddings for. TYPE: list[str]
options	Additional options to pass to the embedding model. TYPE: EmbedderOptionsT | 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/fastembed.py

async def embed_text(self, data: list[str], options: EmbedderOptionsT | None = None) -> list[list[float]]:
    """
    Embeds a list of strings into a list of embeddings.

    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
    with trace(
        data=data, model_name=self.model_name, model_obj=repr(self._model), options=merged_options.dict()
    ) as outputs:
        embeddings = [[float(x) for x in result] for result in self._model.embed(data, **merged_options.dict())]
        outputs.embeddings = embeddings
    return embeddings

ragbits.core.embeddings.fastembed.FastEmbedSparseEmbedder

FastEmbedSparseEmbedder(model_name: str, use_gpu: bool = False, default_options: FastEmbedOptions | None = None)

Bases: SparseEmbedder[FastEmbedOptions]

Class for creating sparse text embeddings using FastEmbed library. For more information, see the FastEmbed GitHub.

Source code in packages/ragbits-core/src/ragbits/core/embeddings/fastembed.py

def __init__(self, model_name: str, use_gpu: bool = False, default_options: FastEmbedOptions | None = None):
    super().__init__(default_options=default_options)
    self.model_name = model_name
    self.use_gpu = use_gpu
    if use_gpu:
        self._model = SparseTextEmbedding(model_name=model_name, providers=["CUDAExecutionProvider"])
    else:
        self._model = SparseTextEmbedding(model_name=model_name)

default_module class-attribute instance-attribute

default_module: ClassVar = embeddings

configuration_key class-attribute instance-attribute

configuration_key: ClassVar = 'sparse_embedder'

default_options instance-attribute

default_options: OptionsT = default_options or options_cls()

options_cls class-attribute instance-attribute

options_cls = FastEmbedOptions

model_name instance-attribute

model_name = model_name

use_gpu instance-attribute

use_gpu = use_gpu

subclass_from_config classmethod

subclass_from_config(config: ObjectConstructionConfig) -> 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: ObjectConstructionConfig

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: ObjectConstructionConfig) -> 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

preferred_subclass classmethod

preferred_subclass(config: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None) -> Self

Tries to create an instance by looking at project's component preferences, either from YAML or from the factory. Takes optional overrides for both, which takes a higher precedence.

PARAMETER	DESCRIPTION
config	The CoreConfig instance containing preferred 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 preferred_subclass(
    cls, config: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None
) -> Self:
    """
    Tries to create an instance by looking at project's component preferences, either from YAML
    or from the factory. Takes optional overrides for both, which takes a higher precedence.

    Args:
        config: The CoreConfig instance containing preferred 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:
        preferences = get_config_from_yaml(yaml_path_override)
        if type_config := preferences.get(cls.configuration_key):
            return cls.subclass_from_config(ObjectConstructionConfig.model_validate(type_config))

    if factory_path_override:
        return cls.subclass_from_factory(factory_path_override)

    if preferred_factory := config.component_preference_factories.get(cls.configuration_key):
        return cls.subclass_from_factory(preferred_factory)

    if preferred_config := config.preferred_instances_config.get(cls.configuration_key):
        return cls.subclass_from_config(ObjectConstructionConfig.model_validate(preferred_config))

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

from_config classmethod

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

Initializes the class with the provided configuration.

PARAMETER	DESCRIPTION
config	A dictionary containing configuration details for the class. TYPE: dict[str, Any]

RETURNS	DESCRIPTION
Self	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]) -> Self:
    """
    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 async

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

Embeds a list of strings into a list of sparse embeddings.

PARAMETER	DESCRIPTION
data	List of strings to get embeddings for. TYPE: list[str]
options	Additional options to pass to the embedding model. TYPE: EmbedderOptionsT | None DEFAULT: None

RETURNS	DESCRIPTION
list[SparseVector]	List of embeddings for the given strings.

Source code in packages/ragbits-core/src/ragbits/core/embeddings/fastembed.py

async def embed_text(self, data: list[str], options: EmbedderOptionsT | None = None) -> list[SparseVector]:
    """
    Embeds a list of strings into a list of sparse embeddings.

    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
    with trace(
        data=data, model_name=self.model_name, model_obj=repr(self._model), options=merged_options.dict()
    ) as outputs:
        outputs.embeddings = [
            SparseVector(values=[float(x) for x in result.values], indices=[int(x) for x in result.indices])
            for result in self._model.embed(data, **merged_options.dict())
        ]
    return outputs.embeddings