Start a CLI chat session. :param rep_queue: A queue to put the prompt and response in. :return:
(self, rep_queue: Queue or None = None)
| 249 | self.__chat_history = [] |
| 250 | |
| 251 | def cli_chat(self, rep_queue: Queue or None = None): |
| 252 | """ |
| 253 | Start a CLI chat session. |
| 254 | :param rep_queue: A queue to put the prompt and response in. |
| 255 | :return: |
| 256 | """ |
| 257 | if rep_queue is not None and not isinstance(rep_queue, Queue): |
| 258 | self.log(f"{Fore.RED}>> Entered a non-queue object to hold responses for another thread.") |
| 259 | raise Exceptions.PyChatGPTException("Cannot enter a non-queue object as the response queue for threads.") |
| 260 | |
| 261 | # Check if the access token is expired |
| 262 | if OpenAI.token_expired(): |
| 263 | self.log(f"{Fore.RED}>> Your access token is expired. {Fore.GREEN}Attempting to recreate it...") |
| 264 | did_create = self._create_access_token() |
| 265 | if did_create: |
| 266 | self.log(f"{Fore.GREEN}>> Successfully recreated access token.") |
| 267 | else: |
| 268 | self.log(f"{Fore.RED}>> Failed to recreate access token.") |
| 269 | raise Exceptions.PyChatGPTException("Failed to recreate access token.") |
| 270 | else: |
| 271 | self.log(f"{Fore.GREEN}>> Access token is valid.") |
| 272 | self.log(f"{Fore.GREEN}>> Starting CLI chat session...") |
| 273 | self.log(f"{Fore.GREEN}>> Type 'exit' to exit the chat session.") |
| 274 | |
| 275 | |
| 276 | # Get access token |
| 277 | access_token = OpenAI.get_access_token() |
| 278 | |
| 279 | while True: |
| 280 | try: |
| 281 | prompt = input("You: ") |
| 282 | if prompt.replace("You: ", "") == "exit": |
| 283 | self.save_data() |
| 284 | break |
| 285 | |
| 286 | spinner = Spinner.Spinner() |
| 287 | spinner.start(Fore.YELLOW + "Chat GPT is typing...") |
| 288 | answer, previous_convo, convo_id = ChatHandler.ask(auth_token=access_token, prompt=prompt, |
| 289 | conversation_id=self.conversation_id, |
| 290 | previous_convo_id=self.previous_convo_id, |
| 291 | proxies=self.options.proxies, |
| 292 | pass_moderation=self.options.pass_moderation) |
| 293 | |
| 294 | if rep_queue is not None: |
| 295 | rep_queue.put((prompt, answer)) |
| 296 | |
| 297 | if answer == "400" or answer == "401": |
| 298 | self.log(f"{Fore.RED}>> Failed to get a response from the API.") |
| 299 | return None |
| 300 | |
| 301 | self.conversation_id = convo_id |
| 302 | self.previous_convo_id = previous_convo |
| 303 | spinner.stop() |
| 304 | print(f"Chat GPT: {answer}") |
| 305 | |
| 306 | if self.options.track: |
| 307 | self.__chat_history.append("You: " + prompt) |
| 308 | self.__chat_history.append("Chat GPT: " + answer) |