Merge branch 'main' into feat/csv-row-mode

Author: xoaryaa

haystack/components/agents/agent.py

@@ -255,6 +255,7 @@ def _initialize_fresh_execution(
         messages: list[ChatMessage],
         streaming_callback: Optional[StreamingCallbackT],
         requires_async: bool,
+        system_prompt: Optional[str] = None,
         **kwargs,
     ) -> _ExecutionContext:
         """
@@ -263,10 +264,12 @@ def _initialize_fresh_execution(
         :param messages: List of ChatMessage objects to start the agent with.
         :param streaming_callback: Optional callback for streaming responses.
         :param requires_async: Whether the agent run requires asynchronous execution.
+        :param system_prompt: System prompt for the agent. If provided, it overrides the default system prompt.
         :param kwargs: Additional data to pass to the State used by the Agent.
         """
-        if self.system_prompt is not None:
-            messages = [ChatMessage.from_system(self.system_prompt)] + messages
+        system_prompt = system_prompt or self.system_prompt
+        if system_prompt is not None:
+            messages = [ChatMessage.from_system(system_prompt)] + messages
 
         if all(m.is_from(ChatRole.SYSTEM) for m in messages):
             logger.warning("All messages provided to the Agent component are system messages. This is not recommended.")
@@ -443,6 +446,7 @@ def run(
         *,
         break_point: Optional[AgentBreakpoint] = None,
         snapshot: Optional[AgentSnapshot] = None,
+        system_prompt: Optional[str] = None,
         **kwargs: Any,
     ) -> dict[str, Any]:
         """
@@ -455,6 +459,7 @@ def run(
             for "tool_invoker".
         :param snapshot: A dictionary containing a snapshot of a previously saved agent execution. The snapshot contains
             the relevant information to restart the Agent execution from where it left off.
+        :param system_prompt: System prompt for the agent. If provided, it overrides the default system prompt.
         :param kwargs: Additional data to pass to the State schema used by the Agent.
             The keys must match the schema defined in the Agent's `state_schema`.
         :returns:
@@ -482,7 +487,11 @@ def run(
             )
         else:
             exe_context = self._initialize_fresh_execution(
-                messages=messages, streaming_callback=streaming_callback, requires_async=False, **kwargs
+                messages=messages,
+                streaming_callback=streaming_callback,
+                requires_async=False,
+                system_prompt=system_prompt,
+                **kwargs,
             )
 
         with self._create_agent_span() as span:
@@ -558,6 +567,7 @@ async def run_async(
         *,
         break_point: Optional[AgentBreakpoint] = None,
         snapshot: Optional[AgentSnapshot] = None,
+        system_prompt: Optional[str] = None,
         **kwargs: Any,
     ) -> dict[str, Any]:
         """
