Execute a single step of the agent loop.
(
self, user_prompt: str | None = None, is_first: bool = False
)
| 134 | self._step_count = 0 |
| 135 | |
| 136 | def _execute_step( |
| 137 | self, user_prompt: str | None = None, is_first: bool = False |
| 138 | ) -> StepResult: |
| 139 | """Execute a single step of the agent loop.""" |
| 140 | self._step_count += 1 |
| 141 | |
| 142 | # Capture current screen state |
| 143 | device_factory = get_device_factory() |
| 144 | screenshot = device_factory.get_screenshot(self.agent_config.device_id) |
| 145 | current_app = device_factory.get_current_app(self.agent_config.device_id) |
| 146 | |
| 147 | # Build messages |
| 148 | if is_first: |
| 149 | self._context.append( |
| 150 | MessageBuilder.create_system_message(self.agent_config.system_prompt) |
| 151 | ) |
| 152 | |
| 153 | screen_info = MessageBuilder.build_screen_info(current_app) |
| 154 | text_content = f"{user_prompt}\n\n{screen_info}" |
| 155 | |
| 156 | self._context.append( |
| 157 | MessageBuilder.create_user_message( |
| 158 | text=text_content, image_base64=screenshot.base64_data |
| 159 | ) |
| 160 | ) |
| 161 | else: |
| 162 | screen_info = MessageBuilder.build_screen_info(current_app) |
| 163 | text_content = f"** Screen Info **\n\n{screen_info}" |
| 164 | |
| 165 | self._context.append( |
| 166 | MessageBuilder.create_user_message( |
| 167 | text=text_content, image_base64=screenshot.base64_data |
| 168 | ) |
| 169 | ) |
| 170 | |
| 171 | # Get model response |
| 172 | try: |
| 173 | msgs = get_messages(self.agent_config.lang) |
| 174 | print("\n" + "=" * 50) |
| 175 | print(f"💭 {msgs['thinking']}:") |
| 176 | print("-" * 50) |
| 177 | response = self.model_client.request(self._context) |
| 178 | except Exception as e: |
| 179 | if self.agent_config.verbose: |
| 180 | traceback.print_exc() |
| 181 | return StepResult( |
| 182 | success=False, |
| 183 | finished=True, |
| 184 | action=None, |
| 185 | thinking="", |
| 186 | message=f"Model error: {e}", |
| 187 | ) |
| 188 | |
| 189 | # Parse action from response |
| 190 | try: |
| 191 | action = parse_action(response.action) |
| 192 | except ValueError: |
| 193 | if self.agent_config.verbose: |
no test coverage detected