v2.0.61

phi/agent/python.py

@@ -1,7 +1,7 @@
 import runpy
 import functools
 from pathlib import Path
-from typing import Optional, List
+from typing import Optional
 
 from phi.agent import Agent
 from phi.utils.log import logger
@@ -17,9 +17,11 @@ def __init__(
         self,
         base_dir: Optional[Path] = None,
         save_and_run: bool = True,
-        read_files: bool = False,
+        pip_install: bool = False,
+        run_code: bool = False,
         list_files: bool = False,
-        run_directly: bool = False,
+        run_files: bool = False,
+        read_files: bool = False,
         safe_globals: Optional[dict] = None,
         safe_locals: Optional[dict] = None,
     ):
@@ -31,20 +33,24 @@ def __init__(
         self.safe_globals: dict = safe_globals or globals()
         self.safe_locals: dict = safe_locals or locals()
 
+        if run_code:
+            self.register(self.run_python_code, sanitize_arguments=False)
         if save_and_run:
             self.register(self.save_to_file_and_run, sanitize_arguments=False)
+        if pip_install:
+            self.register(self.pip_install_package)
+        if run_files:
+            self.register(self.run_python_file_return_variable)
         if read_files:
             self.register(self.read_file)
         if list_files:
             self.register(self.list_files)
-        if run_directly:
-            self.register(self.run_python_code, sanitize_arguments=False)
 
     def save_to_file_and_run(
         self, file_name: str, code: str, variable_to_return: Optional[str] = None, overwrite: bool = True
     ) -> str:
         """This function saves Python code to a file called `file_name` and then runs it.
-        If successful, returns the value of `variable_to_return` if provided otherwise returns the file name.
+        If successful, returns the value of `variable_to_return` if provided otherwise returns a success message.
         If failed, returns an error message.
 
         :param file_name: The name of the file the code will be saved to.
@@ -73,11 +79,38 @@ def save_to_file_and_run(
                 logger.debug(f"Variable {variable_to_return} value: {variable_value}")
                 return str(variable_value)
             else:
-                return str(file_name)
+                return f"successfully ran {str(file_path)}"
         except Exception as e:
             logger.error(f"Error saving and running code: {e}")
             return f"Error saving and running code: {e}"
 
+    def run_python_file_return_variable(self, file_name: str, variable_to_return: Optional[str] = None) -> str:
+        """This function runs code in a Python file.
+        If successful, returns the value of `variable_to_return` if provided otherwise returns a success message.
+        If failed, returns an error message.
+
+        :param file_name: The name of the file to run.
+        :param variable_to_return: The variable to return.
+        :return: if run is successful, the value of `variable_to_return` if provided else file name.
+        """
+        try:
+            warn()
+            file_path = self.base_dir.joinpath(file_name)
+
+            logger.info(f"Running {file_path}")
+            globals_after_run = runpy.run_path(str(file_path), init_globals=self.safe_globals, run_name="__main__")
+            if variable_to_return:
+                variable_value = globals_after_run.get(variable_to_return)
+                if variable_value is None:
+                    return f"Variable {variable_to_return} not found"
+                logger.debug(f"Variable {variable_to_return} value: {variable_value}")
+                return str(variable_value)
+            else:
+                return f"successfully ran {str(file_path)}"
+        except Exception as e:
+            logger.error(f"Error running file: {e}")
+            return f"Error running file: {e}"
+
     def read_file(self, file_name: str) -> str:
         """Reads the contents of the file `file_name` and returns the contents if successful.
 
@@ -93,20 +126,24 @@ def read_file(self, file_name: str) -> str:
             logger.error(f"Error reading file: {e}")
             return f"Error reading file: {e}"
 
-    def list_files(self) -> List[str]:
+    def list_files(self) -> str:
         """Returns a list of files in the base directory
 
-        :return: The contents of the file if successful, otherwise returns an error message.
+        :return: Comma separated list of files in the base directory.
         """
         try:
             logger.info(f"Reading files in : {self.base_dir}")
-            return [str(file_path) for file_path in self.base_dir.iterdir()]
+            files = [str(file_path.name) for file_path in self.base_dir.iterdir()]
+            return ", ".join(files)
         except Exception as e:
             logger.error(f"Error reading files: {e}")
-            return [f"Error reading files: {e}"]
+            return f"Error reading files: {e}"
+
+    def run_python_code(self, code: str, variable_to_return: Optional[str] = None) -> str:
+        """This function to runs Python code in the current environment.
+        If successful, returns the value of `variable_to_return` if provided otherwise returns a success message.
+        If failed, returns an error message.
 
-    def run_python_code(self, code: str, variable_to_return: str) -> str:
-        """Function to run Python code.
         Returns the value of `variable_to_return` if successful, otherwise returns an error message.
 
         :param code: The code to run.
@@ -115,13 +152,39 @@ def run_python_code(self, code: str, variable_to_return: str) -> str:
         """
         try:
             warn()
+
             logger.debug(f"Running code:\n\n{code}\n\n")
             exec(code, self.safe_globals, self.safe_locals)
-            result = self.safe_locals.get(variable_to_return)
-            if result is None:
-                return f"Variable {variable_to_return} not found"
-            logger.debug(f"Result: {result}")
-            return str(result)
+
+            if variable_to_return:
+                variable_value = self.safe_locals.get(variable_to_return)
+                if variable_value is None:
+                    return f"Variable {variable_to_return} not found"
+                logger.debug(f"Variable {variable_to_return} value: {variable_value}")
+                return str(variable_value)
+            else:
+                return "successfully ran python code"
         except Exception as e:
             logger.error(f"Error running python code: {e}")
             return f"Error running python code: {e}"
+
+    def pip_install_package(self, package_name: str) -> str:
+        """This function installs a package using pip in the current environment.
+        If successful, returns a success message.
+        If failed, returns an error message.
+
+        :param package_name: The name of the package to install.
+        :return: success message if successful, otherwise returns an error message.
+        """
+        try:
+            warn()
+
+            logger.debug(f"Installing package {package_name}")
+            import sys
+            import subprocess
+
+            subprocess.check_call([sys.executable, "-m", "pip", "install", package_name])
+            return f"successfully installed package {package_name}"
+        except Exception as e:
+            logger.error(f"Error installing package {package_name}: {e}")
+            return f"Error installing package {package_name}: {e}"

phi/tool/function.py

@@ -61,16 +61,14 @@ def get_call_str(self) -> str:
         """Returns a string representation of the function call."""
         if self.arguments is None:
             return f"{self.function.name}()"
-        call_str = f"{self.function.name}({', '.join([f'{k}={v}' for k, v in self.arguments.items()])})"
-        if len(call_str) > 500:
-            trimmed_arguments = {}
-            for k, v in self.arguments.items():
-                if isinstance(v, str):
-                    trimmed_arguments[k] = f"{v[:50]}..."
-                else:
-                    trimmed_arguments[k] = v
-            call_str = f"{self.function.name}({', '.join([f'{k}={v}' for k, v in trimmed_arguments.items()])})"
 
+        trimmed_arguments = {}
+        for k, v in self.arguments.items():
+            if isinstance(v, str) and len(v) > 50:
+                trimmed_arguments[k] = "..."
+            else:
+                trimmed_arguments[k] = v
+        call_str = f"{self.function.name}({', '.join([f'{k}={v}' for k, v in trimmed_arguments.items()])})"
         return call_str
 
     def execute(self) -> bool: