Module haystack_experimental.core.super_component.super_component

InvalidMappingError

Raised when input or output mappings are invalid or type conflicts are found.

SuperComponent

A class for creating super components that wrap around a Pipeline.

This component allows for remapping of input and output socket names between the wrapped pipeline and the external interface. It handles type checking and verification of all mappings.

Arguments:

• pipeline: The pipeline wrapped by the component
• input_mapping: Mapping from component input names to lists of pipeline socket paths in format "component_name.socket_name"
• output_mapping: Mapping from pipeline socket paths to component output names

Raises:

• InvalidMappingError: If any input or output mappings are invalid or if type conflicts are detected
• ValueError: If no pipeline is provided

SuperComponent.__init__

def __init__(pipeline: Pipeline,
             input_mapping: Optional[Dict[str, List[str]]] = None,
             output_mapping: Optional[Dict[str, str]] = None) -> None

Initialize the component with optional I/O mappings.

Arguments:

• pipeline: The pipeline to wrap
• input_mapping: Optional input name mapping configuration
• output_mapping: Optional output name mapping configuration

SuperComponent.warm_up

def warm_up() -> None

Warms up the pipeline if it has not been warmed up before.

SuperComponent.to_dict

def to_dict() -> Dict[str, Any]

Convert the SuperComponent to a dictionary representation.

Must be overwritten for custom component implementations that inherit from SuperComponent.

Returns:

Dictionary containing serialized super component data

SuperComponent.from_dict

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "SuperComponent"

Create the SuperComponent instance from a dictionary representation.

Must be overwritten for custom component implementations that inherit from SuperComponent.

Arguments:

• data: Dictionary containing serialized super component data

Returns:

New PipelineWrapper instance

SuperComponent.run

def run(**kwargs: Any) -> Dict[str, Any]

Run the wrapped pipeline with the given inputs.

This method:

1. Maps input kwargs to pipeline component inputs
2. Executes the pipeline
3. Maps pipeline outputs back to wrapper outputs

Arguments:

• kwargs: Keyword arguments matching wrapper input names

Raises:

• ValueError: If no pipeline is configured
• InvalidMappingError: If output conflicts occur during auto-mapping

Returns:

Dictionary mapping wrapper output names to values