tools deprecation and new format support

src/openagi/actions/tools/arxiv_search.py

@@ -0,0 +1,72 @@
+from openagi.actions.base import BaseAction
+from pydantic import Field
+from openagi.exception import OpenAGIException
+from typing import ClassVar, Dict, Any
+
+try:
+    import arxiv
+except ImportError:
+    raise OpenAGIException("Install arxiv with cmd `pip install arxiv`")
+
+class ConfigurableAction(BaseAction):
+    config: ClassVar[Dict[str, Any]] = {}
+
+    @classmethod
+    def set_config(cls, *args, **kwargs):
+        if args:
+            if len(args) == 1 and isinstance(args[0], dict):
+                cls.config.update(args[0])
+            else:
+                raise ValueError("If using positional arguments, a single dictionary must be provided.")
+        cls.config.update(kwargs)
+
+    @classmethod
+    def get_config(cls, key: str, default: Any = None) -> Any:
+        return cls.config.get(key, default)
+
+class ArxivSearch(ConfigurableAction):
+    """
+    Arxiv Search is a tool used to search articles in Physics, Mathematics, Computer Science, Quantitative Biology, Quantitative Finance, and Statistics
+    """
+    query: str = Field(..., description="User query or question")
+    max_results: int = Field(10, description="Total results, in int, to be executed from the search. Defaults to 10.")
+
+    def execute(self):
+        search = arxiv.Search(
+        query = self.query,
+        max_results = self.max_results,
+                              )
+        client = arxiv.Client()
+        results = client.results(search)
+        meta_data = ""
+        for result in results:
+            meta_data += f"title : {result.title}\n "
+            meta_data += f"summary : {result.summary}\n "
+            meta_data += f"published : {result.published}\n "
+            meta_data += f"authors : {result.authors}\n "
+            meta_data += f"pdf_url : {result.pdf_url}\n "
+            meta_data += f"entry_id : {result.entry_id}\n\n "
+        return meta_data.strip()
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+

src/openagi/actions/tools/document_loader.py

@@ -1,39 +1,59 @@
 from openagi.actions.base import BaseAction
 from langchain_community.document_loaders import TextLoader
 from langchain_community.document_loaders.csv_loader import CSVLoader
+from langchain_community.document_loaders.pdf import PyPDFLoader
 from pydantic import Field
+from typing import ClassVar, Dict, Any
 
+class ConfigurableAction(BaseAction):
+    config: ClassVar[Dict[str, Any]] = {}
 
-class DocumentLoader(BaseAction):
-    """Use this Action to extract content from documents"""
+    @classmethod
+    def set_config(cls, *args, **kwargs):
+        if args:
+            if len(args) == 1 and isinstance(args[0], dict):
+                cls.config.update(args[0])
+            else:
+                raise ValueError("If using positional arguments, a single dictionary must be provided.")
+        cls.config.update(kwargs)
 
-    file_path: str = Field(
-        default_factory=str,
-        description="File from which content is extracted",
-    )
-
-    def text_loader(self):
-        loader = TextLoader(file_path=self.file_path)
+    @classmethod
+    def get_config(cls, key: str, default: Any = None) -> Any:
+        return cls.config.get(key, default)
+
+class TextLoaderTool(ConfigurableAction):
+    """Use this Action to load the content from .text file"""
+    def execute(self):
+        file_path = self.get_config('filename')        
+        #print(file_path)
+        loader = TextLoader(file_path=file_path)
+        data = loader.load()
+        page_content = data[0].page_content
+        meta_data = data[0].metadata["source"]
+        context = meta_data + " " + page_content
+        return context
+
+class PDFLoaderTool(ConfigurableAction):
+    """Use this Action to load the content from .pdf file"""
+    def execute(self):
+        file_path = self.get_config('filename')        
+        loader = PyPDFLoader(file_path=file_path)
         data = loader.load()
         page_content = data[0].page_content
         meta_data = data[0].metadata["source"]
         context = meta_data + " " + page_content
         return context
 
-    def csv_loader(self):
+class CSVLoaderTool(ConfigurableAction):
+    """Use this Action to load the content from .csv file"""
+    def execute(self):
+        file_path = self.get_config('filename')
         content = ""
