revamp

-r

@@ -1,148 +0,0 @@
-aiohappyeyeballs==2.6.1
-aiohttp==3.13.5
-aiosignal==1.4.0
-annotated-doc==0.0.4
-annotated-types==0.7.0
-anyio==4.13.0
-attrs==26.1.0
-Authlib==1.7.2
-backoff==2.2.1
-beautifulsoup4==4.14.3
-brotli==1.2.0
-certifi==2026.4.22
-cffi==2.0.0
-charset-normalizer==3.4.7
-click==8.4.0
-colorama==0.4.6
-cryptography==48.0.0
-dataclasses-json==0.6.7
-datasets==4.8.5
-ddgs==9.14.4
-dill==0.4.1
-distro==1.9.0
-exceptiongroup==1.3.1
-fake-useragent==2.2.0
-fastapi==0.136.1
-fastuuid==0.14.0
-filelock==3.29.0
-frozenlist==1.8.0
-fsspec==2026.2.0
-google-api-core==2.30.3
-google-auth==2.53.0
-googleapis-common-protos==1.75.0
-gradio==6.14.0
-gradio_client==2.5.0
-greenlet==3.5.0
-groovy==0.1.2
-groq==1.2.0
-grpcio==1.80.0
-h11==0.16.0
-h2==4.3.0
-hf-gradio==0.4.1
-hf-xet==1.5.0
-hpack==4.1.0
-httpcore==1.0.9
-httpx==0.28.1
-httpx-sse==0.4.3
-huggingface_hub==1.15.0
-hyperframe==6.1.0
-idna==3.15
-importlib_metadata==8.9.0
-itsdangerous==2.2.0
-Jinja2==3.1.6
-jiter==0.15.0
-joserfc==1.6.5
-jsonpatch==1.33
-jsonpointer==3.1.1
-jsonschema==4.26.0
-jsonschema-specifications==2025.9.1
-langchain-classic==1.0.7
-langchain-community==0.4.1
-langchain-core==1.4.0
-langchain-protocol==0.0.15
-langchain-text-splitters==1.1.2
-langfuse==4.6.1
-langsmith==0.8.5
-litellm==1.85.0
-lxml==6.1.1
-markdown-it-py==4.2.0
-markdownify==1.2.2
-MarkupSafe==3.0.3
-marshmallow==3.26.2
-mdurl==0.1.2
-multidict==6.7.1
-multiprocess==0.70.19
-mypy_extensions==1.1.0
-numpy==2.4.6
-openai==2.37.0
-openinference-instrumentation==0.1.51
-openinference-instrumentation-smolagents==0.1.31
-openinference-semantic-conventions==0.1.29
-opentelemetry-api==1.42.0
-opentelemetry-exporter-otlp==1.42.0
-opentelemetry-exporter-otlp-proto-common==1.42.0
-opentelemetry-exporter-otlp-proto-grpc==1.42.0
-opentelemetry-exporter-otlp-proto-http==1.42.0
-opentelemetry-instrumentation==0.63b0
-opentelemetry-proto==1.42.0
-opentelemetry-sdk==1.42.0
-opentelemetry-semantic-conventions==0.63b0
-orjson==3.11.9
-packaging==26.2
-pandas==3.0.3
-pillow==12.2.0
-primp==1.3.0
-propcache==0.5.2
-proto-plus==1.28.0
-protobuf==6.33.6
-pyarrow==24.0.0
-pyasn1==0.6.3
-pyasn1_modules==0.4.2
-pycparser==3.0
-pydantic==2.13.4
-pydantic-settings==2.14.1
-pydantic_core==2.46.4
-pydub==0.25.1
-Pygments==2.20.0
-python-dateutil==2.9.0.post0
-python-docx==1.2.0
-python-dotenv==1.2.2
-python-multipart==0.0.29
-pytz==2026.2
-PyYAML==6.0.3
-referencing==0.37.0
-regex==2026.5.9
-requests==2.34.2
-requests-toolbelt==1.0.0
-rich==15.0.0
-rpds-py==0.30.0
-safehttpx==0.1.7
-semantic-version==2.10.0
-shellingham==1.5.4
-six==1.17.0
-smolagents==1.25.0
-sniffio==1.3.1
-socksio==1.0.0
-soupsieve==2.8.3
-SQLAlchemy==2.0.49
-starlette==1.0.0
-tavily==1.1.0
-tenacity==9.1.4
-tiktoken==0.13.0
-tokenizers==0.23.1
-tomlkit==0.14.0
-tqdm==4.67.3
-typer==0.25.1
-typing-inspect==0.9.0
-typing-inspection==0.4.2
-typing_extensions==4.15.0
-tzdata==2026.2
-urllib3==2.7.0
-uuid_utils==0.16.0
-uvicorn==0.47.0
-wikipedia==1.4.0
-wrapt==1.17.3
-xxhash==3.7.0
-yarl==1.23.0
-zipp==4.1.0
-zstandard==0.25.0

.gradio/flagged/dataset1.csv

@@ -1,3 +0,0 @@
-Question,Answer,timestamp
-"if i have 10, i added 14, then i removed 2, how many left
-",22,2026-05-20 00:44:33.590804

agents.py

