Upload 17 files

README.md

@@ -1,13 +1,120 @@
 ---
 title: Code Interpreter
-emoji: 🏃
-colorFrom: yellow
-colorTo: gray
+emoji: 🔥
+colorFrom: blue
+colorTo: pink
 sdk: gradio
-sdk_version: 3.42.0
+sdk_version: 3.41.2
 app_file: app.py
 pinned: false
 license: mit
+duplicated_from: dongsiqie/Code-Interpreter
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Local-Code-Interpreter
+A local implementation of OpenAI's ChatGPT Code Interpreter.
+
+## Introduction
+
+OpenAI's Code Interpreter plugin for ChatGPT is a revolutionary feature that allows the execution of Python code within the AI model. However, it execute code within an online sandbox and has certain limitations. In this project, we present Local Code Interpreter – which enables code execution on your local device, offering enhanced flexibility, security, and convenience.
+
+## Key Advantages
+
+- **Custom Environment**: Execute code in a customized environment of your choice, ensuring you have the right packages and settings.
+
+- **Seamless Experience**: Say goodbye to file size restrictions and internet issues while uploading. With Local Code Interpreter, you're in full control.
+
+- **GPT-3.5 Availability**: While official Code Interpreter is only available for GPT-4 model, the Local Code Interpreter offers the flexibility to switch between both GPT-3.5 and GPT-4 models.
+
+- **Enhanced Data Security**: Keep your data more secure by running code locally, minimizing data transfer over the internet.
+
+## Note
+Executing AI-generated code without human review on your own device is not safe. You are responsible for taking measures to protect the security of your device and data (such as using a virtural machine) before launching this program. All consequences caused by using this program shall be borne by youself.
+
+## Usage
+
+### Getting Started
+
+1. Clone this repository to your local device
+   ```shell
+   git clone https://github.com/MrGreyfun/Local-Code-Interpreter.git
+   ```
+
+2. Install the necessary dependencies. The program has been tested on Windows 10 and CentOS Linux 7.8, with Python 3.9.16. Required packages include:
+   ```text
+   Jupyter Notebook    6.5.4
+   gradio              3.39.0
+   openai              0.27.8
+   ```
+   Other system or package version may also work.
+### Configuration
+
+1. Create a `config.json` file in the `src` directory, following the examples provided in the `config_example` directory.
+
+2. Configure your API key in the `config.json` file.
+
+Please Note:
+1. **Set the `model_name` Correctly**
+    This program relies on the function calling capability of two specific models:
+    - `gpt-3.5-turbo-0613`
+    - `gpt-4-0613`
+
+    Older versions of the models will not work. 
+
+    For Azure OpenAI service users:
+    - Set the `model_name` as your deployment name.
+    - Confirm that the deployed model corresponds to the `0613` version.
+
+2. **API Version Settings**
+    If you're using Azure OpenAI service, set the `API_VERSION` to `2023-07-01-preview` in the `config.json` file. Note that other API versions do not support the necessary function calls for this program.
+
+3. **Alternate API Key Handling**
+    If you prefer not to store your API key in the `config.json` file, you can opt for an alternate approach:
+    - Leave the `API_KEY` field in `config.json` as an empty string:
+        ```json
+        "API_KEY": ""
+        ```
+    - Set the environment variable `OPENAI_API_KEY` with your API key before running the program:
+        - On Windows:
+        ```shell
+        set OPENAI_API_KEY=<YOUR-API-KEY>
+        ```
+        - On Linux:
+        ```shell
+        export OPENAI_API_KEY=<YOUR-API-KEY>
+        ```
+
+## Getting Started
+
+1. Navigate to the `src` directory.
+
+2. Run the command:
+   ```shell
+   python web_ui.py
+   ```
+
+3. Access the generated link in your browser to start using the Local Code Interpreter.
+
+## Example
+
+Imagine uploading a data file and requesting the model to perform linear regression and visualize the data. See how Local Code Interpreter provides a seamless experience:
+
+1. Upload the data and request linear regression:
+   ![Example 1](example_img/1.jpg)
+
+2. Encounter an error in the generated code:
+   ![Example 2](example_img/2.jpg)
+
+3. ChatGPT automatically checks the data structure and fixes the bug:
+   ![Example 3](example_img/3.jpg)
+
+4. The corrected code runs successfully:
+   ![Example 4](example_img/4.jpg)
+
+5. The final result meets your requirements:
+   ![Example 5](example_img/5.jpg)
+   ![Example 6](example_img/6.jpg)
+
+
+
+

app.py

