How to run an async pipeline in Haystack

Pipeline graphs often have independent I/O branches, such as retrievers or API-backed components that become ready at the same point. Haystack 3 runs an async pipeline through Pipeline so one slow request does not force another ready branch to wait.

The Pipeline.run_async() method waits for the graph to finish and returns its leaf outputs. A custom component must still provide run() for the component contract; adding run_async() gives the pipeline a non-blocking implementation to await. The run_async_generator() method is the alternative when the caller needs component outputs as they arrive.

The concurrency_limit parameter caps the number of components that may run together. Only components whose required inputs are available can start; a dependent branch still waits for its upstream result. Asynchronous execution overlaps waiting work without making CPU-bound Python code run in parallel.

1. Create async_pipeline_demo.py with the required imports and shared branch gate.

   async_pipeline_demo.py

   import asyncio
    
   from haystack import Pipeline, component
    
    
   class BranchGate:
       def __init__(self, branch_count: int):
           self.branch_count = branch_count
           self.ready = set()
           self.all_ready = asyncio.Event()
    
       async def wait_for_peers(self, branch_name: str):
           self.ready.add(branch_name)
           if len(self.ready) == self.branch_count:
               self.all_ready.set()
           await asyncio.wait_for(self.all_ready.wait(), timeout=1)

   The active Python 3.11+ environment needs haystack-ai 3.x, where Pipeline replaces the separate AsyncPipeline class. The event opens only after both branch names are present, so sequential execution times out.
   Related: How to install Haystack with pip

2. Add ConcurrentResponder below BranchGate with synchronous and asynchronous component paths.

   @component
   class ConcurrentResponder:
       def __init__(self, name: str, gate: BranchGate):
           self.name = name
           self.gate = gate
    
       @component.output_types(reply=str)
       def run(self, query: str):
           raise RuntimeError("Pipeline used the synchronous component path")
    
       @component.output_types(reply=str)
       async def run_async(self, query: str):
           await self.gate.wait_for_peers(self.name)
           await asyncio.sleep(0.05)
           return {"reply": f"{self.name}: {query}"}

   run() satisfies the component contract and fails visibly if the synchronous path is selected. run_async() waits at the gate and then simulates an I/O delay without blocking the event loop.

3. Add main() below ConcurrentResponder to create two independent pipeline leaves.

   async def main():
       gate = BranchGate(branch_count=2)
       pipeline = Pipeline()
       pipeline.add_component("docs_branch", ConcurrentResponder("docs", gate))
       pipeline.add_component("faq_branch", ConcurrentResponder("faq", gate))

   Neither leaf depends on the other. Both receive top-level input and become ready together.

4. Extend main() with the asynchronous pipeline call and result output.

       result = await pipeline.run_async(
           data={
               "docs_branch": {"query": "reset password"},
               "faq_branch": {"query": "reset password"},
           },
           concurrency_limit=2,
       )
    
       for branch_name in sorted(result):
           print(result[branch_name]["reply"])
       print(f"branches ready together: {', '.join(sorted(gate.ready))}")

   A limit of two allows both ready leaves to enter run_async(). A limit of one makes the first branch time out at the gate because the second branch cannot start. Lower limits are suitable for independent work that does not wait for another running component.

5. Append the asyncio entry point after main().

   if __name__ == "__main__":
       asyncio.run(main())

   asyncio.run() creates the event loop for a Python script. In a notebook or application with an active event loop, await main() runs the same coroutine without starting another loop.

6. Run the completed async pipeline script to confirm both branches pass the synchronization gate.

   $ python3 async_pipeline_demo.py
   docs: reset password
   faq: reset password
   branches ready together: docs, faq

   The two replies prove that both leaf outputs returned. The final line is derived from the shared gate, which cannot open until both asynchronous component calls are active.