wip

wip

wip

wip

Author: kharvd

.gitignore

@@ -159,4 +159,6 @@ cython_debug/
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
 
-.vscode/
+.vscode/
+
+gpt.yml

Dockerfile

@@ -0,0 +1,17 @@
+FROM python:3.9.17-alpine
+WORKDIR /app
+
+RUN apk add build-base linux-headers clang 
+COPY requirements.txt requirements_docker.txt ./
+RUN --mount=type=cache,target=/root/.cache/pip pip install -r requirements.txt
+RUN --mount=type=cache,target=/root/.cache/pip pip install -r requirements_docker.txt
+
+COPY . .
+
+RUN adduser -D gpt
+USER gpt
+RUN mkdir -p $HOME/.config/gpt-cli
+RUN cp /app/gpt.yml $HOME/.config/gpt-cli/gpt.yml
+
+ENV GPTCLI_ALLOW_CODE_EXECUTION=1
+ENTRYPOINT ["python", "gpt.py"]

gpt.py

@@ -1,11 +1,16 @@
 #!/usr/bin/env python
 import os
+import subprocess
+import tempfile
+import traceback
 from typing import cast
 import openai
+import random
 import argparse
 import sys
 import logging
-import google.generativeai as genai
+
+# import google.generativeai as genai
 import gptcli.anthropic
 from gptcli.assistant import (
     Assistant,
@@ -24,6 +29,7 @@
     choose_config_file,
     read_yaml_config,
 )
+from gptcli.interpreter import CodeInterpreterListener
 from gptcli.llama import init_llama_models
 from gptcli.logging import LoggingChatListener
 from gptcli.cost import PriceChatListener
@@ -89,6 +95,12 @@ def parse_args(config: GptCliConfig):
         default=config.log_file,
         help="The file to write logs to",
     )
