Coverage for haystack/dataclasses/breakpoints.py: 100%
39 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 13:53 +0000
1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2#
3# SPDX-License-Identifier: Apache-2.0
5from dataclasses import asdict, dataclass, field
6from datetime import datetime
7from typing import Any
9from haystack.utils.dataclasses import _warn_on_inplace_mutation
11# Value of `PipelineState.inputs_format` used by every snapshot Haystack creates now. See that field for the shape
12# each format implies.
13INTERNAL_INPUTS_FORMAT = "internal"
16@dataclass(frozen=True)
17class Breakpoint:
18 """
19 A dataclass to hold a breakpoint for a component.
21 :param component_name: The name of the component where the breakpoint is set.
22 :param visit_count: The number of times the component must be visited before the breakpoint is triggered.
23 :param snapshot_file_path: Optional path to store a snapshot of the pipeline when the breakpoint is hit.
24 This is useful for debugging purposes, allowing you to inspect the state of the pipeline at the time of the
25 breakpoint and to resume execution from that point.
26 """
28 component_name: str
29 visit_count: int = 0
30 snapshot_file_path: str | None = None
32 def to_dict(self) -> dict[str, Any]:
33 """
34 Convert the Breakpoint to a dictionary representation.
36 :return: A dictionary containing the component name, visit count, and debug path.
37 """
38 return asdict(self)
40 @classmethod
41 def from_dict(cls, data: dict) -> "Breakpoint":
42 """
43 Populate the Breakpoint from a dictionary representation.
45 :param data: A dictionary containing the component name, visit count, and debug path.
46 :return: An instance of Breakpoint.
47 """
48 return cls(**data)
51@_warn_on_inplace_mutation
52@dataclass
53class PipelineState:
54 """
55 A dataclass to hold the state of the pipeline at a specific point in time.
57 :param component_visits: A dictionary mapping component names to their visit counts.
58 :param inputs: The inputs processed by the pipeline at the time of the snapshot.
59 :param pipeline_outputs: Dictionary containing the final outputs of the pipeline up to the breakpoint.
60 :param inputs_format: Which format `inputs` are stored in. `"internal"` means each input records the component
61 that sent it, in the order it arrived, as in `{component: {socket: [{"sender": ..., "value": ...}]}}`.
62 `None` marks snapshots taken before Haystack recorded the sender, which hold one flattened value per socket,
63 `{component: {socket: value}}`, and can only be resumed on a component's first visit.
64 """
66 inputs: dict[str, Any]
67 component_visits: dict[str, int]
68 pipeline_outputs: dict[str, Any]
69 inputs_format: str | None = None
71 def to_dict(self) -> dict[str, Any]:
72 """
73 Convert the PipelineState to a dictionary representation.
75 :return: A dictionary containing the inputs, component visits,
76 and pipeline outputs.
77 """
78 return asdict(self)
80 @classmethod
81 def from_dict(cls, data: dict) -> "PipelineState":
82 """
83 Populate the PipelineState from a dictionary representation.
85 :param data: A dictionary containing the inputs, component visits,
86 and pipeline outputs.
87 :return: An instance of PipelineState.
88 """
89 return cls(**data)
92@_warn_on_inplace_mutation
93@dataclass
94class PipelineSnapshot:
95 """
96 A dataclass to hold a snapshot of the pipeline at a specific point in time.
98 :param original_input_data: The original input data provided to the pipeline.
99 :param ordered_component_names: A list of component names in the order they were visited.
100 :param pipeline_state: The state of the pipeline at the time of the snapshot.
101 :param break_point: The breakpoint that triggered the snapshot.
102 :param timestamp: A timestamp indicating when the snapshot was taken.
103 :param include_outputs_from: Set of component names whose outputs should be included in the pipeline results.
104 """
106 original_input_data: dict[str, Any]
107 ordered_component_names: list[str]
108 pipeline_state: PipelineState
109 break_point: Breakpoint
110 timestamp: datetime | None = None
111 include_outputs_from: set[str] = field(default_factory=set)
113 def __post_init__(self) -> None:
114 # Validate consistency between component_visits and ordered_component_names
115 components_in_state = set(self.pipeline_state.component_visits.keys())
116 components_in_order = set(self.ordered_component_names)
118 if components_in_state != components_in_order:
119 raise ValueError(
120 f"Inconsistent state: components in PipelineState.component_visits {components_in_state} "
121 f"do not match components in PipelineSnapshot.ordered_component_names {components_in_order}"
122 )
124 def to_dict(self) -> dict[str, Any]:
125 """
126 Convert the PipelineSnapshot to a dictionary representation.
128 :return: A dictionary containing the pipeline state, timestamp, breakpoint, agent snapshot, original input data,
129 ordered component names, include_outputs_from, and pipeline outputs.
130 """
131 return {
132 "pipeline_state": self.pipeline_state.to_dict(),
133 "break_point": self.break_point.to_dict(),
134 "timestamp": self.timestamp.isoformat() if self.timestamp else None,
135 "original_input_data": self.original_input_data,
136 "ordered_component_names": self.ordered_component_names,
137 "include_outputs_from": list(self.include_outputs_from),
138 }
140 @classmethod
141 def from_dict(cls, data: dict) -> "PipelineSnapshot":
142 """
143 Populate the PipelineSnapshot from a dictionary representation.
145 :param data: A dictionary containing the pipeline state, timestamp, breakpoint, agent snapshot, original input
146 data, ordered component names, include_outputs_from, and pipeline outputs.
147 """
148 include_outputs_from = set(data.get("include_outputs_from", []))
150 return cls(
151 pipeline_state=PipelineState.from_dict(data=data["pipeline_state"]),
152 break_point=Breakpoint.from_dict(data=data["break_point"]),
153 timestamp=datetime.fromisoformat(data["timestamp"]) if data.get("timestamp") else None,
154 original_input_data=data.get("original_input_data", {}),
155 ordered_component_names=data.get("ordered_component_names", []),
156 include_outputs_from=include_outputs_from,
157 )