Add code interpreter module (#400)

Co-authored-by: Zecheng Zhang <[email protected]>
Co-authored-by: lig <[email protected]>

Author: 3 people

camel/agents/embodied_agent.py

@@ -11,15 +11,20 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 # =========== Copyright 2023 @ CAMEL-AI.org. All Rights Reserved. ===========
-from typing import Any, Dict, List, Optional
+from typing import Any, List, Optional
 
 from colorama import Fore
 
-from camel.agents import BaseToolAgent, ChatAgent, HuggingFaceToolAgent
+from camel.agents import BaseToolAgent, ChatAgent
+from camel.interpreters import (
+    BaseInterpreter,
+    InternalPythonInterpreter,
+    SubprocessInterpreter,
+)
 from camel.messages import BaseMessage
 from camel.responses import ChatAgentResponse
 from camel.types import ModelType
-from camel.utils import PythonInterpreter, print_text_animated
+from camel.utils import print_text_animated
 
 
 class EmbodiedAgent(ChatAgent):
@@ -34,8 +39,13 @@ class EmbodiedAgent(ChatAgent):
         message_window_size (int, optional): The maximum number of previous
             messages to include in the context window. If `None`, no windowing
             is performed. (default: :obj:`None`)
-        action_space (List[Any], optional): The action space for the embodied
-            agent. (default: :obj:`None`)
+        tool_agents (List[BaseToolAgent], optional): The tools agents to use in
+            the embodied agent. (default: :obj:`None`)
+        code_interpreter (BaseInterpreter, optional): The code interpreter to
+            execute codes. If `code_interpreter` and `tool_agent` are both
+            `None`, default to `SubProcessInterpreter`. If `code_interpreter`
+            is `None` and `tool_agents` is not `None`, default to
+            `InternalPythonInterpreter`.  (default: :obj:`None`)
         verbose (bool, optional): Whether to print the critic's messages.
         logger_color (Any): The color of the logger displayed to the user.
             (default: :obj:`Fore.MAGENTA`)
@@ -47,18 +57,22 @@ def __init__(
         model_type: ModelType = ModelType.GPT_4,
         model_config: Optional[Any] = None,
         message_window_size: Optional[int] = None,
-        action_space: Optional[List[BaseToolAgent]] = None,
+        tool_agents: Optional[List[BaseToolAgent]] = None,
+        code_interpreter: Optional[BaseInterpreter] = None,
         verbose: bool = False,
         logger_color: Any = Fore.MAGENTA,
     ) -> None:
-        default_action_space = [
-            HuggingFaceToolAgent('hugging_face_tool_agent',
-                                 model_type=model_type.value),
-        ]
-        self.action_space = action_space or default_action_space
-        action_space_prompt = self.get_action_space_prompt()
-        system_message.content = system_message.content.format(
-            action_space=action_space_prompt)
+        self.tool_agents = tool_agents
+        self.code_interpreter: BaseInterpreter
+        if code_interpreter is not None:
+            self.code_interpreter = code_interpreter
+        elif self.tool_agents:
+            self.code_interpreter = InternalPythonInterpreter()
+        else:
+            self.code_interpreter = SubprocessInterpreter()
+
+        if self.tool_agents:
+            system_message = self._set_tool_agents(system_message)
         self.verbose = verbose
         self.logger_color = logger_color
         super().__init__(
@@ -68,16 +82,41 @@ def __init__(
             message_window_size=message_window_size,
         )
 
-    def get_action_space_prompt(self) -> str:
+    def _set_tool_agents(self, system_message: BaseMessage) -> BaseMessage:
+        action_space_prompt = self._get_tool_agents_prompt()
+        result_message = system_message.create_new_instance(
+            content=system_message.content.format(
+                action_space=action_space_prompt))
+        if self.tool_agents is not None:
+            self.code_interpreter.update_action_space(
+                {tool.name: tool
+                 for tool in self.tool_agents})
+        return result_message
+
+    def _get_tool_agents_prompt(self) -> str:
         r"""Returns the action space prompt.
 
         Returns:
             str: The action space prompt.
         """
-        return "\n".join([
-            f"*** {action.name} ***:\n {action.description}"
-            for action in self.action_space
-        ])
+        if self.tool_agents is not None:
+            return "\n".join([
+                f"*** {tool.name} ***:\n {tool.description}"
+                for tool in self.tool_agents
+            ])
+        else:
+            return ""
+
+    def get_tool_agent_names(self) -> List[str]:
+        r"""Returns the names of tool agents.
+
+        Returns:
+            List[str]: The names of tool agents.
+        """
+        if self.tool_agents is not None:
+            return [tool.name for tool in self.tool_agents]
+        else:
+            return []
 
     def step(
         self,
@@ -111,28 +150,24 @@ def step(
 
             if len(explanations) > len(codes):
                 print_text_animated(self.logger_color +
-                                    f"> Explanation:\n{explanations}")
+                                    f"> Explanation:\n{explanations[-1]}")
 
         content = response.msg.content
 
         if codes is not None:
-            content = "\n> Executed Results:"
-            action_space: Dict[str, Any] = {
-                action.name: action
-                for action in self.action_space
-            }
-            action_space.update({"print": print, "enumerate": enumerate})
-            interpreter = PythonInterpreter(action_space=action_space)
-            for block_idx, code in enumerate(codes):
-                executed_outputs, _ = code.execute(interpreter)
-                content += (f"Executing code block {block_idx}:\n"
-                            f"  - execution output:\n{executed_outputs}\n"
-                            f"  - Local variables:\n{interpreter.state}\n")
-                content += "*" * 50 + "\n"
+            try:
+                content = "\n> Executed Results:\n"
+                for block_idx, code in enumerate(codes):
+                    executed_output = self.code_interpreter.run(
+                        code, code.code_type)
+                    content += (f"Executing code block {block_idx}: {{\n" +
+                                executed_output + "}\n")
+            except InterruptedError as e:
+                content = (f"\n> Running code fail: {e}\n"
+                           "Please regenerate the code.")
 
         # TODO: Handle errors
-        content = input_message.content + (Fore.RESET +
-                                           f"\n> Embodied Actions:\n{content}")
+        content = input_message.content + f"\n> Embodied Actions:\n{content}"
         message = BaseMessage(input_message.role_name, input_message.role_type,
                               input_message.meta_dict, content)
         return ChatAgentResponse([message], response.terminated, response.info)

camel/interpreters/__init__.py

@@ -0,0 +1,25 @@
+# =========== Copyright 2023 @ CAMEL-AI.org. All Rights Reserved. ===========
+# Licensed under the Apache License, Version 2.0 (the “License”);
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an “AS IS” BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# =========== Copyright 2023 @ CAMEL-AI.org. All Rights Reserved. ===========
+
+from .base import BaseInterpreter
+from .interpreter_error import InterpreterError
+from .internal_python_interpreter import InternalPythonInterpreter
+from .subprocess_interpreter import SubprocessInterpreter
+
+__all__ = [
+    'BaseInterpreter',
+    'InterpreterError',
+    'InternalPythonInterpreter',
+    'SubprocessInterpreter',
+]

camel/interpreters/base.py

@@ -0,0 +1,49 @@
+# =========== Copyright 2023 @ CAMEL-AI.org. All Rights Reserved. ===========
+# Licensed under the Apache License, Version 2.0 (the “License”);
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an “AS IS” BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# =========== Copyright 2023 @ CAMEL-AI.org. All Rights Reserved. ===========
+from abc import ABC, abstractmethod
+from typing import Any, Dict, List
+
+
+class BaseInterpreter(ABC):
+    r"""An abstract base class for code interpreters."""
+
+    @abstractmethod
+    def run(self, code: str, code_type: str) -> str:
+        r"""Executes the given code based on its type.
+
+        Args:
+            code (str): The code to be executed.
+            code_type (str): The type of the code, which must be one of the
+                types returned by `supported_code_types()`.
+
+        Returns:
+            str: The result of the code execution. If the execution fails, this
+                should include sufficient information to diagnose and correct
+                the issue.
+
+        Raises:
+            InterpreterError: If the code execution encounters errors that
+                could be resolved by modifying or regenerating the code.
+        """
+        pass
+
+    @abstractmethod
+    def supported_code_types(self) -> List[str]:
+        r"""Provides supported code types by the interpreter."""
+        pass
+
+    @abstractmethod
+    def update_action_space(self, action_space: Dict[str, Any]) -> None:
+        r"""Updates action space for *python* interpreter"""
+        pass

camel/utils/python_interpreter.py renamed to camel/interpreters/internal_python_interpreter.py

@@ -17,20 +17,14 @@
 import typing
 from typing import Any, Dict, List, Optional
 
-
-class InterpreterError(ValueError):
-    r"""An error raised when the interpreter cannot evaluate a Python
-    expression, due to syntax error or unsupported operations.
-    """
-
-    pass
+from camel.interpreters import BaseInterpreter, InterpreterError
 
 
-class PythonInterpreter():
+class InternalPythonInterpreter(BaseInterpreter):
     r"""A customized python interpreter to control the execution of
     LLM-generated codes. The interpreter makes sure the code can only execute
     functions given in action space and import white list. It also supports
-    fuzzy variable matching to reveive uncertain input variable name.
+    fuzzy variable matching to retrieve uncertain input variable name.
 
     .. highlight:: none
 
@@ -64,35 +58,88 @@ class PythonInterpreter():
     Modifications copyright (C) 2023 CAMEL-AI.org
 
     Args:
-        action_space (Dict[str, Any]): A dictionary that maps action names to
-            their corresponding functions or objects. The interpreter can only
-            execute functions that are either directly listed in this
+        action_space (Dict[str, Any], optional): A dictionary that maps action
+            names to their corresponding functions or objects. The interpreter
+            can only execute functions that are either directly listed in this
             dictionary or are member functions of objects listed in this
             dictionary. The concept of :obj:`action_space` is derived from
             EmbodiedAgent, representing the actions that an agent is capable of
-            performing.
-        import_white_list (Optional[List[str]], optional): A list that stores
+            performing. If `None`, set to empty dict. (default: :obj:`None`)
+        import_white_list (List[str], optional): A list that stores
             the Python modules or functions that can be imported in the code.
             All submodules and functions of the modules listed in this list are
             importable. Any other import statements will be rejected. The
             module and its submodule or function name are separated by a period
             (:obj:`.`). (default: :obj:`None`)
+        unsafe_mode (bool, optional): If `True`, the interpreter runs the code
+            by `eval()` without any security check. (default: :obj:`False`)
         raise_error (bool, optional): Raise error if the interpreter fails.
+            (default: :obj:`False`)
     """
 
-    def __init__(self, action_space: Dict[str, Any],
-                 import_white_list: Optional[List[str]] = None,
-                 raise_error: bool = False) -> None:
-        self.action_space = action_space
+    _CODE_TYPES = ["python", "py", "python3", "python2"]
+
+    def __init__(
+        self,
+        action_space: Optional[Dict[str, Any]] = None,
+        import_white_list: Optional[List[str]] = None,
+        unsafe_mode: bool = False,
+        raise_error: bool = False,
+    ) -> None:
+        self.action_space = action_space or dict()
         self.state = self.action_space.copy()
-        self.fuzz_state: Dict[str, Any] = {}
-        self.import_white_list = import_white_list or []
+        self.fuzz_state: Dict[str, Any] = dict()
+        self.import_white_list = import_white_list or list()
         self.raise_error = raise_error
+        self.unsafe_mode = unsafe_mode
+
+    def run(self, code: str, code_type: str) -> str:
+        r"""Executes the given code with specified code type in the
+        interpreter.
+
+        This method takes a string of code and its type, checks if the code
+        type is supported, and then executes the code. If `unsafe_mode` is
+        set to `False`, the code is executed in a controlled environment using
+        the `execute` method. If `unsafe_mode` is `True`, the code is executed
+        using `eval()` with the action space as the global context. An
+        `InterpreterError` is raised if the code type is unsupported or if any
+        runtime error occurs during execution.
+
+        Args:
+            code (str): The python code to be executed.
+            code_type (str): The type of the code, which should be one of the
+            supported code types (`python`, `py`, `python3`, `python2`).
+
+
+        Returns:
+            str: The string representation of the output of the executed code.
+
+        Raises:
+            InterpreterError: If the `code_type` is not supported or if any
+                runtime error occurs during the execution of the code.
+    """
+        if code_type not in self._CODE_TYPES:
+            raise InterpreterError(
+                f"Unsupported code type {code_type}. "
+                f"`{self.__class__.__name__}` only supports "
+                f"{', '.join(self._CODE_TYPES)}.")
+        if not self.unsafe_mode:
+            return str(self.execute(code))
+        else:
+            return str(eval(code, self.action_space))
+
+    def update_action_space(self, action_space: Dict[str, Any]) -> None:
+        r"""Updates action space for *python* interpreter."""
+        self.action_space.update(action_space)
+
+    def supported_code_types(self) -> List[str]:
+        r"""Provides supported code types by the interpreter."""
+        return self._CODE_TYPES
 
     def execute(self, code: str, state: Optional[Dict[str, Any]] = None,
                 fuzz_state: Optional[Dict[str, Any]] = None,
                 keep_state: bool = True) -> Any:
-        r""" Execute the input python codes in a security environment.
+        r"""Execute the input python codes in a security environment.
 
         Args:
             code (str): Generated python code to be executed.
@@ -154,7 +201,7 @@ def execute(self, code: str, state: Optional[Dict[str, Any]] = None,
         return result
 
     def clear_state(self) -> None:
-        r"""Initialize :obj:`state` and :obj:`fuzz_state`"""
+        r"""Initialize :obj:`state` and :obj:`fuzz_state`."""
         self.state = self.action_space.copy()
         self.fuzz_state = {}
 

camel/interpreters/interpreter_error.py

@@ -0,0 +1,18 @@
+# =========== Copyright 2023 @ CAMEL-AI.org. All Rights Reserved. ===========
+# Licensed under the Apache License, Version 2.0 (the “License”);
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an “AS IS” BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# =========== Copyright 2023 @ CAMEL-AI.org. All Rights Reserved. ===========
+
+
+class InterpreterError(Exception):
+    r"""Exception raised for errors that can be solved by regenerating code """
+    pass