refactor!: Remove agent breakpoint and agent snapshot (#11202)

Author: sjrl

MIGRATION.md

@@ -70,3 +70,58 @@ from haystack.dataclasses import Document
 
 doc = Document(content="col\n1\n2\n3")
 ```
+
+### Agent
+
+**What changed:** The agent-specific breakpoint API has been removed. The `AgentBreakpoint`, `ToolBreakpoint`, and `AgentSnapshot` dataclasses are no longer exported from `haystack.dataclasses`, and the `break_point`, `snapshot`, and `snapshot_callback` parameters have been removed from `Agent.run` and `Agent.run_async`. `Pipeline.run` no longer accepts an `AgentBreakpoint` for its `break_point` argument, and the `agent_snapshot` field has been removed from `PipelineSnapshot`. Pausing and resuming execution inside an Agent (at the chat generator or tool invoker) is no longer supported.
+
+**Why:** Simplifies the breakpoint and snapshot machinery and removes the special-cased agent-internal control flow. Pipeline-level breakpoints still cover the common debugging use cases.
+
+**How to migrate:**
+
+Before (v2.x):
+```python
+from haystack.components.agents import Agent
+from haystack.dataclasses import AgentBreakpoint, Breakpoint, ToolBreakpoint
+
+agent = Agent(chat_generator=..., tools=[...])
+
+# Pause before the chat generator runs
+chat_break_point = AgentBreakpoint(
+    agent_name="agent",
+    break_point=Breakpoint(component_name="chat_generator", visit_count=0),
+)
+
+# Or pause before a specific tool is invoked
+tool_break_point = AgentBreakpoint(
+    agent_name="agent",
+    break_point=ToolBreakpoint(component_name="tool_invoker", tool_name="my_tool"),
+)
+
+agent.run(messages=[...], break_point=chat_break_point)
+```
+
+After (v3.0):
+```python
+# Pausing inside an Agent is no longer supported. To inspect an Agent's behavior,
+# use tracing instead (https://docs.haystack.deepset.ai/docs/tracing). For example,
+# you can wire up a Langfuse tracer for a standalone Agent by instantiating a
+# LangfuseConnector — its constructor registers the tracer globally, so any
+# subsequent Agent.run call will be traced.
+
+# NOTE: install the langfuse integration first with `pip install langfuse-haystack` to run this example.
+import os
+
+os.environ["HAYSTACK_CONTENT_TRACING_ENABLED"] = "true"
+
+from haystack.components.agents import Agent
+from haystack.components.generators.chat import OpenAIChatGenerator
+from haystack.dataclasses import ChatMessage
+from haystack_integrations.components.connectors.langfuse import LangfuseConnector
+
+# Instantiating the connector enables tracing globally — no need to add it to a pipeline.
+LangfuseConnector("Standalone Agent example")
+
+agent = Agent(chat_generator=OpenAIChatGenerator(model="gpt-4o-mini"), tools=[...])
+agent.run(messages=[ChatMessage.from_user("What's the weather in Berlin?")])
+```

haystack/components/agents/agent.py

@@ -4,7 +4,7 @@
 
 from typing import Any
 
-from haystack.dataclasses.breakpoints import AgentBreakpoint, Breakpoint, PipelineSnapshot, ToolBreakpoint
+from haystack.dataclasses.breakpoints import Breakpoint, PipelineSnapshot
 
 
 class PipelineError(Exception):
@@ -115,7 +115,7 @@ def __init__(
         pipeline_snapshot: PipelineSnapshot | None = None,
         pipeline_snapshot_file_path: str | None = None,
         *,
-        break_point: AgentBreakpoint | Breakpoint | ToolBreakpoint | None = None,
+        break_point: Breakpoint | None = None,
     ) -> None:
         super().__init__(message)
         self.component = component
@@ -127,7 +127,7 @@ def __init__(
             raise ValueError("Either pipeline_snapshot or break_point must be provided.")
 
     @classmethod
-    def from_triggered_breakpoint(cls, break_point: Breakpoint | ToolBreakpoint) -> "BreakpointException":
+    def from_triggered_breakpoint(cls, break_point: Breakpoint) -> "BreakpointException":
         """
         Create a BreakpointException from a triggered breakpoint.
         """
@@ -137,37 +137,25 @@ def from_triggered_breakpoint(cls, break_point: Breakpoint | ToolBreakpoint) ->
     @property
     def inputs(self) -> dict[str, Any] | None:
         """
-        Returns the inputs of the pipeline or agent at the breakpoint.
-
-        If an AgentBreakpoint caused this exception, returns the inputs of the agent's internal components.
-        Otherwise, returns the current inputs of the pipeline.
+        Returns the current inputs of the pipeline at the breakpoint.
         """
         if not self.pipeline_snapshot:
             return None
-
-        if self.pipeline_snapshot.agent_snapshot:
-            return self.pipeline_snapshot.agent_snapshot.component_inputs
         return self.pipeline_snapshot.pipeline_state.inputs
 
     @property
     def results(self) -> dict[str, Any] | None:
         """
-        Returns the results of the pipeline or agent at the breakpoint.
-
-        If an AgentBreakpoint caused this exception, returns the current results of the agent.
-        Otherwise, returns the current outputs of the pipeline.
+        Returns the current outputs of the pipeline at the breakpoint.
         """
         if not self.pipeline_snapshot:
             return None
-
-        if self.pipeline_snapshot.agent_snapshot:
-            return self.pipeline_snapshot.agent_snapshot.component_inputs["tool_invoker"]["serialized_data"]["state"]
         return self.pipeline_snapshot.pipeline_state.pipeline_outputs
 
     @property
-    def break_point(self) -> AgentBreakpoint | Breakpoint | ToolBreakpoint:
+    def break_point(self) -> Breakpoint:
         """
-        Returns the Breakpoint or AgentBreakpoint that caused this exception, if available.
+        Returns the Breakpoint that caused this exception.
 
         If a specific break point was provided during initialization, it is returned.
         Otherwise, if the pipeline snapshot contains a break point, that is returned.