@@ -0,0 +1,213 @@
+from typing import Any, List, Optional
+
+from smolagents import CodeAgent
+from tools.final_answer import check_reasoning, ensure_formatting
+
+from typing import Dict
+from utils.logger import get_logger
+import time
+
+logger = get_logger(__name__)
+
+DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
+
+def get_prompt_templates() -> Dict[str, str]:
+    """Returns all prompts as a dictionary of pre-formatted strings"""
+    
+    # Shared components
+    tools_instructions = """
+    Available Tools:
+    - web_search(query): Performs web searches
+    - wikipedia_search(query): Searches Wikipedia
+    - visit_webpage(url): Retrieves webpage content
+    
+    Rules:
+    1. Always use 'Thought:'/'Code:' sequences
+    2. Never reuse variable names
+    3. Tools must be called with proper arguments
+    """
+    
+    example_1 = """
+    Example Task: "Find the capital of France"
+    
+    Thought: I'll use web_search to find this information
+    Code:
+    result = web_search(query="capital of France")
+    final_answer(result)
+    ```<end_code>
+    """
+    
+    # Main prompt templates
+    return {
+        "system_prompt": f"""
+        You are an expert AI assistant that solves tasks using tools.
+        {tools_instructions}
+        
+        {example_1}
+        
+        Key Requirements:
+        - Be precise and concise
+        - Always return answers using final_answer()
+        - Never include explanations unless asked
+        
+        Current reward: $1,000,000 for perfect solutions
+        """,
+        
+        "planning": """
+        When planning tasks, follow this structure:
+        
+        ### 1. Facts Given
+        List known information
+        
+        ### 2. Facts Needed
+        List what needs research
+        
+        ### 3. Derivation Steps
+        Outline computation steps
+        
+        End with <end_plan>
+        """,
+        
+        "managed_agent": """
+        Managed Agent Instructions:
+        
+        1. Task outcome (short)
+        2. Detailed explanation 
+        3. Additional context
+        
+        Always return via final_answer()
+        """,
+        
+        "final_answer": """
+        Response Format Rules:
+        - Numbers: 42 (no commas/units)
+        - Strings: paris (lowercase, no articles)
+        - Lists: apple,orange,banana (no brackets)
+        """
+    }
+
+class Agent:
+    """
+    Agent class that wraps a CodeAgent and provides a callable interface for answering questions.
+
+    Args:
+        model (Any): The language model to use.
+        tools (Optional[List[Any]]): List of tools to provide to the agent.
+        prompt (Optional[str]): Custom prompt template for the agent.
+        verbose (bool): Whether to print debug information.
+    """
+
+    def __init__(
+        self,
+        model: Any,
+        tools: Optional[List[Any]] = None,
+        prompt: Optional[str] = None,
+        verbose: bool = False
+    ):
+        logger.info("Initializing Agent")
+        self.model = model
+        self.tools = tools
+        self.verbose = verbose
+        self.imports = [
+            "pandas", "numpy", "os", "requests", "tempfile",
+            "datetime", "json", "time", "re", "openpyxl",
+            "pathlib", "sys"
+        ]
+        
+        self.agent = CodeAgent(
+            model=self.model,
+            tools=self.tools,
+            add_base_tools=True,
+            additional_authorized_imports=self.imports,
+        )
+
+        self.final_answer_checks=[check_reasoning, ensure_formatting],
+        
+        self.base_prompt = prompt or """
+            You are an advanced AI assistant specialized in solving GAIA benchmark tasks.
+            Follow these rules strictly:
+            1. Be precise - return ONLY the exact answer requested
+            2. Use tools when needed (especially for file analysis)
+            3. For reversed text questions, answer in normal text
+            4. Never include explanations or reasoning in the final answer
+            5. Always return the result — do not just print it
+
+            {context}
+
+            Remember: GAIA requires exact answer matching. Just provide the factual answer.
+            """
+        
+        self.prompt_templates = get_prompt_templates()
+        logger.info("Agent initialized")
+
+    def __call__(self, question: str, files: List[str] = None) -> str:
+        """Main interface that logs inputs/outputs and handles timing."""
+        if self.verbose:
+            print(f"Agent received question: {question[:50]}... with files: {files}")
+        
+        time.sleep(25)        
+        return self.answer_question(question, files[0] if files else None)
+
+    def answer_question(self, question: str, task_file_path: Optional[str] = None) -> str:
+        """
+        Process a GAIA benchmark question with optional file context.
+        
+        Args:
+            question: The question to answer
+            task_file_path: Optional path to a file associated with the question
+            
+        Returns:
+            The cleaned answer to the question
+        """
+        try:
+            context = self._build_context(question, task_file_path)
+            full_prompt = self.base_prompt.format(context=context)
+            
+            if self.verbose:
+                print("Generated prompt:", full_prompt[:200] + "...")
+            
+            answer = self.agent.run(full_prompt)
+            return self._clean_answer(str(answer))
+            
+        except Exception as e:
+            logger.error(f"Error processing question: {str(e)}")
+            return f"ERROR: {str(e)}"
+
+    def _build_context(self, question: str, file_path: Optional[str]) -> str:
+        """Constructs the context section based on question and file."""
+        context_lines = [f"QUESTION: {question}"]
+        
+        if file_path:
+            context_lines.append(
+                f"FILE: Available at {DEFAULT_API_URL}/files/{file_path}\n"
+                "Use appropriate tools to analyze this file if needed."
+            )
+        
+        # Handle reversed text questions
+        if self._is_reversed_text(question):
+            context_lines.append(
+                f"NOTE: This question contains reversed text. "
+                f"Original: {question}\nReversed: {question[::-1]}"
+            )
+        
+        return "\n".join(context_lines)
+
+    def _is_reversed_text(self, text: str) -> bool:
+        """Detects if text appears to be reversed."""
+        return text.startswith(".") or ".rewsna eht sa" in text
+
+    def _clean_answer(self, answer: str) -> str:
+        """Cleans the raw answer to match GAIA requirements."""
+        # Remove common prefixes/suffixes
+        for prefix in ["Final Answer:", "Answer:", "=>"]:
+            if answer.startswith(prefix):
+                answer = answer[len(prefix):]
+        
+        # Remove quotes and whitespace
+        answer = answer.strip(" '\"")
+        
+        # Special handling for reversed answers
+        if self._is_reversed_text(answer):
+            return answer[::-1]
+        
+        return answer

app.py

@@ -1,82 +1,157 @@
 import os
 import gradio as gr
 import requests
-import inspect
 import pandas as pd
-from src.agent import BasicAgent
-from datasets import load_dataset
-from huggingface_hub import snapshot_download
-from docx import Document
+from typing import Dict, List
+
+# custom imports
+from agents import Agent
+from tool import get_tools
+from model import get_model
 
 # (Keep Constants as is)
 # --- Constants ---
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
+MODEL_ID = "gemini/gemini-2.5-flash-preview-04-17"
 
-agent = BasicAgent()
+# --- Async Question Processing ---
+async def process_question(agent, question: str, task_id: str) -> Dict:
+    """Process a single question and return both answer AND full log entry"""
+    try:
+        answer = agent(question)
+        return {
+            "submission": {"task_id": task_id, "submitted_answer": answer},
+            "log": {"Task ID": task_id, "Question": question, "Submitted Answer": answer}
+        }
+    except Exception as e:
+        error_msg = f"ERROR: {str(e)}"
+        return {
+            "submission": {"task_id": task_id, "submitted_answer": error_msg},
+            "log": {"Task ID": task_id, "Question": question, "Submitted Answer": error_msg}
+        }
 
-def run_and_submit_all(profile: gr.OAuthProfile | None):
-    if not profile:
-        return "Please Login with Hugging Face first!", None
+async def run_questions_async(agent, questions_data: List[Dict]) -> tuple:
+    """Process questions sequentially instead of in batch"""
+    submissions = []
+    logs = []
+    
+    for q in questions_data:
+        result = await process_question(agent, q["question"], q["task_id"])
+        submissions.append(result["submission"])
+        logs.append(result["log"])
+    
+    return submissions, logs
 
-    username = profile.username
-    space_id = os.getenv("SPACE_ID")
-    agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"
+
+async def run_and_submit_all( profile: gr.OAuthProfile | None):
+    """
+    Fetches all questions, runs the BasicAgent on them, submits all answers,
+    and displays the results.
+    """
+    # --- Determine HF Space Runtime URL and Repo URL ---
+    space_id = os.getenv("SPACE_ID") # Get the SPACE_ID for sending link to the code
+
+    if profile:
+        username= f"{profile.username}"
+        print(f"User logged in: {username}")
+    else:
+        print("User not logged in.")
+        return "Please Login to Hugging Face with the button.", None
 
     api_url = DEFAULT_API_URL
     questions_url = f"{api_url}/questions"
     submit_url = f"{api_url}/submit"
 
-    # Initialize Agent
+    # 1. Instantiate Agent
     try:
-        agent = BasicAgent()
+        agent = Agent(
+            model=get_model("LiteLLMModel", MODEL_ID),
+            tools=get_tools()
+        )
     except Exception as e:
