Execute a single step of the agent loop.
(
self, user_prompt: str | None = None, is_first: bool = False
)
| 152 | self._step_count = 0 |
| 153 | |
| 154 | def _execute_step( |
| 155 | self, user_prompt: str | None = None, is_first: bool = False |
| 156 | ) -> StepResult: |
| 157 | """Execute a single step of the agent loop.""" |
| 158 | self._step_count += 1 |
| 159 | |
| 160 | # Capture current screen state |
| 161 | screenshot = get_screenshot( |
| 162 | wda_url=self.agent_config.wda_url, |
| 163 | session_id=self.agent_config.session_id, |
| 164 | device_id=self.agent_config.device_id, |
| 165 | ) |
| 166 | current_app = get_current_app( |
| 167 | wda_url=self.agent_config.wda_url, session_id=self.agent_config.session_id |
| 168 | ) |
| 169 | |
| 170 | # Build messages |
| 171 | if is_first: |
| 172 | self._context.append( |
| 173 | MessageBuilder.create_system_message(self.agent_config.system_prompt) |
| 174 | ) |
| 175 | |
| 176 | screen_info = MessageBuilder.build_screen_info(current_app) |
| 177 | text_content = f"{user_prompt}\n\n{screen_info}" |
| 178 | |
| 179 | self._context.append( |
| 180 | MessageBuilder.create_user_message( |
| 181 | text=text_content, image_base64=screenshot.base64_data |
| 182 | ) |
| 183 | ) |
| 184 | else: |
| 185 | screen_info = MessageBuilder.build_screen_info(current_app) |
| 186 | text_content = f"** Screen Info **\n\n{screen_info}" |
| 187 | |
| 188 | self._context.append( |
| 189 | MessageBuilder.create_user_message( |
| 190 | text=text_content, image_base64=screenshot.base64_data |
| 191 | ) |
| 192 | ) |
| 193 | |
| 194 | # Get model response |
| 195 | try: |
| 196 | response = self.model_client.request(self._context) |
| 197 | except Exception as e: |
| 198 | if self.agent_config.verbose: |
| 199 | traceback.print_exc() |
| 200 | return StepResult( |
| 201 | success=False, |
| 202 | finished=True, |
| 203 | action=None, |
| 204 | thinking="", |
| 205 | message=f"Model error: {e}", |
| 206 | ) |
| 207 | |
| 208 | # Parse action from response |
| 209 | try: |
| 210 | action = parse_action(response.action) |
| 211 | except ValueError: |
no test coverage detected