feat: AsyncPipeline that can schedule components to run concurrently

[mathislucka]

Related Issues

• closes components should run concurrently when not explicitly waiting on inputs #8453

Proposed Changes:

Implements an AsyncPipeline that supports:

• running pipelines asynchronously
• step-by-step execution through an async generator
• concurrent execution of components whenever possible (e.g. hybrid retrieval, multiple generators that can run in parallel)
• sync run-method with concurrent execution of components

How did you test it?

• unit tests
• adapted behavioral tests to use Pipeline and AsyncPipeline

Notes for the reviewer

Review after #8707
Code was reviewed here before: deepset-ai/haystack-experimental#180

Checklist

• I have read the contributors guidelines and the code of conduct
• I have updated the related issue with new insights and changes
• I added unit tests and updated the docstrings
• I've used one of the conventional commit types for my PR title: fix:, feat:, build:, chore:, ci:, docs:, style:, refactor:, perf:, test: and added ! in case the PR includes breaking changes.
• I documented my code
• I ran pre-commit hooks and fixed any issue

[mathislucka]

@Amnah199 @davidsbatista much smaller diff now that the other PR is merged.

This is largely the same as the PR that we already merged to experimental with the following differences:

• fixed bug where we didn't wait long enough for DEFER(_LAST)
• added pipeline type to telemetry

[davidsbatista]

(nit) suggestion: keeping these imports ordered alphabetically helps locate something as the list grows

__all__ = [
    "Answer",
    "AsyncPipeline",
    "ComponentError",
    "DeserializationError",
    "Document",
    "ExtractedAnswer",
    "GeneratedAnswer",
    "Pipeline",
    "PredefinedPipeline",
    "component",
    "default_from_dict",
    "default_to_dict",
]

[davidsbatista]

I did another quick review, although most of this was already reviewed before

From my side it's approved, but to play safe, let's wait for Amna to also do another quick review before merging.

[davidsbatista]

LGTM

[Amnah199]

Here as well, we can avoid manual handling of loops by using asyncio.run, if you feel that would be better.

[Amnah199]

Something like this? (although I didnt test it)

Suggested change

	def test_async_pipeline_reentrance(waiting_component, spying_tracer):
	pp = AsyncPipeline()
	pp.add_component("wait", waiting_component())
	run_data = [{"wait_for": 1}, {"wait_for": 2}]
	async_loop = asyncio.new_event_loop()
	asyncio.set_event_loop(async_loop)
	async def run_all():
	# Create concurrent tasks for each pipeline run
	tasks = [pp.run_async(data) for data in run_data]
	await asyncio.gather(*tasks)
	try:
	async_loop.run_until_complete(run_all())
	component_spans = [sp for sp in spying_tracer.spans if sp.operation_name == "haystack.component.run_async"]
	for span in component_spans:
	assert span.tags["haystack.component.visits"] == 1
	finally:
	async_loop.close()
	def test_async_pipeline_reentrance(waiting_component, spying_tracer):
	"""
	Test that the AsyncPipeline can execute multiple runs concurrently and that
	each component is called exactly once per run (as indicated by the 'visits' tag).
	"""
	async_pipeline = AsyncPipeline()
	async_pipeline.add_component("wait", waiting_component())
	run_data = [{"wait_for": 1}, {"wait_for": 2}]
	async def run_all():
	tasks = [async_pipeline.run_async(data) for data in run_data]
	await asyncio.gather(*tasks)
	component_spans = [
	sp for sp in spying_tracer.spans
	if sp.operation_name == "haystack.component.run_async"
	]
	for span in component_spans:
	expected_visits = 1
	actual_visits = span.tags.get("haystack.component.visits")
	assert actual_visits == expected_visits, (
	f"Expected {expected_visits} visit, got {actual_visits} for span {span}"
	)
	# Use asyncio.run to manage the event loop.
	asyncio.run(run_all())

[Amnah199]

LGTM! Thanks again @mathislucka.
Much appreciated!