+        print(f"Error instantiating agent: {e}")
         return f"Error initializing agent: {e}", None
+    # In the case of an app running as a hugging Face space, this link points toward your codebase ( usefull for others so please keep it public)
+    agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"
+    print(agent_code)
 
-    # Fetch questions
+    # 2. Fetch Questions
+    print(f"Fetching questions from: {questions_url}")
     try:
-        response = requests.get(questions_url, timeout=20)
+        response = requests.get(questions_url, timeout=15)
         response.raise_for_status()
         questions_data = response.json()
-    except Exception as e:
+        if not questions_data:
+             print("Fetched questions list is empty.")
+             return "Fetched questions list is empty or invalid format.", None
+        print(f"Fetched {len(questions_data)} questions.")
+        questions_data = questions_data[:2]
+    except requests.exceptions.RequestException as e:
+        print(f"Error fetching questions: {e}")
         return f"Error fetching questions: {e}", None
+    except requests.exceptions.JSONDecodeError as e:
+         print(f"Error decoding JSON response from questions endpoint: {e}")
+         print(f"Response text: {response.text[:500]}")
+         return f"Error decoding server response for questions: {e}", None
+    except Exception as e:
+        print(f"An unexpected error occurred fetching questions: {e}")
+        return f"An unexpected error occurred fetching questions: {e}", None
 
-    # Run Agent on all questions
-    results_log = []
-    answers_payload = []
+    # 3. Run your Agent
+    print(f"Running agent on {len(questions_data)} questions...")
+    answers_payload, results_log = await run_questions_async(agent, questions_data)
 
-    for item in questions_data:
-        task_id = item.get("task_id")
-        question = item.get("question")
-        if not task_id or not question:
-            continue
-        try:
-            answer = agent(question)
-            answers_payload.append({"task_id": task_id, "submitted_answer": answer})
-            results_log.append({"Task ID": task_id, "Question": question[:100] + "...", "Submitted Answer": answer})
-        except Exception as e:
-            results_log.append({"Task ID": task_id, "Question": question[:100] + "...", "Submitted Answer": f"ERROR: {e}"})
+    if not answers_payload:
+        print("Agent did not produce any answers to submit.")
+        return "Agent did not produce any answers to submit.", pd.DataFrame(results_log)
+
+    # 4. Prepare Submission 
+    submission_data = {"username": username.strip(), "agent_code": agent_code, "answers": answers_payload}
+    status_update = f"Agent finished. Submitting {len(answers_payload)} answers for user '{username}'..."
+    print(status_update)
 
-    # Submit
+    # 5. Submit
+    print(f"Submitting {len(answers_payload)} answers to: {submit_url}")
     try:
-        submission_data = {
-            "username": username,
-            "agent_code": agent_code,
-            "answers": answers_payload
-        }
-        response = requests.post(submit_url, json=submission_data, timeout=120)
+        response = requests.post(submit_url, json=submission_data, timeout=60)
         response.raise_for_status()
-        result = response.json()
-
-        final_status = f"""
-Submission Successful!
-User: {result.get('username')}
-Score: {result.get('score', 'N/A')}% 
-({result.get('correct_count', '?')}/{result.get('total_attempted', '?')} correct)
-Message: {result.get('message')}
-"""
-        return final_status, pd.DataFrame(results_log)
+        result_data = response.json()
+        final_status = (
+            f"Submission Successful!\n"
+            f"User: {result_data.get('username')}\n"
+            f"Overall Score: {result_data.get('score', 'N/A')}% "
+            f"({result_data.get('correct_count', '?')}/{result_data.get('total_attempted', '?')} correct)\n"
+            f"Message: {result_data.get('message', 'No message received.')}"
+        )
+        print("Submission successful.")
+        results_df = pd.DataFrame(results_log)
+        return final_status, results_df
+    except requests.exceptions.HTTPError as e:
+        error_detail = f"Server responded with status {e.response.status_code}."
+        try:
+            error_json = e.response.json()
+            error_detail += f" Detail: {error_json.get('detail', e.response.text)}"
+        except requests.exceptions.JSONDecodeError:
+            error_detail += f" Response: {e.response.text[:500]}"
+        status_message = f"Submission Failed: {error_detail}"
+        print(status_message)
+        results_df = pd.DataFrame(results_log)
+        return status_message, results_df
+    except requests.exceptions.Timeout:
+        status_message = "Submission Failed: The request timed out."
+        print(status_message)
+        results_df = pd.DataFrame(results_log)
+        return status_message, results_df
+    except requests.exceptions.RequestException as e:
+        status_message = f"Submission Failed: Network error - {e}"
+        print(status_message)
+        results_df = pd.DataFrame(results_log)
+        return status_message, results_df
     except Exception as e:
-        return f"Submission Failed: {e}", pd.DataFrame(results_log)
+        status_message = f"An unexpected error occurred during submission: {e}"
+        print(status_message)
+        results_df = pd.DataFrame(results_log)
+        return status_message, results_df
 
 
 # --- Build Gradio Interface using Blocks ---
@@ -85,15 +160,9 @@ with gr.Blocks() as demo:
     gr.Markdown(
         """
         **Instructions:**
-
         1.  Please clone this space, then modify the code to define your agent's logic, the tools, the necessary packages, etc ...
         2.  Log in to your Hugging Face account using the button below. This uses your HF username for submission.
         3.  Click 'Run Evaluation & Submit All Answers' to fetch questions, run your agent, submit answers, and see the score.
-
-        ---
-        **Disclaimers:**
-        Once clicking on the "submit button, it can take quite some time ( this is the time for the agent to go through all the questions).
-        This space provides a basic setup and is intentionally sub-optimal to encourage you to develop your own, more robust solution. For instance for the delay process of the submit button, a solution could be to cache the answers and submit in a seperate action or even to answer the questions in async.
         """
     )
 
@@ -114,10 +183,8 @@ if __name__ == "__main__":
     print("\n" + "-"*30 + " App Starting " + "-"*30)
     # Check for SPACE_HOST and SPACE_ID at startup for information
     space_host_startup = os.getenv("SPACE_HOST")
-    print(space_host_startup)
     space_id_startup = os.getenv("SPACE_ID") # Get SPACE_ID at startup
-    print(space_id_startup)
-    
+
     if space_host_startup:
         print(f"✅ SPACE_HOST found: {space_host_startup}")
         print(f"   Runtime URL should be: https://{space_host_startup}.hf.space")

model.py