@@ -0,0 +1,197 @@
+from response_parser import *
+import gradio as gr
+
+
+def initialization(state_dict: Dict) -> None:
+    if not os.path.exists('cache'):
+        os.mkdir('cache')
+    if state_dict["bot_backend"] is None:
+        state_dict["bot_backend"] = BotBackend()
+        if 'OPENAI_API_KEY' in os.environ:
+            del os.environ['OPENAI_API_KEY']
+
+
+def get_bot_backend(state_dict: Dict) -> BotBackend:
+    return state_dict["bot_backend"]
+
+
+def switch_to_gpt4(state_dict: Dict, whether_switch: bool) -> None:
+    bot_backend = get_bot_backend(state_dict)
+    if whether_switch:
+        bot_backend.update_gpt_model_choice("GPT-4")
+    else:
+        bot_backend.update_gpt_model_choice("GPT-3.5")
+
+
+def add_text(state_dict: Dict, history: List, text: str) -> Tuple[List, Dict]:
+    bot_backend = get_bot_backend(state_dict)
+    bot_backend.add_text_message(user_text=text)
+
+    history = history + [(text, None)]
+
+    return history, gr.update(value="", interactive=False)
+
+
+def add_file(state_dict: Dict, history: List, file) -> List:
+    bot_backend = get_bot_backend(state_dict)
+    path = file.name
+    filename = os.path.basename(path)
+
+    bot_msg = [f'📁[{filename}]', None]
+    history.append(bot_msg)
+
+    bot_backend.add_file_message(path=path, bot_msg=bot_msg)
+
+    return history
+
+
+def undo_upload_file(state_dict: Dict, history: List) -> Tuple[List, Dict]:
+    bot_backend = get_bot_backend(state_dict)
+    bot_msg = bot_backend.revoke_file()
+
+    if bot_msg is None:
+        return history, gr.Button.update(interactive=False)
+
+    else:
+        assert history[-1] == bot_msg
+        del history[-1]
+        if bot_backend.revocable_files:
+            return history, gr.Button.update(interactive=True)
+        else:
+            return history, gr.Button.update(interactive=False)
+
+
+def refresh_file_display(state_dict: Dict) -> List[str]:
+    bot_backend = get_bot_backend(state_dict)
+    work_dir = bot_backend.jupyter_work_dir
+    filenames = os.listdir(work_dir)
+    paths = []
+    for filename in filenames:
+        paths.append(
+            os.path.join(work_dir, filename)
+        )
+    return paths
+
+
+def restart_ui(history: List) -> Tuple[List, Dict, Dict, Dict, Dict]:
+    history.clear()
+    return (
+        history,
+        gr.Textbox.update(value="", interactive=False),
+        gr.Button.update(interactive=False),
+        gr.Button.update(interactive=False),
+        gr.Button.update(interactive=False)
+    )
+
+
+def restart_bot_backend(state_dict: Dict) -> None:
+    bot_backend = get_bot_backend(state_dict)
+    bot_backend.restart()
+
+
+def bot(state_dict: Dict, history: List) -> List:
+    bot_backend = get_bot_backend(state_dict)
+
+    while bot_backend.finish_reason in ('new_input', 'function_call'):
+        if history[-1][0] is None:
+            history.append(
+                [None, ""]
+            )
+        else:
+            history[-1][1] = ""
+
+        response = chat_completion(bot_backend=bot_backend)
+        for chunk in response:
+            history, weather_exit = parse_response(
+                chunk=chunk,
+                history=history,
+                bot_backend=bot_backend
+            )
+            yield history
+            if weather_exit:
+                exit(-1)
+
+    yield history
+
+
+if __name__ == '__main__':
+    config = get_config()
+    with gr.Blocks(theme=gr.themes.Base()) as block:
+        """
+        Reference: https://www.gradio.app/guides/creating-a-chatbot-fast
+        """
+        # UI components
+        state = gr.State(value={"bot_backend": None})
+        with gr.Tab("Chat"):
+            chatbot = gr.Chatbot([], elem_id="chatbot", label="Local Code Interpreter", height=750)
+            with gr.Row():
+                with gr.Column(scale=0.85):
+                    text_box = gr.Textbox(
+                        show_label=False,
+                        placeholder="Enter text and press enter, or upload a file",
+                        container=False
+                    )
+                with gr.Column(scale=0.15, min_width=0):
+                    file_upload_button = gr.UploadButton("📁", file_types=['file'])
+            with gr.Row(equal_height=True):
+                with gr.Column(scale=0.7):
+                    check_box = gr.Checkbox(label="Use GPT-4", interactive=config['model']['GPT-4']['available'])
+                    check_box.change(fn=switch_to_gpt4, inputs=[state, check_box])
+                with gr.Column(scale=0.15, min_width=0):
+                    restart_button = gr.Button(value='🔄 Restart')
+                with gr.Column(scale=0.15, min_width=0):
+                    undo_file_button = gr.Button(value="↩️Undo upload file", interactive=False)
+        with gr.Tab("Files"):
+            file_output = gr.Files()
+        gr.Markdown(
+            '''
+            <center>
+              <a href="https://github.com/MrGreyfun/Local-Code-Interpreter">
+                <img src="file=github-log.png" width="2.5%">
+                <br>
+                Open source on GitHub
+              </a>
+            </center>
+            '''
+        )
+
+
+        # Components function binding
+        txt_msg = text_box.submit(add_text, [state, chatbot, text_box], [chatbot, text_box], queue=False).then(
+            bot, [state, chatbot], chatbot
+        )
+        txt_msg.then(fn=refresh_file_display, inputs=[state], outputs=[file_output])
+        txt_msg.then(lambda: gr.update(interactive=True), None, [text_box], queue=False)
+        txt_msg.then(lambda: gr.Button.update(interactive=False), None, [undo_file_button], queue=False)
+
+        file_msg = file_upload_button.upload(
+            add_file, [state, chatbot, file_upload_button], [chatbot], queue=False
+        ).then(
+            bot, [state, chatbot], chatbot
+        )
+        file_msg.then(lambda: gr.Button.update(interactive=True), None, [undo_file_button], queue=False)
+        file_msg.then(fn=refresh_file_display, inputs=[state], outputs=[file_output])
+
+        undo_file_button.click(
+            fn=undo_upload_file, inputs=[state, chatbot], outputs=[chatbot, undo_file_button]
+        ).then(
+            fn=refresh_file_display, inputs=[state], outputs=[file_output]
+        )
+
+        restart_button.click(
+            fn=restart_ui, inputs=[chatbot],
+            outputs=[chatbot, text_box, restart_button, file_upload_button, undo_file_button]
+        ).then(
+            fn=restart_bot_backend, inputs=[state], queue=False
+        ).then(
+            fn=refresh_file_display, inputs=[state], outputs=[file_output]
+        ).then(
+            fn=lambda: (gr.Textbox.update(interactive=True), gr.Button.update(interactive=True),
+                        gr.Button.update(interactive=True)),
+            inputs=None, outputs=[text_box, restart_button, file_upload_button], queue=False
+        )
+
+        block.load(fn=initialization, inputs=[state])
+
+    block.queue()
+    block.launch(inbrowser=True)

