Upload 6 files

README.md

@@ -1,15 +1,58 @@
 ---
-title: Template Final Assignment
-emoji: 🕵🏻‍♂️
-colorFrom: indigo
+title: Agent GAIA
+emoji: 🏆
+colorFrom: pink
 colorTo: indigo
 sdk: gradio
-sdk_version: 5.25.2
+sdk_version: 5.33.0
 app_file: app.py
 pinned: false
 hf_oauth: true
-# optional, default duration is 8 hours/480 minutes. Max duration is 30 days/43200 minutes.
 hf_oauth_expiration_minutes: 480
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# GAIA Benchmark Agent
+
+This project is an AI agent built for the GAIA benchmark as part of the Hugging Face Agents course. It combines different LLM models and multimodal tools to reason over text, audio, images and video to solve complex tasks.
+
+
+## Tools
+
+The agent includes a variety of tools for handling diverse input types:
+
+- **Vision Tool:** Analyze images using Gemini Vision.
+- **YouTube Frame Extractor:** Sample video frames from YouTube at regular intervals.
+- **YouTube QA Tool:** Ask questions about video content using Gemini via file URI.
+- **OCR Tool:** Extract text from images using Tesseract.
+- **Audio Transcriber:** Transcribe audio files and YouTube videos using Whisper.
+- **File Tools:** Read plain text, download files from URLs, and summarize CSV or Excel files.
+
+These tools are defined using the `@tool` decorator from the `smolagents` library, making them callable by the agent during task execution.
+
+## Models Used
+
+- `Gemini 2.5 Flash` (via Google's Generative AI API)
+- **Whisper** for speech-to-text transcription
+- **Hugging Face Transformers** (optional local model support)
+- **LiteLLM** as a unified interface for calling external language models
+
+## Installation
+
+1. Install all required dependencies using
+
+```bash
+pip install -r requirements.txt
+```
+
+2. Convfigure environment with API_KEYS
+
+```bash
+echo "GEMINI_API_KEY=your_key_here" > .env
+echo "HF_TOKEN=your_hf_token" >> .env
+```
+
+3. Run the app
+
+```bash
+python app.py
+```

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,183 +1,117 @@
-import time
 import os
 import gradio as gr
 import requests
 import pandas as pd
-import PyPDF2
-from smolagents import CodeAgent, DuckDuckGoSearchTool, InferenceClientModel, tool
-from bs4 import BeautifulSoup
+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"
 
-# --- CUSTOM TOOLS ---
-@tool
-def document_parser_tool(file_path_or_url: str) -> str:
-    """Parses and extracts text from a financial PDF document.
-    
-    Args:
-        file_path_or_url: The local file path or URL of the PDF document to parse.
-    """
+# --- 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:
-        local_path = file_path_or_url
-        if file_path_or_url.startswith('http'):
-            local_path = "temp_financial_doc.pdf"
-            response = requests.get(file_path_or_url, timeout=15)
-            response.raise_for_status()
-            with open(local_path, 'wb') as f:
-                f.write(response.content)
-            
-        extracted_text = ""
-        with open(local_path, 'rb') as file:
-            reader = PyPDF2.PdfReader(file)
-            num_pages = min(len(reader.pages), 2) 
-            for page_num in range(num_pages):
-                text = reader.pages[page_num].extract_text()
-                if text:
-                    extracted_text += f"--- Page {page_num + 1} ---\n{text}\n"
-        
-        if file_path_or_url.startswith('http') and os.path.exists(local_path):
-            os.remove(local_path)
-            
-        return extracted_text.strip()[:4000] if extracted_text else "No text found in document."
+        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:
-        return f"Error reading document: {str(e)}"
-
-@tool
-def visit_webpage_tool(url: str) -> str:
-    """Visits a standard HTML webpage and extracts the text content.
+        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}
+        }
+
+async def run_questions_async(agent, questions_data: List[Dict]) -> tuple:
+    """Process questions sequentially instead of in batch"""
+    submissions = []
+    logs = []
     
-    Args:
-        url: The URL of the webpage to visit.
-    """
-    try:
-        headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'}
-        response = requests.get(url, headers=headers, timeout=15)
-        response.raise_for_status()
-        soup = BeautifulSoup(response.content, 'html.parser')
-        for script in soup(["script", "style"]):
-            script.extract()
-        text = soup.get_text(separator='\n', strip=True)
-        return text[:4000] 
-    except Exception as e:
-        return f"Failed to read webpage: {str(e)}"
-
-# --- Basic Agent Definition ---
-class BasicAgent:
-    def __init__(self):
-        print("Initializing Financial Due Diligence Agent...")
-        self.search_tool = DuckDuckGoSearchTool()
-        
-        # Native Hugging Face model - massive context, no token limits!
-        self.model = InferenceClientModel(
-            model_id="Qwen/Qwen2.5-Coder-32B-Instruct",
-            token=os.environ.get("HF_TOKEN")
-        )
-        
-        self.agent = CodeAgent(
-            tools=[self.search_tool, document_parser_tool, visit_webpage_tool],
-            model=self.model,
-            max_steps=4,
-            additional_authorized_imports=[
-                "math", "pandas", "datetime", "requests", "bs4", "re", "time", "wikipedia"
-            ]
-        )
+    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
 
-    def __call__(self, task_id: str, question: str) -> str:
-        print(f"Analyzing Task {task_id}: {question[:50]}...")
-        file_url = f"{DEFAULT_API_URL}/files/{task_id}"
-        
-        prompt = f"""
-        You are solving a GAIA benchmark question. You must use tools to find the answer.
-        
-        CRITICAL INSTRUCTION:
-        The final output MUST be exactly the requested answer and NOTHING else. 
-        Do not use conversational text. Do not say "The answer is" or "Final Answer:". 
-        Just provide the exact raw value, word, number, or comma-separated list.
-        
-        If the question references an attached file or document, use your `document_parser_tool` 
-        to read it by passing this exact URL: {file_url}
-        
-        Question: {question}
-        """
-        
-        try:
-            answer = self.agent.run(prompt)
-            final_answer = str(answer).strip()
-            print(f"Agent computed answer: {final_answer}")
-            return final_answer
-        except Exception as e:
-            error_msg = str(e)
-            print(f"Agent failed: {error_msg}")
-            return f"ERROR: {error_msg}"
-
-def run_and_submit_all(profile: gr.OAuthProfile | None):
-    space_id = os.getenv("SPACE_ID")
+
+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.")
-        yield "Please Login to Hugging Face with the button.", pd.DataFrame()
-        return
+        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"
 
+    # 1. Instantiate Agent
     try:
-        agent = BasicAgent()
+        agent = Agent(
+            model=get_model("LiteLLMModel", MODEL_ID),
+            tools=get_tools()
+        )
     except Exception as e:
-        yield f"Error initializing agent: {e}", pd.DataFrame()
-        return
-    
+        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)
 