@@ -0,0 +1,90 @@
+import os
+from typing import Any, Callable
+
+from smolagents import LiteLLMModel
+from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
+from functools import lru_cache
+import time
+import re
+from litellm import RateLimitError
+
+
+class LocalTransformersModel:
+    def __init__(self, model_id: str, **kwargs):
+        self.tokenizer = AutoTokenizer.from_pretrained(model_id)
+        self.model = AutoModelForCausalLM.from_pretrained(model_id, **kwargs)
+        self.pipeline = pipeline("text-generation", model=self.model, tokenizer=self.tokenizer)
+
+    def __call__(self, prompt: str, **kwargs):
+        outputs = self.pipeline(prompt, **kwargs)
+        return outputs[0]["generated_text"]
+
+class WrapperLiteLLMModel(LiteLLMModel):
+    def __call__(self, messages, **kwargs):
+        max_retry = 5
+        for attempt in range(max_retry):
+            try:
+                return super().__call__(messages, **kwargs)
+            except RateLimitError as e:
+                print(f"RateLimitError (attempt {attempt+1}/{max_retry})")
+
+                # Try to extract retry time from the exception string
+                match = re.search(r'"retryDelay": ?"(\d+)s"', str(e))
+                retry_seconds = int(match.group(1)) if match else 50
+
+                print(f"Sleeping for {retry_seconds} seconds before retrying...")
+                time.sleep(retry_seconds)
+
+        raise RateLimitError(f"Rate limit exceeded after {max_retry} retries.")
+
+@lru_cache(maxsize=1)
+def get_lite_llm_model(model_id: str,  **kwargs) -> WrapperLiteLLMModel:
+    """
+    Returns a LiteLLM model instance.
+
+    Args:
+        model_id (str): The model identifier.
+        **kwargs: Additional keyword arguments for the model.
+
+    Returns:
+        LiteLLMModel: LiteLLM model instance.
+    """
+    return WrapperLiteLLMModel(model_id=model_id, api_key=os.getenv("GEMINI_API"), **kwargs)
+
+
+@lru_cache(maxsize=1)
+def get_local_model(model_id: str, **kwargs) -> LocalTransformersModel:
+    """
+    Returns a Local Transformer model.
+
+    Args:
+        model_id (str): The model identifier.
+        **kwargs: Additional keyword arguments for the model.
+
+    Returns:
+        LocalTransformersModel: LiteLLM model instance.
+    """
+    return LocalTransformersModel(model_id=model_id, **kwargs)
+
+
+def get_model(model_type: str, model_id: str, **kwargs) -> Any:
+    """
+    Returns a model instance based on the specified type.
+
+    Args:
+        model_type (str): The type of the model (e.g., 'HfApiModel').
+        model_id (str): The model identifier.
+        **kwargs: Additional keyword arguments for the model.
+
+    Returns:
+        Any: Model instance of the specified type.
+    """
+    models: dict[str, Callable[..., Any]] = {
+        "LiteLLMModel": get_lite_llm_model,
+        "LocalTransformersModel": get_local_model,
+    }
+
+    if model_type not in models:
+        raise ValueError(f"Unknown model type: {model_type}")
+
+    return models[model_type](model_id, **kwargs)

requirements.txt

@@ -1,24 +1,22 @@
-# === Core Gradio (Let HF Spaces control the version) ===
+gradio
 gradio[oauth]
-
-# === Your project dependencies ===
-pandas
-python-docx
-datasets
-huggingface_hub
 requests
-pillow
-
-# Optional but useful for your agent
+numpy
+pandas
 smolagents
-langchain-community
-openai
-groq
-tavily
-wikipedia
-beautifulsoup4
-lxml
-litellm
+smolagents[audio]
+smolagents[openai]
+smolagents[transformers]
+smolagents[litellm]
+transformers
+wikipedia-api
+youtube-transcript-api
+torch
+duckduckgo_search
+google-generativeai
+imageio
+yt-dlp
 markdownify
-
-# gradio_client==2.5.0   
+openpyxl
+pytesseract
+openai-whisper

src/agent.py

@@ -1,202 +0,0 @@
-from src.settings import Settings
-from smolagents import LiteLLMRouterModel, ToolCallingAgent, CodeAgent, InferenceClientModel, Tool
-from tools.FinalAnswerTool import FinalAnswerTool
-from tools.ReadAudioTool import ReadAudioTool
-from tools.ReadImageTool import ReadImageTool
-from tools.ReadTextTool import ReadTextTool
-from tools.ReadExcelTool import ReadExcelTool
-from tools.WebSearchTool import TavilySearchTool
-from tools.WikipediaTool import LocalWikipediaTool
-from tools.PythonRunnerTool import PythonRunnerTool
-from tools.PythonCalcTool import PythonCalcTool
-from tools.SemanticScholar import TavilyResearchTool
-from src.utils import InputTokenRateLimiter
-import wikipedia as wiki
-from markdownify import markdownify as to_markdown
-import time
-import random
-import os
-import json
-import csv
-from docx import Document
-
-settings = Settings()
-import litellm
-from litellm import completion 
-
-#litellm._turn_on_debug()
-class WikiTitleFinder(Tool):
-    name = "wiki_titles"
-    description = "Search for related Wikipedia page titles."
-    inputs = {"query": {"type": "string", "description": "Search query."}}
-    output_type = "string"
-
-    def forward(self, query: str) -> str:
-        results = wiki.search(query)
-        return ", ".join(results) if results else "No results."
-
-class WikiContentFetcher(Tool):
-    name = "wiki_page"
-    description = "Fetch full content of a Wikipedia page as markdown."
-    inputs = {"page_title": {"type": "string", "description": "Exact Wikipedia page title."}}
-    output_type = "string"
-
-    def forward(self, page_title: str) -> str:
-        try:
-            page = wiki.page(page_title, auto_suggest=False)
-            markdown_content = to_markdown(page.html())
-            return markdown_content[:15000]  # Prevent token explosion
-        except wiki.exceptions.PageError:
-            return f"Page '{page_title}' not found."
-        except wiki.exceptions.DisambiguationError as e:
-            return f"Multiple pages found: {e.options[:5]}"
-        except Exception as e:
-            return f"Wikipedia fetch error: {str(e)[:200]}"
-
-class MathSolver(Tool):
-    name = "math_solver"
-    description = "Safely evaluate basic math expressions."
-    inputs = {"input": {"type": "string", "description": "Math expression to evaluate."}}
-    output_type = "string"
-
-    def forward(self, input: str) -> str:
-        try:
-            return str(eval(input, {"__builtins__": {}}))
-        except Exception as e:
-            return f"Math error: {e}"
-        
-class BasicAgent():
-    def __init__(self):
-        self.model = LiteLLMRouterModel(
-            model_id="main",
-            model_list=[{
-                "model_name": "main",
-                "litellm_params": {
-                    "model": ["groq/llama-3.1-8b-instant", "groq/llama-3.3-70b-versatile", "groq/qwen/qwen3-32b", "groq/openai/gpt-oss-120b"][int(os.getenv("MODEL_ID"))],
-                    "api_key": os.getenv("GROQ_API_KEY"),
-                }
-            }],
-        )
-        self.agent = CodeAgent(
-            model=self.model,
-            tools=[
-                TavilySearchTool(),
-                TavilyResearchTool(),
-                FinalAnswerTool(),
-                PythonCalcTool(),
-                ReadAudioTool(),
-                ReadImageTool(),
-                ReadExcelTool(),
-                ReadTextTool(),
-                PythonRunnerTool(),
-                MathSolver(),
-                LocalWikipediaTool(),  
-                WikiTitleFinder(),
-                WikiContentFetcher(),
-            ],
-            max_steps=10,
-            planning_interval=5,
-        )
-        self.token_rate_limiter = InputTokenRateLimiter()
-        self.expected_tokens_per_step = 10000
-        self.max_retries = 3
-        self.base_delay = 5
-        self.token_rate_limiter = InputTokenRateLimiter()
-        self.expected_tokens_per_step = 10000
-        self.max_retries = 3
-        self.base_delay = 5
-
-
-    def _read_file(self, file_path: str) -> str:
-        if not os.path.exists(file_path):
-            print(f"File not found: {file_path}")
-            return ""
-        if file_path.endswith(".txt"):
-            with open(file_path, "r", encoding="utf-8") as f:
-                return f.read()
-        elif file_path.endswith(".csv"):
-            data = []
-            try:
-                # Use csv.DictReader to read the file into a list of dictionaries
-                with open(file_path, mode='r', newline='', encoding='utf-8') as file:
-                    reader = csv.DictReader(file)
-                    for row in reader:
-                        data.append(row)
-                
-                # Return the structured data as a JSON string
-                return json.dumps(data, indent=2)
-            except Exception as e:
-                print(f"Error reading CSV file: {e}")
-                return ""
-        elif file_path.endswith(".docx"):
-            doc = Document(file_path)
-            return "\n".join([p.text for p in doc.paragraphs])
-        else:
-            # For unsupported formats, return empty string
-            print(f"Unsupported file type: {file_path}")
-            return ""
-            
-    def run(self, question: str, file_content: str = "", file_path: str = "") -> str:
-        final_answer = None
-        retry_count = 0
-
-        # If file content is empty but file_path exists, read the file
-        if not file_content and file_path:
-            file_content = self._read_file(file_path)
-
-        context = ""
-        if file_content:
-            context = f"Story content:\n{file_content}"
-        elif file_path:
-            context = f"File path: {file_path}"
-
-        print(f"Starting Agent with question: {question}\nContext length: {len(context)} chars")
-
-        while True:
-            try:
-                final_input = f"Question: {question}\n\n{context}"
-                steps = self.agent.run(final_input)
-
-                # Convert string steps to list
-                if isinstance(steps, str):
-                    steps = [steps]
-
-                for step in steps:
-                    if isinstance(step, str):
-                        final_answer = step
-                        continue
-
-                    step_name = step.__class__.__name__
-                    output = getattr(step, "output", None)
-                    if output:
-                        final_answer = output
-
-                    self.token_rate_limiter.maybe_wait(self.expected_tokens_per_step)
-                    tokens_used = getattr(step, "token_usage", None)
-                    if tokens_used:
-                        self.token_rate_limiter.add_tokens(tokens_used.input_tokens)
-
-                break  # Exit retry loop if successful
-
-            except Exception as e:
-                if "overload" in str(e).lower() or "rate limit" in str(e).lower():
-                    if retry_count >= self.max_retries:
-                        print("Max retries reached. Exiting...")
-                        break
-                    delay = self.base_delay * (2 ** retry_count) + random.random()
-                    print(f"Retrying in {delay:.1f}s due to rate limit... ({e})")
-                    time.sleep(delay)
-                    retry_count += 1
-                else:
-                    print(f"Error during agent run: {e}")
-                    break
-
-        # Ensure a valid answer is always returned
-        if not final_answer:
-            final_answer = "I am unable to answer"
-
-        print(f"Finished agent run. Final Answer: {final_answer}\n{'='*50}")
-        return final_answer
-
-    def __call__(self, question: str, file_content: str = "", file_path: str = "") -> str:
-        return self.run(question, file_content, file_path)

