deepset-ai/haystack · error · InvalidMappingTypeError

Type conflict for input '{socket_name}' from component '{com

Error message

Type conflict for input '{socket_name}' from component '{comp_name}'. Existing type: {existing_socket_info['type']}, new type: {socket_info['type']}.

What it means

When multiple pipeline paths map to the same wrapper input, their socket types must be compatible; otherwise the SuperComponent cannot determine a single input type and raises InvalidMappingTypeError with both conflicting types.

Source

Thrown at haystack/core/super_component/super_component.py:247

        """
        aggregated_inputs: dict[str, dict[str, Any]] = {}
        for wrapper_input_name, pipeline_input_paths in input_mapping.items():
            for path in pipeline_input_paths:
                comp_name, socket_name = self._split_component_path(path)
                socket_info = pipeline_inputs[comp_name][socket_name]

                # Add to aggregated inputs
                existing_socket_info = aggregated_inputs.get(wrapper_input_name)
                if existing_socket_info is None:
                    aggregated_inputs[wrapper_input_name] = {"type": socket_info["type"]}
                    if not socket_info["is_mandatory"]:
                        aggregated_inputs[wrapper_input_name]["default"] = _delegate_default
                    continue

                is_compatible, common_type = _is_compatible(existing_socket_info["type"], socket_info["type"])

                if not is_compatible:
                    raise InvalidMappingTypeError(
                        f"Type conflict for input '{socket_name}' from component '{comp_name}'. "
                        f"Existing type: {existing_socket_info['type']}, new type: {socket_info['type']}."
                    )

                # Use the common type for the aggregated input
                aggregated_inputs[wrapper_input_name]["type"] = common_type

                # If any socket requires mandatory inputs then the aggregated input is also considered mandatory.
                # So we use the type of the mandatory input and remove the default value if it exists.
                if socket_info["is_mandatory"]:
                    aggregated_inputs[wrapper_input_name].pop("default", None)

        return aggregated_inputs

    @staticmethod
    def _create_input_mapping(pipeline_inputs: dict[str, dict[str, Any]]) -> dict[str, list[str]]:
        """
        Create an input mapping from pipeline inputs.

View on GitHub (pinned to e318778c9b)

Solutions

Align the socket types: change the components' socket types or wrap with a component that converts types
Map the conflicting paths to different wrapper input names instead of aggregating them
Only group paths whose sockets share a compatible type

Example fix

// before
input_mapping = {"q": ["retriever.query", "ranker.documents"]}  # str vs List[Document]
// after
input_mapping = {"query": ["retriever.query"], "documents": ["ranker.documents"]}

Defensive patterns

Strategy: validation

Validate before calling

from haystack.core.super_component.super_component import _is_compatible

def check_no_type_conflicts(pipe, input_mapping):
    inputs = pipe.inputs()
    seen = {}
    for name, paths in input_mapping.items():
        for p in paths:
            comp, sock = p.split(".")
            t = inputs[comp][sock]["type"]
            if name in seen:
                ok, _ = _is_compatible(seen[name], t)
                if not ok:
                    raise TypeError(f"Conflicting types for '{name}': {seen[name]} vs {t}")
            seen[name] = t

Type guard

def types_are_compatible(t1, t2) -> bool:
    from haystack.core.super_component.super_component import _is_compatible
    ok, _ = _is_compatible(t1, t2)
    return ok

Try / catch

from haystack.core.errors import InvalidMappingTypeError
try:
    super_comp = SuperComponent(pipeline=pipe, input_mapping=input_mapping)
except InvalidMappingTypeError as e:
    if "Type conflict for input" in str(e):
        print("Split into separate wrapper inputs or align socket types:", e)
    else:
        raise

Prevention

Only aggregate paths whose sockets share the same or compatible types
Split conflicting paths into distinct wrapper input names
Keep component socket type annotations consistent across the pipeline

When it happens

Trigger: input_mapping = {"q": ["retriever.query", "ranker.query"]} where one expects str and the other expects List[Document], and no common type exists.

Common situations: Aggregating inputs from components with mismatched typed sockets; pipeline edits changed a socket's type annotation; connecting str and int-typed sockets.

Related errors

AI-assisted analysis of deepset-ai/haystack@e318778c9b (2026-08-30). Data as JSON: /api/errors/7410adfd877adf29. Report an issue: GitHub.