+    # 2. Fetch Questions
+    print(f"Fetching questions from: {questions_url}")
     try:
         response = requests.get(questions_url, timeout=15)
         response.raise_for_status()
         questions_data = response.json()
         if not questions_data:
-             yield "Fetched questions list is empty or invalid format.", pd.DataFrame()
-             return
+             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:
-        yield f"Error fetching questions: {e}", pd.DataFrame()
-        return
+        print(f"An unexpected error occurred fetching questions: {e}")
+        return f"An unexpected error occurred fetching questions: {e}", None
 
-    results_log = []
-    answers_payload = []
-    
-    # LIVE UPDATE LOOP
-    # LIVE UPDATE LOOP
-    for i, item in enumerate(questions_data):
-        task_id = item.get("task_id")
-        question_text = item.get("question")
-        if not task_id or question_text is None:
-            continue
-            
-        yield f"Processing question {i+1} of {len(questions_data)}... Please wait.", pd.DataFrame(results_log)
-            
-        try:
-            # ---> THE FIX: Instantiate a brand new, empty-brained agent for EACH question <---
-            fresh_agent = BasicAgent()
-            submitted_answer = fresh_agent(task_id, question_text)
-            
-            answers_payload.append({"task_id": task_id, "submitted_answer": submitted_answer})
-            results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": submitted_answer})
-        except Exception as e:
-             results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": f"AGENT ERROR: {e}"})
-             
-        yield f"Finished question {i+1}. Moving to next...", pd.DataFrame(results_log)
-        time.sleep(5)
+    # 3. Run your Agent
+    print(f"Running agent on {len(questions_data)} questions...")
+    answers_payload, results_log = await run_questions_async(agent, questions_data)
 
     if not answers_payload:
