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 can schedule those branches concurrently so one slow request does not force another ready branch to wait.

The AsyncPipeline.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 asynchronous 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. The two leaf components in async_pipeline_demo.py share a synchronization gate: each branch waits until the other branch has started, so a sequential execution times out instead of printing a false success result.

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

   async_pipeline_demo.py

   import asyncio
    
   from haystack import AsyncPipeline, 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 event opens only after both branch names are present. A pipeline that runs these components one at a time cannot pass the gate.

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("AsyncPipeline 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() performs the non-blocking work used by AsyncPipeline.

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

   async def main():
       gate = BranchGate(branch_count=2)
       pipeline = AsyncPipeline()
       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 lower limit suits components with stricter rate or resource constraints.

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

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

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.