Coverage for haystack/core/super_component/super_component.py: 96%
194 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 13:53 +0000
1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2#
3# SPDX-License-Identifier: Apache-2.0
5import functools
6from pathlib import Path
7from types import new_class
8from typing import Any, TypeVar
10from haystack import logging
11from haystack.core.component.component import component
12from haystack.core.pipeline.pipeline import Pipeline
13from haystack.core.pipeline.utils import parse_connect_string
14from haystack.core.serialization import default_from_dict, default_to_dict, generate_qualified_class_name
15from haystack.core.super_component.utils import _delegate_default, _is_compatible
17logger = logging.getLogger(__name__)
19T = TypeVar("T")
22class InvalidMappingTypeError(Exception):
23 """Raised when input or output mappings have invalid types or type conflicts."""
25 pass
28class InvalidMappingValueError(Exception):
29 """Raised when input or output mappings have invalid values or missing components/sockets."""
31 pass
34@component
35class _SuperComponent:
36 def __init__(
37 self,
38 pipeline: Pipeline,
39 input_mapping: dict[str, list[str]] | None = None,
40 output_mapping: dict[str, str] | None = None,
41 ) -> None:
42 """
43 Creates a SuperComponent with optional input and output mappings.
45 :param pipeline: The pipeline instance to be wrapped
46 :param input_mapping: A dictionary mapping component input names to pipeline input socket paths.
47 If not provided, a default input mapping will be created based on all pipeline inputs.
48 Example:
49 ```python
50 input_mapping={
51 "query": ["retriever.query", "prompt_builder.query"],
52 }
53 ```
54 :param output_mapping: A dictionary mapping pipeline output socket paths to component output names.
55 If not provided, a default output mapping will be created based on all pipeline outputs.
56 Example:
57 ```python
58 output_mapping={
59 "retriever.documents": "documents",
60 "generator.replies": "replies",
61 }
62 ```
63 :raises InvalidMappingError: Raised if any mapping is invalid or type conflicts occur
64 :raises ValueError: Raised if no pipeline is provided
65 """
66 if pipeline is None:
67 raise ValueError("Pipeline must be provided to SuperComponent.")
69 self.pipeline: Pipeline = pipeline
71 # Determine input types based on pipeline and mapping
72 pipeline_inputs = self.pipeline.inputs()
73 resolved_input_mapping = (
74 input_mapping if input_mapping is not None else self._create_input_mapping(pipeline_inputs)
75 )
76 self._validate_input_mapping(pipeline_inputs, resolved_input_mapping)
77 input_types = self._resolve_input_types_from_mapping(pipeline_inputs, resolved_input_mapping)
78 # Set input types on the component
79 for input_name, info in input_types.items():
80 component.set_input_type(self, name=input_name, **info)
82 self.input_mapping: dict[str, list[str]] = resolved_input_mapping
83 self._original_input_mapping = input_mapping
85 # Set output types based on pipeline and mapping
86 leaf_pipeline_outputs = self.pipeline.outputs()
87 all_possible_pipeline_outputs = self.pipeline.outputs(include_components_with_connected_outputs=True)
89 resolved_output_mapping = (
90 output_mapping if output_mapping is not None else self._create_output_mapping(leaf_pipeline_outputs)
91 )
92 self._validate_output_mapping(all_possible_pipeline_outputs, resolved_output_mapping)
93 output_types = self._resolve_output_types_from_mapping(all_possible_pipeline_outputs, resolved_output_mapping)
94 # Set output types on the component
95 component.set_output_types(self, **output_types)
96 self.output_mapping: dict[str, str] = resolved_output_mapping
97 self._original_output_mapping = output_mapping
99 def warm_up(self) -> None:
100 """
101 Warms up the SuperComponent by warming up the wrapped pipeline.
102 """
103 self.pipeline.warm_up()
105 async def warm_up_async(self) -> None:
106 """
107 Warms up the SuperComponent by warming up the wrapped pipeline on the serving event loop.
108 """
109 await self.pipeline.warm_up_async()
111 def close(self) -> None:
112 """
113 Releases the synchronous resources held by the wrapped pipeline's components.
114 """
115 self.pipeline.close()
117 async def close_async(self) -> None:
118 """
119 Releases the async resources held by the wrapped pipeline's components.
120 """
121 await self.pipeline.close_async()
123 def run(self, **kwargs: Any) -> dict[str, Any]:
124 """
125 Runs the wrapped pipeline with the provided inputs.
127 Steps:
128 1. Maps the inputs from kwargs to pipeline component inputs
129 2. Runs the pipeline
130 3. Maps the pipeline outputs to the SuperComponent's outputs
132 :param kwargs: Keyword arguments matching the SuperComponent's input names
133 :returns:
134 Dictionary containing the SuperComponent's output values
135 """
136 # `is not`, not `!=`: numpy/pandas/torch override `__ne__` element-wise and would crash here.
137 filtered_inputs = {param: value for param, value in kwargs.items() if value is not _delegate_default}
138 pipeline_inputs = self._map_explicit_inputs(input_mapping=self.input_mapping, inputs=filtered_inputs)
139 include_outputs_from = self._get_include_outputs_from()
140 pipeline_outputs = self.pipeline.run(data=pipeline_inputs, include_outputs_from=include_outputs_from)
141 return self._map_explicit_outputs(pipeline_outputs, self.output_mapping)
143 def _get_include_outputs_from(self) -> set[str]:
144 # Collecting the component names from output_mapping
145 return {self._split_component_path(path)[0] for path in self.output_mapping.keys()}
147 async def run_async(self, **kwargs: Any) -> dict[str, Any]:
148 """
149 Runs the wrapped pipeline with the provided inputs async.
151 Steps:
152 1. Maps the inputs from kwargs to pipeline component inputs
153 2. Runs the pipeline async
154 3. Maps the pipeline outputs to the SuperComponent's outputs
156 :param kwargs: Keyword arguments matching the SuperComponent's input names
157 :returns:
158 Dictionary containing the SuperComponent's output values
159 """
160 # `is not`, not `!=`: numpy/pandas/torch override `__ne__` element-wise and would crash here.
161 filtered_inputs = {param: value for param, value in kwargs.items() if value is not _delegate_default}
162 pipeline_inputs = self._map_explicit_inputs(input_mapping=self.input_mapping, inputs=filtered_inputs)
163 pipeline_outputs = await self.pipeline.run_async(data=pipeline_inputs)
164 return self._map_explicit_outputs(pipeline_outputs, self.output_mapping)
166 @staticmethod
167 def _split_component_path(path: str) -> tuple[str, str]:
168 """
169 Splits a component path into a component name and a socket name.
171 :param path: A string in the format "component_name.socket_name".
172 :returns:
173 A tuple containing (component_name, socket_name).
174 :raises InvalidMappingValueError:
175 If the path format is incorrect.
176 """
177 comp_name, socket_name = parse_connect_string(path)
178 if socket_name is None:
179 raise InvalidMappingValueError(f"Invalid path format: '{path}'. Expected 'component_name.socket_name'.")
180 return comp_name, socket_name
182 def _validate_input_mapping(
183 self, pipeline_inputs: dict[str, dict[str, Any]], input_mapping: dict[str, list[str]]
184 ) -> None:
185 """
186 Validates the input mapping to ensure that specified components and sockets exist in the pipeline.
188 :param pipeline_inputs: A dictionary containing pipeline input specifications.
189 :param input_mapping: A dictionary mapping wrapper input names to pipeline socket paths.
190 :raises InvalidMappingTypeError:
191 If the input mapping is of invalid type or contains invalid types.
192 :raises InvalidMappingValueError:
193 If the input mapping contains nonexistent components or sockets.
194 """
195 if not isinstance(input_mapping, dict):
196 raise InvalidMappingTypeError("input_mapping must be a dictionary")
198 for wrapper_input_name, pipeline_input_paths in input_mapping.items():
199 if not isinstance(pipeline_input_paths, list):
200 raise InvalidMappingTypeError(f"Input paths for '{wrapper_input_name}' must be a list of strings.")
201 for path in pipeline_input_paths:
202 comp_name, socket_name = self._split_component_path(path)
203 if comp_name not in pipeline_inputs:
204 raise InvalidMappingValueError(
205 f"Component '{comp_name}' not found in pipeline inputs.\n"
206 f"Available components: {list(pipeline_inputs.keys())}"
207 )
208 if socket_name not in pipeline_inputs[comp_name]:
209 raise InvalidMappingValueError(
210 f"Input socket '{socket_name}' not found in component '{comp_name}'.\n"
211 f"Available inputs for '{comp_name}': {list(pipeline_inputs[comp_name].keys())}"
212 )
214 def _resolve_input_types_from_mapping(
215 self, pipeline_inputs: dict[str, dict[str, Any]], input_mapping: dict[str, list[str]]
216 ) -> dict[str, dict[str, Any]]:
217 """
218 Resolves and validates input types based on the provided input mapping.
220 This function ensures that all mapped pipeline inputs are compatible, consolidating types
221 when multiple mappings exist. It also determines whether an input is mandatory or has a default value.
223 :param pipeline_inputs: A dictionary containing pipeline input specifications.
224 :param input_mapping: A dictionary mapping SuperComponent inputs to pipeline socket paths.
225 :returns:
226 A dictionary specifying the resolved input types and their properties.
227 :raises InvalidMappingTypeError:
228 If the input mapping contains incompatible types.
229 """
230 aggregated_inputs: dict[str, dict[str, Any]] = {}
231 for wrapper_input_name, pipeline_input_paths in input_mapping.items():
232 for path in pipeline_input_paths:
233 comp_name, socket_name = self._split_component_path(path)
234 socket_info = pipeline_inputs[comp_name][socket_name]
236 # Add to aggregated inputs
237 existing_socket_info = aggregated_inputs.get(wrapper_input_name)
238 if existing_socket_info is None:
239 aggregated_inputs[wrapper_input_name] = {"type": socket_info["type"]}
240 if not socket_info["is_mandatory"]:
241 aggregated_inputs[wrapper_input_name]["default"] = _delegate_default
242 continue
244 is_compatible, common_type = _is_compatible(existing_socket_info["type"], socket_info["type"])
246 if not is_compatible:
247 raise InvalidMappingTypeError(
248 f"Type conflict for input '{socket_name}' from component '{comp_name}'. "
249 f"Existing type: {existing_socket_info['type']}, new type: {socket_info['type']}."
250 )
252 # Use the common type for the aggregated input
253 aggregated_inputs[wrapper_input_name]["type"] = common_type
255 # If any socket requires mandatory inputs then the aggregated input is also considered mandatory.
256 # So we use the type of the mandatory input and remove the default value if it exists.
257 if socket_info["is_mandatory"]:
258 aggregated_inputs[wrapper_input_name].pop("default", None)
260 return aggregated_inputs
262 @staticmethod
263 def _create_input_mapping(pipeline_inputs: dict[str, dict[str, Any]]) -> dict[str, list[str]]:
264 """
265 Create an input mapping from pipeline inputs.
267 :param pipeline_inputs: Dictionary of pipeline input specifications
268 :returns:
269 Dictionary mapping SuperComponent input names to pipeline socket paths
270 """
271 input_mapping: dict[str, list[str]] = {}
272 for comp_name, inputs_dict in pipeline_inputs.items():
273 for socket_name in inputs_dict.keys():
274 existing_socket_info = input_mapping.get(socket_name)
275 if existing_socket_info is None:
276 input_mapping[socket_name] = [f"{comp_name}.{socket_name}"]
277 continue
278 input_mapping[socket_name].append(f"{comp_name}.{socket_name}")
279 return input_mapping
281 def _validate_output_mapping(
282 self, pipeline_outputs: dict[str, dict[str, Any]], output_mapping: dict[str, str]
283 ) -> None:
284 """
285 Validates the output mapping to ensure that specified components and sockets exist in the pipeline.
287 :param pipeline_outputs: A dictionary containing pipeline output specifications.
288 :param output_mapping: A dictionary mapping pipeline socket paths to wrapper output names.
289 :raises InvalidMappingTypeError:
290 If the output mapping is of invalid type or contains invalid types.
291 :raises InvalidMappingValueError:
292 If the output mapping contains nonexistent components or sockets.
293 """
294 for pipeline_output_path, wrapper_output_name in output_mapping.items():
295 if not isinstance(wrapper_output_name, str):
296 raise InvalidMappingTypeError("Output names in output_mapping must be strings.")
297 comp_name, socket_name = self._split_component_path(pipeline_output_path)
298 if comp_name not in pipeline_outputs:
299 raise InvalidMappingValueError(f"Component '{comp_name}' not found among pipeline outputs.")
300 if socket_name not in pipeline_outputs[comp_name]:
301 raise InvalidMappingValueError(f"Output socket '{socket_name}' not found in component '{comp_name}'.")
303 def _resolve_output_types_from_mapping(
304 self, pipeline_outputs: dict[str, dict[str, Any]], output_mapping: dict[str, str]
305 ) -> dict[str, Any]:
306 """
307 Resolves and validates output types based on the provided output mapping.
309 This function ensures that all mapped pipeline outputs are correctly assigned to
310 the corresponding SuperComponent outputs while preventing duplicate output names.
312 :param pipeline_outputs: A dictionary containing pipeline output specifications.
313 :param output_mapping: A dictionary mapping pipeline output socket paths to SuperComponent output names.
314 :returns:
315 A dictionary mapping SuperComponent output names to their resolved types.
316 :raises InvalidMappingValueError:
317 If the output mapping contains duplicate output names.
318 """
319 resolved_outputs = {}
320 for pipeline_output_path, wrapper_output_name in output_mapping.items():
321 comp_name, socket_name = self._split_component_path(pipeline_output_path)
322 if wrapper_output_name in resolved_outputs:
323 raise InvalidMappingValueError(f"Duplicate output name '{wrapper_output_name}' in output_mapping.")
324 resolved_outputs[wrapper_output_name] = pipeline_outputs[comp_name][socket_name]["type"]
325 return resolved_outputs
327 @staticmethod
328 def _create_output_mapping(pipeline_outputs: dict[str, dict[str, Any]]) -> dict[str, str]:
329 """
330 Create an output mapping from pipeline outputs.
332 :param pipeline_outputs: Dictionary of pipeline output specifications
333 :returns:
334 Dictionary mapping pipeline socket paths to SuperComponent output names
335 :raises InvalidMappingValueError:
336 If there are output name conflicts between components
337 """
338 output_mapping = {}
339 used_output_names: set[str] = set()
340 for comp_name, outputs_dict in pipeline_outputs.items():
341 for socket_name in outputs_dict.keys():
342 if socket_name in used_output_names:
343 raise InvalidMappingValueError(
344 f"Output name conflict: '{socket_name}' is produced by multiple components. "
345 "Please provide an output_mapping to resolve this conflict."
346 )
347 used_output_names.add(socket_name)
348 output_mapping[f"{comp_name}.{socket_name}"] = socket_name
349 return output_mapping
351 def _map_explicit_inputs(
352 self, input_mapping: dict[str, list[str]], inputs: dict[str, Any]
353 ) -> dict[str, dict[str, Any]]:
354 """
355 Map inputs according to explicit input mapping.
357 :param input_mapping: Mapping configuration for inputs
358 :param inputs: Input arguments provided to wrapper
359 :return: Dictionary of mapped pipeline inputs
360 """
361 pipeline_inputs: dict[str, dict[str, Any]] = {}
362 for wrapper_input_name, pipeline_input_paths in input_mapping.items():
363 if wrapper_input_name not in inputs:
364 continue
366 for socket_path in pipeline_input_paths:
367 comp_name, input_name = self._split_component_path(socket_path)
368 if comp_name not in pipeline_inputs:
369 pipeline_inputs[comp_name] = {}
370 pipeline_inputs[comp_name][input_name] = inputs[wrapper_input_name]
372 return pipeline_inputs
374 def _map_explicit_outputs(
375 self, pipeline_outputs: dict[str, dict[str, Any]], output_mapping: dict[str, str]
376 ) -> dict[str, Any]:
377 """
378 Map outputs according to explicit output mapping.
380 :param pipeline_outputs: Raw outputs from pipeline execution
381 :param output_mapping: Output mapping configuration
382 :return: Dictionary of mapped outputs
383 """
384 outputs: dict[str, Any] = {}
385 for pipeline_output_path, wrapper_output_name in output_mapping.items():
386 comp_name, socket_name = self._split_component_path(pipeline_output_path)
387 if comp_name in pipeline_outputs and socket_name in pipeline_outputs[comp_name]:
388 outputs[wrapper_output_name] = pipeline_outputs[comp_name][socket_name]
389 return outputs
391 def _to_super_component_dict(self) -> dict[str, Any]:
392 """
393 Convert to a SuperComponent dictionary representation.
395 :return: Dictionary containing serialized SuperComponent data
396 """
397 serialized_pipeline = self.pipeline.to_dict()
398 serialized = default_to_dict(
399 self,
400 pipeline=serialized_pipeline,
401 input_mapping=self._original_input_mapping,
402 output_mapping=self._original_output_mapping,
403 )
404 serialized["type"] = generate_qualified_class_name(SuperComponent)
405 return serialized
408@component
409class SuperComponent(_SuperComponent):
410 """
411 A class for creating super components that wrap around a Pipeline.
413 This component allows for remapping of input and output socket names between the wrapped pipeline and the
414 SuperComponent's input and output names. This is useful for creating higher-level components that abstract
415 away the details of the wrapped pipeline.
417 ### Usage example
419 ```python
420 from haystack import Pipeline, SuperComponent
421 from haystack.components.generators.chat import OpenAIChatGenerator
422 from haystack.components.builders import ChatPromptBuilder
423 from haystack.components.retrievers import InMemoryBM25Retriever
424 from haystack.dataclasses.chat_message import ChatMessage
425 from haystack.document_stores.in_memory import InMemoryDocumentStore
426 from haystack.dataclasses import Document
428 document_store = InMemoryDocumentStore()
429 documents = [
430 Document(content="Paris is the capital of France."),
431 Document(content="London is the capital of England."),
432 ]
433 document_store.write_documents(documents)
435 prompt_template = [
436 ChatMessage.from_user(
437 '''
438 According to the following documents:
439 {% for document in documents %}
440 {{document.content}}
441 {% endfor %}
442 Answer the given question: {{query}}
443 Answer:
444 '''
445 )
446 ]
448 prompt_builder = ChatPromptBuilder(template=prompt_template, required_variables="*")
450 pipeline = Pipeline()
451 pipeline.add_component("retriever", InMemoryBM25Retriever(document_store=document_store))
452 pipeline.add_component("prompt_builder", prompt_builder)
453 pipeline.add_component("llm", OpenAIChatGenerator())
454 pipeline.connect("retriever.documents", "prompt_builder.documents")
455 pipeline.connect("prompt_builder.prompt", "llm.messages")
457 # Create a super component with simplified input/output mapping
458 wrapper = SuperComponent(
459 pipeline=pipeline,
460 input_mapping={
461 "query": ["retriever.query", "prompt_builder.query"],
462 },
463 output_mapping={"llm.replies": "replies"}
464 )
466 # Run the pipeline with simplified interface
467 result = wrapper.run(query="What is the capital of France?")
468 print(result)
469 {'replies': [ChatMessage(_role=<ChatRole.ASSISTANT: 'assistant'>,
470 _content=[TextContent(text='The capital of France is Paris.')],...)
471 ```
473 """
475 def to_dict(self) -> dict[str, Any]:
476 """
477 Serializes the SuperComponent into a dictionary.
479 :returns:
480 Dictionary with serialized data.
481 """
482 return self._to_super_component_dict()
484 @classmethod
485 def from_dict(cls, data: dict[str, Any]) -> "SuperComponent":
486 """
487 Deserializes the SuperComponent from a dictionary.
489 :param data: The dictionary to deserialize from.
490 :returns:
491 The deserialized SuperComponent.
492 """
493 # `is_pipeline_async` is a legacy key kept only for backward compatibility.
494 data["init_parameters"].pop("is_pipeline_async", None)
495 pipeline = Pipeline.from_dict(data["init_parameters"]["pipeline"])
496 data["init_parameters"]["pipeline"] = pipeline
497 return default_from_dict(cls, data)
499 def show(self, server_url: str = "https://mermaid.ink", params: dict | None = None, timeout: int = 30) -> None:
500 """
501 Display an image representing this SuperComponent's underlying pipeline in a Jupyter notebook.
503 This function generates a diagram of the Pipeline using a Mermaid server and displays it directly in
504 the notebook.
506 :param server_url:
507 The base URL of the Mermaid server used for rendering (default: 'https://mermaid.ink').
508 See https://github.com/jihchi/mermaid.ink and https://github.com/mermaid-js/mermaid-live-editor for more
509 info on how to set up your own Mermaid server.
511 :param params:
512 Dictionary of customization parameters to modify the output. Refer to Mermaid documentation for more details
513 Supported keys:
514 - format: Output format ('img', 'svg', or 'pdf'). Default: 'img'.
515 - type: Image type for /img endpoint ('jpeg', 'png', 'webp'). Default: 'png'.
516 - theme: Mermaid theme ('default', 'neutral', 'dark', 'forest'). Default: 'neutral'.
517 - bgColor: Background color in hexadecimal (e.g., 'FFFFFF') or named format (e.g., '!white').
518 - width: Width of the output image (integer).
519 - height: Height of the output image (integer).
520 - scale: Scaling factor (1–3). Only applicable if 'width' or 'height' is specified.
521 - fit: Whether to fit the diagram size to the page (PDF only, boolean).
522 - paper: Paper size for PDFs (e.g., 'a4', 'a3'). Ignored if 'fit' is true.
523 - landscape: Landscape orientation for PDFs (boolean). Ignored if 'fit' is true.
525 :param timeout:
526 Timeout in seconds for the request to the Mermaid server.
528 :raises PipelineDrawingError:
529 If the function is called outside of a Jupyter notebook or if there is an issue with rendering.
530 """
531 self.pipeline.show(server_url=server_url, params=params, timeout=timeout)
533 def draw(
534 self, path: Path, server_url: str = "https://mermaid.ink", params: dict | None = None, timeout: int = 30
535 ) -> None:
536 """
537 Save an image representing this SuperComponent's underlying pipeline to the specified file path.
539 This function generates a diagram of the Pipeline using the Mermaid server and saves it to the provided path.
541 :param path:
542 The file path where the generated image will be saved.
543 :param server_url:
544 The base URL of the Mermaid server used for rendering (default: 'https://mermaid.ink').
545 See https://github.com/jihchi/mermaid.ink and https://github.com/mermaid-js/mermaid-live-editor for more
546 info on how to set up your own Mermaid server.
547 :param params:
548 Dictionary of customization parameters to modify the output. Refer to Mermaid documentation for more details
549 Supported keys:
550 - format: Output format ('img', 'svg', or 'pdf'). Default: 'img'.
551 - type: Image type for /img endpoint ('jpeg', 'png', 'webp'). Default: 'png'.
552 - theme: Mermaid theme ('default', 'neutral', 'dark', 'forest'). Default: 'neutral'.
553 - bgColor: Background color in hexadecimal (e.g., 'FFFFFF') or named format (e.g., '!white').
554 - width: Width of the output image (integer).
555 - height: Height of the output image (integer).
556 - scale: Scaling factor (1–3). Only applicable if 'width' or 'height' is specified.
557 - fit: Whether to fit the diagram size to the page (PDF only, boolean).
558 - paper: Paper size for PDFs (e.g., 'a4', 'a3'). Ignored if 'fit' is true.
559 - landscape: Landscape orientation for PDFs (boolean). Ignored if 'fit' is true.
561 :param timeout:
562 Timeout in seconds for the request to the Mermaid server.
564 :raises PipelineDrawingError:
565 If there is an issue with rendering or saving the image.
566 """
567 self.pipeline.draw(path=path, server_url=server_url, params=params, timeout=timeout)
570def super_component(cls: type[T]) -> type[T]:
571 """
572 Decorator that converts a class into a SuperComponent.
574 This decorator:
575 1. Creates a new class that inherits from SuperComponent
576 2. Copies all methods and attributes from the original class
577 3. Adds initialization logic to properly set up the SuperComponent
579 The decorated class should define:
580 - pipeline: A Pipeline instance in the __init__ method
581 - input_mapping: Dictionary mapping component inputs to pipeline inputs (optional)
582 - output_mapping: Dictionary mapping pipeline outputs to component outputs (optional)
583 """
584 logger.debug("Registering {cls} as a super_component", cls=cls)
586 # Store the original __init__ method
587 original_init = cls.__init__
589 # Create a new __init__ method that will initialize both the original class and SuperComponent
590 def init_wrapper(self: Any, *args: Any, **kwargs: Any) -> None:
591 # Call the original __init__ to set up pipeline and mappings
592 original_init(self, *args, **kwargs)
594 # Verify required attributes
595 if not hasattr(self, "pipeline"):
596 raise ValueError(f"Class {cls.__name__} decorated with @super_component must define a 'pipeline' attribute")
598 # Initialize SuperComponent
599 _SuperComponent.__init__(
600 self,
601 pipeline=self.pipeline,
602 input_mapping=getattr(self, "input_mapping", None),
603 output_mapping=getattr(self, "output_mapping", None),
604 )
606 # Preserve original init's signature for IDEs/docs/tools
607 init_wrapper = functools.wraps(original_init)(init_wrapper)
609 # Function to copy namespace from the original class
610 def copy_class_namespace(namespace: dict[str, Any]) -> None:
611 """Copy all attributes from the original class except special ones."""
612 for key, val in dict(cls.__dict__).items():
613 # Skip special attributes that should be recreated
614 if key in ("__dict__", "__weakref__"):
615 continue
617 # Override __init__ with our wrapper
618 if key == "__init__":
619 namespace["__init__"] = init_wrapper
620 continue
622 namespace[key] = val
624 # Create a new class inheriting from SuperComponent with the original methods
625 # We use (SuperComponent,) + cls.__bases__ to make the new class inherit from
626 # SuperComponent and all the original class's bases
627 new_cls = new_class(cls.__name__, (_SuperComponent,) + cls.__bases__, {}, copy_class_namespace)
629 # Copy other class attributes
630 new_cls.__module__ = cls.__module__
631 new_cls.__qualname__ = cls.__qualname__
632 new_cls.__doc__ = cls.__doc__
634 # Apply the component decorator to the new class
635 return component(new_cls)