feat(forge): Implement a very very bad, but "functional" agent by default

autogpts/forge/forge/__main__.py

@@ -17,9 +17,9 @@
   d88P   888 888  888 888   888  888 888    888 888           888     
  d8888888888 Y88b 888 Y88b. Y88..88P Y88b  d88P 888           888     
 d88P     888  "Y88888  "Y888 "Y88P"   "Y8888P88 888           888     
-                                                                      
-                                                                      
-                                                                      
+
+
+
                 8888888888                                            
                 888                                                   
                 888                                                   

autogpts/forge/forge/actions/file_system/files.py

@@ -1,22 +1,24 @@
 from typing import List
 
-from ..registry import action
+from forge.sdk.agent import Agent
+
+from ..registry import ActionParameter, action
 
 
 @action(
     name="list_files",
     description="List files in a directory",
     parameters=[
-        {
-            "name": "path",
-            "description": "Path to the directory",
-            "type": "string",
-            "required": True,
-        }
+        ActionParameter(
+            name="path",
+            description="Path to the directory",
+            type="string",
+            required=True,
+        )
     ],
     output_type="list[str]",
 )
-async def list_files(agent, task_id: str, path: str) -> List[str]:
+async def list_files(agent: Agent, task_id: str, path: str) -> List[str]:
     """
     List files in a workspace directory
     """
@@ -27,22 +29,22 @@ async def list_files(agent, task_id: str, path: str) -> List[str]:
     name="write_file",
     description="Write data to a file",
     parameters=[
-        {
-            "name": "file_path",
-            "description": "Path to the file",
-            "type": "string",
-            "required": True,
-        },
-        {
-            "name": "data",
-            "description": "Data to write to the file",
-            "type": "bytes",
-            "required": True,
-        },
+        ActionParameter(
+            name="file_path",
+            description="Path to the file",
+            type="string",
+            required=True,
+        ),
+        ActionParameter(
+            name="data",
+            description="Data to write to the file",
+            type="bytes",
+            required=True,
+        ),
     ],
     output_type="None",
 )
-async def write_file(agent, task_id: str, file_path: str, data: bytes):
+async def write_file(agent: Agent, task_id: str, file_path: str, data: bytes):
     """
     Write data to a file
     """
@@ -62,16 +64,16 @@ async def write_file(agent, task_id: str, file_path: str, data: bytes):
     name="read_file",
     description="Read data from a file",
     parameters=[
-        {
-            "name": "file_path",
-            "description": "Path to the file",
-            "type": "string",
-            "required": True,
-        },
+        ActionParameter(
+            name="file_path",
+            description="Path to the file",
+            type="string",
+            required=True,
+        )
     ],
     output_type="bytes",
 )
-async def read_file(agent, task_id: str, file_path: str) -> bytes:
+async def read_file(agent: Agent, task_id: str, file_path: str) -> bytes:
     """
     Read data from a file
     """

autogpts/forge/forge/actions/finish.py

@@ -1,6 +1,7 @@
+from forge.sdk.agent import Agent
 from forge.sdk.forge_log import ForgeLogger
 
-from .registry import action
+from .registry import ActionParameter, action
 
 logger = ForgeLogger(__name__)
 
@@ -11,17 +12,17 @@
     " or when there are insurmountable problems that make it impossible"
     " for you to finish your task.",
     parameters=[
-        {
-            "name": "reason",
-            "description": "A summary to the user of how the goals were accomplished",
-            "type": "string",
-            "required": True,
-        }
+        ActionParameter(
+            name="reason",
+            description="A summary to the user of how the goals were accomplished",
+            type="string",
+            required=True,
+        )
     ],
     output_type="None",
 )
 async def finish(
-    agent,
+    agent: Agent,
     task_id: str,
     reason: str,
 ) -> str:
