Update `SuperComponent` to specify if Pipeline or AsyncPipeline should be used in `from_dict`

[sjrl]

Currently in SuperComponent, the from_dict method does not specify whether the component should be loaded using Pipeline or AsyncPipeline. It just always uses Pipeline.from_dict. We should update the internal logic so to_dict specifies whether the component should be loaded using Pipeline or AsyncPipeline, and then from_dict should use that information.

This is especially problematic if loading a SuperComponent from yaml within an async pipeline. Since SuperComponent has a run_async method defined if it's placed in an AsyncPipeline at runtime AsyncPipeline will try and run the run_async method which will throw an error. This error is caused by SuperComponent complaining that it was given a normal Pipeline in it's init and not a AsyncPipeline. This means that it's not possible currently to use SuperComponent within an AsyncPipeline when loading from yaml.

Related to this issue #9435

[anakin87]

I started to investigate this issue, but I found it hard to reproduce this behavior.

What I tried, with no success:

from haystack.core.super_component import super_component
from haystack.core.component import component
from typing import Dict
from haystack import AsyncPipeline
import asyncio

@component
class HelloWorldComponent:
    @component.output_types(output=str)
    def run(self) -> Dict[str, str]:
        return {"output": "Hello world"}

    @component.output_types(output=str)
    async def run_async(self) -> Dict[str, str]:
        return {"output": "Hello world"}
    
@component
class AddFriendsComponent:
    @component.output_types(output=str)
    def run(self, initial_string: str) -> Dict[str, str]:
        return {"output": f"{initial_string} my friends"}

    @component.output_types(output=str)
    async def run_async(self, initial_string: str) -> Dict[str, str]:
        return {"output": f"{initial_string} my friends"}

@super_component
class MySuperComponent:
    def __init__(self):
       
        pipeline = AsyncPipeline()
        pipeline.add_component("hello", HelloWorldComponent())
        pipeline.add_component("add_friends", AddFriendsComponent())

        pipeline.connect("hello", "add_friends")

        self.pipeline = pipeline

my_super_component = MySuperComponent()
print(asyncio.run(my_super_component.run_async()))
# {'output': 'Hello world my friends'}

mypipe = AsyncPipeline()
mypipe.add_component("mysuper", MySuperComponent())
print(asyncio.run(mypipe.run_async({})))
# {'mysuper': {'output': 'Hello world my friends'}}

dumped = mypipe.dumps()
print(dumped)
# components:
#   mysuper:
#     init_parameters: {}
#     type: __main__.MySuperComponent
# connection_type_validation: true
# connections: []
# max_runs_per_component: 100
# metadata: {}
        
reloaded = AsyncPipeline.loads(dumped)
for name, component in reloaded.walk():
    print(name, type(component.pipeline))
# mysuper <class 'haystack.core.pipeline.async_pipeline.AsyncPipeline'>    
print(asyncio.run(reloaded.run_async({})))
# {'mysuper': {'output': 'Hello world my friends'}}

@sjrl if you can provide a code example, it would be helpful...

[sjrl]

@anakin87 good catch! You're right when using the @super_component decorator this won't be a problem since we are never calling SuperComponent().from_dict() we are just using the default component_to_dict and component_from_dict in this case and you always use whatever pipeline you define within the init method.

The error can reproduced if you define a super component using the class SuperComponent and passing in a pipeline in at init time. So something like

from haystack import AsyncPipeline
from haystack.core.super_component import SuperComponent

my_pipeline = AsyncPipeline()
...
sc = SuperComponent(pipeline=my_pipeline)
SuperComponent.from_dict(sc.to_dict())