bot_backend.py

@@ -0,0 +1,232 @@
+import json
+import openai
+import os
+import copy
+import shutil
+from jupyter_backend import *
+from typing import *
+
+functions = [
+    {
+        "name": "execute_code",
+        "description": "This function allows you to execute Python code and retrieve the terminal output. If the code "
+                       "generates image output, the function will return the text '[image]'. The code is sent to a "
+                       "Jupyter kernel for execution. The kernel will remain active after execution, retaining all "
+                       "variables in memory.",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "code": {
+                    "type": "string",
+                    "description": "The code text"
+                }
+            },
+            "required": ["code"],
+        }
+    }
+]
+
+system_msg = '''You are an AI code interpreter.
+Your goal is to help users do a variety of jobs by executing Python code.
+
+You should:
+1. Comprehend the user's requirements carefully & to the letter. 
+2. Give a brief description for what you plan to do & call the execute_code function to run code
+3. Provide results analysis based on the execution output. 
+4. If error occurred, try to fix it.
+
+Note: If the user uploads a file, you will receive a system message "User uploaded a file: filename". Use the filename as the path in the code. '''
+
+with open('config.json') as f:
+    config = json.load(f)
+
+if not config['API_KEY']:
+    config['API_KEY'] = os.getenv('OPENAI_API_KEY')
+    os.unsetenv('OPENAI_API_KEY')
+
+
+def get_config():
+    return config
+
+
+def config_openai_api(api_type, api_base, api_version, api_key):
+    openai.api_type = api_type
+    openai.api_base = api_base
+    openai.api_version = api_version
+    openai.api_key = api_key
+
+
+class GPTResponseLog:
+    def __init__(self):
+        self.assistant_role_name = ''
+        self.content = ''
+        self.function_name = None
+        self.function_args_str = ''
+        self.display_code_block = ''
+        self.finish_reason = 'stop'
+        self.bot_history = None
+
+    def reset_gpt_response_log_values(self, exclude=None):
+        if exclude is None:
+            exclude = []
+
+        attributes = {'assistant_role_name': '',
+                      'content': '',
+                      'function_name': None,
+                      'function_args_str': '',
+                      'display_code_block': '',
+                      'finish_reason': 'stop',
+                      'bot_history': None}
+
+        for attr_name in exclude:
+            del attributes[attr_name]
+        for attr_name, value in attributes.items():
+            setattr(self, attr_name, value)
+
+    def set_assistant_role_name(self, assistant_role_name: str):
+        self.assistant_role_name = assistant_role_name
+
+    def add_content(self, content: str):
+        self.content += content
+
+    def set_function_name(self, function_name: str):
+        self.function_name = function_name
+
+    def copy_current_bot_history(self, bot_history: List):
+        self.bot_history = copy.deepcopy(bot_history)
+
+    def add_function_args_str(self, function_args_str: str):
+        self.function_args_str += function_args_str
+
+    def update_display_code_block(self, display_code_block):
+        self.display_code_block = display_code_block
+
+    def update_finish_reason(self, finish_reason: str):
+        self.finish_reason = finish_reason
+
+
+class BotBackend(GPTResponseLog):
+    def __init__(self):
+        super().__init__()
+        self.unique_id = hash(id(self))
+        self.jupyter_work_dir = f'cache/work_dir_{self.unique_id}'
+        self.jupyter_kernel = JupyterKernel(work_dir=self.jupyter_work_dir)
+        self.gpt_model_choice = "GPT-3.5"
+        self.revocable_files = []
+        self._init_conversation()
+        self._init_api_config()
+        self._init_kwargs_for_chat_completion()
+
+    def _init_conversation(self):
+        first_system_msg = {'role': 'system', 'content': system_msg}
+        if hasattr(self, 'conversation'):
+            self.conversation.clear()
+            self.conversation.append(first_system_msg)
+        else:
+            self.conversation: List[Dict] = [first_system_msg]
+
+    def _init_api_config(self):
+        self.config = get_config()
+        api_type = self.config['API_TYPE']
+        api_base = self.config['API_base']
+        api_version = self.config['API_VERSION']
+        api_key = config['API_KEY']
+        config_openai_api(api_type, api_base, api_version, api_key)
+
+    def _init_kwargs_for_chat_completion(self):
+        self.kwargs_for_chat_completion = {
+            'stream': True,
+            'messages': self.conversation,
+            'functions': functions,
+            'function_call': 'auto'
+        }
+
+        model_name = self.config['model'][self.gpt_model_choice]['model_name']
+
+        if self.config['API_TYPE'] == 'azure':
+            self.kwargs_for_chat_completion['engine'] = model_name
+        else:
+            self.kwargs_for_chat_completion['model'] = model_name
+
+    def _clear_all_files_in_work_dir(self):
+        for filename in os.listdir(self.jupyter_work_dir):
+            os.remove(
+                os.path.join(self.jupyter_work_dir, filename)
+            )
+
+    def add_gpt_response_content_message(self):
+        self.conversation.append(
+            {'role': self.assistant_role_name, 'content': self.content}
+        )
+
+    def add_text_message(self, user_text):
+        self.conversation.append(
+            {'role': 'user', 'content': user_text}
+        )
+        self.revocable_files.clear()
+        self.update_finish_reason(finish_reason='new_input')
+
+    def add_file_message(self, path, bot_msg):
+        filename = os.path.basename(path)
+        work_dir = self.jupyter_work_dir
+
+        shutil.copy(path, work_dir)
+
+        gpt_msg = {'role': 'system', 'content': f'User uploaded a file: {filename}'}
+        self.conversation.append(gpt_msg)
+        self.revocable_files.append(
+            {
+                'bot_msg': bot_msg,
+                'gpt_msg': gpt_msg,
+                'path': os.path.join(work_dir, filename)
+            }
+        )
+
+    def add_function_call_response_message(self, function_response: str, save_tokens=True):
+        self.conversation.append(
+            {
+                "role": self.assistant_role_name,
+                "name": self.function_name,
+                "content": self.function_args_str
+            }
+        )
+
+        if save_tokens and len(function_response) > 500:
+            function_response = f'{function_response[:200]}\n[Output too much, the middle part output is omitted]\n ' \
+                                f'End part of output:\n{function_response[-200:]}'
+        self.conversation.append(
+            {
+                "role": "function",
+                "name": self.function_name,
+                "content": function_response,
+            }
+        )
+
+    def revoke_file(self):
+        if self.revocable_files:
+            file = self.revocable_files[-1]
+            bot_msg = file['bot_msg']
+            gpt_msg = file['gpt_msg']
+            path = file['path']
+
+            assert self.conversation[-1] is gpt_msg
+            del self.conversation[-1]
+
+            os.remove(path)
+
+            del self.revocable_files[-1]
+
+            return bot_msg
+        else:
+            return None
+
+    def update_gpt_model_choice(self, model_choice):
+        self.gpt_model_choice = model_choice
+        self._init_kwargs_for_chat_completion()
+
+    def restart(self):
+        self._clear_all_files_in_work_dir()
+        self.revocable_files.clear()
+        self._init_conversation()
+        self.reset_gpt_response_log_values()
+        self.jupyter_kernel.restart_jupyter_kernel()

