Skip to content

Pipelines#

ragbits.evaluate.pipelines.base.EvaluationPipeline #

EvaluationPipeline(evaluation_target: EvaluationTargetT)

Bases: WithConstructionConfig, Generic[EvaluationTargetT, EvaluationDataT, EvaluationResultT], ABC

Evaluation pipeline.

Initialize the evaluation pipeline.

PARAMETER DESCRIPTION
evaluation_target

Evaluation target instance.

TYPE: EvaluationTargetT

Source code in packages/ragbits-evaluate/src/ragbits/evaluate/pipelines/base.py 
38
39
40
41
42
43
44
45
46
def __init__(self, evaluation_target: EvaluationTargetT) -> None:
    """
    Initialize the evaluation pipeline.

    Args:
        evaluation_target: Evaluation target instance.
    """
    super().__init__()
    self.evaluation_target = evaluation_target

default_module class-attribute #

default_module: ModuleType | None = pipelines

configuration_key class-attribute #

configuration_key: str = 'pipeline'

evaluation_target instance-attribute #

evaluation_target = evaluation_target

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 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
@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. Supports both synchronous and asynchronous factory functions.

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 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
@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. Supports both synchronous and asynchronous factory functions.

    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)

    if asyncio.iscoroutinefunction(factory):
        try:
            loop = asyncio.get_running_loop()
            obj = asyncio.run_coroutine_threadsafe(factory, loop).result()
        except RuntimeError:
            obj = asyncio.run(factory())
    else:
        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 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
@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) -> Self

Initializes the class with the provided configuration.

PARAMETER DESCRIPTION
config

A dictionary containing configuration details for the class.