+    parser.add_argument(
+        "--history_file",
+        type=str,
+        default=config.history_file,
+        help="The file to write chat history to",
+    )
     parser.add_argument(
         "--log_level",
         type=str,
@@ -166,8 +178,8 @@ def main():
     if config.anthropic_api_key:
         gptcli.anthropic.api_key = config.anthropic_api_key
 
-    if config.google_api_key:
-        genai.configure(api_key=config.google_api_key)
+    #  if config.google_api_key:
+    #      genai.configure(api_key=config.google_api_key)
 
     if config.llama_models is not None:
         init_llama_models(config.llama_models)
@@ -215,16 +227,24 @@ def __init__(self, assistant: Assistant, markdown: bool, show_price: bool):
         if show_price:
             listeners.append(PriceChatListener(assistant))
 
+        if os.environ.get("GPTCLI_ALLOW_CODE_EXECUTION") == "1":
+            listeners.append(CodeInterpreterListener("python_eval"))
+
         listener = CompositeChatListener(listeners)
-        super().__init__(assistant, listener)
+        super().__init__(
+            assistant,
+            listener,
+        )
 
 
 def run_interactive(args, assistant):
     logger.info("Starting a new chat session. Assistant config: %s", assistant.config)
     session = CLIChatSession(
         assistant=assistant, markdown=args.markdown, show_price=args.show_price
     )
-    history_filename = os.path.expanduser("~/.config/gpt-cli/history")
+    history_filename = args.history_file or os.path.expanduser(
+        "~/.config/gpt-cli/history"
+    )
     os.makedirs(os.path.dirname(history_filename), exist_ok=True)
     input_provider = CLIUserInputProvider(history_filename=history_filename)
     session.loop(input_provider)

gpt.yml.template

@@ -0,0 +1,12 @@
+markdown: True
+openai_api_key: <YOUR_OPENAI_API_KEY_HERE>
+log_file: /mnt/gpt.log
+log_level: DEBUG
+history_file: /mnt/history
+assistants:
+  python:
+    model: gpt-4-0613
+    enable_code_execution: True
+    messages:
+      - { role: "system", content: "You are a helpful assistant. You have access to a Python environment. You can install missing packages. You have access to the internet. The user can see the code you are executing and its output: do not repeat them to the user verbatim. Pre-installed packages: numpy, matplotlib, ipython, ipykernel." }
+

gptcli/anthropic.py

@@ -1,6 +1,7 @@
 import os
 from typing import Iterator, List
-import anthropic
+
+# import anthropic
 
 from gptcli.completion import CompletionProvider, Message
 

gptcli/assistant.py

@@ -4,7 +4,7 @@
 import platform
 from typing import Any, Dict, Iterator, Optional, TypedDict, List
 
-from gptcli.completion import CompletionProvider, ModelOverrides, Message
+from gptcli.completion import Completion, CompletionProvider, ModelOverrides, Message
 from gptcli.google import GoogleCompletionProvider
 from gptcli.llama import LLaMACompletionProvider
 from gptcli.openai import OpenAICompletionProvider
@@ -16,12 +16,14 @@ class AssistantConfig(TypedDict, total=False):
     model: str
     temperature: float
     top_p: float
+    enable_code_execution: bool
 
 
 CONFIG_DEFAULTS = {
     "model": "gpt-3.5-turbo",
     "temperature": 0.7,
     "top_p": 1.0,
+    "enable_code_execution": False,
 }
 
 DEFAULT_ASSISTANTS: Dict[str, AssistantConfig] = {
@@ -89,7 +91,7 @@ def init_messages(self) -> List[Message]:
         return self.config.get("messages", [])[:]
 
     def supported_overrides(self) -> List[str]:
-        return ["model", "temperature", "top_p"]
+        return ["model", "temperature", "top_p", "enable_code_execution"]
 
     def _param(self, param: str, override_params: ModelOverrides) -> Any:
         # If the param is in the override_params, use that value
@@ -101,9 +103,15 @@ def _param(self, param: str, override_params: ModelOverrides) -> Any:
 
     def complete_chat(
         self, messages, override_params: ModelOverrides = {}, stream: bool = True
-    ) -> Iterator[str]:
+    ) -> Iterator[Completion]:
         model = self._param("model", override_params)
         completion_provider = get_completion_provider(model)
+
+        enable_code_execution = (
+            bool(self._param("enable_code_execution", override_params))
+            and os.environ.get("GPTCLI_ALLOW_CODE_EXECUTION") == "1"
+        )
+
         return completion_provider.complete(
             messages,
             {
@@ -112,6 +120,7 @@ def complete_chat(
                 "top_p": float(self._param("top_p", override_params)),
             },
             stream,
+            enable_code_execution,
         )
 
 

gptcli/cli.py

@@ -1,4 +1,8 @@
+import base64
+import logging
 import re
+import json
+from imgcat import imgcat
 from prompt_toolkit import PromptSession
 from prompt_toolkit.history import FileHistory
 from openai import OpenAIError, InvalidRequestError
@@ -9,6 +13,7 @@
 from typing import Any, Dict, Optional, Tuple
 
 from rich.text import Text
+from gptcli.completion import FunctionCall, Message, merge_dicts
 from gptcli.session import (
     ALL_COMMANDS,
     COMMAND_CLEAR,
@@ -32,7 +37,7 @@
 class StreamingMarkdownPrinter:
     def __init__(self, console: Console, markdown: bool):
         self.console = console
-        self.current_text = ""
+        self.current_message = {}
         self.markdown = markdown
         self.live: Optional[Live] = None
 
@@ -44,15 +49,50 @@ def __enter__(self) -> "StreamingMarkdownPrinter":
             self.live.__enter__()
         return self
 
-    def print(self, text: str):
-        self.current_text += text
+    def _format_function_call(self, function_call: FunctionCall) -> str:
+        text = ""
+        if function_call.get("name") == "python_eval":
+            source = function_call.get("arguments", "")
+            try:
+                source = json.loads(source).get("source", "")
+            except:
+                source = source + '"}'
+                try:
+                    source = json.loads(source).get("source", "")
+                except:
+                    source = ""
+
+            text += "\n\nExecuting Python code:\n"
+            text += f"```python\n{source}\n```"
+        else:
+            function_name = function_call.get("name", "?")
+            function_arguments = function_call.get("arguments", {})
+            text += f"""\n
+Calling function:
+
+```
+{function_name}({function_arguments})
+```"""
+        return text
+
+    def print(self, message_delta: Message):
+        self.current_message = merge_dicts(self.current_message, message_delta)
+
         if self.markdown:
             assert self.live
-            content = Markdown(self.current_text, style="green")
+            text = self.current_message.get("content", "")
+
+            function_call = self.current_message.get("function_call")
+            if function_call:
+                text += self._format_function_call(function_call)
+
+            content = Markdown(text, style="green")
             self.live.update(content)
             self.live.refresh()
         else:
-            self.console.print(Text(text, style="green"), end="")
+            self.console.print(
+                Text(message_delta.get("content", ""), style="green"), end=""
+            )
 
     def __exit__(self, *args):
         if self.markdown:
@@ -66,17 +106,29 @@ def __init__(self, console: Console, markdown: bool):
         self.console = console
         self.markdown = markdown
         self.printer = StreamingMarkdownPrinter(self.console, self.markdown)
-        self.first_token = True
 
     def __enter__(self):
         self.printer.__enter__()
         return self
 
-    def on_next_token(self, token: str):
-        if self.first_token and token.startswith(" "):
-            token = token[1:]
-        self.first_token = False
-        self.printer.print(token)
+    def on_message_delta(self, message_delta: Message):
+        self.printer.print(message_delta)
+
+    def on_function_result(self, result: dict):
+        self.console.print(Text("Function result:", style="yellow"))
+        if "image/png" in result:
+            image_base64 = result["image/png"]
+            image_bytes = base64.b64decode(image_base64)
+            imgcat(image_bytes)
+        if "text/plain" in result:
+            text = result["text/plain"]
+            if self.markdown:
+                content = Markdown(
+                    f"```\n{text}\n```",
+                )
+            else:
+                content = Text(text, style="yellow")
+            self.console.print(content)
 
     def __exit__(self, *args):
         self.printer.__exit__(*args)

gptcli/completion.py

@@ -1,21 +1,57 @@
 from abc import abstractmethod
-from typing import Iterator, List, TypedDict
+from typing import Iterator, List, Optional, TypedDict
+from typing_extensions import Required
 
 
-class Message(TypedDict):
-    role: str
-    content: str
+class FunctionCall(TypedDict, total=False):
+    name: str
+    arguments: str
+
+
+class Message(TypedDict, total=False):
+    role: Required[str]
+    content: Optional[str]
+    name: str
+    function_call: FunctionCall
+
+
+def merge_dicts(a: dict, b: dict):
+    """
+    Given two nested dicts with string values, merge dict `b` into dict `a`, concatenating
+    string values.
+    """
+    for key, value in b.items():
+        if isinstance(value, dict):
+            a[key] = merge_dicts(a.get(key, {}), value)
+        elif value is not None:
+            a[key] = a.get(key, "") + value
+    return a
 
 
 class ModelOverrides(TypedDict, total=False):
     model: str
     temperature: float
     top_p: float
+    enable_code_execution: bool
+
+
+class CompletionDelta(TypedDict):
+    content: Optional[str]
+    function_call: Optional[FunctionCall]
+
+
+class Completion(TypedDict):
+    delta: Message
+    finish_reason: Optional[str]
 
 
 class CompletionProvider:
     @abstractmethod
     def complete(
-        self, messages: List[Message], args: dict, stream: bool = False
-    ) -> Iterator[str]:
+        self,
+        messages: List[Message],
+        args: dict,
+        stream: bool = False,
+        enable_code_execution: bool = False,
+    ) -> Iterator[Completion]:
         pass

gptcli/composite.py

@@ -2,7 +2,7 @@
 from gptcli.session import ChatListener, ResponseStreamer
 
 
-from typing import List
+from typing import List, Optional
 
 
 class CompositeResponseStreamer(ResponseStreamer):
@@ -14,9 +14,13 @@ def __enter__(self):
             streamer.__enter__()
         return self
 
-    def on_next_token(self, token: str):
+    def on_message_delta(self, message_delta: Message):
         for streamer in self.streamers:
-            streamer.on_next_token(token)
+            streamer.on_message_delta(message_delta)
+
+    def on_function_result(self, result: dict):
+        for streamer in self.streamers:
+            streamer.on_function_result(result)
 
     def __exit__(self, *args):
         for streamer in self.streamers:
@@ -57,3 +61,9 @@ def on_chat_response(
     ):
         for listener in self.listeners:
             listener.on_chat_response(messages, response, overrides)
+
+    def on_function_call(self, function_name: str, **kwargs) -> Optional[str]:
+        for listener in self.listeners:
+            result = listener.on_function_call(function_name, **kwargs)
+            if result is not None:
+                return result