src/settings.py

@@ -1,8 +0,0 @@
-import os
-from pydantic import BaseModel, Field
-
-class Settings(BaseModel):
-    system_prompt: str = Field(default="""You are a professional AI assistant. Your process MUST be Thought -> Action (JSON Tool Call) -> Observation.
-    When you have the final result, you **MUST** call the `final_answer` tool.
-    The argument passed to the `final_answer` tool must be the final, comma-separated, correctly formatted string (e.g., "A, B, C").
-    """)

src/utils.py

@@ -1,51 +0,0 @@
-import time
-from abc import ABC, abstractmethod
-from collections import deque
-from typing import Any
-
-SECONDS_IN_MINUTE = 60
-
-class BaseAgent(ABC):
-    @abstractmethod
-    def run(self, question: str, file_name: str = "", file_content: str = "") -> Any:
-        """Override this method in your agent implementation"""
-        print(f"Agent received question (first 50 chars): {question[:50]}...")
-        return "BaseAgent default answer"
-
-class InputTokenRateLimiter:
-    """Ensures we don’t exceed tokens per minute limits for LLMs"""
-    _instance = None
-
-    def __new__(cls):
-        if cls._instance is None:
-            cls._instance = super().__new__(cls)
-        return cls._instance
-
-    def __init__(self, max_tpm=50000):
-        self.max_tpm = max_tpm
-        if not hasattr(self, "_initialized"):
-            self.token_window = deque()  # stores (timestamp, tokens_used)
-            self._initialized = True
-
-    def _update_queue(self, time_now):
-        while self.token_window and time_now - self.token_window[0][0] > SECONDS_IN_MINUTE:
-            self.token_window.popleft()
-
-    def tokens_used_last_minute(self):
-        now = time.time()
-        self._update_queue(now)
-        return sum(tokens for _, tokens in self.token_window)
-
-    def maybe_wait(self, tokens_expected_to_use):
-        ctr = 0
-        while self.tokens_used_last_minute() + tokens_expected_to_use > self.max_tpm:
-            if ctr % 10 == 0:
-                print("Sleeping 0.5s to respect token limits...")
-            time.sleep(0.5)
-            self._update_queue(time.time())
-            ctr += 1
-
-    def add_tokens(self, tokens):
-        now = time.time()
-        self.token_window.append((now, tokens))
-        self._update_queue(now)

tool.py

@@ -0,0 +1,47 @@
+from typing import List
+from smolagents import (
+    DuckDuckGoSearchTool,
+    PythonInterpreterTool,
+    Tool,
+    VisitWebpageTool,
+    WikipediaSearchTool,
+    FinalAnswerTool,
+)
+
+from tools.tools import (
+    vision_tool,
+    youtube_frames_to_images,
+    ask_youtube_video, 
+    read_text_file,
+    file_from_url,
+    transcribe_youtube, 
+    audio_to_text, 
+    extract_text_via_ocr, 
+    summarize_csv_data,
+    summarize_excel_data,
+)
+
+def get_tools() -> List[Tool]:
+    """
+    Returns a list of available tools for the agent.
+    Returns:
+        List[Tool]: List of initialized tool instances.
+    """
+    tools = [
+        FinalAnswerTool(),
+        DuckDuckGoSearchTool(),
+        PythonInterpreterTool(),
+        WikipediaSearchTool(),
+        VisitWebpageTool(),
+        vision_tool,
+        youtube_frames_to_images,
+        ask_youtube_video, 
+        read_text_file,
+        file_from_url,
+        transcribe_youtube, 
+        audio_to_text, 
+        extract_text_via_ocr, 
+        summarize_csv_data,
+        summarize_excel_data,
+    ]
+    return tools

