REPL mode for chat sessions (#94)

README.md

@@ -7,7 +7,7 @@ A command-line productivity tool powered by OpenAI's ChatGPT (GPT-3.5). As devel
 
 ## Installation
 ```shell
-pip install shell-gpt==0.8.2
+pip install shell-gpt==0.8.3
 ```
 You'll need an OpenAI API key, you can generate one [here](https://beta.openai.com/account/api-keys).
 
@@ -121,7 +121,7 @@ python fizz_buzz.py
 ```
 
 ### Chat
-To start a chat session, use the `--chat` option followed by a unique session name and a prompt:
+To start a chat session, use the `--chat` option followed by a unique session name and a prompt. You can also use "temp" as a session name to start a temporary chat session.
 ```shell
 sgpt --chat number "please remember my favorite number: 4"
 # -> I will remember that your favorite number is 4.
@@ -164,6 +164,48 @@ sgpt --chat sh "Convert the resulting file into an MP3"
 # -> ffmpeg -i output.mp4 -vn -acodec libmp3lame -ac 2 -ab 160k -ar 48000 final_output.mp3
 ```
 
+### REPL
+There is very handy REPL (read–eval–print loop) mode, which allows you to interactively chat with GPT models. To start a chat session in REPL mode, use the `--repl` option followed by a unique session name. You can also use "temp" as a session name to start a temporary REPL session. Note that `--chat` and `--repl` are using same chat sessions, so you can use `--chat` to start a chat session and then use `--repl` to continue the conversation in REPL mode. REPL mode will also show history of your conversation in the beginning.
+
+<p align="center">
+  <img src="https://s10.gifyu.com/images/repl-demo.gif" alt="gif">
+</p>
+
+```text
+sgpt --repl temp
+Entering REPL mode, press Ctrl+C to exit.
+>>> What is REPL?
+REPL stands for Read-Eval-Print Loop. It is a programming environment ...
+>>> How can I use Python with REPL?
+To use Python with REPL, you can simply open a terminal or command prompt ...
+```
+REPL mode can work with `--shell` and `--code` options, which makes it very handy for interactive shell commands and code generation:
+```text
+sgpt --repl temp --shell
+Entering shell REPL mode, type [e] to execute commands or press Ctrl+C to exit.
+>>> What is in current folder?
+ls
+>>> Show file sizes
+ls -lh
+>>> Sort them by file sizes
+ls -lhS
+>>> e (enter just e to execute commands)
+...
+```
+Example of using REPL mode to generate code:
+```text
+sgpt --repl temp --code
+Entering REPL mode, press Ctrl+C to exit.
+>>> Using Python request localhost:80
+import requests
+response = requests.get('http://localhost:80')
+print(response.text)
+>>> Change port to 443
+import requests
+response = requests.get('https://localhost:443')
+print(response.text)
+```
+
 ### Chat sessions
 To list all the current chat sessions, use the `--list-chat` option:
 ```shell

setup.py

@@ -3,7 +3,7 @@
 # pylint: disable=consider-using-with
 setup(
     name="shell_gpt",
-    version="0.8.2",
+    version="0.8.3",
     packages=find_packages(),
     install_requires=[
         "typer~=0.7.0",

sgpt/__init__.py

@@ -3,6 +3,7 @@
 from .client import OpenAIClient as OpenAIClient
 from .handlers.chat_handler import ChatHandler as ChatHandler
 from .handlers.default_handler import DefaultHandler as DefaultHandler
+from .handlers.repl_handler import ReplHandler as ReplHandler
 from . import utils as utils
 from .app import main as main
 from .app import entry_point as cli

sgpt/app.py

@@ -13,12 +13,15 @@
 
 import os
 
+# To allow users to use arrow keys in the REPL.
+import readline  # pylint: disable=unused-import
+
 import typer
 
 # Click is part of typer.
 from click import MissingParameter, BadArgumentUsage
 from sgpt import config, OpenAIClient
-from sgpt import ChatHandler, DefaultHandler
+from sgpt import ChatHandler, DefaultHandler, ReplHandler
 from sgpt.utils import get_edited_prompt
 
 
@@ -40,23 +43,6 @@ def main(  # pylint: disable=too-many-arguments
         max=1.0,
         help="Limits highest probable tokens (words).",
     ),
-    chat: str = typer.Option(
-        None,
-        help="Follow conversation with id (chat mode).",
-        rich_help_panel="Chat Options",
-    ),
-    show_chat: str = typer.Option(  # pylint: disable=W0613
-        None,
-        help="Show all messages from provided chat id.",
-        callback=ChatHandler.show_messages,
-        rich_help_panel="Chat Options",
-    ),
-    list_chat: bool = typer.Option(  # pylint: disable=W0613
-        False,
-        help="List all existing chat ids.",
-        callback=ChatHandler.list_ids,
-        rich_help_panel="Chat Options",
-    ),
     shell: bool = typer.Option(
         False,
         "--shell",
@@ -77,20 +63,55 @@ def main(  # pylint: disable=too-many-arguments
         True,
         help="Cache completion results.",
     ),
+    chat: str = typer.Option(
+        None,
+        help="Follow conversation with id, " 'use "temp" for quick session.',
+        rich_help_panel="Chat Options",
+    ),
+    repl: str = typer.Option(
+        None,
+        help="Start a REPL (Read–eval–print loop) session.",
+        rich_help_panel="Chat Options",
+    ),
+    show_chat: str = typer.Option(  # pylint: disable=W0613
+        None,
+        help="Show all messages from provided chat id.",
+        callback=ChatHandler.show_messages_callback,
+        rich_help_panel="Chat Options",
+    ),
+    list_chat: bool = typer.Option(  # pylint: disable=W0613
+        False,
+        help="List all existing chat ids.",
+        callback=ChatHandler.list_ids,
+        rich_help_panel="Chat Options",
+    ),
 ) -> None:
-    if not prompt and not editor:
+    if not prompt and not editor and not repl:
         raise MissingParameter(param_hint="PROMPT", param_type="string")
 
     if shell and code:
         raise BadArgumentUsage("--shell and --code options cannot be used together.")
 
+    if chat and repl:
+        raise BadArgumentUsage("--chat and --repl options cannot be used together.")
+
     if editor:
         prompt = get_edited_prompt()
 
     api_host = config.get("OPENAI_API_HOST")
     api_key = config.get("OPENAI_API_KEY")
     client = OpenAIClient(api_host, api_key)
 
+    if repl:
+        # Will be in infinite loop here until user exits with Ctrl+C.
+        ReplHandler(client, repl, shell, code).handle(
+            prompt,
+            temperature=temperature,
+            top_probability=top_probability,
+            chat_id=repl,
+            caching=cache,
+        )
+
     if chat:
         full_completion = ChatHandler(client, chat, shell, code).handle(
             prompt,

sgpt/config.py

@@ -7,6 +7,7 @@
 
 CONFIG_FOLDER = os.path.expanduser("~/.config")
 CONFIG_PATH = Path(CONFIG_FOLDER) / "shell_gpt" / ".sgptrc"
+# TODO: Refactor it to CHAT_STORAGE_PATH.
 CHAT_CACHE_PATH = Path(gettempdir()) / "shell_gpt" / "chat_cache"
 CACHE_PATH = Path(gettempdir()) / "shell_gpt" / "cache"
 CHAT_CACHE_LENGTH = 100

sgpt/handlers/chat_handler.py

@@ -16,14 +16,14 @@
 class ChatSession:
     """
     This class is used as a decorator for OpenAI chat API requests.
-    The ChatCache class caches chat messages and keeps track of the
+    The ChatSession class caches chat messages and keeps track of the
     conversation history. It is designed to store cached messages
     in a specified directory and in JSON format.
     """
 
     def __init__(self, length: int, storage_path: Path):
         """
-        Initialize the ChatCache decorator.
+        Initialize the ChatSession decorator.
 
         :param length: Integer, maximum number of cached messages to keep.
         """
@@ -71,7 +71,7 @@ def _write(self, messages: List[Dict], chat_id: str):
 
     def invalidate(self, chat_id: str):
         file_path = self.storage_path / chat_id
-        file_path.unlink()
+        file_path.unlink(missing_ok=True)
 
     def get_messages(self, chat_id):
         messages = self._read(chat_id)
@@ -104,10 +104,9 @@ def __init__(  # pylint: disable=too-many-arguments
         self.mode = CompletionModes.get_mode(shell, code)
         self.model = model
 
-        chat_history = self.chat_session.get_messages(self.chat_id)
-        self.is_shell_chat = chat_history and chat_history[0].endswith("###\nCommand:")
-        self.is_code_chat = chat_history and chat_history[0].endswith("###\nCode:")
-        self.is_default_chat = chat_history and chat_history[0].endswith("###")
+        if chat_id == "temp":
+            # If the chat id is "temp", we don't want to save the chat session.
+            self.chat_session.invalidate(chat_id)
 
         self.validate()
 
@@ -120,16 +119,40 @@ def list_ids(cls, value) -> None:
             typer.echo(chat_id)
         raise typer.Exit()
 
+    @property
+    def initiated(self) -> bool:
+        return self.chat_session.exists(self.chat_id)
+
+    @property
+    def is_shell_chat(self) -> bool:
+        # TODO: Should be optimized for REPL mode.
+        chat_history = self.chat_session.get_messages(self.chat_id)
+        return chat_history and chat_history[0].endswith("###\nCommand:")
+
+    @property
+    def is_code_chat(self) -> bool:
+        chat_history = self.chat_session.get_messages(self.chat_id)
+        return chat_history and chat_history[0].endswith("###\nCode:")
+
+    @property
+    def is_default_chat(self) -> bool:
+        chat_history = self.chat_session.get_messages(self.chat_id)
+        return chat_history and chat_history[0].endswith("###")
+
     @classmethod
-    def show_messages(cls, chat_id: str) -> None:
+    def show_messages_callback(cls, chat_id) -> None:
         if not chat_id:
             return
+        cls.show_messages(chat_id)
+        raise typer.Exit()
+
+    @classmethod
+    def show_messages(cls, chat_id: str) -> None:
         # Prints all messages from a specified chat ID to the console.
         for index, message in enumerate(cls.chat_session.get_messages(chat_id)):
             message = message.replace("\nCommand:", "").replace("\nCode:", "")
             color = "cyan" if index % 2 == 0 else "green"
             typer.secho(message, fg=color)
-        raise typer.Exit()
 
     def validate(self) -> None:
         if self.initiated:
@@ -155,10 +178,6 @@ def validate(self) -> None:
                 elif self.is_code_chat:
                     self.mode = CompletionModes.CODE
 
-    @property
-    def initiated(self) -> bool:
-        return self.chat_session.exists(self.chat_id)
-
     def make_prompt(self, prompt: str) -> str:
         prompt = prompt.strip()
         if self.initiated:

sgpt/handlers/handler.py

@@ -6,7 +6,7 @@
 
 
 class Handler:
-    def __init__(self, client: OpenAIClient):
+    def __init__(self, client: OpenAIClient) -> None:
         self.client = client
 
     def make_prompt(self, prompt) -> str:

sgpt/handlers/repl_handler.py

@@ -0,0 +1,54 @@
+import os
+
+import typer
+
+from rich import print as rich_print
+from rich.rule import Rule
+
+from sgpt.handlers.chat_handler import ChatHandler
+from sgpt.client import OpenAIClient
+from sgpt.utils import CompletionModes
+
+
+class ReplHandler(ChatHandler):
+    def __init__(  # pylint: disable=useless-parent-delegation,too-many-arguments
+        self,
+        client: OpenAIClient,
+        chat_id: str,
+        shell: bool = False,
+        code: bool = False,
+        model: str = "gpt-3.5-turbo",
+    ):
+        super().__init__(client, chat_id, shell, code, model)
+
+    def handle(self, prompt: str, **kwargs) -> None:
+        if self.initiated:
+            rich_print(Rule(title="Chat History", style="bold magenta"))
+            self.show_messages(self.chat_id)
+            rich_print(Rule(style="bold magenta"))
+
+        info_message = (
+            "Entering REPL mode, press Ctrl+C to exit."
+            if not self.mode == CompletionModes.SHELL
+            else "Entering shell REPL mode, type [e] to execute commands or press Ctrl+C to exit."
+        )
+        typer.secho(info_message, fg="yellow")
+
+        if not prompt:
+            prompt = typer.prompt(">>>", prompt_suffix=" ")
+        else:
+            typer.echo(f">>> {prompt}")
+
+        while True:
+            full_completion = super().handle(prompt, **kwargs)
+            prompt = typer.prompt(">>>", prompt_suffix=" ")
+            if prompt == "exit()":
+                # This is also useful during tests.
+                raise typer.Exit()
+            if self.mode == CompletionModes.SHELL:
+                if prompt == "e":
+                    typer.echo()
+                    os.system(full_completion)
+                    typer.echo()
+                    rich_print(Rule(style="bold magenta"))
+                    prompt = typer.prompt(">>> ", prompt_suffix=" ")