-        loader = CSVLoader(file_path=self.file_path)
+        loader = CSVLoader(file_path=file_path)
         data = loader.load()
 
         for i in range(len(data)):
             row_content = data[i].page_content
             row_no = data[i].metadata["row"]
             content += "row_no" + " " + str(row_no) + ": " + str(row_content)
-        return content
-
-    def execute(self):
-        if self.file_path.endswith(".txt"):
-            context = self.text_loader()
-        elif self.file_path.endswith(".csv"):
-            context = self.csv_loader()
-        return context
+        return content

src/openagi/actions/tools/exasearch.py

@@ -1,30 +1,75 @@
 from openagi.actions.base import BaseAction
-import os
 from pydantic import Field
 from openagi.exception import OpenAGIException
+from typing import ClassVar, Dict, Any
+import os
+import warnings
 
 try:
-   from exa_py import Exa
+    from exa_py import Exa
 except ImportError:
-  raise OpenAGIException("Install Exa Py with cmd `pip install exa_py`")
+    raise OpenAGIException("Install Exa Py with cmd `pip install exa_py`")
 
-class ExaSearch(BaseAction):
+class ConfigurableAction(BaseAction):
+    config: ClassVar[Dict[str, Any]] = {}
+
+    @classmethod
+    def set_config(cls, *args, **kwargs):
+        if args:
+            if len(args) == 1 and isinstance(args[0], dict):
+                cls.config.update(args[0])
+            else:
+                raise ValueError("If using positional arguments, a single dictionary must be provided.")
+        cls.config.update(kwargs)
+
+    @classmethod
+    def get_config(cls, key: str, default: Any = None) -> Any:
+        return cls.config.get(key, default)
+
+class ExaSearch(ConfigurableAction):
     """
     Exa Search is a tool used when user needs to ask the question in terms of query to get response 
     """
-    query: str = Field(..., description="User query or question ")
-
+    query: str = Field(..., description="User query or question")
+
+    def __init__(self, **data):
+        super().__init__(**data)
+        self._check_deprecated_usage()
+
+    def _check_deprecated_usage(self):
+        if 'EXA_API_KEY' in os.environ and not self.get_config('api_key'):
+            warnings.warn(
+                "Using environment variables for API keys is deprecated and will be removed in a future version. "
+                "Please use ExaSearch.set_config(api_key='your_key') instead of setting environment variables.",
+                DeprecationWarning,
+                stacklevel=2
+            )
+            self.set_config(api_key=os.environ['EXA_API_KEY'])
+
     def execute(self):
-        api_key = os.environ["EXA_API_KEY"]
+        api_key = self.get_config('api_key')
+
+        if not api_key:
+            if 'EXA_API_KEY' in os.environ:
+                api_key = os.environ['EXA_API_KEY']
+                warnings.warn(
+                    "Using environment variables for API keys is deprecated and will be removed in a future version. "
+                    "Please use ExaSearch.set_config(api_key='your_key') instead of setting environment variables.",
+                    DeprecationWarning,
+                    stacklevel=2
+                )
+            else:
+                raise OpenAGIException("API KEY NOT FOUND. Use ExaSearch.set_config(api_key='your_key') to set the API key.")
+
+        exa = Exa(api_key=api_key)
+        results = exa.search_and_contents(
+            self.query,
+            text={"max_characters": 512},
+        )
 
-        exa = Exa(api_key = api_key)
-        results = exa.search_and_contents(self.query, 
-                                    text={"max_characters": 512}, 
-                                )
         content = ""
         for idx in results.results:
             content += idx.text.strip()
-
-        content = content.replace("<|endoftext|>","")
-        content = content.replace("NaN","")
-        return content
+        content = content.replace("<|endoftext|>", "")
+        content = content.replace("NaN", "")
+        return content

src/openagi/actions/tools/google_search_tool.py

