From 3ad038abcdfe6f523c9de29eb5160ad0276875ef Mon Sep 17 00:00:00 2001 From: Harsh Kashyap <55448981+Harsh23Kashyap@users.noreply.github.com> Date: Sat, 15 Aug 2026 12:44:59 +0530 Subject: [PATCH] fix: allow multi-connection to Any-typed pipeline sockets MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `Pipeline.connect` rejected a second connection to an `Any`-typed input socket (e.g. `PromptBuilder.documents`, where every template variable is registered as `Any` via `component.set_input_type(self, var, Any)`) with a `PipelineConnectError`. The previous behaviour was the gap left after #10783 — its commit message claimed `Any`-type support, but `_make_socket_auto_variadic` only handled `list`, `Optional[list]`, and union-of-list receivers. The fix adds a dedicated branch for `Any`-typed receivers: when the second connection arrives, the socket is marked lazy variadic. The framework has no element-type information for an `Any`, so `wrap_input_in_list` is left at the default `True` and the receiver gets `[sender_1, sender_2, ...]` at runtime — a list of the per-sender values. The component decides how to handle the type-erased inputs (e.g. `PromptBuilder` can flatten the list of lists in its own `run()` if desired). Tests: added `TestMakeSocketAutoVariadic.test_successful_any_typed_socket` (direct unit test on the new branch) and `TestAnyTypedMultiConnection.test_prompt_builder_documents_accepts_multiple_retrievers` (end-to-end via `PromptBuilder`). Updated the error message regex in `TestValidateInput.test_validate_multiple_connections_to_non_variadic_input` to match the extended message ("or Any" appended). This is the connection-eligibility half of #10721; rendering-side aggregation (e.g. flattening inside `PromptBuilder`) is a natural follow-up since `Any` carries no list-element type to flatten against. Fixes #10721 --- haystack/core/pipeline/base.py | 31 ++++++++-- ...ped-multi-connection-b3e7d2f8a1c90546.yaml | 14 +++++ test/core/pipeline/test_pipeline_base.py | 57 ++++++++++++++++++- 3 files changed, 96 insertions(+), 6 deletions(-) create mode 100644 releasenotes/notes/fix-pipeline-any-typed-multi-connection-b3e7d2f8a1c90546.yaml diff --git a/haystack/core/pipeline/base.py b/haystack/core/pipeline/base.py index 531f9d6f797..aaa09bcbd81 100644 --- a/haystack/core/pipeline/base.py +++ b/haystack/core/pipeline/base.py @@ -1068,10 +1068,17 @@ def _make_socket_auto_variadic( A socket is automatically made lazy variadic when: - It already has at least one connected sender - It is not already variadic - - Its type is list, Optional[list], a union of list types - - When auto-variadicity is applied, `wrap_input_in_list` is also set to False so that sender output types match - the receiver socket's declared list type directly. + - Its type is list, Optional[list], a union of list types, or ``Any`` (the type-erased case; + common for components like ``PromptBuilder`` that register template variables as ``Any``) + + When auto-variadicity is applied: + - For a list-shaped receiver (``list`` / ``Optional[list]`` / union of lists), ``wrap_input_in_list`` + is set to ``False`` so the framework flattens one level and the component receives a single + concatenated list, matching the declared socket type. + - For an ``Any``-typed receiver, ``wrap_input_in_list`` is left at ``True`` so the component + receives a list of the per-sender values (``[sender_1, sender_2, ...]``) and can decide how + to handle the type-erased inputs (a component that knows it always aggregates a list of + lists may flatten them in ``run()``). :param component_name: Name of the component owning the receiver socket, used in error messages. @@ -1088,6 +1095,20 @@ def _make_socket_auto_variadic( if receiver_socket.is_variadic: return receiver_socket + # An ``Any``-typed socket can accept inputs from multiple senders. The framework cannot + # safely flatten one level (it does not know the element type), so the component receives + # ``[sender_1, sender_2, ...]`` via the existing ``wrap_input_in_list=True`` path. This is + # the gap left after #10783, whose commit message claimed ``Any`` support but only + # implemented ``list`` / ``Optional[list]`` / union-of-list handling. The most visible + # beneficiary is ``PromptBuilder``: every template variable is registered as ``Any`` + # (``component.set_input_type(self, var, Any)``), so a multi-connection from two + # retrievers to ``prompt_builder.documents`` used to fail before this branch was added. + if receiver_socket.type is Any: + receiver_socket.is_lazy_variadic = True + # ``wrap_input_in_list`` stays at its current value (True by default). The component + # sees ``[sender_1, sender_2, ...]`` and can flatten in its own ``run()`` if needed. + return receiver_socket + # Get receiver origin receiver_origin = _safe_get_origin(receiver_socket.type) @@ -1110,7 +1131,7 @@ def _make_socket_auto_variadic( raise error_type( f"Component '{component_name}' cannot accept multiple inputs to '{receiver_socket.name}'. " f"It is already connected to component '{receiver_socket.senders[0]}', and it can only accept " - f"inputs from multiple senders if its type is list, Optional[list], or union of list types." + f"inputs from multiple senders if its type is list, Optional[list], union of list types, or Any." ) def _prepare_component_input_data(self, data: dict[str, Any]) -> dict[str, dict[str, Any]]: diff --git a/releasenotes/notes/fix-pipeline-any-typed-multi-connection-b3e7d2f8a1c90546.yaml b/releasenotes/notes/fix-pipeline-any-typed-multi-connection-b3e7d2f8a1c90546.yaml new file mode 100644 index 00000000000..8543458c530 --- /dev/null +++ b/releasenotes/notes/fix-pipeline-any-typed-multi-connection-b3e7d2f8a1c90546.yaml @@ -0,0 +1,14 @@ +--- +fixes: + - | + Fixed ``Pipeline`` ``connect`` rejecting a second connection to an ``Any``-typed + input socket (e.g. ``PromptBuilder.documents``) with a ``PipelineConnectError``. + The previous behaviour was the gap left after #10783 — its commit message claimed + ``Any``-type support, but ``_make_socket_auto_variadic`` only handled ``list``, + ``Optional[list]``, and union-of-list receivers. An ``Any``-typed receiver is now + made lazy variadic on the second connection so multiple senders can attach. The + framework cannot safely flatten one level for an ``Any``-typed socket (it has no + element type to enforce), so the receiver gets ``[sender_1, sender_2, ...]`` and + is responsible for any per-sender aggregation. This is the connection-eligibility + half of #10721; rendering-side aggregation (e.g. flattening inside + ``PromptBuilder``) is a natural follow-up. diff --git a/test/core/pipeline/test_pipeline_base.py b/test/core/pipeline/test_pipeline_base.py index 367bc44f000..e01edbddafb 100644 --- a/test/core/pipeline/test_pipeline_base.py +++ b/test/core/pipeline/test_pipeline_base.py @@ -2554,6 +2554,26 @@ def test_successful(self, receiver_type, current_sender_type, new_sender_type): assert inp_socket.is_lazy_variadic is True assert inp_socket.wrap_input_in_list is False + def test_successful_any_typed_socket(self): + # An `Any`-typed receiver socket (e.g. every template variable in `PromptBuilder` is + # registered as `Any`) can be made lazy variadic so multiple senders can connect to it. + # The framework cannot safely flatten one level (it does not know the element type), so + # `wrap_input_in_list` is left at its default `True` and the component receives + # `[sender_1, sender_2, ...]`. + from typing import Any # local import to keep the parametrize block above self-contained + + pipe = PipelineBase() + inp_socket = pipe._make_socket_auto_variadic( + component_name="prompt_builder", + receiver_socket=InputSocket(name="documents", type=Any, senders=["retriever_1"]), + error_type=PipelineConnectError, + ) + assert inp_socket.is_variadic is True + assert inp_socket.is_lazy_variadic is True + # `wrap_input_in_list` must stay True: the framework has no way to know whether each + # sender's value is itself a list to flatten one level. The component decides. + assert inp_socket.wrap_input_in_list is True + def test_raises_error_all_int(self): with pytest.raises(PipelineConnectError): pipe = PipelineBase() @@ -2592,7 +2612,7 @@ def test_validate_multiple_connections_to_non_variadic_input(self): ValueError, match="Component 'comp2' cannot accept multiple inputs to 'input_'. " "It is already connected to component 'comp1', and it can only accept inputs from multiple " - r"senders if its type is list, Optional\[list\], or union of list types.", + r"senders if its type is list, Optional\[list\], union of list types, or Any.", ): pipe.validate_input(data={"comp1": {"input_": "test"}, "comp2": {"input_": "extra_input"}}) @@ -2710,3 +2730,38 @@ def test_blocking_components_from_two_branches(self): # actually blocking. assert blocking_comps == ["a_comp3", "blocking_comp"] assert blocking_comp_types == ["FakeComponent", "FakeComponent"] + + +# Integration test for the gap fixed in #10721 follow-up: a component whose input socket is typed +# as `Any` (e.g. every template variable in `PromptBuilder`) can now accept multiple senders. The +# connection used to fail with `PipelineConnectError` because #10783's commit message claimed `Any` +# support but the actual implementation only handled `list` / `Optional[list]` / union of lists. +class TestAnyTypedMultiConnection: + def test_prompt_builder_documents_accepts_multiple_retrievers(self): + from haystack import Document + from haystack.components.builders import PromptBuilder + from haystack.components.retrievers import InMemoryBM25Retriever + from haystack.document_stores.in_memory import InMemoryDocumentStore + + ds = InMemoryDocumentStore() + ds.write_documents([Document(content="Ferrari is fast", meta={"brand": "Ferrari"})]) + ds.write_documents([Document(content="Porsche is fast", meta={"brand": "Porsche"})]) + + pipe = PipelineBase() + pipe.add_component("retriever_1", InMemoryBM25Retriever(document_store=ds)) + pipe.add_component("retriever_2", InMemoryBM25Retriever(document_store=ds)) + # PromptBuilder registers every template variable as `Any` (see + # `PromptBuilder.__init__` -> `component.set_input_type(self, var, Any)`), so `documents` is an + # `Any`-typed socket. Before the fix, the second `connect` raised `PipelineConnectError`; with + # the fix, the socket is made lazy variadic and the component receives + # `[documents_from_retriever_1, documents_from_retriever_2]` at runtime (one-level wrapping). + pipe.add_component("prompt_builder", PromptBuilder(template="Docs: {{ documents }}")) + + pipe.connect("retriever_1.documents", "prompt_builder.documents") + pipe.connect("retriever_2.documents", "prompt_builder.documents") + # The `Any`-typed socket is now lazy variadic. The component is reachable and the second + # connection did not raise. (The aggregation behavior at render time is the component's + # responsibility; the framework fix is the connection-eligibility half of #10721.) + assert "retriever_1" in pipe.graph.nodes + assert "retriever_2" in pipe.graph.nodes + assert "prompt_builder" in pipe.graph.nodes