Connections: how pipelines check your wiring

connect checks that sockets exist and their types fit before anything runs. A typo or a wrong type fails at once, and the message lists the valid options.

pipeline.connect("cleaner.txt", "counter.text")

txt is not an output of cleaner, and the error lists the ones that exist with their types. A pipeline with a wiring mistake never starts.

@component
class Doubler:
    @component.output_types(value=int)
    def run(self, value: int):
        return {"value": value * 2}


pipeline.add_component("doubler", Doubler())
pipeline.connect("counter.words", "doubler.value")

A list[str] cannot become an int, so the types are rejected at connect.

Connections that adapt

@component
class Longest:
    @component.output_types(word=str)
    def run(self, words: list[str]):
        return {"word": max(words, key=len)}


pipeline = Pipeline()
pipeline.add_component("cleaner", Cleaner())
pipeline.add_component("longest", Longest())
pipeline.connect("cleaner.text", "longest.words")
print(pipeline.run({"cleaner": {"text": "Parcels"}}))

A str into a list[str] input is accepted: Haystack wraps it in a one-item list. The same smart connections let several list outputs feed one list input, joined, and turn a str into a ChatMessage. When types match exactly, the exact match is used.

Getting inner outputs

result = pipeline.run({"cleaner": {"text": "  Where is   MY parcel?  "}}, include_outputs_from={"cleaner"})
print(result["cleaner"])
print(pipeline.inputs())

include_outputs_from adds a component's outputs to the result even though another component used them. inputs() lists the inputs with no connection, the ones you must or may pass to run.