@@ -574,6 +584,7 @@ async def run_async(
             for "tool_invoker".
         :param snapshot: A dictionary containing a snapshot of a previously saved agent execution. The snapshot contains
             the relevant information to restart the Agent execution from where it left off.
+        :param system_prompt: System prompt for the agent. If provided, it overrides the default system prompt.
         :param kwargs: Additional data to pass to the State schema used by the Agent.
             The keys must match the schema defined in the Agent's `state_schema`.
         :returns:
@@ -601,7 +612,11 @@ async def run_async(
             )
         else:
             exe_context = self._initialize_fresh_execution(
-                messages=messages, streaming_callback=streaming_callback, requires_async=False, **kwargs
+                messages=messages,
+                streaming_callback=streaming_callback,
+                requires_async=False,
+                system_prompt=system_prompt,
+                **kwargs,
             )
 
         with self._create_agent_span() as span:

haystack/components/rankers/sentence_transformers_diversity.py

@@ -184,7 +184,7 @@ def __init__(  # noqa: PLR0913 # pylint: disable=too-many-positional-arguments
         self.meta_fields_to_embed = meta_fields_to_embed or []
         self.embedding_separator = embedding_separator
         self.strategy = DiversityRankingStrategy.from_str(strategy) if isinstance(strategy, str) else strategy
-        self.lambda_threshold = lambda_threshold or 0.5
+        self.lambda_threshold = lambda_threshold
         self._check_lambda_threshold(self.lambda_threshold, self.strategy)
         self.model_kwargs = model_kwargs
         self.tokenizer_kwargs = tokenizer_kwargs

haystack/core/component/component.py

@@ -605,7 +605,7 @@ def copy_class_namespace(namespace):
 
         return new_cls
 
-    # Call signature when the the decorator is usead without parens (@component).
+    # Call signature when the decorator is used without parens (@component).
     @overload
     def __call__(self, cls: type[T]) -> type[T]: ...
 

haystack/core/pipeline/breakpoint.py

@@ -7,7 +7,7 @@
 from dataclasses import replace
 from datetime import datetime
 from pathlib import Path
-from typing import Any, Optional, Union
+from typing import TYPE_CHECKING, Any, Optional, Union
 
 from networkx import MultiDiGraph
 
@@ -22,9 +22,12 @@
     PipelineState,
     ToolBreakpoint,
 )
-from haystack.tools import Tool, Toolset
 from haystack.utils.base_serialization import _serialize_value_with_schema
 
+if TYPE_CHECKING:
+    from haystack.tools.tool import Tool
+    from haystack.tools.toolset import Toolset
+
 logger = logging.getLogger(__name__)
 
 
@@ -323,7 +326,9 @@ def _create_agent_snapshot(
     )
 
 
-def _validate_tool_breakpoint_is_valid(agent_breakpoint: AgentBreakpoint, tools: Union[list[Tool], Toolset]) -> None:
+def _validate_tool_breakpoint_is_valid(
+    agent_breakpoint: AgentBreakpoint, tools: Union[list["Tool"], "Toolset"]
+) -> None:
     """
     Validates the AgentBreakpoint passed to the agent.
 

haystack/core/pipeline/utils.py

@@ -10,7 +10,6 @@
 
 from haystack import logging
 from haystack.core.component import Component
-from haystack.tools import Tool, Toolset
 
 logger = logging.getLogger(__name__)
 
@@ -29,6 +28,10 @@ def _deepcopy_with_exceptions(obj: Any) -> Any:
     :returns:
         A deep-copied version of the object, or the original object if deepcopying fails.
     """
+    # Import here to avoid circular imports
+    from haystack.tools.tool import Tool
+    from haystack.tools.toolset import Toolset
+
     if isinstance(obj, (list, tuple, set)):
         return type(obj)(_deepcopy_with_exceptions(v) for v in obj)
 

haystack/core/super_component/super_component.py

@@ -46,8 +46,21 @@ def __init__(
         :param pipeline: The pipeline instance or async pipeline instance to be wrapped
         :param input_mapping: A dictionary mapping component input names to pipeline input socket paths.
             If not provided, a default input mapping will be created based on all pipeline inputs.
+            Example:
+            ```python
+            input_mapping={
+                "query": ["retriever.query", "prompt_builder.query"],
+            }
+            ```
         :param output_mapping: A dictionary mapping pipeline output socket paths to component output names.
             If not provided, a default output mapping will be created based on all pipeline outputs.
+            Example:
+            ```python
+            output_mapping={
+                "retriever.documents": "documents",
+                "generator.replies": "replies",
+            }
+            ```
         :raises InvalidMappingError: Raised if any mapping is invalid or type conflicts occur
         :raises ValueError: Raised if no pipeline is provided
         """
@@ -177,10 +190,14 @@ def _validate_input_mapping(
             for path in pipeline_input_paths:
                 comp_name, socket_name = self._split_component_path(path)
                 if comp_name not in pipeline_inputs:
-                    raise InvalidMappingValueError(f"Component '{comp_name}' not found in pipeline inputs.")
+                    raise InvalidMappingValueError(
+                        f"Component '{comp_name}' not found in pipeline inputs.\n"
+                        f"Available components: {list(pipeline_inputs.keys())}"
+                    )
                 if socket_name not in pipeline_inputs[comp_name]:
                     raise InvalidMappingValueError(
-                        f"Input socket '{socket_name}' not found in component '{comp_name}'."
+                        f"Input socket '{socket_name}' not found in component '{comp_name}'.\n"
+                        f"Available inputs for '{comp_name}': {list(pipeline_inputs[comp_name].keys())}"
                     )
 
     def _resolve_input_types_from_mapping(

haystack/tools/__init__.py

@@ -4,18 +4,20 @@
 
 # NOTE: we do not use LazyImporter here because it creates conflicts between the tool module and the tool decorator
 
-# ruff: noqa: I001 (ignore import order as we need to import Tool before ComponentTool)
-from .from_function import create_tool_from_function, tool
-from .tool import Tool, _check_duplicate_tool_names
-from .component_tool import ComponentTool
-from .serde_utils import deserialize_tools_or_toolset_inplace, serialize_tools_or_toolset
-from .toolset import Toolset
+# ruff: noqa: I001 (ignore import order as we need to import Tool before ComponentTool and PipelineTool)
+from haystack.tools.from_function import create_tool_from_function, tool
+from haystack.tools.tool import Tool, _check_duplicate_tool_names
+from haystack.tools.toolset import Toolset
+from haystack.tools.component_tool import ComponentTool
+from haystack.tools.pipeline_tool import PipelineTool
+from haystack.tools.serde_utils import deserialize_tools_or_toolset_inplace, serialize_tools_or_toolset
 
 __all__ = [
     "_check_duplicate_tool_names",
     "ComponentTool",
     "create_tool_from_function",
     "deserialize_tools_or_toolset_inplace",
+    "PipelineTool",
     "serialize_tools_or_toolset",
     "Tool",
     "Toolset",

haystack/tools/component_tool.py

@@ -222,13 +222,9 @@ def to_dict(self) -> dict[str, Any]:
             "description": self.description,
             "parameters": self._unresolved_parameters,
             "inputs_from_state": self.inputs_from_state,
-            # This is soft-copied as to not modify the attributes in place
-            "outputs_to_state": self.outputs_to_state.copy() if self.outputs_to_state else None,
+            "outputs_to_state": _serialize_outputs_to_state(self.outputs_to_state) if self.outputs_to_state else None,
         }
 
-        if self.outputs_to_state is not None:
-            serialized["outputs_to_state"] = _serialize_outputs_to_state(self.outputs_to_state)
-
         if self.outputs_to_string is not None and self.outputs_to_string.get("handler") is not None:
             # This is soft-copied as to not modify the attributes in place
             serialized["outputs_to_string"] = self.outputs_to_string.copy()
@@ -239,7 +235,7 @@ def to_dict(self) -> dict[str, Any]:
         return {"type": generate_qualified_class_name(type(self)), "data": serialized}
 
     @classmethod
-    def from_dict(cls, data: dict[str, Any]) -> "Tool":
+    def from_dict(cls, data: dict[str, Any]) -> "ComponentTool":
         """
         Deserializes the ComponentTool from a dictionary.
         """