@@ -0,0 +1,57 @@
+from openagi.actions.base import BaseAction
+from pydantic import Field
+from openagi.exception import OpenAGIException
+import logging
+import requests
+from bs4 import BeautifulSoup
+from typing import ClassVar, Dict, Any
+
+try:
+   from googlesearch import search
+except ImportError:
+  raise OpenAGIException("Install googlesearch-python with cmd `pip install googlesearch-python`")
+
+class ConfigurableAction(BaseAction):
+    config: ClassVar[Dict[str, Any]] = {}
+
+    @classmethod
+    def set_config(cls, *args, **kwargs):
+        if args:
+            if len(args) == 1 and isinstance(args[0], dict):
+                cls.config.update(args[0])
+            else:
+                raise ValueError("If using positional arguments, a single dictionary must be provided.")
+        cls.config.update(kwargs)
+
+    @classmethod
+    def get_config(cls, key: str, default: Any = None) -> Any:
+        return cls.config.get(key, default)
+
+class GoogleSearchTool(ConfigurableAction):
+    """
+    Google Search is a tool used for scraping the Google search engine. Extract information from Google search results.
+    """
+    query: str = Field(..., description="User query or question ")
+
+    max_results: int = Field(
+        default=10,
+        description="Total results, in int, to be executed from the search. Defaults to 10. The limit should be 10 and not execeed more than 10",
+    )
+
+    lang: str = Field(
+        default="en",
+        description = "specify the langauge for your search results."
+    )
+
+    def execute(self):
+        if self.max_results > 15:
+            logging.info("Over threshold value... Limiting the Max results to 15")
+            self.max_results = 15
+
+        context = ""
+        search_results = search(self.query,num_results=self.max_results,lang=self.lang,advanced=True)
+        for info in search_results:
+            context += f"Title: {info.title}. Description: {info.description}. URL: {info.url}"
+
+        return context
+

src/openagi/actions/tools/searchapi_search.py

@@ -8,33 +8,61 @@
 
 from openagi.actions.base import BaseAction
 from openagi.exception import OpenAGIException
+from typing import ClassVar, Dict, Any
+import warnings
 
+class ConfigurableAction(BaseAction):
+    config: ClassVar[Dict[str, Any]] = {}
+
+    @classmethod
+    def set_config(cls, *args, **kwargs):
+        if args:
+            if len(args) == 1 and isinstance(args[0], dict):
+                cls.config.update(args[0])
+            else:
+                raise ValueError("If using positional arguments, a single dictionary must be provided.")
+        cls.config.update(kwargs)
+
+    @classmethod
+    def get_config(cls, key: str, default: Any = None) -> Any:
+        return cls.config.get(key, default)
 
 class SearchApiSearch(BaseAction):
     """SearchApi.io provides a real-time API to access search results from Google (default), Google Scholar, Bing, Baidu, and other search engines."""
     query: str = Field(
         ..., description="User query of type string used to fetch web search results from a search engine."
     )
 
+    def __init__(self, **data):
+        super().__init__(**data)
+        self._check_deprecated_usage()
+
+    def _check_deprecated_usage(self):
+        if 'SEARCHAPI_API_KEY' in os.environ and not self.get_config('api_key'):
+            warnings.warn(
+                "Using environment variables for API keys is deprecated and will be removed in a future version. "
+                "Please use SearchApiSearch.set_config(api_key='your_key') instead of setting environment variables.",
+                DeprecationWarning,
+                stacklevel=2
+            )
+            self.set_config(api_key=os.environ['SEARCHAPI_API_KEY'])
+
     def execute(self):
         base_url = "https://www.searchapi.io/api/v1/search"
-        searchapi_api_key = os.environ["SEARCHAPI_API_KEY"]
-        engine = os.environ.get("SEARCHAPI_ENGINE") or "google"
+        api_key = self.get_config('api_key')
+
         search_dict = {
             "q": self.query,
-            "engine": engine,
-            "api_key": searchapi_api_key
+            "engine": "google",
+            "api_key": api_key
         }
+
         logging.debug(f"{search_dict=}")
+
         url = f"{base_url}?{urlencode(search_dict)}"
         response = requests.request("GET", url)
         json_response = response.json()
 
-        # if not json_response:
-        #     raise OpenAGIException(f"Unable to generate result for the query {self.query}")
-
-        # logging.debug(json_response)
-
         organic_results = json_response.get("organic_results", [])
 
         meta_data = ""