-        yield "Agent did not produce any answers to submit.", pd.DataFrame(results_log)
-        return
+        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}
-    yield f"Agent finished. Submitting {len(answers_payload)} answers for user '{username}'...", pd.DataFrame(results_log)
+    status_update = f"Agent finished. Submitting {len(answers_payload)} answers for user '{username}'..."
+    print(status_update)
 
+    # 5. Submit
+    print(f"Submitting {len(answers_payload)} answers to: {submit_url}")
     try:
         response = requests.post(submit_url, json=submission_data, timeout=60)
         response.raise_for_status()
@@ -189,23 +123,55 @@ def run_and_submit_all(profile: gr.OAuthProfile | None):
             f"({result_data.get('correct_count', '?')}/{result_data.get('total_attempted', '?')} correct)\n"
             f"Message: {result_data.get('message', 'No message received.')}"
         )
-        yield final_status, pd.DataFrame(results_log)
+        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:
-        yield 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 ---
 with gr.Blocks() as demo:
-    gr.Markdown("# Financial Due Diligence Agent Evaluation Runner")
+    gr.Markdown("# Basic Agent Evaluation Runner")
     gr.Markdown(
         """
         **Instructions:**
-        1.  Log in to your Hugging Face account using the button below.
-        2.  Click 'Run Evaluation & Submit All Answers' to fetch questions, run your agent, submit answers, and see the score.
+        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.
         """
     )
 
     gr.LoginButton()
+
     run_button = gr.Button("Run Evaluation & Submit All Answers")
+
     status_output = gr.Textbox(label="Run Status / Submission Result", lines=5, interactive=False)
+    # Removed max_rows=10 from DataFrame constructor
     results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True)
 
     run_button.click(
@@ -214,4 +180,25 @@ with gr.Blocks() as demo:
     )
 
 if __name__ == "__main__":
-    demo.launch(debug=True, ssr_mode=False)
+    print("\n" + "-"*30 + " App Starting " + "-"*30)
+    # Check for SPACE_HOST and SPACE_ID at startup for information
+    space_host_startup = os.getenv("SPACE_HOST")
+    space_id_startup = os.getenv("SPACE_ID") # Get SPACE_ID at 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")
+    else:
+        print("ℹ️  SPACE_HOST environment variable not found (running locally?).")
+
+    if space_id_startup: # Print repo URLs if SPACE_ID is found
+        print(f"✅ SPACE_ID found: {space_id_startup}")
+        print(f"   Repo URL: https://huggingface.co/spaces/{space_id_startup}")
+        print(f"   Repo Tree URL: https://huggingface.co/spaces/{space_id_startup}/tree/main")
+    else:
+        print("ℹ️  SPACE_ID environment variable not found (running locally?). Repo URL cannot be determined.")
+
+    print("-"*(60 + len(" App Starting ")) + "\n")
+
+    print("Launching Gradio Interface for Basic Agent Evaluation...")
+    demo.launch(debug=True, share=False)

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)

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