config.json

@@ -0,0 +1,16 @@
+{
+  "API_TYPE": "open_ai",
+  "API_base": "https://api.nova-oss.com/v1",
+  "API_VERSION": null,
+  "API_KEY": "",
+  "model": {
+    "GPT-3.5": {
+      "model_name": "gpt-3.5-turbo",
+      "available": true
+    },
+    "GPT-4": {
+      "model_name": "gpt-4-0613",
+      "available": true
+    }
+  }
+}

config.json.example

@@ -0,0 +1,16 @@
+{
+  "API_TYPE": "open_ai",
+  "API_base": "https://api.chatanywhere.cn/v1",
+  "API_VERSION": null,
+  "API_KEY": "",
+  "model": {
+    "GPT-3.5": {
+      "model_name": "gpt-3.5-turbo",
+      "available": true
+    },
+    "GPT-4": {
+      "model_name": "",
+      "available": false
+    }
+  }
+}

functional.py

@@ -0,0 +1,116 @@
+from bot_backend import *
+import base64
+import time
+
+
+def chat_completion(bot_backend: BotBackend):
+    model_choice = bot_backend.gpt_model_choice
+    config = bot_backend.config
+    kwargs_for_chat_completion = bot_backend.kwargs_for_chat_completion
+
+    assert config['model'][model_choice]['available'], f"{model_choice} is not available for you API key"
+
+    response = openai.ChatCompletion.create(**kwargs_for_chat_completion)
+    return response
+
+
+def add_function_response_to_bot_history(content_to_display, history, unique_id):
+    images, text = [], []
+
+    # terminal output
+    error_occurred = False
+    for mark, out_str in content_to_display:
+        if mark in ('stdout', 'execute_result_text', 'display_text'):
+            text.append(out_str)
+        elif mark in ('execute_result_png', 'execute_result_jpeg', 'display_png', 'display_jpeg'):
+            if 'png' in mark:
+                images.append(('png', out_str))
+            else:
+                images.append(('jpg', out_str))
+        elif mark == 'error':
+            text.append(delete_color_control_char(out_str))
+            error_occurred = True
+    text = '\n'.join(text).strip('\n')
+    if error_occurred:
+        history.append([None, f'❌Terminal output:\n```shell\n\n{text}\n```'])
+    else:
+        history.append([None, f'✔️Terminal output:\n```shell\n{text}\n```'])
+
+    # image output
+    for filetype, img in images:
+        image_bytes = base64.b64decode(img)
+        temp_path = f'cache/temp_{unique_id}'
+        if not os.path.exists(temp_path):
+            os.mkdir(temp_path)
+        path = f'{temp_path}/{hash(time.time())}.{filetype}'
+        with open(path, 'wb') as f:
+            f.write(image_bytes)
+        history.append(
+            [
+                None,
+                f'<img src=\"file={path}\" style=\'width: 600px; max-width:none; max-height:none\'>'
+            ]
+        )
+
+
+def parse_json(function_args: str, finished: bool):
+    """
+    GPT may generate non-standard JSON format string, which contains '\n' in string value, leading to error when using
+    `json.loads()`.
+    Here we implement a parser to extract code directly from non-standard JSON string.
+    :return: code string if successfully parsed otherwise None
+    """
+    parser_log = {
+        'met_begin_{': False,
+        'begin_"code"': False,
+        'end_"code"': False,
+        'met_:': False,
+        'met_end_}': False,
+        'met_end_code_"': False,
+        "code_begin_index": 0,
+        "code_end_index": 0
+    }
+    try:
+        for index, char in enumerate(function_args):
+            if char == '{':
+                parser_log['met_begin_{'] = True
+            elif parser_log['met_begin_{'] and char == '"':
+                if parser_log['met_:']:
+                    if finished:
+                        parser_log['code_begin_index'] = index + 1
+                        break
+                    else:
+                        if index + 1 == len(function_args):
+                            return ''
+                        else:
+                            temp_code_str = function_args[index + 1:]
+                            if '\n' in temp_code_str:
+                                return temp_code_str.strip('\n')
+                            else:
+                                return json.loads(function_args + '"}')['code']
+                elif parser_log['begin_"code"']:
+                    parser_log['end_"code"'] = True
+                else:
+                    parser_log['begin_"code"'] = True
+            elif parser_log['end_"code"'] and char == ':':
+                parser_log['met_:'] = True
+            else:
+                continue
+        if finished:
+            for index, char in enumerate(function_args[::-1]):
+                back_index = -1 - index
+                if char == '}':
+                    parser_log['met_end_}'] = True
+                elif parser_log['met_end_}'] and char == '"':
+                    parser_log['code_end_index'] = back_index - 1
+                    break
+                else:
+                    continue
+            code_str = function_args[parser_log['code_begin_index']: parser_log['code_end_index'] + 1]
+            if '\n' in code_str:
+                return code_str.strip('\n')
+            else:
+                return json.loads(function_args)['code']
+
+    except Exception as e:
+        return None

