Created distincts file for langchain and smolagent tools

tools_langchain_browser.py

@@ -0,0 +1,80 @@
+from bs4 import BeautifulSoup
+
+from langchain_community.document_loaders import WikipediaLoader
+from langchain_community.document_loaders import ArxivLoader
+from langchain_community.tools.tavily_search import TavilySearchResults
+
+from langchain_core.tools import tool
+# from smolagents import tool
+
+from playwright.sync_api import sync_playwright
+
+
+@tool
+def wiki_search(query: str) -> str:
+    """Search Wikipedia for a query and return maximum 2 results.
+    Args:
+        query: The search query."""
+    search_docs = WikipediaLoader(query=query, load_max_docs=2).load()
+    formatted_search_docs = "\n\n---\n\n".join(
+        [
+            f'<Document source="{doc.metadata["source"]}" page="{doc.metadata.get("page", "")}"/>\n{doc.page_content}\n</Document>'
+            for doc in search_docs
+        ]
+    )
+    return {"wiki_results": formatted_search_docs}
+
+
+@tool
+def web_search(query: str) -> str:
+    """Search Tavily for a query and return maximum 3 results.
+    Args:
+        query: The search query."""
+    search_docs = TavilySearchResults(max_results=3).invoke(query=query)
+    formatted_search_docs = "\n\n---\n\n".join(
+        [
+            f'<Document source="{doc.metadata["source"]}" page="{doc.metadata.get("page", "")}"/>\n{doc.page_content}\n</Document>'
+            for doc in search_docs
+        ]
+    )
+    return {"web_results": formatted_search_docs}
+
+
+@tool
+def arxiv_search(query: str) -> str:
+    """Search Arxiv for a query and return maximum 3 result.
+    Args:
+        query: The search query."""
+    search_docs = ArxivLoader(query=query, load_max_docs=3).load()
+    formatted_search_docs = "\n\n---\n\n".join(
+        [
+            f'<Document source="{doc.metadata["source"]}" page="{doc.metadata.get("page", "")}"/>\n{doc.page_content[:1000]}\n</Document>'
+            for doc in search_docs
+        ]
+    )
+    return {"arxiv_results": formatted_search_docs}
+
+
+@tool
+def website_scrape(url: str, question: str) -> str:
+    """Scrapes a website and returns the text.
+    Args:
+        url: the URL to the website to scrape.
+        question: the question to answer when searching for the answer in the website.
+    Returns:
+        str: The text of the website.
+    """
+
+    with sync_playwright() as p:
+        browser = p.chromium.launch(headless=True)
+        page = browser.new_page()
+        page.goto(url)
+        html_content = page.content()
+        browser.close()
+
+    soup = BeautifulSoup(html_content, "html.parser")
+
+    # Extract text from the website
+    text = soup.get_text()
+
+    return text

tools_smolagent_audio.py

@@ -0,0 +1,44 @@
+# Audio Transcription Tool
+import os
+
+from google import genai
+from google.genai import types
+from smolagents import tool
+
+
+@tool
+def transcribe_audio(audio_file_path: str, mime_type: str) -> str:
+    """Transcribes an audio file using Gemini's audio capabilities.
+    Args:
+        audio_file_path (str): the path to the audio file to transcribe.
+        mime_type (str): the mime type of the audio file.
+    Returns:
+        str: The transcript of the audio file.
+    """
+    try:
+        # Initialize the model
+        client = genai.Client(api_key=os.getenv("GEMINI_KEY"))
+        model = "models/gemini-1.5-flash-8b"
+
+        # Read and encode the audio file
+        with open(audio_file_path, "rb") as audio_file:
+            audio_data = audio_file.read()
+
+        # Create the content with audio data
+        contents = types.Content(
+            parts=[
+                types.Part.from_bytes(
+                    data=audio_data,
+                    mime_type=mime_type,
+                ),
+                types.Part(text="Please transcribe this audio file."),
+            ]
+        )
+
+        # Generate transcription
+        response = client.models.generate_content(
+            model=model, contents=contents
+        )
+        return response.text
+    except Exception as e:
+        return f"Error transcribing audio: {str(e)}"

tools_browser.py → tools_smolagent_browser.py

@@ -4,8 +4,7 @@ from langchain_community.document_loaders import WikipediaLoader
 from langchain_community.document_loaders import ArxivLoader
 from langchain_community.tools.tavily_search import TavilySearchResults
 
-from langchain_core.tools import tool
-# from smolagents import tool
+from smolagents import tool
 
 from playwright.sync_api import sync_playwright
 
@@ -56,7 +55,7 @@ def arxiv_search(query: str) -> str:
 
 
 @tool
