How to create a pipeline in Haystack

Reusable Haystack applications depend on a graph that makes each processing boundary explicit. A Pipeline holds that graph so component inputs, outputs, and execution can be assembled once and invoked with different data.

Each component receives a unique name, while connect() joins one named output socket to a compatible input socket. Runtime data enters only through inputs that do not already receive a value from another component.

The local pipeline uses two custom components and no network service or API key. Its final answer contains a normalized form of the supplied query, so changed spacing and capitalization prove that the value crossed the connection instead of coming from a fixed success message.

1. Create pipeline_create_demo.py with the Haystack imports and query-normalizing component.

   pipeline_create_demo.py

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

   The query parameter becomes an input socket, and the returned clean_query key matches the output declared by @component.output_types.

2. Add the AnswerTemplate component below QueryNormalizer.

   @component
   class AnswerTemplate:
       @component.output_types(answer=str)
       def run(self, clean_query: str):
           return {"answer": f"Pipeline received: {clean_query}"}

   The clean_query parameter gives the second component an input socket that can accept the first component's string output.

3. Append the named pipeline graph below both component classes.

   pipeline = Pipeline()
   pipeline.add_component("normalizer", QueryNormalizer())
   pipeline.add_component("responder", AnswerTemplate())
   pipeline.connect("normalizer.clean_query", "responder.clean_query")

   The connection starts at the normalizer.clean_query output socket and ends at the responder.clean_query input socket.

4. Append the runtime query and downstream result output below the connection.

   query = " ".join(sys.argv[1:]) or "How do Haystack pipelines connect components?"
   result = pipeline.run({"normalizer": {"query": query}})
    
   print(f"input={query}")
   print(f"answer={result['responder']['answer']}")

   Pipeline.run() receives the unconnected normalizer.query input. The connected responder.clean_query socket receives its value from QueryNormalizer instead of the runtime dictionary.

5. Review the completed pipeline_create_demo.py file before execution.

   pipeline_create_demo.py

   import sys
    
   from haystack import Pipeline, component
    
    
   @component
   class QueryNormalizer:
       @component.output_types(clean_query=str)
       def run(self, query: str):
           clean_query = " ".join(query.strip().split()).casefold()
           return {"clean_query": clean_query}
    
    
   @component
   class AnswerTemplate:
       @component.output_types(answer=str)
       def run(self, clean_query: str):
           return {"answer": f"Pipeline received: {clean_query}"}
    
    
   pipeline = Pipeline()
   pipeline.add_component("normalizer", QueryNormalizer())
   pipeline.add_component("responder", AnswerTemplate())
   pipeline.connect("normalizer.clean_query", "responder.clean_query")
    
   query = " ".join(sys.argv[1:]) or "How do Haystack pipelines connect components?"
   result = pipeline.run({"normalizer": {"query": query}})
    
   print(f"input={query}")
   print(f"answer={result['responder']['answer']}")

6. Verify the completed pipeline with a query containing mixed case and repeated spaces.

   $ python pipeline_create_demo.py 'Route   this QUERY'
   input=Route   this QUERY
   answer=Pipeline received: route this query

   The lowercase, single-spaced answer proves that the runtime query passed through QueryNormalizer and reached AnswerTemplate over the named connection.
   Related: How to install Haystack with pip