gitattributes.txt

@@ -0,0 +1,35 @@
+*.7z filter=lfs diff=lfs merge=lfs -text
+*.arrow filter=lfs diff=lfs merge=lfs -text
+*.bin filter=lfs diff=lfs merge=lfs -text
+*.bz2 filter=lfs diff=lfs merge=lfs -text
+*.ckpt filter=lfs diff=lfs merge=lfs -text
+*.ftz filter=lfs diff=lfs merge=lfs -text
+*.gz filter=lfs diff=lfs merge=lfs -text
+*.h5 filter=lfs diff=lfs merge=lfs -text
+*.joblib filter=lfs diff=lfs merge=lfs -text
+*.lfs.* filter=lfs diff=lfs merge=lfs -text
+*.mlmodel filter=lfs diff=lfs merge=lfs -text
+*.model filter=lfs diff=lfs merge=lfs -text
+*.msgpack filter=lfs diff=lfs merge=lfs -text
+*.npy filter=lfs diff=lfs merge=lfs -text
+*.npz filter=lfs diff=lfs merge=lfs -text
+*.onnx filter=lfs diff=lfs merge=lfs -text
+*.ot filter=lfs diff=lfs merge=lfs -text
+*.parquet filter=lfs diff=lfs merge=lfs -text
+*.pb filter=lfs diff=lfs merge=lfs -text
+*.pickle filter=lfs diff=lfs merge=lfs -text
+*.pkl filter=lfs diff=lfs merge=lfs -text
+*.pt filter=lfs diff=lfs merge=lfs -text
+*.pth filter=lfs diff=lfs merge=lfs -text
+*.rar filter=lfs diff=lfs merge=lfs -text
+*.safetensors filter=lfs diff=lfs merge=lfs -text
+saved_model/**/* filter=lfs diff=lfs merge=lfs -text
+*.tar.* filter=lfs diff=lfs merge=lfs -text
+*.tar filter=lfs diff=lfs merge=lfs -text
+*.tflite filter=lfs diff=lfs merge=lfs -text
+*.tgz filter=lfs diff=lfs merge=lfs -text
+*.wasm filter=lfs diff=lfs merge=lfs -text
+*.xz filter=lfs diff=lfs merge=lfs -text
+*.zip filter=lfs diff=lfs merge=lfs -text
+*.zst filter=lfs diff=lfs merge=lfs -text
+*tfevents* filter=lfs diff=lfs merge=lfs -text

