Discussion on AsyncPipeline and async-enabled components

[julian-risch]

This is the discussion board for the experimental AsyncPipeline class and async-enabled components.

This experiment allows individual components to opt into async support. Not all components benefit from being async-enabled - I/O-bound components are the most suitable candidates. It also provides a backward-compatible way to execute Haystack pipelines containing both async and non-async components.

🧑‍🍳 Try the notebook here

Note: Experimental features live in this repository for a fixed period of time. We don't guarantee that we will continue maintaining experimental features. But if they are successful, stable and if you like it, we will move the feature to the core Haystack package.

[bilgeyucel]

The notebook is published as cookbook now 👉 https://haystack.deepset.ai/cookbook/async_pipeline

[d-kleine]

@julian-risch @bilgeyucel Might be a bit (too) late, but I just tested out the new AsyncPipeline feature of haystack 2.10 on a hybrid pipeline (the one you have provided with the Advent challenge, if you are interested to see that)
https://docs.haystack.deepset.ai/docs/asyncpipeline

From a user's perspective, I think it would be great to have some kind of validation once the async pipeline has been set up (ideally, before running it). Currently, you set up the pipeline and run it as, for instance...

async def run_inner(data, include_outputs_from):
    return await rag_pipeline.run_async(data=data, include_outputs_from=include_outputs_from)

data = {
    "retriever": {"query": question},
    "prompt_builder": {"question": question},
}

results = asyncio.run(run_inner(data, include_outputs_from={"retriever", "llm"}))

Currently, users do not receive clear feedback on the asynchronous processing of components or the order in which they flow through the pipeline. From my point of view, I think it would be beneficial to provide textual feedback to enhance visibility into the process, helping users validate if and how components are being handled asynchronously in the pipeline.

[phiweger]

Can all document stores be used with async retrievers? Or is this limited to certain classes such as the in-memory document store for now? If so, when do you plan to add support for other stores, I'm particularly interested in the pgvector one :) Thanks.

[julian-risch]

We are adding support with run_async methods in all document stores and their respective retrievers step-by-step. For pgvector, we're tracking the task with this issue and you can expect us to work on that issue within the next 3-4 weeks.

[phiweger]

that would be awesome!

[julian-risch]

This experiment was adopted into Haystack in version 2.10. With version 2.11 we rolled out run_async implementations for all retrievers and ChatGenerators in the core repository and we are continuing with retrievers and chat generators in haystack-core-integrations.