Upload 6 files

Tools/code_reader.py

@@ -0,0 +1,17 @@
+from smolagents import tool
+
+@tool
+def code_reader(python_file_path:str) -> str:
+    """
+    This tool reads python file and returns the code as a string.
+
+    Args:
+        python_file_path: Path of a local python file that needs to be accessed.
+    """
+
+    try:
+        with open(python_file_path) as f:
+            code = f.read()
+        return code
+    except Exception as e:
+        return f'An unexpected error occurred: {str(e)}'

Tools/mywebpagevisit.py

@@ -0,0 +1,42 @@
+from smolagents import VisitWebpageTool
+
+
+class CustomVisitWebpageTool(VisitWebpageTool):
+
+    def forward(self, url: str) -> str:
+        try:
+            import re
+            import requests
+            from markdownify import markdownify
+            from requests.exceptions import RequestException
+        except ImportError as e:
+            raise ImportError(
+                "You must install packages `markdownify` and `requests` to run this tool: for instance run `pip install markdownify requests`."
+            ) from e
+        try:
+            headers = {
+                "User-Agent": (
+                    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+                    "AppleWebKit/537.36 (KHTML, like Gecko) "
+                    "Chrome/122.0.0.0 Safari/537.36"
+                ),
+                "Accept-Language": "en-US,en;q=0.9",
+                "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
+            }
+            # Send a GET request to the URL with a 20-second timeout
+            response = requests.get(url, timeout=20, headers=headers)
+            response.raise_for_status()  # Raise an exception for bad status codes
+
+            # Convert the HTML content to Markdown
+            markdown_content = markdownify(response.text).strip()
+
+            # Remove multiple line breaks
+            #markdown_content = re.sub(r"\n{3,}", "\n\n", markdown_content)
+            return self._truncate_content(markdown_content, self.max_output_length)
+
+        except requests.exceptions.Timeout:
+            return "The request timed out. Please try again later or check the URL."
+        except RequestException as e:
+            return f"Error fetching the webpage: {str(e)}"
+        except Exception as e:
+            return f"An unexpected error occurred: {str(e)}"

Tools/searchtool.py

