Skip to content

Tools Module

The tools module provides utilities for web search and other agent-based capabilities.

WebSearchTool

The WebSearchTool enables real-time web search capabilities for RAG applications.

Usage

from rankify.tools.Tools import WebSearchTool

# Initialize the tool
search_tool = WebSearchTool(
    search_provider='serper',
    search_provider_api_key='your-api-key'
)

# Setup (loads necessary components)
search_tool.setup()

# Perform search
results = search_tool.forward("What is RAG in NLP?", num_result=10)
print(results)

API Reference

rankify.tools.Tools

Tool

A base class for the functions used by the agent. Subclass this and implement the forward method as well as the following class attributes:

  • description (str) -- A short description of what your tool does, the inputs it expects and the output(s) it will return. For instance, 'This is a tool that downloads a file from a url. It takes the url as input, and returns the text contained in the file'.
  • name (str) -- A performative name that will be used for your tool in the prompt to the agent. For instance "text-classifier" or "image_generator".
  • inputs (Dict[str, Dict[str, Union[str, type, bool]]]) -- The dict of modalities expected for the inputs. It has one type' key and adescriptionkey. This is used bylaunch_gradio_demo` or to make a nice space from your tool, and also can be used in the generated description for your tool.
  • output_type (type) -- The type of the tool output. This is used by launch_gradio_demo or to make a nice space from your tool, and also can be used in the generated description for your tool.
Source code in rankify/tools/Tools.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
class Tool:
    """
        A base class for the functions used by the agent. Subclass this and implement the `forward` method as well as the
        following class attributes:

        - **description** (`str`) -- A short description of what your tool does, the inputs it expects and the output(s) it
          will return. For instance, 'This is a tool that downloads a file from a `url`. It takes the `url` as input, and
          returns the text contained in the file'.
        - **name** (`str`) -- A performative name that will be used for your tool in the prompt to the agent. For instance
          `"text-classifier"` or `"image_generator"`.
        - **inputs** (`Dict[str, Dict[str, Union[str, type, bool]]]`) -- The dict of modalities expected for the inputs.
          It has one `type' key and a `description`key.
          This is used by `launch_gradio_demo` or to make a nice space from your tool, and also can be used in the generated
          description for your tool.
        - **output_type** (`type`) -- The type of the tool output. This is used by `launch_gradio_demo`
          or to make a nice space from your tool, and also can be used in the generated description for your tool.
"""
    name: str
    description: str
    inputs: dict[str, dict[str, str | type | bool]]
    output_type: str

    def __init__(self):
        self.is_initialized = False

    def validate_arguments(self):
        required_attributes = {
            "description": str,
            "name": str,
            "inputs": dict[str, dict[str, str | type | bool]],
            "output_type": str
        }
        for attr, expected_type in required_attributes.items():
            attr_value = getattr(self, attr, None)
            if attr_value is None:
                raise ValueError(f"Attribute {attr} is required for tool {self.name}")
            if not isinstance(attr_value, expected_type):# noqa
                raise ValueError(f"Attribute {attr} for tool {self.name} must be of type {expected_type}")

    def forward(self, query: str):
        return NotImplementedError("Write this method in your subclass of `Tool`.")

    def setup(self):
        """
        Overwrite this method here for any operation that is expensive and needs to be executed before you start using
        your tool. Such as loading a big model.
        """
        self.is_initialized = True

setup()

Overwrite this method here for any operation that is expensive and needs to be executed before you start using your tool. Such as loading a big model.

Source code in rankify/tools/Tools.py 
54
55
56
57
58
59
def setup(self):
    """
    Overwrite this method here for any operation that is expensive and needs to be executed before you start using
    your tool. Such as loading a big model.
    """
    self.is_initialized = True

WebSearchTool

Bases: Tool

Source code in rankify/tools/Tools.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
class WebSearchTool(Tool):

    name = "web_search"
    description = "Performs web search using user query and returns the top n results."
    inputs = {
        'query': {
            'type': 'string',
            'description': 'The user search query'
        }
    }
    output_type = "string"

    def __init__(self, model: Optional[str]=None, search_provider: str = 'serper',
                 search_provider_api_key: Optional[str] = None):
        super().__init__()
        self.search_client = None
        self.model = model
        self.search_provider = search_provider
        self.search_provider_api_key = search_provider_api_key
