new

agent/plan_agent.py

@@ -1,6 +1,7 @@
 import logging
 from agent.agent_base import AgentBase
 from tools.tools_base import ToolsBase
+from tools.tools_factory import ToolsFactory
 from work_principle.okr_principle import OKR_Object
 
 class PlanAgent(AgentBase):
@@ -12,8 +13,8 @@ class PlanAgent(AgentBase):
     # 根据object和已有的工具能力拆解成key result，key
     def aligning(self, okr_object: OKR_Object):
         raw = okr_object.raw_user_task
-        tools_list = [i().get_info() for i in self.get_subclasses(ToolsBase)]
-        r = self.call_gpt(f"object为'{raw}'， 工具列表为{tools_list}")
+        r = self.call_gpt(f"object为'{raw}'， 工具列表为{list(ToolsFactory().get_tools())}")
+        self.logger.info(f"prompt：object为'{raw}'， 工具列表为{list(ToolsFactory().get_tools())}")
         self.logger.info(f"Alignment result: {r}")
         # return r
 

tests/test_plan_agent.py

@@ -2,14 +2,18 @@ import unittest
 
 from agent.plan_agent import PlanAgent
 from work_principle.okr_principle import OKR_Object
+
+
 class TestPlanAgent(unittest.TestCase):
     def test_aligning(self):
         # Create an instance of the PlanAgent class
         plan_agent = PlanAgent()
         # Create a mock OKR_Object with a raw_user_task
-        okr_object = OKR_Object("因为想要增加编程效率，对比一下copilot和curson谁更好用，比较提示词数量、安装易用性，给出不少于100字的文")
+        okr_object = OKR_Object(
+            "因为想要增加编程效率，对比一下copilot和curson谁更好用，比较提示词数量、安装易用性，给出不少于100字的文")
         # Call the aligning method
         plan_agent.aligning(okr_object)
 
+
 if __name__ == '__main__':
-    unittest.main()
+    unittest.main()

tools/llm_tools.py

@@ -5,6 +5,7 @@ from tools.tools_base import ToolsBase
 
 class LLMTools(ToolsBase):
     def __init__(self):
+        super().__init__()
         self.name = "llm_tools"
         self.description = "利用大模型进行回答"
         self.request_param='字典，如{"content": "天气怎么样"}'
@@ -20,11 +21,8 @@ class LLMTools(ToolsBase):
         Returns:
             str: The generated text from the chat API.
         """
-        with open("config.yaml", "r") as file:
-            config = yaml.safe_load(file).get("openai")
-
-        openai.api_key = config.get("api_key")
-        openai.base_url = config.get("api_url")
+        openai.api_key = self.config.OPEN_AI.get("api_key")
+        openai.base_url = self.config.OPEN_AI.get("api_url")
         messages = []
         if "role" in param:
             messages.append({"role": "system", "content": param["role"]})

tools/tools_base.py

@@ -1,9 +1,13 @@
+from utils.config import Config
+
+
 class ToolsBase:
     def __init__(self):
         self.name = ""
         self.description = ""
         self.request_param = ""
         self.return_content = ""
+        self.config = Config()
         
     def get_info(self):
         return {"name": self.name, "description": self.description, "param": self.request_param, "return_content": self.return_content}

tools/tools_factory.py

@@ -0,0 +1,16 @@
+from tools.llm_tools import LLMTools
+from tools.web_browser_element_tools import WebBrowserElement
+from tools.web_browser_url_tools import WebBrowserUrl
+
+
+class ToolsFactory:
+    def __init__(self):
+        self.tools = [
+            WebBrowserElement(),
+            WebBrowserUrl(),
+            LLMTools()
+        ]
+
+    def get_tools(self):
+        for tool in self.tools:
+            yield tool.get_info()

tools/web_browser_element_tools.py

@@ -13,6 +13,7 @@ from tools.tools_base import ToolsBase
 
 class WebBrowserElement(ToolsBase):
     def __init__(self):
+        super().__init__()
         self.name = "web_element"
         self.description = "利用selenium对指定xpath进行点击或者输入内容"
         self.request_param = '字典，如{"driver": "driver", "xpath": "", "action":"click或send_text"}'

tools/web_browser_url_tools.py

@@ -1,56 +1,42 @@
 import os
-import yaml
 from selenium import webdriver
-from selenium.webdriver.common.by import By
 from webdriver_manager.chrome import ChromeDriverManager
-from webdriver_manager.firefox import GeckoDriverManager
 from webdriver_manager.microsoft import EdgeChromiumDriverManager
-from webdriver_manager.opera import OperaDriverManager
 from selenium.webdriver.edge.service import Service as EdgeService
-from selenium.webdriver.common.by import By
+from selenium.webdriver.chrome.service import Service as ChromeService
 
 from tools.tools_base import ToolsBase
 
+
 class WebBrowserUrl(ToolsBase):
     def __init__(self):
+        super().__init__()
         self.name = "web_browser"
         self.description = "利用selenium对指定URL进行访问"
         self.request_param = '字典，如{"usrl": ""}'
-        self.return_content = '{"driver": "selenium的webdriver，driver用于继续在此见面上进行操作，可作为web_element工具的入参", "content": "网页xml结构"'
+        self.return_content = ('{"driver": "selenium的webdriver，driver用于继续在此见面上进行操作，可作为web_element工具的入参", "content": '
+                               '"网页xml结构"')
 
     def run(self, param=None):
         # Load browser configuration from YAML file
-        with open("config.yaml", "r") as file:
-            config = yaml.safe_load(file)
-
+        driver = None
         # Check if webdriver is available
         if not os.path.exists("webdriver.exe"):
             # Download webdriver based on browser type
-            browser_type = config.get("browser_type")
-            driver = None
+            browser_type = self.config.BROWSER.get("browser_type")
             if browser_type == "chrome":
                 options = webdriver.ChromeOptions()
                 options.add_argument("--headless")  # Enable headless mode
                 webdriver_manager = ChromeDriverManager()
-                driver = webdriver.Chrome(webdriver_manager.install(), options=options)
-            elif browser_type == "firefox":
-                options = webdriver.FirefoxOptions()
-                options.add_argument("--headless")  # Enable headless mode
-                webdriver_manager = GeckoDriverManager()
-                driver = webdriver.Firefox(webdriver_manager.install(), options=options)
+                driver = webdriver.Chrome(service=ChromeService(webdriver_manager.install()), options=options)
             elif browser_type == "edge":
                 options = webdriver.EdgeOptions()
                 options.add_argument("--headless")  # Enable headless mode
                 webdriver_manager = EdgeChromiumDriverManager()
                 driver = webdriver.Edge(service=EdgeService(webdriver_manager.install()), options=options)
-            elif browser_type == "opera":
-                options = webdriver.ChromeOptions()
-                options.add_argument("--headless")  # Enable headless mode
-                webdriver_manager = OperaDriverManager()
-                driver = webdriver.Opera(webdriver_manager.install(), options=options)
+            else:
+                return
         driver.implicitly_wait(10)
-        # browser_url = config.get("browser_url")
-        # driver.get(browser_url)
         driver.get(param["url"])
         driver.quit()
         return driver.page_source

utils/config.py

@@ -0,0 +1,14 @@
+import yaml
+
+
+class Config:
+    def __init__(self):
+        self.config = self._load_config('../config.yaml')
+        self.OPEN_AI = self.config["openai"]
+        self.BROWSER = self.config["browser"]
+
+    # Load content from a yaml file and return as variables
+    def _load_config(self, file_path):
+        with open(file_path, 'r') as file:
+            config = yaml.safe_load(file)
+        return config