File size: 5,229 Bytes
25a8ac6
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
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
60
61
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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
import random
from httpcore import TimeoutException
from huggingface_hub import list_models
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_core.tools import Tool

from selenium import webdriver
driver = webdriver.Chrome()
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

import os
try:
    from langchain_tavily import TavilySearch
    tavily_api_key = os.getenv("TAVILY_API_KEY")
    if tavily_api_key:
        tavilytool = TavilySearch(
            max_results=2,
            search_depth="advanced",   # deep crawl mode
            include_raw_content=True,  # returns full page text, not just snippets
            tavily_api_key=tavily_api_key
        )
    else:
        tavilytool = None
except ImportError:
    tavilytool = None



class ToolRegistry:
    """
    A registry that defines and exposes all LangChain tools.
    Import this class in downstream code and access tools via class methods or attributes.
    """

    # ------------------------------------------------------------------ #
    #  Internal implementations                                           #
    # ------------------------------------------------------------------ #

    @staticmethod
    def _weather_info(location: str) -> str:
        """Fetches dummy weather information for a given location."""
        weather_conditions = [
            {"condition": "Rainy",  "temp_c": 15},
            {"condition": "Clear",  "temp_c": 25},
            {"condition": "Windy",  "temp_c": 20},
        ]
        data = random.choice(weather_conditions)
        return f"Weather in {location}: {data['condition']}, {data['temp_c']}°C"

    @staticmethod
    def _get_hub_stats(author: str) -> str:
        """Fetches the most downloaded model from a specific author on the Hugging Face Hub."""
        try:
            models = list(list_models(author=author, sort="downloads", limit=1))
            if models:
                model = models[0]
                return (
                    f"The most downloaded model by {author} is {model.id} "
                    f"with {model.downloads:,} downloads."
                )
            return f"No models found for author {author}."
        except Exception as e:
            return f"Error fetching models for {author}: {str(e)}"
        
    @staticmethod
    def _browse_url(url: str) -> str:
        """Opens a URL and returns the page text"""
        try:
            driver.get(url)
            WebDriverWait(driver, 20).until(EC.presence_of_all_elements_located(("tag name", "body")))
            return driver.find_element("tag name", "body").text[:3000]
        except TimeoutException as e:
            return f"Timeout loading {url}: {str(e)}. Current URL: {driver.current_url}"     

    @staticmethod
    def _click_element(selector: str) -> str:
        """Clicks an element by CSS selector"""
        try:
            element = WebDriverWait(driver, 20).until(
                EC.element_to_be_clickable(("css selector", selector))
            )
            element.click()
            return f"Clicked {selector}"
        except TimeoutException as e:
            return f"Timeout clicking {selector}: {str(e)}. Current URL: {driver.current_url}"
    # ------------------------------------------------------------------ #
    #  Tool factories                                                      #
    # ------------------------------------------------------------------ #

    @classmethod
    def weather_info_tool(cls) -> Tool:
        """Returns a LangChain Tool for fetching weather information."""
        return Tool(
            name="WeatherInfoTool",
            func=cls._weather_info,
            description="Retrieves weather information for a given location.",
        )

    @classmethod
    def hub_stats_tool(cls) -> Tool:
        """Returns a LangChain Tool for fetching Hugging Face Hub stats."""
        return Tool(
            name="get_hub_stats",
            func=cls._get_hub_stats,
            description=(
                "Fetches the most downloaded model from a specific author "
                "on the Hugging Face Hub."
            ),
        )

    @classmethod
    def duck_duck_go_search_tool(cls) -> DuckDuckGoSearchRun:
        """Returns a LangChain DuckDuckGo search tool."""
        return DuckDuckGoSearchRun()

    @classmethod
    def tavily_search_tool(cls) -> TavilySearch:
        """Returns a LangChain Tavily search tool."""
        if tavilytool is None:
            raise ValueError("Tavily tool not available. Please install langchain-tavily and set TAVILY_API_KEY environment variable.")
        return tavilytool
    
    # ------------------------------------------------------------------ #
    #  Convenience: get all tools at once                                  #
    # ------------------------------------------------------------------ #

    @classmethod
    def get_all_tools(cls) -> list[Tool]:
        """Returns a list of all registered tools — ready to pass to an agent."""
        return [
            cls.weather_info_tool(),
            cls.hub_stats_tool(),
            cls.duck_duck_go_search_tool(),
        ]