#        self.validate_arguments()

    @staticmethod
    def create_search_api(search_provider: str = 'SERPER', api_key: Optional[str] = None) -> SearchAPIClient:
        """
        Instantiate a search API client to fetch SERP results from a SERP provider e.g., SERPER.dev.
        Args:
            search_provider (str): The name of the search provider.
            Api_key (str): The API key for the search provider. Can be stored and loaded from env vars.
        Returns:
            SearchAPIClient: An instance of the search API client. E.g., SerpAPIClient.
        """
        if search_provider.lower() == 'serper':
            #print(api_key)
            return SerperApiClient(api_key=api_key)
        else:
            raise ValueError(f'Invalid search provider{search_provider}')

    def forward(self, query: str,num_result:int=10):
        sources = self.search_client.search_web(query=query,num_results=num_result)
        #print(sources, len(sources.data['organic']), num_result)


        #asdadsasdad
        try:
            loop = asyncio.get_event_loop()
            if loop.is_running():
                nest_asyncio.apply()
        except RuntimeError:
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)

        return loop.run_until_complete(build_search_context(sources))



    def setup(self):
        """
        Performs time-consuming setup operations here e.g., model loading.
        """
        self.search_client = self.create_search_api(api_key=self.search_provider_api_key)
        self.is_initialized = True

create_search_api(search_provider='SERPER', api_key=None) staticmethod

Instantiate a search API client to fetch SERP results from a SERP provider e.g., SERPER.dev. Args: search_provider (str): The name of the search provider. Api_key (str): The API key for the search provider. Can be stored and loaded from env vars. Returns: SearchAPIClient: An instance of the search API client. E.g., SerpAPIClient.

Source code in rankify/tools/Tools.py 
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
@staticmethod
def create_search_api(search_provider: str = 'SERPER', api_key: Optional[str] = None) -> SearchAPIClient:
    """
    Instantiate a search API client to fetch SERP results from a SERP provider e.g., SERPER.dev.
    Args:
        search_provider (str): The name of the search provider.
        Api_key (str): The API key for the search provider. Can be stored and loaded from env vars.
    Returns:
        SearchAPIClient: An instance of the search API client. E.g., SerpAPIClient.
    """
    if search_provider.lower() == 'serper':
        #print(api_key)
        return SerperApiClient(api_key=api_key)
    else:
        raise ValueError(f'Invalid search provider{search_provider}')

setup()

Performs time-consuming setup operations here e.g., model loading.

Source code in rankify/tools/Tools.py 
117
118
119
120
121
122
def setup(self):
    """
    Performs time-consuming setup operations here e.g., model loading.
    """
    self.search_client = self.create_search_api(api_key=self.search_provider_api_key)
    self.is_initialized = True

build_search_context(sources) async

Source code in rankify/tools/websearch/context/build_search_context.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
async def build_search_context(sources: T) -> T:
    if sources is None:
        return []
    stra= ['no_extraction'] #, 'cosine'
    # Filter only Wikipedia sources
    filtered_sources = [
        (i, source)
        for i, source in enumerate(sources.data['organic'])
        #if 'wikipedia.org' in source.get('link', '')
    ]

    urls = [source[1]['link'] for source in filtered_sources] 
    print(f"Filtered Wikipedia URLs: {urls}")

    scraper = WebScraper(strategies=stra) #strategies=['no_extraction']
    multi_results = await scraper.scrape_many(urls)

    sources_with_html: List[T] = []
    for _, source in filtered_sources:
        url = source['link']
        if url in multi_results:
            source['html'] = multi_results[url]['no_extraction'].content
            source['fit_markdown'] = multi_results[url]['no_extraction'].fit_markdown
            #print(multi_results[url]['no_extraction'] , source['fit_markdown'])
            sources_with_html.append(source)

    return sources_with_html