How to create a branching pipeline in Haystack

A pipeline branch sends one component output into more than one downstream path. In Haystack, this keeps shared preprocessing in one place while separate components perform different work with the same value.

Haystack validates output and input socket types through Pipeline.connect() while it builds the graph. Connecting normalizer.text to two compatible input sockets creates the fan-out point without duplicating the producer component.

Three local custom components keep the branch independent of model credentials and document stores. The final run proves the branch by returning a keyword list and a summary from two leaf components that received the normalized query.

1. Create pipeline_create_branch_demo.py with the QueryNormalizer producer.

   pipeline_create_branch_demo.py

   from haystack import Pipeline, component
    
    
   @component
   class QueryNormalizer:
       @component.output_types(text=str)
       def run(self, query: str):
           normalized = " ".join(query.strip().split()).lower()
           return {"text": normalized}

   The text output socket carries one normalized string to every connected consumer.

2. Add the KeywordScanner component below QueryNormalizer.

    
   @component
   class KeywordScanner:
       @component.output_types(keywords=list[str])
       def run(self, text: str):
           tracked_terms = ["branches", "components", "pipelines"]
           matches = [term for term in tracked_terms if term in text]
           return {"keywords": matches}

3. Add the BranchSummary component below KeywordScanner.

    
   @component
   class BranchSummary:
       @component.output_types(summary=str)
       def run(self, text: str):
           return {"summary": " ".join(text.split()[:6])}

4. Register the producer and both branch components below the class definitions.

    
   pipeline = Pipeline()
   pipeline.add_component("normalizer", QueryNormalizer())
   pipeline.add_component("keyword_scanner", KeywordScanner())
   pipeline.add_component("branch_summary", BranchSummary())

5. Connect the shared producer output to both downstream inputs below the component registrations.

    
   pipeline.connect("normalizer.text", "keyword_scanner.text")
   pipeline.connect("normalizer.text", "branch_summary.text")

   Each connect() call adds one outgoing edge from normalizer.text. Haystack rejects an incompatible socket type when the connection is created.

6. Add the Pipeline.run() call below the branch connections.

    
   result = pipeline.run(
       {
           "normalizer": {
               "query": "Haystack pipelines create branches across components",
           }
       }
   )

7. Add the result display statements below the Pipeline.run() call.

    
   print(
       "keyword_scanner.keywords="
       + ", ".join(result["keyword_scanner"]["keywords"])
   )
   print("branch_summary.summary=" + result["branch_summary"]["summary"])

8. Run the completed script to confirm that both downstream components return values from the normalized query.

   $ python pipeline_create_branch_demo.py
   keyword_scanner.keywords=branches, components, pipelines
   branch_summary.summary=haystack pipelines create branches across components

   The script requires a Python environment with haystack-ai installed.
   Related: How to install Haystack with pip