jupyter_backend.py

@@ -0,0 +1,100 @@
+import jupyter_client
+import re
+
+
+def delete_color_control_char(string):
+    ansi_escape = re.compile(r'(\x9B|\x1B\[)[0-?]*[ -\/]*[@-~]')
+    return ansi_escape.sub('', string)
+
+
+class JupyterKernel:
+    def __init__(self, work_dir):
+        self.kernel_manager, self.kernel_client = jupyter_client.manager.start_new_kernel(kernel_name='python3')
+        self.work_dir = work_dir
+        self._create_work_dir()
+        self.available_functions = {
+            'execute_code': self.execute_code,
+            'python': self.execute_code
+        }
+
+    def execute_code_(self, code):
+        msg_id = self.kernel_client.execute(code)
+
+        # Get the output of the code
+        iopub_msg = self.kernel_client.get_iopub_msg()
+
+        all_output = []
+        while True:
+            if iopub_msg['msg_type'] == 'stream':
+                if iopub_msg['content'].get('name') == 'stdout':
+                    output = iopub_msg['content']['text']
+                    all_output.append(('stdout', output))
+                iopub_msg = self.kernel_client.get_iopub_msg()
+            elif iopub_msg['msg_type'] == 'execute_result':
+                if 'data' in iopub_msg['content']:
+                    if 'text/plain' in iopub_msg['content']['data']:
+                        output = iopub_msg['content']['data']['text/plain']
+                        all_output.append(('execute_result_text', output))
+                    if 'text/html' in iopub_msg['content']['data']:
+                        output = iopub_msg['content']['data']['text/html']
+                        all_output.append(('execute_result_html', output))
+                    if 'image/png' in iopub_msg['content']['data']:
+                        output = iopub_msg['content']['data']['image/png']
+                        all_output.append(('execute_result_png', output))
+                    if 'image/jpeg' in iopub_msg['content']['data']:
+                        output = iopub_msg['content']['data']['image/jpeg']
+                        all_output.append(('execute_result_jpeg', output))
+                iopub_msg = self.kernel_client.get_iopub_msg()
+            elif iopub_msg['msg_type'] == 'display_data':
+                if 'data' in iopub_msg['content']:
+                    if 'text/plain' in iopub_msg['content']['data']:
+                        output = iopub_msg['content']['data']['text/plain']
+                        all_output.append(('display_text', output))
+                    if 'text/html' in iopub_msg['content']['data']:
+                        output = iopub_msg['content']['data']['text/html']
+                        all_output.append(('display_html', output))
+                    if 'image/png' in iopub_msg['content']['data']:
+                        output = iopub_msg['content']['data']['image/png']
+                        all_output.append(('display_png', output))
+                    if 'image/jpeg' in iopub_msg['content']['data']:
+                        output = iopub_msg['content']['data']['image/jpeg']
+                        all_output.append(('display_jpeg', output))
+                iopub_msg = self.kernel_client.get_iopub_msg()
+            elif iopub_msg['msg_type'] == 'error':
+                if 'traceback' in iopub_msg['content']:
+                    output = '\n'.join(iopub_msg['content']['traceback'])
+                    all_output.append(('error', output))
+                iopub_msg = self.kernel_client.get_iopub_msg()
+            elif iopub_msg['msg_type'] == 'status' and iopub_msg['content'].get('execution_state') == 'idle':
+                break
+            else:
+                iopub_msg = self.kernel_client.get_iopub_msg()
+
+        return all_output
+
+    def execute_code(self, code):
+        text_to_gpt = []
+        content_to_display = self.execute_code_(code)
+        for mark, out_str in content_to_display:
+            if mark in ('stdout', 'execute_result_text', 'display_text'):
+                text_to_gpt.append(out_str)
+            elif mark in ('execute_result_png', 'execute_result_jpeg', 'display_png', 'display_jpeg'):
+                text_to_gpt.append('[image]')
+            elif mark == 'error':
+                text_to_gpt.append(delete_color_control_char(out_str))
+
+        return '\n'.join(text_to_gpt), content_to_display
+
+    def _create_work_dir(self):
+        # set work dir in jupyter environment
+        init_code = f"import os\n" \
+                    f"if not os.path.exists('{self.work_dir}'):\n" \
+                    f"    os.mkdir('{self.work_dir}')\n" \
+                    f"os.chdir('{self.work_dir}')\n" \
+                    f"del os"
+        self.execute_code_(init_code)
+
+    def restart_jupyter_kernel(self):
+        self.kernel_client.shutdown()
+        self.kernel_manager, self.kernel_client = jupyter_client.manager.start_new_kernel(kernel_name='python3')
+        self._create_work_dir()

