Check whether need to early stop
(
trajectory: Trajectory, max_steps: int, thresholds: dict[str, int]
)
| 163 | |
| 164 | |
| 165 | def early_stop( |
| 166 | trajectory: Trajectory, max_steps: int, thresholds: dict[str, int] |
| 167 | ) -> tuple[bool, str]: |
| 168 | """Check whether need to early stop""" |
| 169 | |
| 170 | # reach the max step |
| 171 | num_steps = (len(trajectory) - 1) / 2 |
| 172 | if num_steps >= max_steps: |
| 173 | return True, f"Reach max steps {max_steps}" |
| 174 | |
| 175 | last_k_actions: list[Action] |
| 176 | action_seq: list[Action] |
| 177 | |
| 178 | # Case: parsing failure for k times |
| 179 | k = thresholds["parsing_failure"] |
| 180 | last_k_actions = trajectory[1::2][-k:] # type: ignore[assignment] |
| 181 | if len(last_k_actions) >= k: |
| 182 | if all( |
| 183 | [ |
| 184 | action["action_type"] == ActionTypes.NONE |
| 185 | for action in last_k_actions |
| 186 | ] |
| 187 | ): |
| 188 | return True, f"Failed to parse actions for {k} times" |
| 189 | |
| 190 | # Case: same action for k times |
| 191 | k = thresholds["repeating_action"] |
| 192 | last_k_actions = trajectory[1::2][-k:] # type: ignore[assignment] |
| 193 | action_seq = trajectory[1::2] # type: ignore[assignment] |
| 194 | |
| 195 | if len(action_seq) == 0: |
| 196 | return False, "" |
| 197 | |
| 198 | last_action: Action = action_seq[-1] |
| 199 | |
| 200 | if last_action["action_type"] != ActionTypes.TYPE: |
| 201 | if len(last_k_actions) >= k: |
| 202 | if all( |
| 203 | [ |
| 204 | is_equivalent(action, last_action) |
| 205 | for action in last_k_actions |
| 206 | ] |
| 207 | ): |
| 208 | return True, f"Same action for {k} times" |
| 209 | |
| 210 | else: |
| 211 | # check the action sequence |
| 212 | if ( |
| 213 | sum([is_equivalent(action, last_action) for action in action_seq]) |
| 214 | >= k |
| 215 | ): |
| 216 | return True, f"Same typing action for {k} times" |
| 217 | |
| 218 | return False, "" |
| 219 | |
| 220 | |
| 221 | def test( |