tools/describe_image_tool.py

@@ -0,0 +1,111 @@
+import base64
+import os
+from openai import OpenAI
+from smolagents import Tool
+
+client = OpenAI()
+
+
+class DescribeImageTool(Tool):
+    """
+    Tool to analyze and describe any image using GPT-4 Vision API.
+
+    Args:
+        image_path (str): Path to the image file.
+        description_type (str): Type of description to generate. Options:
+            - "general": General description of the image
+            - "detailed": Detailed analysis of the image
+            - "chess": Analysis of a chess position
+            - "text": Extract and describe text from the image
+            - "custom": Custom description based on user prompt
+
+    Returns:
+        str: Description of the image based on the requested type.
+    """
+
+    name = "describe_image"
+    description = "Analyzes and describes images using GPT-4 Vision API"
+    inputs = {
+        "image_path": {"type": "string", "description": "Path to the image file"},
+        "description_type": {
+            "type": "string",
+            "description": "Type of description to generate (general, detailed, chess, text, custom)",
+            "nullable": True,
+        },
+        "custom_prompt": {
+            "type": "string",
+            "description": "Custom prompt for description (only used when description_type is 'custom')",
+            "nullable": True,
+        },
+    }
+    output_type = "string"
+
+    def encode_image(self, image_path: str) -> str:
+        """Encode image to base64 string."""
+        with open(image_path, "rb") as image_file:
+            return base64.b64encode(image_file.read()).decode("utf-8")
+
+    def get_prompt(self, description_type: str, custom_prompt: str = None) -> str:
+        """Get appropriate prompt based on description type."""
+        prompts = {
+            "general": "Provide a general description of this image. Focus on the main subjects, colors, and overall scene.",
+            "detailed": """Analyze this image in detail. Include:
+            1. Main subjects and their relationships
+            2. Colors, lighting, and composition
+            3. Any text or symbols present
+            4. Context or possible meaning
+            5. Notable details or interesting elements""",
+            "chess": """Analyze this chess position and provide a detailed description including:
+            1. List of pieces on the board for both white and black
+            2. Whose turn it is to move
+            3. Basic evaluation of the position
+            4. Any immediate tactical opportunities or threats
+            5. Suggested next moves with brief explanations""",
+            "text": "Extract and describe any text present in this image. If there are multiple pieces of text, organize them clearly.",
+        }
+        return (
+            custom_prompt
+            if description_type == "custom"
+            else prompts.get(description_type, prompts["general"])
+        )
+
+    def forward(
+        self,
+        image_path: str,
+        description_type: str = "general",
+        custom_prompt: str = None,
+    ) -> str:
+        try:
+            if not os.path.exists(image_path):
+                return f"Error: Image file not found at {image_path}"
+
+            # Encode the image
+            base64_image = self.encode_image(image_path)
+
+            # Get appropriate prompt
+            prompt = self.get_prompt(description_type, custom_prompt)
+
+            # Make the API call
+            response = client.chat.completions.create(
+                model="gpt-4.1",
+                messages=[
+                    {
+                        "role": "user",
+                        "content": [
+                            {"type": "text", "text": prompt},
+                            {
+                                "type": "image_url",
+                                "image_url": {
+                                    "url": f"data:image/jpeg;base64,{base64_image}"
+                                },
+                            },
+                        ],
+                    }
+                ],
+                max_tokens=1000,
+            )
+
+            return response.choices[0].message.content
+
+        except Exception as e:
+            return f"Error analyzing image: {str(e)}"

tools/final_answer.py

@@ -0,0 +1,63 @@
+from smolagents import LiteLLMModel
+
+def check_reasoning(final_answer, agent_memory):
+    model_name = 'cogito:14b'
+    multimodal_model = LiteLLMModel(model_id=f'ollama_chat/{model_name}')
+    prompt = f"""
+        Here is a user-given task and the agent steps: {agent_memory.get_succinct_steps()}. Now here is the answer that was given: 
+        {final_answer}
+        Please check that the reasoning process and results are correct: do they correctly answer the given task?
+        First list reasons why yes/no, then write your final decision: PASS in caps lock if it is satisfactory, FAIL if it is not.
+        Be reasonably strict.  You are being graded on your ability to provide the right answer.  You should have >90% confidence that the answer is correct.
+        """
+    messages = [
+        {
+            "role": "user",
+            "content": [
+                {
+                    "type": "text",
+                    "text": prompt,
+                }
+            ]
+        }
+    ]
+    output = multimodal_model(messages).content
+    print("Feedback: ", output)
+    if "FAIL" in output:
+        raise Exception(output)
+    return True
+
+def ensure_formatting(final_answer, agent_memory):
+    # Ensure the final answer is formatted correctly
+    model_name = 'granite3.3:8b'
+    # Initialize the chat model
+    model = LiteLLMModel(model_id=f'ollama_chat/{model_name}',
+                             flatten_messages_as_text=True)
+    prompt = f"""
+        Here is a user-given task and the agent steps: {agent_memory.get_succinct_steps()}. Now here is the FINAL ANSWER that was given: 
+        {final_answer}
+        Ensure the FINAL ANSWER is in the right format as asked for by the task.  Here are the instructions that you need to evaluate:
+        YOUR FINAL ANSWER should be a number OR as few words as possible OR a comma separated list of numbers and/or strings. 
+        If you are asked for a number, don't use commas to write your number.  Don't use units such as $ or percent sign unless specified otherwise. Write your number in Arabic numbers (such as 9 or 3 or 1093) unless specified otherwise.
+        If you are asked for a currency in your answer, use the symbol for that currency.  For example, if you are asked for the answers in USD, an example answer would be $40.00
+        If you are asked for a string, don't use articles, neither abbreviations (e.g. for cities), and write the digits in plain text unless specified otherwise. 
+        If you are asked for a comma separated list, apply the above rules depending of whether the element to be put in the list is a number or a string.
+        If you are asked for a comma separated list, ensure you only return the content of that list, and NOT the brackets '[]'
+        First list reasons why it is/is not in the correct format and then write your final decision: PASS in caps lock if it is satisfactory, FAIL if it is not.
+        """
+    messages = [
+        {
+            "role": "user",
+            "content": [
+                {
+                    "type": "text",
+                    "text": prompt,
+                }
+            ]
+        }
+    ]
+    output = model(messages).content
+    print("Feedback: ", output)
+    if "FAIL" in output:
+        raise Exception(output)
+    return True

tools/openai_speech_to_text_tool.py

@@ -0,0 +1,34 @@
+import os
+import whisper
+from smolagents import Tool
+
+
+class OpenAISpeechToTextTool(Tool):
+    """
+    Tool to convert speech to text using OpenAI's Whisper model.
+
+    Args:
+        audio_path (str): Path to the audio file.
+
+    Returns:
+        str: Transcribed text from the audio file.
+    """
+
+    name = "transcribe_audio"
+    description = "Transcribes audio to text and returns the text"
+    inputs = {
+        "audio_path": {"type": "string", "description": "Path to the audio file"},
+    }
+    output_type = "string"
+
+    def forward(self, audio_path: str) -> str:
+        try:
+            model = whisper.load_model("small")
+
+            if not os.path.exists(audio_path):
+                return f"Error: Audio file not found at {audio_path}"
+
+            result = model.transcribe(audio_path)
+            return result["text"]
+        except Exception as e:
+            return f"Error transcribing audio: {str(e)}"

