feat: adding debugging breakpoints to Pipeline and Agent

haystack/components/tools/tool_invoker.py

@@ -9,9 +9,10 @@
 from functools import partial
 from typing import Any, Dict, List, Optional, Set, Union
 
-from haystack import component, default_from_dict, default_to_dict, logging
 from haystack.components.agents import State
+from haystack.core.component.component import component
 from haystack.core.component.sockets import Sockets
+from haystack.core.serialization import default_from_dict, default_to_dict, logging
 from haystack.dataclasses import ChatMessage, ToolCall
 from haystack.dataclasses.streaming_chunk import StreamingCallbackT, StreamingChunk, select_streaming_callback
 from haystack.tools import (

haystack/core/errors.py

@@ -2,7 +2,7 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-from typing import Any, Optional, Type
+from typing import Any, Dict, Optional, Type
 
 
 class PipelineError(Exception):
@@ -89,3 +89,30 @@ class DeserializationError(Exception):
 
 class SerializationError(Exception):
     pass
+
+
+class BreakpointException(Exception):
+    """
+    Exception raised when a pipeline breakpoint is triggered.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        component: Optional[str] = None,
+        inputs: Optional[Dict[str, Any]] = None,
+        results: Optional[Dict[str, Any]] = None,
+    ):
+        super().__init__(message)
+        self.component = component
+        self.inputs = inputs
+        self.results = results
+
+
+class PipelineInvalidPipelineSnapshotError(Exception):
+    """
+    Exception raised when a pipeline is resumed from an invalid snapshot.
+    """
+
+    def __init__(self, message: str):
+        super().__init__(message)

haystack/core/pipeline/base.py

@@ -1079,7 +1079,9 @@ def _convert_to_internal_format(pipeline_inputs: Dict[str, Any]) -> Dict[str, Di
         return inputs
 
     @staticmethod
-    def _consume_component_inputs(component_name: str, component: Dict, inputs: Dict) -> Dict[str, Any]:
+    def _consume_component_inputs(
+        component_name: str, component: Dict, inputs: Dict, is_resume: bool = False
+    ) -> Dict[str, Any]:
         """
         Extracts the inputs needed to run for the component and removes them from the global inputs state.
 
@@ -1094,6 +1096,11 @@ def _consume_component_inputs(component_name: str, component: Dict, inputs: Dict
         for socket_name, socket in component["input_sockets"].items():
             socket_inputs = component_inputs.get(socket_name, [])
             socket_inputs = [sock["value"] for sock in socket_inputs if sock["value"] is not _NO_OUTPUT_PRODUCED]
+
+            # if we are resuming a component, the inputs are already consumed, so we just return the first input
+            if is_resume:
+                consumed_inputs[socket_name] = socket_inputs[0]
+                continue
             if socket_inputs:
                 if not socket.is_variadic:
                     # We only care about the first input provided to the socket.