@@ -0,0 +1,230 @@
+# ##not implemented include_images and include_image_descriptions attributes
+
+# import os
+
+# from tavily import TavilyClient
+# from smolagents import Tool
+# #from pydantic import BaseModel, Field 
+# from typing import Optional,List, Dict, Any, Literal
+
+
+# # class TavilySearchInput(BaseModel):
+# #     """Input for Tavily Search."""
+# #     query : str = Field(description="The search query string.")
+# #     max_results : Optional[int] = Field(default = 5,
+# #                                         description="Maximum number of search results to return.")
+# #     topic : Optional[str] = Field(default = "general",
+# #                                   description="Category of the search. Can be 'general', 'news', or 'finance'.")
+# #     include_answer : Optional[bool] = Field(default = False,
+# #                                             description="Whether to include an answer to original query in results.")
+# #     include_raw_content : Optional[bool] = Field(default = False,
+# #                                                  description="Whether to include cleaned and parsed HTML of each search result.")
+# #     search_depth: Optional[str] = Field(default = "basic",
+# #                                         description="Depth of the search, either 'basic' or 'advanced'.")
+# #     time_range: Optional[str] = Field(default = None,
+# #                                       description="The time range back from the current date to filter results - 'day', 'week', 'month', or 'year'.")
+# #     include_domains: Optional[List[str]] = Field(default = None,
+# #                                                  description="List of domains to specifically include.")
+# #     exclude_domains: Optional[List[str]] = Field(default = None,
+# #                                                  description="List of domains to specifically exclude.")
+
+
+# # class TavilySearchTool(Tool):
+# #     name: str = "tavily_search"
+# #     description = """
+# #     A tool to perform comprehensive web searches using the Tavily Search API.
+# #     It can retrieve real-time, accurate, and factual information from across the web.
+# #     Useful for answering questions, gathering information, and reducing hallucinations."""
+
+# #     inputs = inputs
+
+# #     output_type = "list"
+
+# #     def __init__(self, api_key:Optional[str] = None, **kwargs):
+# #         super().__init__(**kwargs)
+# #         if api_key is None:
+# #             api_key = os.getenv("TRAVILY_SECRET_KEY")
+# #         if not api_key:
+# #             raise ValueError(
+# #                 "Tavily API key not found. Please set the TAVILY_API_KEY environment variable "
+# #                 "or pass it directly to TavilySearchTool(api_key='YOUR_API_KEY')."
+# #             )
+# #         self.tavily_client = TavilyClient(api_key=api_key)
+# #         print('TavilyClient is initiated')
+
+# #     def forward(self, query: str, 
+# #                 max_results: int = 5, 
+# #                 include_answer: bool = False, 
+# #                 include_raw_content: bool = False,
+# #                 search_depth: str = "basic",
+# #                 topic: str = "general",
+# #                 time_range: Optional[str] = None,
+# #                 include_domains: Optional[List[str]] = None,
+# #                 exclude_domains: Optional[List[str]] = None) -> List[Dict[str, Any]]:
+# #         """
+# #         Executes a search query using the Tavily API.
+
+# #         Args:
+# #             query (str): The search query.
+# #             max_results (int): Maximum number of search results to return.
+# #             include_answer (bool): Whether to include a direct answer synthesized from the search results.
+# #             include_raw_content (bool): Whether to include the raw HTML content of the searched pages.
+# #             search_depth (str): Depth of the search, either 'basic' or 'advanced'.
+# #             topic (str): Category of the search. Can be 'general', 'news', or 'finance'.
+# #             time_range (Optional[str]): The time range back from the current date to filter results - 'day', 'week', 'month', or 'year'.
+# #             include_domains (Optional[List[str]]): List of domains to specifically include.
+# #             exclude_domains (Optional[List[str]]): List of domains to specifically exclude.
+
+# #         Returns:
+# #             List[Dict[str, Any]]: A list of dictionaries, where each dictionary represents a search result.
+# #                                   Each result typically contains 'title', 'url', and 'content'.
+# #                                   If include_answer is True, it will also include the 'answer' field
+# #                                   in the top-level response.
+# #         """
+# #         try:
+# #             response = self.tavily_client.search(
+# #                 query=query,
+# #                 max_results=max_results,
+# #                 include_answer=include_answer,
+# #                 include_raw_content=include_raw_content,
+# #                 search_depth=search_depth,
+# #                 topic=topic,
+# #                 time_range=time_range,
+# #                 include_domains=include_domains,
+# #                 exclude_domains=exclude_domains
+# #             )
+
+# #             return response
+
+# #         except Exception as e:
+# #             return [{"error": f"Tavily Search Error: {e}"}]
+        
+
+
+# # class TavilySearchTool(Tool):
+# #     name = "tavily_search"
+# #     description = """
+# #     A tool to perform comprehensive web searches using the Tavily Search API.
+# #     It can retrieve real-time, accurate, and factual information from across the web.
+# #     Useful for answering questions, gathering information, and reducing hallucinations."""
+    
+# #     inputs = {
+# #         "query": {"type": "string", "description": "The search query string."},
+# #         "max_results": {"type": "integer", "description": "Maximum number of results.", "default": 5, "nullable": False},
+# #         "topic": {"type": "string", "description": "Search category: 'general', 'news', or 'finance'.", "default": "general", "nullable": False},
+# #         "include_answer": {"type": "boolean", "description": "Whether to include answer in results.", "default": False, "nullable": False},
+# #         "include_raw_content": {"type": "boolean", "description": "Whether to include HTML content.", "default": False, "nullable": False},
+# #         "search_depth": {"type": "string", "description": "Search depth: 'basic' or 'advanced'.", "default": "basic", "nullable": False},
+# #         "time_range": {"type": "string", "description": "Filter by time range.", "default": None, "nullable": True},
+# #         "include_domains": {"type": "array", "description": "List of domains to include.", "default": None, "nullable": True},
+# #         "exclude_domains": {"type": "array", "description": "List of domains to exclude.", "default": None, "nullable": True},
+# #     }
+
+# #     output_type = "any"
+
+# #     def __init__(self, api_key:Optional[str] = None, **kwargs):
+# #         super().__init__(**kwargs)
+# #         if api_key is None:
+# #             api_key = os.getenv("TRAVILY_SECRET_KEY")
+# #         if not api_key:
+# #             raise ValueError(
+# #                 "Tavily API key not found. Please set the TAVILY_API_KEY environment variable "
+# #                 "or pass it directly to TavilySearchTool(api_key='YOUR_API_KEY')."
+# #             )
+# #         self.tavily_client = TavilyClient(api_key=api_key)
+# #         print('TavilyClient is initiated')
+
+# #     def forward(self, query: str, 
+# #                 max_results: int = 5, 
+# #                 include_answer: bool = False, 
+# #                 include_raw_content: bool = False,
+# #                 search_depth: str = "basic",
+# #                 topic: str = "general",
+# #                 time_range: Optional[str] = None,
+# #                 include_domains: Optional[List[str]] = None,
+# #                 exclude_domains: Optional[List[str]] = None) -> List[Dict[str, Any]]:
+# #         """
+# #         Executes a search query using the Tavily API.
+
+# #         Args:
+# #             query (str): The search query.
+# #             max_results (int): Maximum number of search results to return.
+# #             include_answer (bool): Whether to include a direct answer synthesized from the search results.
+# #             include_raw_content (bool): Whether to include the raw HTML content of the searched pages.
+# #             search_depth (str): Depth of the search, either 'basic' or 'advanced'.
+# #             topic (str): Category of the search. Can be 'general', 'news', or 'finance'.
+# #             time_range (Optional[str]): The time range back from the current date to filter results - 'day', 'week', 'month', or 'year'.
+# #             include_domains (Optional[List[str]]): List of domains to specifically include.
+# #             exclude_domains (Optional[List[str]]): List of domains to specifically exclude.
+
+# #         Returns:
+# #             List[Dict[str, Any]]: A list of dictionaries, where each dictionary represents a search result.
+# #                                   Each result typically contains 'title', 'url', and 'content'.
+# #                                   If include_answer is True, it will also include the 'answer' field
+# #                                   in the top-level response.
+# #         """
+# #         try:
+# #             response = self.tavily_client.search(
+# #                 query=query,
+# #                 max_results=max_results,
+# #                 include_answer=include_answer,
+# #                 include_raw_content=include_raw_content,
+# #                 search_depth=search_depth,
+# #                 topic=topic,
+# #                 time_range=time_range,
+# #                 include_domains=include_domains,
+# #                 exclude_domains=exclude_domains
+# #             )
+
+# #             return response
+
+# #         except Exception as e:
+# #             return [{"error": f"Tavily Search Error: {e}"}]
+
+
+
+# class WebSearchTool(Tool):
+#     name = "web_search"
+#     description = """
+#     A tool to perform comprehensive web searches using the Tavily Search API.
+#     It can retrieve real-time, accurate, and factual information from across the web.
+#     Useful for answering questions, gathering information, and reducing hallucinations."""
+    
+#     inputs = {
+#         "query": {"type": "string",
+#                   "description": "The search query string."
+#                  }
+#      }
+
+#     output_type = "any"
+
+#     def __init__(self, api_key:Optional[str] = None, **kwargs):
+#         super().__init__(**kwargs)
+#         if api_key is None:
+#             api_key = os.getenv("TRAVILY_SECRET_KEY")
+#         if not api_key:
+#             raise ValueError(
+#                 "Tavily API key not found. Please set the TAVILY_API_KEY environment variable "
+#                 "or pass it directly to TavilySearchTool(api_key='YOUR_API_KEY')."
+#             )
+#         self.tavily_client = TavilyClient(api_key=api_key)
+#         print('TavilyClient is initiated')
+
+#     def forward(self, 
+#                 query: str) -> Any:
+#         try:
+#             print(f'trying travily search...')
+#             response = self.tavily_client.search(
+#                 query=query,
+#                 max_results=5,
+#                 search_depth="advanced",
+#                 topic='general',
+#                 )
+#             #print(response)
+#             if "results" in response:
+#                 return response.get("results", [])
+#             else:
+#                 return('try shorter query or different approach.')
+
+#         except Exception as e:
+#             return [{"error": f"Tavily Search Error: {e}"}]

