(self, step_id: str)
| 152 | self._iteration_index: Dict[tuple, AgentIteration] = {} |
| 153 | |
| 154 | def _get_step(self, step_id: str) -> Step: |
| 155 | step = self._step_index.get(step_id) |
| 156 | if step: |
| 157 | return step |
| 158 | step = Step( |
| 159 | step_id=step_id, |
| 160 | number=step_id, |
| 161 | name=f"Step {step_id}", |
| 162 | step_type="step", |
| 163 | ) |
| 164 | self._step_index[step_id] = step |
| 165 | self.workflow.steps.append(step) |
| 166 | return step |
| 167 | |
| 168 | def _get_iteration(self, step_id: str, iteration_number: int) -> AgentIteration: |
| 169 | key = (step_id, iteration_number) |
no test coverage detected