Parse a single step
(
self, step: Dict[str, Any]
)
| 135 | return result |
| 136 | |
| 137 | def _parse_step( |
| 138 | self, step: Dict[str, Any] |
| 139 | ) -> Union[Dict[str, Any], List[Dict[str, Any]], None]: |
| 140 | """Parse a single step""" |
| 141 | step_type = list(step.keys())[0] |
| 142 | step_data = step[step_type] |
| 143 | |
| 144 | if step_type == "step": |
| 145 | return self._parse_basic_step(step_data) |
| 146 | |
| 147 | elif step_type == "task": |
| 148 | return self._parse_task_step(step_data) |
| 149 | |
| 150 | elif step_type == "for_each": |
| 151 | return self._parse_for_each_step(step_data) |
| 152 | |
| 153 | elif step_type == "if": |
| 154 | return self._parse_if_step(step_data) |
| 155 | |
| 156 | elif step_type == "while": |
| 157 | return self._parse_while_step(step_data) |
| 158 | |
| 159 | elif step_type == "switch": |
| 160 | return self._parse_switch_step(step_data) |
| 161 | |
| 162 | elif step_type == "parallel": |
| 163 | return self._parse_parallel_step(step_data) |
| 164 | |
| 165 | elif step_type == "increment": |
| 166 | return self._parse_increment_step(step_data) |
| 167 | |
| 168 | else: |
| 169 | raise ValueError(f"Unknown step type: {step_type}") |
| 170 | |
| 171 | def _parse_basic_step(self, step_data: Dict[str, Any]) -> Dict[str, Any]: |
| 172 | """Parse step (with conversation history)""" |
no test coverage detected