Tools/transcriber.py

@@ -0,0 +1,50 @@
+from smolagents import tool
+import torch
+from transformers import AutoModelForSpeechSeq2Seq, AutoProcessor, pipeline
+
+@tool
+def transcriber(audio_file_path:str) -> str:
+    """
+    This tool transcribes an audio file and returns the generated transcription.
+
+    Args:
+        audio_file_path: Path of a local audio file that needs to be transcribed.
+    """
+
+    try:
+        device = "cuda:0" if torch.cuda.is_available() else "cpu"
+        torch_dtype = torch.float16 if torch.cuda.is_available() else torch.float32
+
+        model_id = "openai/whisper-small"
+
+        model = AutoModelForSpeechSeq2Seq.from_pretrained(
+            model_id, torch_dtype=torch_dtype, low_cpu_mem_usage=True, use_safetensors=True
+        )
+        model.to(device)
+
+        processor = AutoProcessor.from_pretrained(model_id)
+
+        pipe = pipeline(
+            "automatic-speech-recognition",
+            model=model,
+            tokenizer=processor.tokenizer,
+            feature_extractor=processor.feature_extractor,
+            torch_dtype=torch_dtype,
+            device=device,
+            return_timestamps=True
+        )
+
+        result = pipe(audio_file_path)
+
+        import gc
+        
+        # After inference
+        del pipe
+        del model
+        del processor
+        gc.collect()                 # Force Python garbage collection
+        torch.cuda.empty_cache()     # Clear cached memory
+        
+        return result["text"]
+    except Exception as e:
+        return f'error occured: {e}'

