Introduction

Haystack pipelines support breakpoints for debugging complex execution flows. A Breakpoint allows you to pause the execution at specific components, inspect the pipeline state, and resume execution from saved snapshots. This feature works for any regular component as well as an Agent component.

You can set a Breakpoint on any component in a pipeline with a specific visit count. When triggered, the system stops the executions of the Pipeline and creates a JSON file containing a snapshot of the current pipeline state. You can inspect and modify the snapshot and use it to resume execution from the exact point where it stopped.

You can also set breakpoints on an Agent, specifically on the ChatGenerator component or on any of the Tool specified in the ToolInvoker component .

Setting a Breakpoint on a Regular Component

Create a Breakpoint by specifying the component name and the visit count at which to trigger it. This is useful for pipelines with loops. The default visit_count value is 0.

from haystack.dataclasses.breakpoints import Breakpoint
from haystack.core.errors import BreakpointException

# Create a breakpoint that triggers on the first visit to the "llm" component
break_point = Breakpoint(
    component_name="llm", 
    visit_count=0,  # 0 = first visit, 1 = second visit, etc.
    snapshot_file_path="/path/to/snapshots"  # Optional: save snapshot to file
)

# Run pipeline with breakpoint
try:
    result = pipeline.run(data=input_data, break_point=break_point)
except BreakpointException as e:
    print(f"Breakpoint triggered at component: {e.component}")
    print(f"Component inputs: {e.inputs}")
    print(f"Pipeline results so far: {e.results}")

A BreakpointException is raised containing the component inputs and the outputs of the pipeline up until the moment where the execution was interrupted, such as just before the execution of component associated with the breakpoint – the llm in the example above.

If a snapshot_file_path is specified in the Breakpoint, the system saves a JSON snapshot with the same information as in the BreakpointException .

To access the pipeline state during the breakpoint we can both catch the exception raised by the breakpoint as well as specify where the JSON file should be saved.

Resuming a Pipeline Execution from a Breakpoint

To resume the execution of a pipeline from the breakpoint, pass the path to the generated JSON file at the run time of the pipeline, using the pipeline_snapshot.

Use the load_pipeline_snapshot() to first load the JSON and then pass it to the pipeline.

from haystack.core.pipeline.breakpoint import load_pipeline_snapshot

# Load the snapshot
snapshot = load_pipeline_snapshot("llm_2025_05_03_11_23_23.json")

# Resume execution from the snapshot
result = pipeline.run(data={}, pipeline_snapshot=snapshot)
print(result["llm"]["replies"])

Setting a Breakpoint on an Agent

You can also set breakpoints in an Agent component. An Agent supports two types of breakpoints:

1. Chat Generator Breakpoint: Pauses before LLM calls.
2. Tool Invoker Breakpoint: Pauses before any tool execution.

A ChatGenerator breakpoint is defined as shown below. You need to define a Breakpoint as for a pipeline breakpoint and then an AgentBreakpoint where you pass the breakpoint defined before and the name of Agent component.

from haystack.dataclasses.breakpoints import AgentBreakpoint, Breakpoint, ToolBreakpoint

# Break at chat generator (LLM calls)
chat_bp = Breakpoint(component_name="chat_generator", visit_count=0)
agent_breakpoint = AgentBreakpoint(
    break_point=chat_bp, 
    agent_name="my_agent"
)

To set a breakpoint on a Tool in an Agent, do the following:

First, define a ToolBreakpoint specifying the ToolInvoker component whose name is tool_invoker and then the tool associated with the breakpoint, in this case – a weather_tool .

Then, define an AgentBreakpoint passing the ToolBreakpoint defined before as the breakpoint.

from haystack.dataclasses.breakpoints import AgentBreakpoint, Breakpoint, ToolBreakpoint

# Break at tool invoker (tool calls)
tool_bp = ToolBreakpoint(
    component_name="tool_invoker", 
    visit_count=0, 
    tool_name="weather_tool"  # Specific tool, or None for any tool
)
agent_breakpoint = AgentBreakpoint(
    break_point=tool_bp, 
    agent_name="my_agent"
)

Resuming Agent Execution

When an Agent breakpoint is triggered, you can resume execution using the saved snapshot. Similar to the regular component in a pipeline, pass the JSON file with the snapshot to the run() method of the pipeline.

from haystack.core.pipeline.breakpoint import load_pipeline_snapshot

# Load the snapshot
snapshot_file = "./agent_debug/agent_chat_generator_2025_07_11_23_23.json"
snapshot = load_pipeline_snapshot(snapshot_file)
    
# Resume pipeline execution
result = pipeline.run(data={}, pipeline_snapshot=snapshot)
print("Pipeline resumed successfully")
print(f"Final result: {result}")