@@ -34,5 +35,6 @@ async def finish(
         A result string from create chat completion. A list of suggestions to
             improve the code.
     """
+
     logger.info(reason, extra={"title": "Shutting down...\n"})
     return reason

autogpts/forge/forge/actions/registry.py

@@ -112,9 +112,11 @@ def register_abilities(self) -> None:
                 "__init__.py",
                 "registry.py",
             ]:
-                action = os.path.relpath(
-                    action_path, os.path.dirname(__file__)
-                ).replace("/", ".")
+                action = (
+                    os.path.relpath(action_path, os.path.dirname(__file__))
+                    .replace("\\", "/")
+                    .replace("/", ".")
+                )
                 try:
                     module = importlib.import_module(
                         f".{action[:-3]}", package="forge.actions"
@@ -185,9 +187,12 @@ async def run_action(
 
 
 if __name__ == "__main__":
+    import asyncio
     import sys
 
-    sys.path.append("/Users/swifty/dev/forge/forge")
-    register = ActionRegister(agent=None)
-    print(register.abilities_description())
-    print(register.run_action("abc", "list_files", "/Users/swifty/dev/forge/forge"))
+    async def run():
+        register = ActionRegister(agent=None)
+        print(register.abilities_description())
+        print(await register.run_action("abc", "finish", "./registry.py"))
+
+    asyncio.run(run())

autogpts/forge/forge/actions/web/web_search.py

@@ -5,7 +5,9 @@
 
 from duckduckgo_search import DDGS
 
-from ..registry import action
+from forge.sdk.agent import Agent
+
+from ..registry import ActionParameter, action
 
 DUCKDUCKGO_MAX_ATTEMPTS = 3
 
@@ -14,16 +16,16 @@
     name="web_search",
     description="Searches the web",
     parameters=[
-        {
-            "name": "query",
-            "description": "The search query",
-            "type": "string",
-            "required": True,
-        }
+        ActionParameter(
+            name="query",
+            description="The search query",
+            type="string",
+            required=True,
+        )
     ],
     output_type="list[str]",
 )
-async def web_search(agent, task_id: str, query: str) -> str:
+async def web_search(agent: Agent, task_id: str, query: str) -> str:
     """Return the results of a Google search
 
     Args:

autogpts/forge/forge/actions/web/web_selenium.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from forge.sdk.agent import Agent
+
 COMMAND_CATEGORY = "web_browse"
 COMMAND_CATEGORY_TITLE = "Web Browsing"
 
@@ -38,7 +40,7 @@
 
 from forge.sdk.errors import CommandExecutionError
 
-from ..registry import action
+from ..registry import ActionParameter, action
 
 
 def extract_hyperlinks(soup: BeautifulSoup, base_url: str) -> list[tuple[str, str]]:
@@ -185,25 +187,25 @@ class BrowsingError(CommandExecutionError):
     name="read_webpage",
     description="Read a webpage, and extract specific information from it if a question is specified. If you are looking to extract specific information from the webpage, you should specify a question.",
     parameters=[
-        {
-            "name": "url",
-            "description": "The URL to visit",
-            "type": "string",
-            "required": True,
-        },
-        {
-            "name": "question",
-            "description": "A question that you want to answer using the content of the webpage.",
-            "type": "string",
-            "required": False,
-        },
+        ActionParameter(
+            name="url",
+            description="The URL to visit",
+            type="string",
+            required=True,
+        ),
+        ActionParameter(
+            name="question",
+            description="A question that you want to answer using the content of the webpage.",
+            type="string",
+            required=False,
+        ),
     ],
     output_type="string",
 )
 @validate_url
 async def read_webpage(
-    agent, task_id: str, url: str, question: str = ""
-) -> Tuple(str, List[str]):
+    agent: Agent, task_id: str, url: str, question: str = ""
+) -> Tuple(str, list[str]):
     """Browse a website and return the answer and links to the user
 
     Args:
@@ -231,7 +233,9 @@ async def read_webpage(
     except WebDriverException as e:
         # These errors are often quite long and include lots of context.
         # Just grab the first line.
-        msg = e.msg.split("\n")[0]
+        msg = "An error occurred while trying to load the page"
+        if e.msg:
+            msg = e.msg.split("\n")[0]
         if "net::" in msg:
             raise BrowsingError(
                 f"A networking error occurred while trying to load the page: "
@@ -340,9 +344,11 @@ def open_page_in_browser(url: str) -> WebDriver:
         chromium_driver_path = Path("/usr/bin/chromedriver")
 
         driver = ChromeDriver(
-            service=ChromeDriverService(str(chromium_driver_path))
-            if chromium_driver_path.exists()
-            else ChromeDriverService(ChromeDriverManager().install()),
+            service=(
+                ChromeDriverService(str(chromium_driver_path))
+                if chromium_driver_path.exists()
+                else ChromeDriverService(ChromeDriverManager().install())
+            ),
             options=options,
         )
     driver.get(url)