Tools/visual_reasoner.py

@@ -0,0 +1,148 @@
+from transformers import Qwen2_5_VLForConditionalGeneration, AutoTokenizer, AutoProcessor
+from qwen_vl_utils import process_vision_info
+from smolagents import tool
+import torch
+
+
+@tool
+def video_reasoner(file_path : str, query : str) -> str: 
+    """
+    This tool performs requested visual reasoning task on the provided video and returns the generated output.
+
+    Args:
+        file_path: Path of a local video file on which visual reasoning is to be done.
+        query: visual reasoning that is to be done.
+    """
+    try:
+        # default: Load the model on the available device(s)
+        model = Qwen2_5_VLForConditionalGeneration.from_pretrained(
+            "Qwen/Qwen2.5-VL-7B-Instruct", torch_dtype="auto", device_map="auto"
+        )
+
+        # default processer
+        processor = AutoProcessor.from_pretrained("Qwen/Qwen2.5-VL-7B-Instruct")
+
+        messages = [
+        {
+        "role": "user",
+        "content": [
+            {
+                "type": "video",
+                "video": file_path,
+                "max_pixels": 360 * 360,
+                "fps": 0.3,
+            },
+            {"type": "text", "text": f"{query}\n\nAdditional instruction: Treat the two types of penguins as distinct species e.g. Adelie and Emperor Penguins are considered two different species of birds."},
+            ],
+        }      
+        ]
+
+        # Preparation for inference
+        text = processor.apply_chat_template(
+            messages, tokenize=False, add_generation_prompt=True
+        )
+        image_inputs, video_inputs = process_vision_info(messages)
+        inputs = processor(
+            text=[text],
+            images=image_inputs,
+            videos=video_inputs,
+            padding=True,
+            return_tensors="pt",
+        )
+        inputs = inputs.to("cuda")
+
+        # Inference: Generation of the output
+        generated_ids = model.generate(**inputs, max_new_tokens=512)
+        generated_ids_trimmed = [
+            out_ids[len(in_ids) :] for in_ids, out_ids in zip(inputs.input_ids, generated_ids)
+        ]
+        output_text = processor.batch_decode(
+            generated_ids_trimmed, skip_special_tokens=True, clean_up_tokenization_spaces=False
+        )
+
+        import gc
+        
+        # After inference
+        del image_inputs
+        del video_inputs
+        del inputs
+        del model
+        del processor
+        gc.collect()                 # Force Python garbage collection
+        torch.cuda.empty_cache()     # Clear cached memory
+        
+        return output_text
+    
+    except Exception as e:
+        return f'error occured: {e}'
+    
+
+@tool
+def image_reasoner(file_path : str, query : str) -> str: 
+    """
+    This tool performs requested visual reasoning task on the provided image and returns the generated output.
+
+    Args:
+        file_path: Path of a local image file on which visual reasoning is to be done.
+        query: visual reasoning that is to be done.
+    """
+    try:
+        # default: Load the model on the available device(s)
+        model = Qwen2_5_VLForConditionalGeneration.from_pretrained(
+            "Qwen/Qwen2.5-VL-7B-Instruct", torch_dtype="auto", device_map="auto"
+        )
+
+        # default processer
+        processor = AutoProcessor.from_pretrained("Qwen/Qwen2.5-VL-7B-Instruct")
+
+        messages = [
+        {
+        "role": "user",
+        "content": [
+            {
+                "type": "image",
+                "image": file_path,
+            },
+            {"type": "text", "text": f"{query}\n\nAdditional instruction: Review your answer for correctness."},
+            ],
+        }      
+        ]
+
+        # Preparation for inference
+        text = processor.apply_chat_template(
+            messages, tokenize=False, add_generation_prompt=True
+        )
+        image_inputs, video_inputs = process_vision_info(messages)
+        inputs = processor(
+            text=[text],
+            images=image_inputs,
+            videos=video_inputs,
+            padding=True,
+            return_tensors="pt",
+        )
+        inputs = inputs.to("cuda")
+
+        # Inference: Generation of the output
+        generated_ids = model.generate(**inputs, max_new_tokens=512)
+        generated_ids_trimmed = [
+            out_ids[len(in_ids) :] for in_ids, out_ids in zip(inputs.input_ids, generated_ids)
+        ]
+        output_text = processor.batch_decode(
+            generated_ids_trimmed, skip_special_tokens=True, clean_up_tokenization_spaces=False
+        )
+
+        import gc
+        
+        # After inference
+        del image_inputs
+        del video_inputs
+        del inputs
+        del model
+        del processor
+        gc.collect()                 # Force Python garbage collection
+        torch.cuda.empty_cache()     # Clear cached memory
+        
+        return output_text
+    
+    except Exception as e:
+        return f'error occured: {e}'

