How to create a custom component in Haystack

Built-in Haystack components cover common retrieval and generation tasks, but project rules often need their own typed place in a pipeline. A custom component turns that Python logic into a reusable pipeline node whose connections Haystack can validate before data starts moving.

The @component decorator registers the class, parameters on run() define its input sockets, and @component.output_types declares the names and types it returns. Each returned dictionary key must match a declared output socket.

The ticket router uses constructor state to hold priority terms and sends two outputs to a separate summary component. Running opposite ticket subjects through the completed graph demonstrates that the sockets carry values produced from the input instead of a fixed success marker.

1. Create custom_component_demo.py with the imports and TicketRouter constructor.

   custom_component_demo.py

   import sys
    
   from haystack import Pipeline, component
    
    
   @component
   class TicketRouter:
       def __init__(self, priority_terms=None):
           self.priority_terms = priority_terms or ["outage", "security", "payment"]

   The constructor keeps the routing terms on each component instance so the same class can support a different policy when it is instantiated.

2. Insert the typed run() method inside TicketRouter after __init__().

       @component.output_types(route=str, reason=str)
       def run(self, subject: str):
           matched_terms = [
               term for term in self.priority_terms if term in subject.casefold()
           ]
           if matched_terms:
               return {
                   "route": "priority-support",
                   "reason": "matched " + ", ".join(matched_terms),
               }
           return {
               "route": "standard-support",
               "reason": "no priority terms",
           }

   The subject annotation defines the input socket. The route and reason keys match the output names declared by @component.output_types.

3. Add RoutingSummary below TicketRouter.

   @component
   class RoutingSummary:
       @component.output_types(summary=str)
       def run(self, route: str, reason: str):
           return {"summary": f"{route}: {reason}"}

   The parameter names give RoutingSummary one receiving socket for each value emitted by TicketRouter.

4. Append the command-line input and pipeline graph below both component classes.

   subject = " ".join(sys.argv[1:]) or "Payment outage in checkout"
    
   pipeline = Pipeline()
   pipeline.add_component("router", TicketRouter())
   pipeline.add_component("summary", RoutingSummary())
   pipeline.connect("router.route", "summary.route")
   pipeline.connect("router.reason", "summary.reason")
    
   result = pipeline.run({"router": {"subject": subject}})
   print(result)

   Each connect() call names one sender output and its compatible receiver input. Pipeline.run() supplies only the external router.subject value.
   Related: How to create a pipeline in Haystack

5. Run the completed script with its default priority ticket.

   $ python custom_component_demo.py
   {'summary': {'summary': 'priority-support: matched outage, payment'}}

   The command requires a Python environment with haystack-ai installed. The leaf output contains values that passed through both explicit router connections.
   Related: How to install Haystack with pip

6. Verify the standard route with a subject that contains no priority term.

   $ python custom_component_demo.py "Password reset request for customer portal"
   {'summary': {'summary': 'standard-support: no priority terms'}}

   The unchanged pipeline graph produces a different leaf summary, confirming that TicketRouter.run() evaluated the supplied subject.