兼容autogpt plugin load

pilot/agent/base_open_ai_plugin.py

@@ -0,0 +1,199 @@
+"""Handles loading of plugins."""
+from typing import Any, Dict, List, Optional, Tuple, TypedDict, TypeVar
+
+from auto_gpt_plugin_template import AutoGPTPluginTemplate
+
+PromptGenerator = TypeVar("PromptGenerator")
+
+
+class Message(TypedDict):
+    role: str
+    content: str
+
+
+class BaseOpenAIPlugin(AutoGPTPluginTemplate):
+    """
+    This is a BaseOpenAIPlugin class for generating Auto-GPT plugins.
+    """
+
+    def __init__(self, manifests_specs_clients: dict):
+        # super().__init__()
+        self._name = manifests_specs_clients["manifest"]["name_for_model"]
+        self._version = manifests_specs_clients["manifest"]["schema_version"]
+        self._description = manifests_specs_clients["manifest"]["description_for_model"]
+        self._client = manifests_specs_clients["client"]
+        self._manifest = manifests_specs_clients["manifest"]
+        self._openapi_spec = manifests_specs_clients["openapi_spec"]
+
+    def can_handle_on_response(self) -> bool:
+        """This method is called to check that the plugin can
+        handle the on_response method.
+        Returns:
+            bool: True if the plugin can handle the on_response method."""
+        return False
+
+    def on_response(self, response: str, *args, **kwargs) -> str:
+        """This method is called when a response is received from the model."""
+        return response
+
+    def can_handle_post_prompt(self) -> bool:
+        """This method is called to check that the plugin can
+        handle the post_prompt method.
+        Returns:
+            bool: True if the plugin can handle the post_prompt method."""
+        return False
+
+    def post_prompt(self, prompt: PromptGenerator) -> PromptGenerator:
+        """This method is called just after the generate_prompt is called,
+            but actually before the prompt is generated.
+        Args:
+            prompt (PromptGenerator): The prompt generator.
+        Returns:
+            PromptGenerator: The prompt generator.
+        """
+        return prompt
+
+    def can_handle_on_planning(self) -> bool:
+        """This method is called to check that the plugin can
+        handle the on_planning method.
+        Returns:
+            bool: True if the plugin can handle the on_planning method."""
+        return False
+
+    def on_planning(
+        self, prompt: PromptGenerator, messages: List[Message]
+    ) -> Optional[str]:
+        """This method is called before the planning chat completion is done.
+        Args:
+            prompt (PromptGenerator): The prompt generator.
+            messages (List[str]): The list of messages.
+        """
+        pass
+
+    def can_handle_post_planning(self) -> bool:
+        """This method is called to check that the plugin can
+        handle the post_planning method.
+        Returns:
+            bool: True if the plugin can handle the post_planning method."""
+        return False
+
+    def post_planning(self, response: str) -> str:
+        """This method is called after the planning chat completion is done.
+        Args:
+            response (str): The response.
+        Returns:
+            str: The resulting response.
+        """
+        return response
+
+    def can_handle_pre_instruction(self) -> bool:
+        """This method is called to check that the plugin can
+        handle the pre_instruction method.
+        Returns:
+            bool: True if the plugin can handle the pre_instruction method."""
+        return False
+
+    def pre_instruction(self, messages: List[Message]) -> List[Message]:
+        """This method is called before the instruction chat is done.
+        Args:
+            messages (List[Message]): The list of context messages.
+        Returns:
+            List[Message]: The resulting list of messages.
+        """
+        return messages
+
+    def can_handle_on_instruction(self) -> bool:
+        """This method is called to check that the plugin can
+        handle the on_instruction method.
+        Returns:
+            bool: True if the plugin can handle the on_instruction method."""
+        return False
+
+    def on_instruction(self, messages: List[Message]) -> Optional[str]:
+        """This method is called when the instruction chat is done.
+        Args:
+            messages (List[Message]): The list of context messages.
+        Returns:
+            Optional[str]: The resulting message.
+        """
+        pass
+
+    def can_handle_post_instruction(self) -> bool:
+        """This method is called to check that the plugin can
+        handle the post_instruction method.
+        Returns:
+            bool: True if the plugin can handle the post_instruction method."""
+        return False
+
+    def post_instruction(self, response: str) -> str:
+        """This method is called after the instruction chat is done.
+        Args:
+            response (str): The response.
+        Returns:
+            str: The resulting response.
+        """
+        return response
+
+    def can_handle_pre_command(self) -> bool:
+        """This method is called to check that the plugin can
+        handle the pre_command method.
+        Returns:
+            bool: True if the plugin can handle the pre_command method."""
+        return False
+
+    def pre_command(
+        self, command_name: str, arguments: Dict[str, Any]
+    ) -> Tuple[str, Dict[str, Any]]:
+        """This method is called before the command is executed.
+        Args:
+            command_name (str): The command name.
+            arguments (Dict[str, Any]): The arguments.
+        Returns:
+            Tuple[str, Dict[str, Any]]: The command name and the arguments.
+        """
+        return command_name, arguments
+
+    def can_handle_post_command(self) -> bool:
+        """This method is called to check that the plugin can
+        handle the post_command method.
+        Returns:
+            bool: True if the plugin can handle the post_command method."""
+        return False
+
+    def post_command(self, command_name: str, response: str) -> str:
+        """This method is called after the command is executed.
+        Args:
+            command_name (str): The command name.
+            response (str): The response.
+        Returns:
+            str: The resulting response.
+        """
+        return response
+
+    def can_handle_chat_completion(
+        self, messages: Dict[Any, Any], model: str, temperature: float, max_tokens: int
+    ) -> bool:
+        """This method is called to check that the plugin can
+          handle the chat_completion method.
+        Args:
+            messages (List[Message]): The messages.
+            model (str): The model name.
+            temperature (float): The temperature.
+            max_tokens (int): The max tokens.
+          Returns:
+              bool: True if the plugin can handle the chat_completion method."""
+        return False
+
+    def handle_chat_completion(
+        self, messages: List[Message], model: str, temperature: float, max_tokens: int
+    ) -> str:
+        """This method is called when the chat completion is done.
+        Args:
+            messages (List[Message]): The messages.
+            model (str): The model name.
+            temperature (float): The temperature.
+            max_tokens (int): The max tokens.
+        Returns:
+            str: The resulting response.
+        """
+        pass