requirements.txt

@@ -0,0 +1,12 @@
+notebook==6.5.4
+gradio==3.39.0
+openai==0.27.8
+pandas
+numpy
+openpyxl
+Pillow
+opencv-python
+PyPDF2
+pdfminer.six
+sympy
+PyMuPDF

response_parser.py

@@ -0,0 +1,200 @@
+from abc import ABCMeta, abstractmethod
+from functional import *
+
+
+class ChoiceStrategy(metaclass=ABCMeta):
+    def __init__(self, choice):
+        self.choice = choice
+        self.delta = choice['delta']
+
+    @abstractmethod
+    def support(self):
+        pass
+
+    @abstractmethod
+    def execute(self, bot_backend: BotBackend, history: List, whether_exit: bool):
+        pass
+
+
+class RoleChoiceStrategy(ChoiceStrategy):
+
+    def support(self):
+        return 'role' in self.delta
+
+    def execute(self, bot_backend: BotBackend, history: List, whether_exit: bool):
+        bot_backend.set_assistant_role_name(assistant_role_name=self.delta['role'])
+        return history, whether_exit
+
+
+class ContentChoiceStrategy(ChoiceStrategy):
+    def support(self):
+        return 'content' in self.delta and self.delta['content'] is not None
+
+    def execute(self, bot_backend: BotBackend, history: List, whether_exit: bool):
+        # null value of content often occur in function call:
+        #     {
+        #       "role": "assistant",
+        #       "content": null,
+        #       "function_call": {
+        #         "name": "python",
+        #         "arguments": ""
+        #       }
+        #     }
+        bot_backend.add_content(content=self.delta.get('content', ''))
+        history[-1][1] = bot_backend.content
+        return history, whether_exit
+
+
+class NameFunctionCallChoiceStrategy(ChoiceStrategy):
+    def support(self):
+        return 'function_call' in self.delta and 'name' in self.delta['function_call']
+
+    def execute(self, bot_backend: BotBackend, history: List, whether_exit: bool):
+        function_dict = bot_backend.jupyter_kernel.available_functions
+        bot_backend.set_function_name(function_name=self.delta['function_call']['name'])
+        bot_backend.copy_current_bot_history(bot_history=history)
+        if bot_backend.function_name not in function_dict:
+            history.append(
+                [
+                    None,
+                    f'GPT attempted to call a function that does '
+                    f'not exist: {bot_backend.function_name}\n '
+                ]
+            )
+            whether_exit = True
+
+        return history, whether_exit
+
+
+class ArgumentsFunctionCallChoiceStrategy(ChoiceStrategy):
+
+    def support(self):
+        return 'function_call' in self.delta and 'arguments' in self.delta['function_call']
+
+    def execute(self, bot_backend: BotBackend, history: List, whether_exit: bool):
+        bot_backend.add_function_args_str(function_args_str=self.delta['function_call']['arguments'])
+
+        if bot_backend.function_name == 'python':  # handle hallucinatory function calls
+            '''
+            In practice, we have noticed that GPT, especially GPT-3.5, may occasionally produce hallucinatory
+            function calls. These calls involve a non-existent function named `python` with arguments consisting 
+            solely of raw code text (not a JSON format).
+            '''
+            temp_code_str = bot_backend.function_args_str
+            bot_backend.update_display_code_block(
+                display_code_block="\n🔴Working:\n```python\n{}\n```".format(temp_code_str)
+            )
+            history = copy.deepcopy(bot_backend.bot_history)
+            history[-1][1] += bot_backend.display_code_block
+        else:
+            temp_code_str = parse_json(function_args=bot_backend.function_args_str, finished=False)
+            if temp_code_str is not None:
+                bot_backend.update_display_code_block(
+                    display_code_block="\n🔴Working:\n```python\n{}\n```".format(
+                        temp_code_str
+                    )
+                )
+                history = copy.deepcopy(bot_backend.bot_history)
+                history[-1][1] += bot_backend.display_code_block
+
+        return history, whether_exit
+
+
+class FinishReasonChoiceStrategy(ChoiceStrategy):
+    def support(self):
+        return self.choice['finish_reason'] is not None
+
+    def execute(self, bot_backend: BotBackend, history: List, whether_exit: bool):
+        function_dict = bot_backend.jupyter_kernel.available_functions
+
+        if bot_backend.content:
+            bot_backend.add_gpt_response_content_message()
+
+        bot_backend.update_finish_reason(finish_reason=self.choice['finish_reason'])
+        if bot_backend.finish_reason == 'function_call':
+            try:
+
+                code_str = self.get_code_str(bot_backend)
+
+                bot_backend.update_display_code_block(
+                    display_code_block="\n🟢Working:\n```python\n{}\n```".format(code_str)
+                )
+                history = copy.deepcopy(bot_backend.bot_history)
+                history[-1][1] += bot_backend.display_code_block
+
+                # function response
+                text_to_gpt, content_to_display = function_dict[
+                    bot_backend.function_name
+                ](code_str)
+
+                # add function call to conversion
+                bot_backend.add_function_call_response_message(function_response=text_to_gpt, save_tokens=True)
+
+                add_function_response_to_bot_history(
+                    content_to_display=content_to_display, history=history, unique_id=bot_backend.unique_id
+                )
+
+            except json.JSONDecodeError:
+                history.append(
+                    [None, f"GPT generate wrong function args: {bot_backend.function_args_str}"]
+                )
+                whether_exit = True
+                return history, whether_exit
+
+            except Exception as e:
+                history.append([None, f'Backend error: {e}'])
+                whether_exit = True
+                return history, whether_exit
+
+        bot_backend.reset_gpt_response_log_values(exclude=['finish_reason'])
+
+        return history, whether_exit
+
+    @staticmethod
+    def get_code_str(bot_backend):
+        if bot_backend.function_name == 'python':
+            code_str = bot_backend.function_args_str
+        else:
+            code_str = parse_json(function_args=bot_backend.function_args_str, finished=True)
+            if code_str is None:
+                raise json.JSONDecodeError
+        return code_str
+
+
+class ChoiceHandler:
+    strategies = [
+        RoleChoiceStrategy, ContentChoiceStrategy, NameFunctionCallChoiceStrategy,
+        ArgumentsFunctionCallChoiceStrategy, FinishReasonChoiceStrategy
+    ]
+
+    def __init__(self, choice):
+        self.choice = choice
+
+    def handle(self, bot_backend: BotBackend, history: List, whether_exit: bool):
+        for Strategy in self.strategies:
+            strategy_instance = Strategy(choice=self.choice)
+            if not strategy_instance.support():
+                continue
+            history, whether_exit = strategy_instance.execute(
+                bot_backend=bot_backend,
+                history=history,
+                whether_exit=whether_exit
+            )
+        return history, whether_exit
+
+
+def parse_response(chunk, history, bot_backend: BotBackend):
+    """
+    :return: history, whether_exit
+    """
+    whether_exit = False
+    if chunk['choices']:
+        choice = chunk['choices'][0]
+        choice_handler = ChoiceHandler(choice=choice)
+        history, whether_exit = choice_handler.handle(
+            history=history,
+            bot_backend=bot_backend,
+            whether_exit=whether_exit
+        )
+
+    return history, whether_exit