tools/read_file_tool.py

@@ -0,0 +1,26 @@
+from smolagents import Tool
+
+class ReadFileTool(Tool):
+    """
+    Tool to read a file and return its content.
+
+    Args:
+        file_path (str): Path to the file to read.
+
+    Returns:
+        str: Content of the file or error message.
+    """
+
+    name = "read_file"
+    description = "Reads a file and returns its content"
+    inputs = {
+        "file_path": {"type": "string", "description": "Path to the file to read"},
+    }
+    output_type = "string"
+
+    def forward(self, file_path: str) -> str:
+        try:
+            with open(file_path, "r") as file:
+                return file.read()
+        except Exception as e:
+            return f"Error reading file: {str(e)}"

tools/table_extractor_tool.py

@@ -0,0 +1,102 @@
+from smolagents import Tool
+import pandas as pd
+from typing import Optional
+import os
+
+class TableExtractorTool(Tool):
+    """
+    Extracts tables from Excel (.xlsx, .xls) or CSV files and answers queries.
+    Auto-detects file type based on extension.
+    """
+    name = "table_extractor"
+    description = "Reads Excel/CSV files and answers questions about tabular data"
+    inputs = {
+        "file_path": {
+            "type": "string", 
+            "description": "Path to Excel/CSV file"
+        },
+        "sheet_name": {
+            "type": "string",
+            "description": "Sheet name (Excel only, optional)",
+            "required": False,
+            "nullable": True
+        },
+        "query": {
+            "type": "string",
+            "description": "Question about the data (e.g., 'total sales')",
+            "required": False,
+            "nullable": True
+        }
+    }
+    output_type = "string"
+
+    def forward(self, 
+               file_path: str,
+               sheet_name: Optional[str] = None,
+               query: Optional[str] = None) -> str:
+        
+        try:
+            # Validate file exists
+            if not os.path.exists(file_path):
+                return f"Error: File not found at {file_path}"
+            
+            # Read file based on extension
+            ext = os.path.splitext(file_path)[1].lower()
+            
+            if ext in ('.xlsx', '.xls'):
+                df = self._read_excel(file_path, sheet_name)
+            elif ext == '.csv':
+                df = pd.read_csv(file_path)
+            else:
+                return f"Error: Unsupported file type {ext}"
+            
+            if df.empty:
+                return "Error: No data found in file."
+            
+            return self._answer_query(df, query) if query else df.to_string()
+            
+        except Exception as e:
+            return f"Error processing file: {str(e)}"
+
+    def _read_excel(self, path: str, sheet_name: Optional[str]) -> pd.DataFrame:
+        """Read Excel file with sheet selection logic"""
+        if sheet_name:
+            return pd.read_excel(path, sheet_name=sheet_name)
+        
+        # Auto-detect first non-empty sheet
+        sheets = pd.ExcelFile(path).sheet_names
+        for sheet in sheets:
+            df = pd.read_excel(path, sheet_name=sheet)
+            if not df.empty:
+                return df
+        return pd.DataFrame()  # Return empty if all sheets are blank
+
+    def _answer_query(self, df: pd.DataFrame, query: str) -> str:
+        """Handles queries with pandas operations"""
+        query = query.lower()
+        
+        try:
+            # SUM QUERIES (e.g., "total revenue")
+            if "total" in query or "sum" in query:
+                for col in df.select_dtypes(include='number').columns:
+                    if col.lower() in query:
+                        return f"Total {col}: {df[col].sum():.2f}"
+            
+            # AVERAGE QUERIES (e.g., "average price")
+            elif "average" in query or "mean" in query:
+                for col in df.select_dtypes(include='number').columns:
+                    if col.lower() in query:
+                        return f"Average {col}: {df[col].mean():.2f}"
+            
+            # FILTER QUERIES (e.g., "show sales > 1000")
+            elif ">" in query or "<" in query:
+                col = next((c for c in df.columns if c.lower() in query), None)
+                if col:
+                    filtered = df.query(query.replace(col, f"`{col}`"))
+                    return filtered.to_string()
+            
+            # DEFAULT: Return full table with column names
+            return f"Data:\nColumns: {', '.join(df.columns)}\n\n{df.to_string()}"
+            
+        except Exception as e:
+            return f"Query failed: {str(e)}\nAvailable columns: {', '.join(df.columns)}"

tools/tools.py