pilot/configs/config.py

@@ -29,22 +29,35 @@ def __init__(self) -> None:
             "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_4) AppleWebKit/537.36"
             " (KHTML, like Gecko) Chrome/83.0.4103.97 Safari/537.36",
         )
-
+
+        self.elevenlabs_api_key = os.getenv("ELEVENLABS_API_KEY")
+        self.elevenlabs_voice_1_id = os.getenv("ELEVENLABS_VOICE_1_ID")
+        self.elevenlabs_voice_2_id = os.getenv("ELEVENLABS_VOICE_2_ID")
+
+        self.use_mac_os_tts = False
+        self.use_mac_os_tts = os.getenv("USE_MAC_OS_TTS")
+
         # milvus or zilliz cloud configuration
         self.milvus_addr = os.getenv("MILVUS_ADDR", "localhost:19530")
         self.milvus_username = os.getenv("MILVUS_USERNAME")
         self.milvus_password = os.getenv("MILVUS_PASSWORD")
         self.milvus_collection = os.getenv("MILVUS_COLLECTION", "dbgpt")
         self.milvus_secure = os.getenv("MILVUS_SECURE") == "True"
 
+        self.plugins_dir = os.getenv("PLUGINS_DIR", "plugins")
+        self.plugins: List[AutoGPTPluginTemplate] = []
+        self.plugins_openai = []
+
         plugins_allowlist = os.getenv("ALLOWLISTED_PLUGINS")
         if plugins_allowlist:
             self.plugins_allowlist = plugins_allowlist.split(",")
         else:
-            self.plugins_allowlist = [] 
+            self.plugins_allowlist = []
 
-        plugins_denylist = os.getenv("DENYLISTED_PLUGINS") 
+        plugins_denylist = os.getenv("DENYLISTED_PLUGINS")
         if plugins_denylist:
+            self.plugins_denylist = plugins_denylist.split(",")
+        else:
             self.plugins_denylist = []
 
     def set_debug_mode(self, value: bool) -> None:

pilot/log/json_handler.py

@@ -0,0 +1,20 @@
+import json
+import logging
+
+
+class JsonFileHandler(logging.FileHandler):
+    def __init__(self, filename, mode="a", encoding=None, delay=False):
+        super().__init__(filename, mode, encoding, delay)
+
+    def emit(self, record):
+        json_data = json.loads(self.format(record))
+        with open(self.baseFilename, "w", encoding="utf-8") as f:
+            json.dump(json_data, f, ensure_ascii=False, indent=4)
+
+
+import logging
+
+
+class JsonFormatter(logging.Formatter):
+    def format(self, record):
+        return record.msg