This function might recurse. user_request: The original user request step_number: the number of times we've called the LLM for this request. Used to keep track of whether it's a fresh request we're processing (step number 0), or if we're already
(self, user_request: str, step_num: int = 0)
| 36 | self.interrupt_execution = True |
| 37 | |
| 38 | def execute(self, user_request: str, step_num: int = 0) -> Optional[str]: |
| 39 | """ |
| 40 | This function might recurse. |
| 41 | |
| 42 | user_request: The original user request |
| 43 | step_number: the number of times we've called the LLM for this request. |
| 44 | Used to keep track of whether it's a fresh request we're processing (step number 0), or if we're already |
| 45 | in the middle of one. |
| 46 | Without it the LLM kept looping after finishing the user request. |
| 47 | Also, it is needed because the LLM we are using doesn't have a stateful/assistant mode. |
| 48 | """ |
| 49 | self.interrupt_execution = False |
| 50 | |
| 51 | if not self.llm: |
| 52 | status = 'Set your OpenAPI API Key in Settings and Restart the App' |
| 53 | self.status_queue.put(status) |
| 54 | return status |
| 55 | |
| 56 | try: |
| 57 | instructions: dict[str, Any] = self.llm.get_instructions_for_objective(user_request, step_num) |
| 58 | |
| 59 | if instructions == {}: |
| 60 | # Sometimes LLM sends malformed JSON response, in that case retry once more. |
| 61 | instructions = self.llm.get_instructions_for_objective(user_request + ' Please reply in valid JSON', |
| 62 | step_num) |
| 63 | |
| 64 | for step in instructions['steps']: |
| 65 | if self.interrupt_execution: |
| 66 | self.status_queue.put('Interrupted') |
| 67 | self.interrupt_execution = False |
| 68 | return 'Interrupted' |
| 69 | |
| 70 | success = self.interpreter.process_command(step) |
| 71 | |
| 72 | if not success: |
| 73 | return 'Unable to execute the request' |
| 74 | |
| 75 | except Exception as e: |
| 76 | status = f'Exception Unable to execute the request - {e}' |
| 77 | self.status_queue.put(status) |
| 78 | return status |
| 79 | |
| 80 | if instructions['done']: |
| 81 | # Communicate Results |
| 82 | self.status_queue.put(instructions['done']) |
| 83 | self.play_ding_on_completion() |
| 84 | return instructions['done'] |
| 85 | else: |
| 86 | # if not done, continue to next phase |
| 87 | self.status_queue.put('Fetching further instructions based on current state') |
| 88 | return self.execute(user_request, step_num + 1) |
| 89 | |
| 90 | def play_ding_on_completion(self): |
| 91 | # Play ding sound to signal completion |
no test coverage detected