Tools/ytdownload.py

@@ -0,0 +1,69 @@
+#!apt update && apt install -y ffmpeg
+import yt_dlp
+from smolagents import tool
+import os
+
+@tool
+def youtube_audio_downloader(video_url:str) ->str :
+    """
+    This tool downloads the audio of a Youtube video given a Youtube Url and returns the downlaoded path.
+
+    Args:
+        video_url: URL of the Youtube video.
+    """
+    ydl_opts = {
+    'format': 'bestaudio/best',
+    'outtmpl': './downloads/%(title)s.%(ext)s',
+    'postprocessors': [{
+        'key': 'FFmpegExtractAudio',
+        'preferredcodec': 'mp3',
+        'preferredquality': '192',
+        }],
+    }
+
+    try:
+        os.makedirs("./downloads", exist_ok=True)
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            info = ydl.extract_info(video_url, download=True)
+            title = info['title']
+            filepath = f"./downloads/{title}.mp3"
+        return filepath
+    except Exception as e:
+        return f'error occured: {e}'
+    
+
+@tool
+def youtube_video_downloader(video_url:str) ->str :
+    """
+    This tool downloads the Youtube video given a Youtube Url and returns the downlaoded path.
+
+    Args:
+        video_url: URL of the Youtube video.
+    """
+    ydl_opts = {
+        #'format': 'bestvideo+bestaudio/best',
+        'format': 'bestvideo[height<=720][vcodec^=avc1]+bestaudio[acodec^=mp4a]/best[ext=mp4]',
+        'outtmpl': 'downloads/%(title)s.%(ext)s',
+        'merge_output_format': 'mp4',
+
+        'writesubtitles': True,              # Download user-provided subtitles
+        'subtitleslangs': ['en'],            # Use your preferred language(s)
+        'embedsubtitles': True,              # Enable embedding into video
+
+        'postprocessors': [
+            {'key': 'FFmpegEmbedSubtitle'}   # Actually does the embedding
+        ]
+    }
+
+    try:
+        os.makedirs("./downloads", exist_ok=True)
+
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            info = ydl.extract_info(video_url, download=True)
+            title = info.get('title')
+            filepath = f"./downloads/{title}.mp4"
+            if not os.path.exists(filepath):
+                return f"Error: File not found after download: {filepath}"
+        return filepath
+    except Exception as e:
+        return f'error occured while downloading the video: {e}'