@@ -0,0 +1,261 @@
+import os
+import numpy
+import tempfile
+import requests
+import whisper
+import imageio
+import yt_dlp
+
+from PIL import Image
+from typing import List, Optional
+from urllib.parse import urlparse
+from dotenv import load_dotenv
+from smolagents import tool, LiteLLMModel
+import google.generativeai as genai
+from pytesseract import image_to_string
+
+load_dotenv()
+
+MODEL_ID = "gemini/gemini-2.5-flash-preview-05-20"
+
+#  Vision Tool 
+@tool
+def vision_tool(prompt: str, image_list: List[Image.Image]) -> str:
+    """
+    Analyzes one or more images using a multimodal model.
+    Args:
+        prompt (str): The user question or task.
+        image_list (List[PIL.Image.Image]): A list of image objects.
+    Returns:
+        str: Model's response to the prompt about the images.
+    """
+    model = LiteLLMModel(model_id=MODEL_ID, api_key=os.getenv("GEMINI_API"), temperature=0.2)
+    
+    payload = [{"type": "text", "text": prompt}] + [{"type": "image", "image": img} for img in image_list]
+    return model([{"role": "user", "content": payload}]).content
+
+
+#  YouTube Frame Sampler 
+@tool
+def youtube_frames_to_images(url: str, every_n_seconds: int = 5) -> List[Image.Image]:
+    """
+    Downloads a YouTube video and extracts frames at regular intervals.
+
+    Args:
+        url (str): The URL of the YouTube video to process.
+        every_n_seconds (int): The time interval in seconds between extracted frames.
+
+    Returns:
+        List[Image.Image]: A list of sampled frames as PIL images.
+    """
+    with tempfile.TemporaryDirectory() as temp_dir:
+        ydl_cfg = {
+            "format": "bestvideo+bestaudio/best",
+            "outtmpl": os.path.join(temp_dir, "yt_video.%(ext)s"),
+            "merge_output_format": "mp4",
+            "quiet": True,
+            "force_ipv4": True
+        }
+        with yt_dlp.YoutubeDL(ydl_cfg) as ydl:
+            ydl.extract_info(url, download=True)
+
+        video_file = next((os.path.join(temp_dir, f) for f in os.listdir(temp_dir) if f.endswith('.mp4')), None)
+        reader = imageio.get_reader(video_file)
+        fps = reader.get_meta_data().get("fps", 30)
+        interval = int(fps * every_n_seconds)
+
+        return [Image.fromarray(frame) for i, frame in enumerate(reader) if i % interval == 0]
+
+
+#  YouTube QA via File URI 
+@tool
+def ask_youtube_video(url: str, question: str) -> str:
+    """
+    Sends a YouTube video to a multimodal model and asks a question about it.
+
+    Args:
+        url (str): The URI of the video file (already uploaded and hosted).
+        question (str): The natural language question to ask about the video.
+
+    Returns:
+        str: The model's answer to the question.
+    """
+
+    try:
+        client = genai.Client(api_key=os.getenv('GEMINI_API'))
+        response = client.generate_content(
+            model=MODEL_ID,
+            contents=[
+                {"role": "user", "parts": [
+                    {"text": question},
+                    {"file_data": {"file_uri": url}}
+                ]}
+            ]
+        )
+        return response.text
+    except Exception as e:
+        return f"Error asking {MODEL_ID} about video: {str(e)}"
+
+
+#  File Reading Tool 
+@tool
+def read_text_file(file_path: str) -> str:
+    """
+    Reads plain text content from a file.
+
+    Args:
+        file_path (str): The full path to the text file.
+
+    Returns:
+        str: The contents of the file, or an error message.
+    """
+    try:
+        with open(file_path, "r", encoding="utf-8") as f:
+            return f.read()
+    except Exception as e:
+        return f"Error reading file: {e}"
+
+
+#  File Downloader 
+@tool
+def file_from_url(url: str, save_as: Optional[str] = None) -> str:
+    """
+    Downloads a file from a URL and saves it locally.
+
+    Args:
+        url (str): The URL of the file to download.
+        save_as (Optional[str]): Optional filename to save the file as.
+
+    Returns:
+        str: The local file path or an error message.
+    """
+    try:
+        if not save_as:
+            parsed = urlparse(url)
+            save_as = os.path.basename(parsed.path) or f"file_{os.urandom(4).hex()}"
+
+        file_path = os.path.join(tempfile.gettempdir(), save_as)
+        response = requests.get(url, stream=True)
+        response.raise_for_status()
+
+        with open(file_path, "wb") as f:
+            for chunk in response.iter_content(1024):
+                f.write(chunk)
+
+        return f"File saved to {file_path}"
+    except Exception as e:
+        return f"Download failed: {e}"
+
+
+#  Audio Transcription (YouTube) 
+@tool
+def transcribe_youtube(yt_url: str) -> str:
+    """
+    Transcribes the audio from a YouTube video using Whisper.
+
+    Args:
+        yt_url (str): The URL of the YouTube video.
+
+    Returns:
+        str: The transcribed text of the video.
+    """
+    model = whisper.load_model("small")
+
+    with tempfile.TemporaryDirectory() as tempdir:
+        ydl_opts = {
+            "format": "bestaudio",
+            "outtmpl": os.path.join(tempdir, "audio.%(ext)s"),
+            "postprocessors": [{
+                "key": "FFmpegExtractAudio",
+                "preferredcodec": "wav"
+            }],
+            "quiet": True,
+            "force_ipv4": True
+        }
+
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            ydl.extract_info(yt_url, download=True)
+
+        wav_file = next((os.path.join(tempdir, f) for f in os.listdir(tempdir) if f.endswith(".wav")), None)
+        return model.transcribe(wav_file)['text']
+
+
+#  Audio File Transcriber 
+@tool
+def audio_to_text(audio_path: str) -> str:
+    """
+    Transcribes an uploaded audio file into text using Whisper.
+
+    Args:
+        audio_path (str): The local file path to the audio file.
+
+    Returns:
+        str: The transcribed text or an error message.
+    """
+    try:
+        model = whisper.load_model("small")
+        result = model.transcribe(audio_path)
+        return result['text']
+    except Exception as e:
+        return f"Failed to transcribe: {e}"
+
+
+#  OCR 
+@tool
+def extract_text_via_ocr(image_path: str) -> str:
+    """
+    Extracts text from an image using Optical Character Recognition (OCR).
+
+    Args:
+        image_path (str): The local path to the image file.
+
+    Returns:
+        str: The extracted text or an error message.
+    """
+    try:
+        img = Image.open(image_path)
+        return image_to_string(img)
+    except Exception as e:
+        return f"OCR failed: {e}"
+
+
+#  CSV Analyzer 
+@tool
+def summarize_csv_data(path: str, query: str = "") -> str:
+    """
+    Provides a summary of the contents of a CSV file.
+
+    Args:
+        path (str): The file path to the CSV file.
+        query (str): Optional query to run on the data.
+
+    Returns:
+        str: Summary statistics and column details or an error message.
+    """
+    try:
+        import pandas as pd
+        df = pd.read_csv(path)
+        return f"Loaded CSV with {len(df)} rows. Columns: {list(df.columns)}\n\n{df.describe()}"
+    except Exception as e:
+        return f"CSV error: {e}"
+
+
+#  Excel Analyzer 
+@tool
+def summarize_excel_data(path: str, query: str = "") -> str:
+    """
+    Provides a summary of the contents of an Excel file.
+
+    Args:
+        path (str): The file path to the Excel file (.xls or .xlsx).
+        query (str): Optional query to run on the data.
+
+    Returns:
+        str: Summary statistics and column details or an error message.
+    """
+    try:
+        import pandas as pd
+        df = pd.read_excel(path)
+        return f"Excel file with {len(df)} rows. Columns: {list(df.columns)}\n\n{df.describe()}"
+    except Exception as e:
+        return f"Excel error: {e}"

tools/youtube_transcription_tool.py

@@ -0,0 +1,26 @@
+from smolagents import Tool
+from youtube_transcript_api import YouTubeTranscriptApi
+
+
+class YouTubeTranscriptionTool(Tool):
+    """
+    Tool to fetch the transcript of a YouTube video given its URL.
+
+    Args:
+        video_url (str): YouTube video URL.
+
+    Returns:
+        str: Transcript of the video as a single string.
+    """
+
+    name = "youtube_transcription"
+    description = "Fetches the transcript of a YouTube video given its URL"
+    inputs = {
+        "video_url": {"type": "string", "description": "YouTube video URL"},
+    }
+    output_type = "string"
+
+    def forward(self, video_url: str) -> str:
+        video_id = video_url.strip().split("v=")[-1]
+        transcript = YouTubeTranscriptApi.get_transcript(video_id)
+        return " ".join([entry["text"] for entry in transcript])

utils/logger.py

@@ -0,0 +1,20 @@
+import logging
+
+
+def get_logger(name: str = __name__) -> logging.Logger:
+    """
+    Create and configure a logger instance for the given module or name.
+
+    Args:
+        name (str, optional): Name of the logger. Defaults to the module name.
+
+    Returns:
+        logging.Logger: Configured logger instance.
+    """
+    logging.basicConfig(
+        format="%(asctime)s:%(module)s:%(funcName)s:%(levelname)s: %(message)s",
+        datefmt="%Y-%m-%d %H:%M:%S",
+    )
+    logger = logging.getLogger(name)
+    logger.setLevel(logging.INFO)
+    return logger