-def website_scrape(url: str, question: str) -> str:
+def website_scrape(url: str) -> str:
     """Scrapes a website and returns the text.
     Args:
         url (str): the URL to the website to scrape.

tools_smolagent_code.py

@@ -0,0 +1,410 @@
+import base64
+import contextlib
+import io
+import os
+import subprocess
+import sqlite3
+import tempfile
+import traceback
+import uuid
+
+from smolagents import tool
+from typing import Dict, Any
+
+# Imports for Python Code interpreter
+import numpy as np
+import pandas as pd
+import matplotlib.pyplot as plt
+from PIL import Image
+
+
+class CodeInterpreter:
+    def __init__(
+        self,
+        allowed_modules=None,
+        max_execution_time=30,
+        working_directory=None,
+    ):
+        """Initialize the code interpreter with safety measures."""
+        self.allowed_modules = allowed_modules or [
+            "numpy",
+            "pandas",
+            "matplotlib",
+            "scipy",
+            "sklearn",
+            "math",
+            "random",
+            "statistics",
+            "datetime",
+            "collections",
+            "itertools",
+            "functools",
+            "operator",
+            "re",
+            "json",
+            "sympy",
+            "networkx",
+            "nltk",
+            "PIL",
+            "pytesseract",
+            "cmath",
+            "uuid",
+            "tempfile",
+            "requests",
+            "urllib",
+        ]
+        self.max_execution_time = max_execution_time
+        self.working_directory = working_directory or os.path.join(os.getcwd())
+        if not os.path.exists(self.working_directory):
+            os.makedirs(self.working_directory)
+
+        self.globals = {
+            "__builtins__": __builtins__,
+            "np": np,
+            "pd": pd,
+            "plt": plt,
+            "Image": Image,
+        }
+        self.temp_sqlite_db = os.path.join(
+            tempfile.gettempdir(), "code_exec.db"
+        )
+
+    def execute_code(
+        self, code: str, language: str = "python"
+    ) -> Dict[str, Any]:
+        """Execute the provided code in the selected programming language."""
+        language = language.lower()
+        execution_id = str(uuid.uuid4())
+
+        result = {
+            "execution_id": execution_id,
+            "status": "error",
+            "stdout": "",
+            "stderr": "",
+            "result": None,
+            "plots": [],
+            "dataframes": [],
+        }
+
+        try:
+            if language == "python":
+                return self._execute_python(code, execution_id)
+            elif language == "bash":
+                return self._execute_bash(code, execution_id)
+            elif language == "sql":
+                return self._execute_sql(code, execution_id)
+            elif language == "c":
+                return self._execute_c(code, execution_id)
+            elif language == "java":
+                return self._execute_java(code, execution_id)
+            else:
+                result["stderr"] = f"Unsupported language: {language}"
+        except Exception as e:
+            result["stderr"] = str(e)
+
+        return result
+
+    def _execute_python(self, code: str, execution_id: str) -> dict:
+        output_buffer = io.StringIO()
+        error_buffer = io.StringIO()
+        result = {
+            "execution_id": execution_id,
+            "status": "error",
+            "stdout": "",
+            "stderr": "",
+            "result": None,
+            "plots": [],
+            "dataframes": [],
+        }
+
+        try:
+            exec_dir = os.path.join(self.working_directory, execution_id)
+            os.makedirs(exec_dir, exist_ok=True)
+            plt.switch_backend("Agg")
+
+            with (
+                contextlib.redirect_stdout(output_buffer),
+                contextlib.redirect_stderr(error_buffer),
+            ):
+                exec_result = exec(code, self.globals)
+
+                if plt.get_fignums():
+                    for i, fig_num in enumerate(plt.get_fignums()):
+                        fig = plt.figure(fig_num)
+                        img_path = os.path.join(exec_dir, f"plot_{i}.png")
+                        fig.savefig(img_path)
+                        with open(img_path, "rb") as img_file:
+                            img_data = base64.b64encode(
+                                img_file.read()
+                            ).decode("utf-8")
+                            result["plots"].append(
+                                {"figure_number": fig_num, "data": img_data}
+                            )
+
+                for var_name, var_value in self.globals.items():
+                    if (
+                        isinstance(var_value, pd.DataFrame)
+                        and len(var_value) > 0
+                    ):
+                        result["dataframes"].append(
+                            {
+                                "name": var_name,
+                                "head": var_value.head().to_dict(),
+                                "shape": var_value.shape,
+                                "dtypes": str(var_value.dtypes),
+                            }
+                        )
+
+            result["status"] = "success"
+            result["stdout"] = output_buffer.getvalue()
+            result["result"] = exec_result
+
+        except Exception as e:
+            result["status"] = "error"
+            result["stderr"] = (
+                f"{error_buffer.getvalue()}\n{traceback.format_exc()}"
+            )
+
+        return result
+
+    def _execute_bash(self, code: str, execution_id: str) -> dict:
+        try:
+            completed = subprocess.run(
+                code,
+                shell=True,
+                capture_output=True,
+                text=True,
+                timeout=self.max_execution_time,
+            )
+            return {
+                "execution_id": execution_id,
+                "status": "success" if completed.returncode == 0 else "error",
+                "stdout": completed.stdout,
+                "stderr": completed.stderr,
+                "result": None,
+                "plots": [],
+                "dataframes": [],
+            }
+        except subprocess.TimeoutExpired:
+            return {
+                "execution_id": execution_id,
+                "status": "error",
+                "stdout": "",
+                "stderr": "Execution timed out.",
+                "result": None,
+                "plots": [],
+                "dataframes": [],
+            }
+
+    def _execute_sql(self, code: str, execution_id: str) -> dict:
+        result = {
+            "execution_id": execution_id,
+            "status": "error",
+            "stdout": "",
+            "stderr": "",
+            "result": None,
+            "plots": [],
+            "dataframes": [],
+        }
+        try:
+            conn = sqlite3.connect(self.temp_sqlite_db)
+            cur = conn.cursor()
+            cur.execute(code)
+            if code.strip().lower().startswith("select"):
+                columns = [description[0] for description in cur.description]
+                rows = cur.fetchall()
+                df = pd.DataFrame(rows, columns=columns)
+                result["dataframes"].append(
+                    {
+                        "name": "query_result",
+                        "head": df.head().to_dict(),
+                        "shape": df.shape,
+                        "dtypes": str(df.dtypes),
+                    }
+                )
+            else:
+                conn.commit()
+
+            result["status"] = "success"
+            result["stdout"] = "Query executed successfully."
+
+        except Exception as e:
+            result["stderr"] = str(e)
+        finally:
+            conn.close()
+
+        return result
+
+    def _execute_c(self, code: str, execution_id: str) -> dict:
+        temp_dir = tempfile.mkdtemp()
+        source_path = os.path.join(temp_dir, "program.c")
+        binary_path = os.path.join(temp_dir, "program")
+
+        try:
+            with open(source_path, "w") as f:
+                f.write(code)
+
+            compile_proc = subprocess.run(
+                ["gcc", source_path, "-o", binary_path],
+                capture_output=True,
+                text=True,
+                timeout=self.max_execution_time,
+            )
+            if compile_proc.returncode != 0:
+                return {
+                    "execution_id": execution_id,
+                    "status": "error",
+                    "stdout": compile_proc.stdout,
+                    "stderr": compile_proc.stderr,
+                    "result": None,
+                    "plots": [],
+                    "dataframes": [],
+                }
+
+            run_proc = subprocess.run(
+                [binary_path],
+                capture_output=True,
+                text=True,
+                timeout=self.max_execution_time,
+            )
+            return {
+                "execution_id": execution_id,
+                "status": "success" if run_proc.returncode == 0 else "error",
+                "stdout": run_proc.stdout,
+                "stderr": run_proc.stderr,
+                "result": None,
+                "plots": [],
+                "dataframes": [],
+            }
+        except Exception as e:
+            return {
+                "execution_id": execution_id,
+                "status": "error",
+                "stdout": "",
+                "stderr": str(e),
+                "result": None,
+                "plots": [],
+                "dataframes": [],
+            }
+
+    def _execute_java(self, code: str, execution_id: str) -> dict:
+        temp_dir = tempfile.mkdtemp()
+        source_path = os.path.join(temp_dir, "Main.java")
+
+        try:
+            with open(source_path, "w") as f:
+                f.write(code)
+
+            compile_proc = subprocess.run(
+                ["javac", source_path],
+                capture_output=True,
+                text=True,
+                timeout=self.max_execution_time,
+            )
+            if compile_proc.returncode != 0:
+                return {
+                    "execution_id": execution_id,
+                    "status": "error",
+                    "stdout": compile_proc.stdout,
+                    "stderr": compile_proc.stderr,
+                    "result": None,
+                    "plots": [],
+                    "dataframes": [],
+                }
+
+            run_proc = subprocess.run(
+                ["java", "-cp", temp_dir, "Main"],
+                capture_output=True,
+                text=True,
+                timeout=self.max_execution_time,
+            )
+            return {
+                "execution_id": execution_id,
+                "status": "success" if run_proc.returncode == 0 else "error",
+                "stdout": run_proc.stdout,
+                "stderr": run_proc.stderr,
+                "result": None,
+                "plots": [],
+                "dataframes": [],
+            }
+        except Exception as e:
+            return {
+                "execution_id": execution_id,
+                "status": "error",
+                "stdout": "",
+                "stderr": str(e),
+                "result": None,
+                "plots": [],
+                "dataframes": [],
+            }
+
+
+@tool
+def execute_code_multilang(code: str, language: str = "python") -> str:
+    """Execute code in multiple languages (Python, Bash, SQL, C, Java) and return results.
+    Args:
+        code (str): The source code to execute.
+        language (str): The language of the code. Supported: "python", "bash", "sql", "c", "java".
+    Returns:
+        A string summarizing the execution results (stdout, stderr, errors, plots, dataframes if any).
+    """
+    interpreter_instance = CodeInterpreter()
+    supported_languages = ["python", "bash", "sql", "c", "java"]
+    language = language.lower()
+
+    if language not in supported_languages:
+        return f"❌ Unsupported language: {language}. Supported languages are: {', '.join(supported_languages)}"
+
+    result = interpreter_instance.execute_code(code, language=language)
+
+    response = []
+
+    if result["status"] == "success":
+        response.append(
+            f"✅ Code executed successfully in **{language.upper()}**"
+        )
+
+        if result.get("stdout"):
+            response.append(
+                "\n**Standard Output:**\n```\n"
+                + result["stdout"].strip()
+                + "\n```"
+            )
+
+        if result.get("stderr"):
+            response.append(
+                "\n**Standard Error (if any):**\n```\n"
+                + result["stderr"].strip()
+                + "\n```"
+            )
+
+        if result.get("result") is not None:
+            response.append(
+                "\n**Execution Result:**\n```\n"
+                + str(result["result"]).strip()
+                + "\n```"
+            )
+
+        if result.get("dataframes"):
+            for df_info in result["dataframes"]:
+                response.append(
+                    f"\n**DataFrame `{df_info['name']}` (Shape: {df_info['shape']})**"
+                )
+                df_preview = pd.DataFrame(df_info["head"])
+                response.append(
+                    "First 5 rows:\n```\n" + str(df_preview) + "\n```"
+                )
+
+        if result.get("plots"):
+            response.append(
+                f"\n**Generated {len(result['plots'])} plot(s)** (Image data returned separately)"
+            )
+
+    else:
+        response.append(f"❌ Code execution failed in **{language.upper()}**")
+        if result.get("stderr"):
+            response.append(
+                "\n**Error Log:**\n```\n" + result["stderr"].strip() + "\n```"
+            )
+
+    return "\n".join(response)

tools_smolagent_doc.py

@@ -0,0 +1,275 @@
+import os
+import pandas as pd
+import re
+import requests
+import tempfile
+import uuid
+
+from smolagents import tool
+
+from google import genai
+from google.genai import types
+
+# from smolagents import tool
+from typing import Optional, Dict, Union
+
+
+@tool
+def save_and_read_file(content: str, filename: Optional[str] = None) -> str:
+    """
+    Save content to a temporary file and return the path.
+    Useful for processing files from the GAIA API.
+
+    Args:
+        content: The content to save to the file
+        filename: Optional filename, will generate a random name if not provided
+
+    Returns:
+        Path to the saved file
+    """
+    temp_dir = tempfile.gettempdir()
+    if filename is None:
+        temp_file = tempfile.NamedTemporaryFile(delete=False)
+        filepath = temp_file.name
+    else:
+        filepath = os.path.join(temp_dir, filename)
+
+    # Write content to the file
+    with open(filepath, "w") as f:
+        f.write(content)
+
+    return f"File saved to {filepath}. You can read this file to process its contents."
+
+
+# File Download Tool
+@tool
+def download_file_from_url(
+    url: str, directory: str
+) -> Dict[str, Union[str, None]]:
+    """Downloads a file from a URL and saves it to a directory.
+    Args:
+        url (str): the URL to download the file from.
+        directory (str): the directory to save the file to.
+    Returns:
+        Dict[str, Union[str, None]]: A dictionary containing the file type and path.
+    """
+
+    try:
+        response = requests.get(url, stream=True, timeout=10)
+        response.raise_for_status()
+
+        content_type = response.headers.get("content-type", "").lower()
+
+        # Try to get filename from headers
+        filename = None
+        cd = response.headers.get("content-disposition", "")
+        match = re.search(r"filename\*=UTF-8\'\'(.+)", cd) or re.search(
+            r'filename="?([^"]+)"?', cd
+        )
+        if match:
+            filename = match.group(1)
+
+        # If not in headers, try URL
+        if not filename:
+            filename = os.path.basename(url.split("?")[0])
+
+        # Fallback to generated filename
+        if not filename:
+            extension = {
+                "image/jpeg": ".jpg",
+                "image/png": ".png",
+                "image/gif": ".gif",
+                "audio/wav": ".wav",
+                "audio/mpeg": ".mp3",
+                "video/mp4": ".mp4",
+                "text/plain": ".txt",
+                "text/csv": ".csv",
+                "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": ".xlsx",
+                "application/vnd.ms-excel": ".xls",
+                "application/octet-stream": ".bin",
+            }.get(content_type, ".bin")
+            filename = f"downloaded_{uuid.uuid4().hex[:8]}{extension}"
+
+        os.makedirs(directory, exist_ok=True)
+        file_path = os.path.join(directory, filename)
+
+        with open(file_path, "wb") as f:
+            for chunk in response.iter_content(chunk_size=8192):
+                f.write(chunk)
+
+        # shutil.copy(file_path, os.getcwd())
+
+        if os.path.exists(file_path) and os.path.getsize(file_path) > 0:
+            return {"type": content_type, "path": file_path}
+        else:
+            return {
+                "type": "error",
+                "path": None,
+                "error": "Failed to save file",
+            }
+
+    except Exception as e:
+        return {
+            "type": "error",
+            "path": None,
+            "error": f"Error downloading file: {str(e)}",
+        }
+
+
+@tool
+def extract_text_from_image(image_path: str) -> str:
+    """
+    Extract text from an image using pytesseract (if available).
+
+    Args:
+        image_path: Path to the image file
+
+    Returns:
+        Extracted text or error message
+    """
+    try:
+        # Try to import pytesseract
+        import pytesseract
+        from PIL import Image
+
+        # Open the image
+        image = Image.open(image_path)
+
+        # Extract text
+        text = pytesseract.image_to_string(image)
+
+        return f"Extracted text from image:\n\n{text}"
+    except ImportError:
+        return "Error: pytesseract is not installed. Please install it with 'pip install pytesseract' and ensure Tesseract OCR is installed on your system."
+    except Exception as e:
+        return f"Error extracting text from image: {str(e)}"
+
+
+# CSV Analysis Tool
+@tool
+def analyze_csv_file(file_path: str, query: str) -> str:
+    """Analyzes a CSV file and answers questions about its contents using Gemini.
+    Args:
+        file_path (str): the path to the CSV file to analyze.
+        query (str): the question to answer about the CSV file.
+    Returns:
+        str: The result of the analysis.
+    """
+    try:
+        # Read the CSV file
+        df = pd.read_csv(file_path)
+
+        # Initialize Gemini
+        client = genai.Client(api_key=os.getenv("GEMINI_KEY"))
+        model = "models/gemini-1.5-flash-8b"
+
+        # Convert DataFrame to a string representation
+        df_str = df.to_string()
+
+        # Create a prompt for Gemini
+        prompt = f"""Analyze this CSV data and provide insights:
+Dimensions: {len(df)} rows × {len(df.columns)} columns
+Data:
+{df_str}
+Please provide:
+1. A summary of the data structure and content
+2. Key patterns and insights
+3. Potential data quality issues
+4. Suggestions for analysis
+User Query: {query}
+Please format your response in a clear, structured way with sections and bullet points."""
+
+        # Get analysis from Gemini
+        response = client.models.generate_content(
+            model=model,
+            contents=types.Content(
+                parts=[
+                    types.Part(text=df_str),
+                    types.Part(text=prompt),
+                ]
+            ),
+        )
+
+        result = f"CSV file loaded with {len(df)} rows and {len(df.columns)} columns.\n\n"
+        result += response.text
+
+        return result
+    except Exception as e:
+        return f"Error analyzing CSV file: {str(e)}"
+
+
+# Excel Analysis Tool
+@tool
+def analyze_excel_file(file_path: str, query: str) -> str:
+    """Analyzes an Excel file and answers questions about its contents using Gemini.
+    Args:
+        file_path (str): the path to the Excel file to analyze.
+        query (str): the question to answer about the Excel file.
+    Returns:
+        str: The result of the analysis.
+    """
+    try:
+        # Read all sheets from the Excel file
+        excel_file = pd.ExcelFile(file_path)
+        sheet_names = excel_file.sheet_names
+
+        # Initialize Gemini
+        client = genai.Client(api_key=os.getenv("GEMINI_KEY"))
+        model = "models/gemini-1.5-flash-8b"
+
+        result = f"Excel file loaded with {len(sheet_names)} sheets: {', '.join(sheet_names)}\n\n"
+
+        # Analyze each sheet
+        for sheet_name in sheet_names:
+            df = pd.read_excel(file_path, sheet_name=sheet_name)
+
+            # Convert DataFrame to a string representation
+            df_str = df.to_string()
+
+            # Create a prompt for Gemini
+            prompt = f"""Analyze this Excel sheet data and provide insights:
+Sheet Name: {sheet_name}
+Dimensions: {len(df)} rows × {len(df.columns)} columns
+Data:
+{df_str}
+Please provide:
+1. A summary of the data structure and content
+2. Key patterns and insights
+3. Potential data quality issues
+4. Suggestions for analysis
+User Query: {query}
+Please format your response in a clear, structured way with sections and bullet points."""
+
+            # Get analysis from Gemini
+            response = client.models.generate_content(
+                model=model,
+                contents=types.Content(
+                    parts=[types.Part(text=df_str), types.Part(text=prompt)]
+                ),
+            )
+
+            result += f"=== Sheet: {sheet_name} ===\n"
+            result += response.text + "\n"
+            result += "=" * 50 + "\n\n"
+
+        return result
+    except Exception as e:
+        return f"Error analyzing Excel file: {str(e)}"
+
+
+@tool
+def read_file(filepath: str) -> str:
+    """Reads the content of a text file.
+    Args:
+        filepath (str): the path to the file to read.
+    Returns:
+        str: The content of the file.
+    """
+    try:
+        with open(filepath, "r", encoding="utf-8") as file:
+            content = file.read()
+        return content
+    except FileNotFoundError:
+        return f"File not found: {filepath}"
+    except IOError as e:
+        return f"Error reading file: {str(e)}"

tools_smolagent_img.py

@@ -0,0 +1,332 @@
+import os
+import io
+import base64
+import numpy as np
+import uuid
+
+from PIL import Image, ImageDraw, ImageFont, ImageEnhance, ImageFilter
+from smolagents import tool
+
+# from smolagents import tool
+from typing import Any, Dict, List, Optional
+
+
+@tool
+def analyze_image(image_base64: str) -> Dict[str, Any]:
+    """
+    Analyze basic properties of an image (size, mode, color analysis, thumbnail preview).
+    Args:
+        image_base64 (str): Base64 encoded image string
+    Returns:
+        Dictionary with analysis result
+    """
+    try:
+        img = decode_image(image_base64)
+        width, height = img.size
+        mode = img.mode
+
+        if mode in ("RGB", "RGBA"):
+            arr = np.array(img)
+            avg_colors = arr.mean(axis=(0, 1))
+            dominant = ["Red", "Green", "Blue"][np.argmax(avg_colors[:3])]
+            brightness = avg_colors.mean()
+            color_analysis = {
+                "average_rgb": avg_colors.tolist(),
+                "brightness": brightness,
+                "dominant_color": dominant,
+            }
+        else:
+            color_analysis = {"note": f"No color analysis for mode {mode}"}
+
+        thumbnail = img.copy()
+        thumbnail.thumbnail((100, 100))
+        thumb_path = save_image(thumbnail, "thumbnails")
+        thumbnail_base64 = encode_image(thumb_path)
+
+        return {
+            "dimensions": (width, height),
+            "mode": mode,
+            "color_analysis": color_analysis,
+            "thumbnail": thumbnail_base64,
+        }
+    except Exception as e:
+        return {"error": str(e)}
+
+
+@tool
+def transform_image(
+    image_base64: str, operation: str, params: Optional[Dict[str, Any]] = None
+) -> Dict[str, Any]:
+    """
+    Apply transformations: resize, rotate, crop, flip, brightness, contrast, blur, sharpen, grayscale.
+    Args:
+        image_base64 (str): Base64 encoded input image
+        operation (str): Transformation operation
+        params (Dict[str, Any], optional): Parameters for the operation
+    Returns:
+        Dictionary with transformed image (base64)
+    """
+    try:
+        img = decode_image(image_base64)
+        params = params or {}
+
+        if operation == "resize":
+            img = img.resize(
+                (
+                    params.get("width", img.width // 2),
+                    params.get("height", img.height // 2),
+                )
+            )
+        elif operation == "rotate":
+            img = img.rotate(params.get("angle", 90), expand=True)
+        elif operation == "crop":
+            img = img.crop(
+                (
+                    params.get("left", 0),
+                    params.get("top", 0),
+                    params.get("right", img.width),
+                    params.get("bottom", img.height),
+                )
+            )
+        elif operation == "flip":
+            if params.get("direction", "horizontal") == "horizontal":
+                img = img.transpose(Image.FLIP_LEFT_RIGHT)
+            else:
+                img = img.transpose(Image.FLIP_TOP_BOTTOM)
+        elif operation == "adjust_brightness":
+            img = ImageEnhance.Brightness(img).enhance(
+                params.get("factor", 1.5)
+            )
+        elif operation == "adjust_contrast":
+            img = ImageEnhance.Contrast(img).enhance(params.get("factor", 1.5))
+        elif operation == "blur":
+            img = img.filter(ImageFilter.GaussianBlur(params.get("radius", 2)))
+        elif operation == "sharpen":
+            img = img.filter(ImageFilter.SHARPEN)
+        elif operation == "grayscale":
+            img = img.convert("L")
+        else:
+            return {"error": f"Unknown operation: {operation}"}
+
+        result_path = save_image(img)
+        result_base64 = encode_image(result_path)
+        return {"transformed_image": result_base64}
+
+    except Exception as e:
+        return {"error": str(e)}
+
+
+@tool
+def draw_on_image(
+    image_base64: str, drawing_type: str, params: Dict[str, Any]
+) -> Dict[str, Any]:
+    """
+    Draw shapes (rectangle, circle, line) or text onto an image.
+    Args:
+        image_base64 (str): Base64 encoded input image
+        drawing_type (str): Drawing type
+        params (Dict[str, Any]): Drawing parameters
+    Returns:
+        Dictionary with result image (base64)
+    """
+    try:
+        img = decode_image(image_base64)
+        draw = ImageDraw.Draw(img)
+        color = params.get("color", "red")
+
+        if drawing_type == "rectangle":
+            draw.rectangle(
+                [
+                    params["left"],
+                    params["top"],
+                    params["right"],
+                    params["bottom"],
+                ],
+                outline=color,
+                width=params.get("width", 2),
+            )
+        elif drawing_type == "circle":
+            x, y, r = params["x"], params["y"], params["radius"]
+            draw.ellipse(
+                (x - r, y - r, x + r, y + r),
+                outline=color,
+                width=params.get("width", 2),
+            )
+        elif drawing_type == "line":
+            draw.line(
+                (
+                    params["start_x"],
+                    params["start_y"],
+                    params["end_x"],
+                    params["end_y"],
+                ),
+                fill=color,
+                width=params.get("width", 2),
+            )
+        elif drawing_type == "text":
+            font_size = params.get("font_size", 20)
+            try:
+                font = ImageFont.truetype("arial.ttf", font_size)
+            except IOError:
+                font = ImageFont.load_default()
+            draw.text(
+                (params["x"], params["y"]),
+                params.get("text", "Text"),
+                fill=color,
+                font=font,
+            )
+        else:
+            return {"error": f"Unknown drawing type: {drawing_type}"}
+
+        result_path = save_image(img)
+        result_base64 = encode_image(result_path)
+        return {"result_image": result_base64}
+
+    except Exception as e:
+        return {"error": str(e)}
+
+
+@tool
+def generate_simple_image(
+    image_type: str,
+    width: int = 500,
+    height: int = 500,
+    params: Optional[Dict[str, Any]] = None,
+) -> Dict[str, Any]:
+    """
+    Generate a simple image (gradient, noise, pattern, chart).
+    Args:
+        image_type (str): Type of image
+        width (int): Width of image,
+        height (int): Height of image,
+        params (Dict[str, Any], optional): Specific parameters
+    Returns:
+        Dictionary with generated image (base64)
+    """
+    try:
+        params = params or {}
+
+        if image_type == "gradient":
+            direction = params.get("direction", "horizontal")
+            start_color = params.get("start_color", (255, 0, 0))
+            end_color = params.get("end_color", (0, 0, 255))
+
+            img = Image.new("RGB", (width, height))
+            draw = ImageDraw.Draw(img)
+
+            if direction == "horizontal":
+                for x in range(width):
+                    r = int(
+                        start_color[0]
+                        + (end_color[0] - start_color[0]) * x / width
+                    )
+                    g = int(
+                        start_color[1]
+                        + (end_color[1] - start_color[1]) * x / width
+                    )
+                    b = int(
+                        start_color[2]
+                        + (end_color[2] - start_color[2]) * x / width
+                    )
+                    draw.line([(x, 0), (x, height)], fill=(r, g, b))
+            else:
+                for y in range(height):
+                    r = int(
+                        start_color[0]
+                        + (end_color[0] - start_color[0]) * y / height
+                    )
+                    g = int(
+                        start_color[1]
+                        + (end_color[1] - start_color[1]) * y / height
+                    )
+                    b = int(
+                        start_color[2]
+                        + (end_color[2] - start_color[2]) * y / height
+                    )
+                    draw.line([(0, y), (width, y)], fill=(r, g, b))
+
+        elif image_type == "noise":
+            noise_array = np.random.randint(
+                0, 256, (height, width, 3), dtype=np.uint8
+            )
+            img = Image.fromarray(noise_array, "RGB")
+
+        else:
+            return {"error": f"Unsupported image_type {image_type}"}
+
+        result_path = save_image(img)
+        result_base64 = encode_image(result_path)
+        return {"generated_image": result_base64}
+
+    except Exception as e:
+        return {"error": str(e)}
+
+
+@tool
+def combine_images(
+    images_base64: List[str],
+    operation: str,
+    params: Optional[Dict[str, Any]] = None,
+) -> Dict[str, Any]:
+    """
+    Combine multiple images (collage, stack, blend).
+    Args:
+        images_base64 (List[str]): List of base64 images
+        operation (str): Combination type
+        params (Dict[str, Any], optional): Parameters
+    Returns:
+        Dictionary with combined image (base64)
+    """
+    try:
+        images = [decode_image(b64) for b64 in images_base64]
+        params = params or {}
+
+        if operation == "stack":
+            direction = params.get("direction", "horizontal")
+            if direction == "horizontal":
+                total_width = sum(img.width for img in images)
+                max_height = max(img.height for img in images)
+                new_img = Image.new("RGB", (total_width, max_height))
+                x = 0
+                for img in images:
+                    new_img.paste(img, (x, 0))
+                    x += img.width
+            else:
+                max_width = max(img.width for img in images)
+                total_height = sum(img.height for img in images)
+                new_img = Image.new("RGB", (max_width, total_height))
+                y = 0
+                for img in images:
+                    new_img.paste(img, (0, y))
+                    y += img.height
+        else:
+            return {"error": f"Unsupported combination operation {operation}"}
+
+        result_path = save_image(new_img)
+        result_base64 = encode_image(result_path)
+        return {"combined_image": result_base64}
+
+    except Exception as e:
+        return {"error": str(e)}
+
+
+# Helper functions for image processing
+def encode_image(image_path: str) -> str:
+    """Convert an image file to base64 string."""
+    with open(image_path, "rb") as image_file:
+        return base64.b64encode(image_file.read()).decode("utf-8")
+
+
+def decode_image(base64_string: str) -> Image.Image:
+    """Convert a base64 string to a PIL Image."""
+    image_data = base64.b64decode(base64_string)
+    return Image.open(io.BytesIO(image_data))
+
+
+def save_image(image: Image.Image, directory: str = "image_outputs") -> str:
+    """Save a PIL Image to disk and return the path."""
+    os.makedirs(directory, exist_ok=True)
+    image_id = str(uuid.uuid4())
+    image_path = os.path.join(directory, f"{image_id}.png")
+    image.save(image_path)
+    return image_path

tools_smolagent_maths.py

@@ -0,0 +1,83 @@
+import cmath
+
+from smolagents import tool
+
+
+@tool
+def multiply(a: int, b: int) -> int:
+    """Multiply two numbers.
+    Args:
+        a: first int
+        b: second int
+    """
+    return a * b
+
+
+@tool
+def add(a: int, b: int) -> int:
+    """Add two numbers.
+
+    Args:
+        a: first int
+        b: second int
+    """
+    return a + b
+
+
+@tool
+def subtract(a: int, b: int) -> int:
+    """Subtract two numbers.
+
+    Args:
+        a: first int
+        b: second int
+    """
+    return a - b
+
+
+@tool
+def divide(a: int, b: int) -> int:
+    """Divide two numbers.
+
+    Args:
+        a: first int
+        b: second int
+    """
+    if b == 0:
+        raise ValueError("Cannot divide by zero.")
+    return a / b
+
+
+@tool
+def modulus(a: int, b: int) -> int:
+    """Get the modulus of two numbers.
+
+    Args:
+        a: first int
+        b: second int
+    """
+    return a % b
+
+
+@tool
+def power(a: float, b: float) -> float:
+    """
+    Get the power of two numbers.
+    Args:
+        a (float): the first number
+        b (float): the second number
+    """
+    return a**b
+
+
+# @tool
+# def square_root(a: float) -> float | complex:
+#     """
+#     Get the square root of a number.
+#     Args:
+#         a (float): the number to get the square root of
+#     """
+#     print(a)
+#     if a >= 0:
+#         return a**0.5
+#     return cmath.sqrt(a)