TYPE: dict

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 
178
179
180
181
182
183
184
185
186
187
188
189
@classmethod
def from_config(cls, config: dict) -> 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.
    """
    return cls(**config)

prepare async #

prepare() -> None

Prepare pipeline for evaluation. Optional step.

Source code in packages/ragbits-evaluate/src/ragbits/evaluate/pipelines/base.py 
48
49
50
51
52
async def prepare(self) -> None:
    """
    Prepare pipeline for evaluation. Optional step.
    """
    pass

ragbits.evaluate.pipelines.document_search.DocumentSearchPipeline #

DocumentSearchPipeline(evaluation_target: DocumentSearch, source: dict | None = None)

Bases: EvaluationPipeline[DocumentSearch, DocumentSearchData, DocumentSearchResult]

Document search evaluation pipeline.

Initialize the document search evaluation pipeline.

PARAMETER DESCRIPTION
evaluation_target

Document Search instance.

TYPE: DocumentSearch

source

Source data config for ingest.

TYPE: dict | None DEFAULT: None

Source code in packages/ragbits-evaluate/src/ragbits/evaluate/pipelines/document_search.py 
43
44
45
46
47
48
49
50
51
52
def __init__(self, evaluation_target: DocumentSearch, source: dict | None = None) -> None:
    """
    Initialize the document search evaluation pipeline.

    Args:
        evaluation_target: Document Search instance.
        source: Source data config for ingest.
    """
    super().__init__(evaluation_target=evaluation_target)
    self.source = source or {}

default_module class-attribute #

default_module: ModuleType | None = pipelines

configuration_key class-attribute #

configuration_key: str = 'pipeline'

evaluation_target instance-attribute #

evaluation_target = evaluation_target

source instance-attribute #

source = source or {}

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 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
@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. Supports both synchronous and asynchronous factory functions.

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 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
@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. Supports both synchronous and asynchronous factory functions.

    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)

    if asyncio.iscoroutinefunction(factory):
        try:
            loop = asyncio.get_running_loop()
            obj = asyncio.run_coroutine_threadsafe(factory, loop).result()
        except RuntimeError:
            obj = asyncio.run(factory())
    else:
        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 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
@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) -> Self

Create an instance of DocumentSearchPipeline from a configuration dictionary.

PARAMETER DESCRIPTION
config

A dictionary containing configuration settings for the pipeline.

TYPE: dict

RETURNS DESCRIPTION
Self

An instance of the pipeline class initialized with the provided configuration.
Source code in packages/ragbits-evaluate/src/ragbits/evaluate/pipelines/document_search.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@classmethod
def from_config(cls, config: dict) -> Self:
    """
    Create an instance of `DocumentSearchPipeline` from a configuration dictionary.

    Args:
        config: A dictionary containing configuration settings for the pipeline.

    Returns:
        An instance of the pipeline class initialized with the provided configuration.
    """
    # At this point, we assume that if the source is set, the pipeline is run in experimental mode
    # and create random indexes for testing
    # TODO: optimize this for cases with duplicated document search configs between runs
    if config.get("source"):
        config["vector_store"]["config"]["index_name"] = str(uuid4())
    evaluation_target: DocumentSearch = DocumentSearch.from_config(config)
    return cls(evaluation_target=evaluation_target, source=config.get("source"))

prepare async #

prepare() -> None

Ingest corpus data for evaluation.

Source code in packages/ragbits-evaluate/src/ragbits/evaluate/pipelines/document_search.py 
73
74
75
76
77
78
79
80
81
82
83
84
async def prepare(self) -> None:
    """
    Ingest corpus data for evaluation.
    """
    if self.source:
        # For now we only support HF sources for pre-evaluation ingest
        # TODO: Make it generic to any data source
        sources = await HuggingFaceSource.list_sources(
            path=self.source["config"]["path"],
            split=self.source["config"]["split"],
        )
        await self.evaluation_target.ingest(sources)

ragbits.evaluate.pipelines.question_answer.QuestionAnswerPipeline #

QuestionAnswerPipeline(evaluation_target: EvaluationTargetT)

Bases: EvaluationPipeline[QuestionAnswerAgent[LLMClientOptionsT, QuestionAnswerPromptInput, QuestionAnswerPromptOutputT], QuestionAnswerData, QuestionAnswerResult]

Question answer evaluation pipeline.

Initialize the evaluation pipeline.

PARAMETER DESCRIPTION
evaluation_target

Evaluation target instance.

TYPE: EvaluationTargetT

Source code in packages/ragbits-evaluate/src/ragbits/evaluate/pipelines/base.py 
38
39
40
41
42
43
44
45
46
def __init__(self, evaluation_target: EvaluationTargetT) -> None:
    """
    Initialize the evaluation pipeline.

    Args:
        evaluation_target: Evaluation target instance.
    """
    super().__init__()
    self.evaluation_target = evaluation_target

default_module class-attribute #

default_module: ModuleType | None = pipelines

configuration_key class-attribute #

configuration_key: str = 'pipeline'

evaluation_target instance-attribute #

evaluation_target = evaluation_target

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 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
@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. Supports both synchronous and asynchronous factory functions.

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 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
@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. Supports both synchronous and asynchronous factory functions.

    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)

    if asyncio.iscoroutinefunction(factory):
        try:
            loop = asyncio.get_running_loop()
            obj = asyncio.run_coroutine_threadsafe(factory, loop).result()
        except RuntimeError:
            obj = asyncio.run(factory())
    else:
        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 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
@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}")

prepare async #

prepare() -> None

Prepare pipeline for evaluation. Optional step.

Source code in packages/ragbits-evaluate/src/ragbits/evaluate/pipelines/base.py 
48
49
50
51
52
async def prepare(self) -> None:
    """
    Prepare pipeline for evaluation. Optional step.
    """
    pass

from_config classmethod #

from_config(config: dict) -> Self

Create an instance of QuestionAnswerPipeline from a configuration dictionary.

PARAMETER DESCRIPTION
config

A dictionary containing configuration settings for the pipeline.

TYPE: dict

RETURNS DESCRIPTION
Self

An instance of the pipeline class initialized with the provided configuration.
Source code in packages/ragbits-evaluate/src/ragbits/evaluate/pipelines/question_answer.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
@classmethod
def from_config(cls, config: dict) -> Self:
    """
    Create an instance of `QuestionAnswerPipeline` from a configuration dictionary.

    Args:
        config: A dictionary containing configuration settings for the pipeline.

    Returns:
        An instance of the pipeline class initialized with the provided configuration.
    """
    config["evaluation_target"] = QuestionAnswerAgent.from_config(config)
    return super().from_config(config)