tools_smolagent_video.py

@@ -0,0 +1,279 @@
+import imageio
+import os
+import re
+import tempfile
+import yt_dlp
+
+from datetime import timedelta
+from google import genai
+from google.genai import types
+
+from smolagents import tool
+from typing import List, Optional
+from youtube_transcript_api import YouTubeTranscriptApi
+
+
+# YouTube Video Review Tool
+@tool
+def review_youtube_video(url: str, question: str) -> str:
+    """Reviews a YouTube video and answers a specific question about that video.
+    Args:
+        url (str): the URL to the YouTube video.
+        question (str): The question you are asking about the video
+    Returns:
+        str: The answer to the question
+    """
+    try:
+        client = genai.Client(api_key=os.getenv("GEMINI_KEY"))
+        model = "models/gemini-1.5-flash-8b"
+
+        response = client.models.generate_content(
+            model=model,
+            contents=types.Content(
+                parts=[
+                    types.Part(file_data=types.FileData(file_uri=url)),
+                    types.Part(text=question),
+                ]
+            ),
+        )
+        return response.text
+    except Exception as e:
+        return f"Error asking {model} about video: {str(e)}"
+
+
+@tool
+def use_vision_model(
+    question: str, image_paths: List[str], mime_type: str
+) -> str:
+    """Use a Vision Model to answer a question about a set of images.
+    Args:
+        question (str): The question you are asking about the images.
+        image_paths (List[str]): The paths to the images to use for the question.
+        mime_type (str): The mime type of the image.
+    Returns:
+        str: The answer to the question
+    """
+    try:
+        client = genai.Client(api_key=os.getenv("GEMINI_KEY"))
+        model = "models/gemini-2.0-flash-001"
+
+        # Prepare the content parts
+        parts = []
+        for image_path in image_paths:
+            with open(image_path, "rb") as f:
+                image_bytes = f.read()
+
+        response = []
+
+        for chunk in client.models.generate_content_stream(
+            model=model,
+            contents=[
+                question,
+                types.Part.from_bytes(data=image_bytes, mime_type=mime_type),
+            ],
+        ):
+            response.append(chunk.text)
+
+        return " ".join(response)
+
+    except Exception as e:
+        return f"Error using vision model: {str(e)}"
+
+
+# YouTube Frames to Images Tool
+@tool
+def video_frames_to_images(
+    url: str,
+    folder_name: str,
+    sample_interval_seconds: int = 5,
+) -> List[str]:
+    """Extracts frames from a video at specified intervals and saves them as images.
+    Args:
+        url (str): the URL to the video.
+        folder_name (str): the name of the folder to save the images to.
+        sample_interval_seconds (int): the interval between frames to sample.
+    Returns:
+        List[str]: A list of paths to the saved image files.
+    """
+    # Create a subdirectory for the frames
+    frames_dir = os.path.join(folder_name, "frames")
+    os.makedirs(frames_dir, exist_ok=True)
+
+    ydl_opts = {
+        "format": "bestvideo[height<=1080]+bestaudio/best[height<=1080]/best",
+        "outtmpl": os.path.join(folder_name, "video.%(ext)s"),
+        "quiet": True,
+        "noplaylist": True,
+        "merge_output_format": "mp4",
+        "force_ipv4": True,
+    }
+
+    try:
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            info = ydl.extract_info(url, download=True)
+            video_path = next(
+                (
+                    os.path.join(folder_name, f)
+                    for f in os.listdir(folder_name)
+                    if f.endswith(".mp4")
+                ),
+                None,
+            )
+
+            if not video_path:
+                raise RuntimeError("Failed to download video as mp4")
+
+            reader = imageio.get_reader(video_path)
+            metadata = reader.get_meta_data()
+            fps = metadata.get("fps")
+
+            if fps is None:
+                reader.close()
+                raise RuntimeError(
+                    "Unable to determine FPS from video metadata"
+                )
+
+            frame_interval = int(fps * sample_interval_seconds)
+            image_paths: List[str] = []
+
+            for idx, frame in enumerate(reader):
+                if idx % frame_interval == 0:
+                    # Save frame as image
+                    image_path = os.path.join(
+                        frames_dir, f"frame_{idx:06d}.jpg"
+                    )
+                    imageio.imwrite(image_path, frame)
+                    image_paths.append(image_path)
+
+            reader.close()
+            return image_paths
+
+    except Exception as e:
+        raise RuntimeError(f"Error processing video frames: {str(e)}") from e
+
+
+@tool
+def transcribe_youtube(url: str) -> str:
+    """Transcribes a YouTube video using YouTube Transcript API or Gemini as fallback.
+    Args:
+        url (str): the URL to the YouTube video.
+    Returns:
+        str: The transcript of the YouTube video.
+    """
+    try:
+        # First try using YouTube Transcript API
+        video_id = _extract_video_id(url)
+        if not video_id:
+            raise ValueError(f"Invalid YouTube URL: {url}")
+
+        try:
+            # Try to get transcript in English
+            transcript_chunks = YouTubeTranscriptApi.get_transcript(
+                video_id, languages=["en"]
+            )
+            # Combine all chunks into a single transcript with timestamps
+            transcript = ""
+            for chunk in transcript_chunks:
+                timestamp = str(timedelta(seconds=int(chunk["start"])))
+                transcript += f"[{timestamp}] {chunk['text']}\n"
+            return transcript
+
+        except Exception as transcript_error:
+            print(
+                f"Failed to get transcript using YouTube API: {str(transcript_error)}"
+            )
+            print("Falling back to Gemini-based transcription...")
+
+            # Fallback to Gemini-based transcription
+            with tempfile.TemporaryDirectory() as tmpdir:
+                # Download audio from YouTube
+                ydl_opts = {
+                    "format": "bestaudio/best",
+                    "outtmpl": os.path.join(tmpdir, "audio.%(ext)s"),
+                    "quiet": True,
+                    "noplaylist": True,
+                    "postprocessors": [
+                        {
+                            "key": "FFmpegExtractAudio",
+                            "preferredcodec": "wav",
+                            "preferredquality": "192",
+                        }
+                    ],
+                }
+
+                try:
+                    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+                        info = ydl.extract_info(url, download=True)
+                        audio_path = next(
+                            (
+                                os.path.join(tmpdir, f)
+                                for f in os.listdir(tmpdir)
+                                if f.endswith(".wav")
+                            ),
+                            None,
+                        )
+
+                        if not audio_path:
+                            raise RuntimeError(
+                                "Failed to download audio"
+                            ) from transcript_error
+
+                        # Use Gemini to transcribe the audio
+                        client = genai.Client(api_key=os.getenv("GEMINI_KEY"))
+                        model = "models/gemini-1.5-flash-8b"
+
+                        # Read the audio file
+                        with open(audio_path, "rb") as audio_file:
+                            audio_data = audio_file.read()
+
+                        # Create the content with audio data
+                        contents = types.Content(
+                            parts=[
+                                types.Part(
+                                    file_data=types.FileData(
+                                        mime_type="audio/wav",
+                                        data=audio_data,
+                                    )
+                                ),
+                                types.Part(
+                                    text="Please transcribe this audio file. Include timestamps if possible."
+                                ),
+                            ]
+                        )
+
+                        # Generate transcription
+                        response = client.models.generate_content(
+                            model=model, contents=contents
+                        )
+                        return response.text
+
+                except yt_dlp.utils.DownloadError as e:
+                    raise RuntimeError(
+                        f"Error downloading YouTube video: {str(e)}"
+                    ) from transcript_error
+                except Exception as e:
+                    raise RuntimeError(
+                        f"Error processing YouTube video: {str(e)}"
+                    ) from transcript_error
+
+    except Exception as e:
+        raise RuntimeError(f"Error in YouTube transcription: {str(e)}") from e
+
+
+def _extract_video_id(url: str) -> Optional[str]:
+    """Extract video ID from YouTube URL.
+    Args:
+        url (str): the URL to the YouTube video.
+    Returns:
+        str: The video ID of the YouTube video.
+    """
+    patterns = [
+        r"(?:youtube\.com\/watch\?v=|youtube\.com\/embed\/|youtu\.be\/)([^&\n?#]+)",
+        r"(?:youtube\.com\/v\/|youtube\.com\/e\/|youtube\.com\/user\/[^\/]+\/|youtube\.com\/[^\/]+\/|youtube\.com\/embed\/|youtu\.be\/)([^&\n?#]+)",
+    ]
+
+    for pattern in patterns:
+        match = re.search(pattern, url)
+        if match:
+            return match.group(1)
+    return None