Upload 6 files

agent.py

@@ -0,0 +1,538 @@
+import os
+from dotenv import load_dotenv
+from typing import TypedDict, List, Dict, Any, Optional
+from urllib.parse import urlparse
+from langgraph.graph import StateGraph, START, END, MessagesState
+from langchain.agents import create_tool_calling_agent, ConversationalAgent, AgentExecutor, initialize_agent, create_react_agent
+from langchain_google_genai import ChatGoogleGenerativeAI
+from langchain_groq import ChatGroq
+from langchain_core.tools import tool, Tool
+from langchain_core.messages import HumanMessage, SystemMessage
+from langchain.memory import ConversationBufferMemory
+from langchain_core.prompts import ChatPromptTemplate, PromptTemplate
+from langgraph.prebuilt import ToolNode
+from langgraph.prebuilt import tools_condition
+
+# 1. Web Browsing
+from langchain_community.tools import DuckDuckGoSearchResults
+from langchain_community.document_loaders import ImageCaptionLoader
+import requests, time, yt_dlp
+import pandas as pd
+from pathlib import Path
+from bs4 import BeautifulSoup
+from langchain_community.tools import WikipediaQueryRun
+from langchain_community.utilities import WikipediaAPIWrapper, DuckDuckGoSearchAPIWrapper
+from langchain_community.document_loaders import YoutubeLoader
+from langchain_community.document_loaders import UnstructuredExcelLoader
+from langchain_community.document_loaders import AssemblyAIAudioTranscriptLoader
+from langchain.text_splitter import CharacterTextSplitter
+from langchain_community.utilities import GoogleSerperAPIWrapper
+
+load_dotenv()
+DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
+
+@tool
+def duckduck_websearch(query: str) -> str:
+    """Allows search through DuckDuckGo.
+    Args:
+        query: what you want to search
+    """
+    try:
+        # search = DuckDuckGoSearchResults()
+        # results = search.invoke(query)
+        search = search = DuckDuckGoSearchAPIWrapper(max_results=5)
+        results = search.run(query)
+        if not results or results.strip() == "":
+            return "No search results found."
+            
+        return results
+    except Exception as e:
+        print(str(e))
+        print('Try to use request method for duckcudckgo Search')
+        base_url = "https://html.duckduckgo.com/html"
+        params = {"q": query}
+        response = requests.get(base_url, params=params, timeout=10)
+        soup = BeautifulSoup(response.text, 'html.parser')
+        for result in soup.find_all('div', {'class': 'result'}):
+            title = result.find('a', {'class': 'result__a'})
+            snippet = result.find('a', {'class': 'result__snippet'})
+            if title and snippet:
+                results.append({
+                    'title': title.get_text(),
+                    'snippet': snippet.get_text(),
+                    'url': title.get('href')
+                })
+
+        # Format results
+        formatted_results = []
+        for r in results[:10]:  # Limit to top 5 results
+            formatted_results.append(f"[{r['title']}]({r['url']})\n{r['snippet']}\n")
+
+        return "## Search Results\n\n" + "\n".join(formatted_results)
+
+@tool
+def serper_websearch(query: str) -> str:
+    """Allows search through Serper.
+    Args:
+        query: what you want to search
+    """
+    search = GoogleSerperAPIWrapper(serper_api_key=os.getenv("SERPER_API_KEY"))
+    results = search.run(query)
+    return results
+
+@tool
+def visit_webpage(url: str) -> str:
+    """Fetches raw HTML content of a web page.
+    Args:
+        url: the webpage url
+    """
+    try:
+        response = requests.get(url, timeout=5)
+        return response.text[:5000]
+    except Exception as e:
+        return f"[ERROR fetching {url}]: {str(e)}"
+
+@tool
+def wiki_search(query: str) -> str:
+    """Wiki search tools.
+    Args:
+        query: what you want to wiki
+    """
+    api_wrapper = WikipediaAPIWrapper(top_k_results=1, doc_content_chars_max=100)
+    wikipediatool = WikipediaQueryRun(api_wrapper=api_wrapper)
+    return wikipediatool.run({"query": query})
+
+@tool
+def text_splitter(text: str) -> List[str]:
+    """Splits text into chunks using LangChain's CharacterTextSplitter.
+    Args:
+        text: A string of text to split.
+    """
+    splitter = CharacterTextSplitter(chunk_size=450, chunk_overlap=10)
+    return splitter.split_text(text)
+
+@tool
+def youtube_transcript(video_url: str) -> str:
+    """Fetched youtube transcript
+    Args:
+        video_url: YouTube video url
+    """
+    try:
+        loader = YoutubeLoader.from_youtube_url(video_url)
+        # video_id = video_url.split("v=")[-1].split("&")[0]
+        # transcript = YouTubeTranscriptApi.get_transcript(video_id)
+        return loader.load()
+    except Exception as e:
+        return f"Error fetching transcript: {str(e)}"
+
+# 4. File Reading
+@tool
+def read_file(task_id: str) -> str:
+    """First download the file, then read its content
+    Args:
+        dir: the task_id
+    """
+    file_url = f'{DEFAULT_API_URL}/files/{task_id}'
+    r = requests.get(file_url, timeout=15, allow_redirects=True)
+    with open('temp', "wb") as fp:
+        fp.write(r.content)
+    with open('temp') as f:
+        return f.read()
+
+@tool
+def excel_read(task_id: str) -> str:
+    """First download the excel file, then read its content
+    Args:
+        dir: the task_id
+    """
+    try:
+        file_url = f'{DEFAULT_API_URL}/files/{task_id}'
+        r = requests.get(file_url, timeout=15, allow_redirects=True)
+        with open('temp.xlsx', "wb") as fp:
+            fp.write(r.content)
+        # Read the Excel file
+        df = pd.read_excel('temp.xlsx')
+        # Run various analyses based on the query
+        result = (
+            f"Excel file loaded with {len(df)} rows and {len(df.columns)} columns.\n"
+        )
+        result += f"Columns: {', '.join(df.columns)}\n\n"
+        # Add summary statistics
+        result += "Summary statistics:\n"
+        result += str(df.describe())
+        return result
+    except Exception as e:
+        return f"Error analyzing Excel file: {str(e)}"
+   
+@tool
+def csv_read(task_id: str) -> str:
+    """First download the csv file, then read its content
+    Args:
+        dir: the task_id
+    """
+    try:
+        file_url = f'{DEFAULT_API_URL}/files/{task_id}'
+        r = requests.get(file_url, timeout=15, allow_redirects=True)
+        with open('temp.csv', "wb") as fp:
+            fp.write(r.content)
+        # Read the CSV file
+        df = pd.read_csv(temp.csv)
+        # Run various analyses based on the query
+        result = (
+            f"Excel file loaded with {len(df)} rows and {len(df.columns)} columns.\n"
+        )
+        result += f"Columns: {', '.join(df.columns)}\n\n"
+        # Add summary statistics
+        result += "Summary statistics:\n"
+        result += str(df.describe())
+        return result
+    except Exception as e:
+        return f"Error analyzing CSV file: {str(e)}"
+        
+@tool
+def mp3_listen(task_id: str) -> str:
+    """First download the mp3 file, then listen to it
+    Args:
+        dir: the task_id
+    """
+    file_url = f'{DEFAULT_API_URL}/files/{task_id}'
+    r = requests.get(file_url, timeout=15, allow_redirects=True)
+    with open('temp.mp3', "wb") as fp:
+        fp.write(r.content)
+    loader = AssemblyAIAudioTranscriptLoader(file_path="temp.mp3", api_key=os.getenv("AssemblyAI_API_KEY"))
+    docs = loader.load()
+    contents = [doc.page_content for doc in docs]
+    return "\n".join(contents)
+    
+# 5. Image Open
+@tool
+def image_caption(dir: str) -> str:
+    """Understand the content of the provided image
+    Args:
+        dir: the image url link
+    """
+    loader = ImageCaptionLoader(images=[dir])
+    metadata = loader.load()
+    return metadata[0].page_content
+
+# 2. Coding
+from langchain_experimental.tools import PythonREPLTool
+@tool
+def run_python(code: str):
+    """ Run the given python code
+    Args:
+        code: the python code
+    """
+    return PythonREPLTool().run(code)
+
+@tool
+def multiply(a: float, b: float) -> float:
+    """Multiply two numbers.
+    Args:
+        a: first float
+        b: second float
+    """
+    return a * b
+
+@tool
+def add(a: float, b: float) -> float:
+    """Add two numbers.
+    Args:
+        a: first float
+        b: second float
+    """
+    return a + b
+
+@tool
+def subtract(a: float, b: float) -> float:
+    """Subtract two numbers.
+    Args:
+        a: first float
+        b: second float
+    """
+    return a - b
+
+@tool
+def divide(a: float, b: float) -> float:
+    """Divide two numbers.
+    Args:
+        a: first float
+        b: second float
+    """
+    if b == 0:
+        raise ValueError("Cannot divide by zero.")
+    return a / b
+
+# 3. Multi-Modality
+# - multiply: multiply two numbers, A and B
+# - add: add two numbers, A and B
+# - subtract: Subtract A by B with passing A as the first argument
+# - divide: Divide A by B with passing A as the first argument
+
+
+
+# ("human", f"Question: {question}\nReport to validate: {final_answer}")
+class BasicAgent:
+    def __init__(self):
+        self.model = ChatGoogleGenerativeAI(
+            model="gemini-2.0-flash-lite",
+            temperature=0,
+            max_tokens=1024,
+            candidate_count=1,
+            google_api_key=os.getenv("GEMINI_API_KEY"),
+        )
+        # System Prompt for few shot prompting
+        self.sys_prompt = """"
+                You are a general AI assistant. I will ask you a question. Report your thoughts, and finish your answer with the following template: 
+                FINAL ANSWER: [YOUR FINAL ANSWER].
+                YOUR FINAL ANSWER should be a number OR as few words as possible OR a comma separared list of numbers and/or strings.
+                If you are asked for a number, don't use comma to write your number neither use units such as $ or percent sign unless specified otherwise.
+                If you are asked for a string, don't use articles, neither abbreviations (eg. 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 put in the list is a number or a string.
+
+                You have access to the following tools:
+                - serper_websearch: web search the content of the query by passing the query as input with Serper Search Engine
+                - duckduck_websearch: web search the content of the query by passing the query as input with DuckDuckGo Search Engine
+                - visit_webpage: visit the given webpage url by passing the url as input
+                - wiki_search: wiki search the content of the query by passing the query as input if the question asks for wiki search it
+                - text_splitter: split text into chunks
+                - youtube_transcript: fetch the transcript of the Youtube video by passing the video url as input if the question asks for watching a Youtube video
+                - read_file: read the content of the attached file by passing the TASK-ID as input
+                - excel_read: read the content of the attached excel file by passing the TASK-ID as input
+                - csv_read: read the content of the attached csv file by passing the TASK-ID as input
+                - mp3_listen: listen to the content of the attached mp3 file by passing the TASK-ID as input
+                - image_caption: understand the visual content of the attached image by passing the TASK-ID as input
+                - run_python: run the python code
+                
+                If Task ID is included in the question, remember to call the relevant read tools [ie. read_file, excel_read, csv_read, mp3_listen, image_caption]
+                Note: python_tool is called when the question mentions the term "Python" or any math calculation.
+        """
+        # self.tools = [duckduck_websearch, serper_websearch, visit_webpage, wiki_search, text_splitter, self._analyze_video, youtube_transcript, read_file, excel_read, csv_read, mp3_listen, image_caption, run_python]
+        self.tools = [
+            Tool(
+                name="duckduck_websearch",
+                func=duckduck_websearch,
+                description="Search the web for information with DuckDuckGo"
+            ),
+            Tool(
+                name="serper_websearch",
+                func=serper_websearch,
+                description="Search the web for information with Serper"
+            ),
+            Tool(
+                name="visit_webpage",
+                func=visit_webpage,
+                description="Directly visit the webpage"
+            ),
+            Tool(
+                name="wiki_search",
+                func=wiki_search,
+                description="Search the information on Wikipedia"
+            ),
+            Tool(
+                name="text_splitter",
+                func=text_splitter,
+                description="Split text into chunks"
+            ),
+            Tool(
+                name="analyze_video",
+                func=self._analyze_video,
+                description="Analyze YouTube video content directly"
+            ),
+            Tool(
+                name="youtube_transcript",
+                func=youtube_transcript,
+                description="Fetch the transcript of YouTube video"
+            ),
+            Tool(
+                name="read_file",
+                func=read_file,
+                description="Read the file content"
+            ),
+            Tool(
+                name="excel_read",
+                func=excel_read,
+                description="Read the content of Excel file"
+            ),
+            Tool(
+                name="csv_read",
+                func=csv_read,
+                description="Read the content of CSV file"
+            ),
+            Tool(
+                name='mp3_listen',
+                func=mp3_listen,
+                description="Listen to the MP3 file"
+            ),
+            Tool(
+                name="image_caption",
+                func=image_caption,
+                description="Understand the image content"
+            ),
+            Tool(
+                name="run_python",
+                func=run_python,
+                description="Run Python code"
+            )
+        ]
+        # Setup memory
+        self.memory = ConversationBufferMemory(
+            memory_key="chat_history",
+            return_messages=True
+        )
+        self.agent = self.__setup_agent__()
+        # self.prompt = ChatPromptTemplate.from_messages([
+        #     ("system", self.sys_prompt),
+        #     ("human", "{input}")
+        # ])
+
+        # self.agent = initialize_agent(
+        #     tools=self.tools,
+        #     llm=self.model,
+        #     agent="zero-shot-react-description",  # ReAct agent type
+        #     verbose=True,
+        #     system_prompt=self.prompt,
+        #     handle_parsing_errors=True,
+        #     max_iterations=30
+        #     # "Check your output and make sure it conforms, use the Action/Action Input syntax"
+        # )
+        print("BasicAgent initialized.")
+    
+    def __call__(self, task: dict) -> str:
+        task_id, question, file_name = task["task_id"], task["question"], task["file_name"]
+        print(f"Agent received question (first 50 chars): {question[:50]}...")
+        
+        if file_name == "" or file_name is None:
+            question = question
+        else:
+            question = f"{question} with TASK-ID: {task_id}"
+            # fixed_answer = self.agent.run(f'{question} with TASK-ID: {task_id}')
+        fixed_answer = "This is a default answer."
+        
+
+        max_retries = 5
+        base_sleep = 1
+        for attempt in range(max_retries):
+            try:
+                fixed_answer = self.agent.run(question)
+                print(f"Agent returning fixed answer: {fixed_answer}")
+                time.sleep(60)
+                return fixed_answer
+            except Exception as e:
+                sleep_time = base_sleep * (attempt + 1)
+                if attempt < max_retries - 1:
+                    print(str(e))
+                    print(f"Attempt {attempt + 1} failed. Retrying in {sleep_time} seconds...")
+                    time.sleep(sleep_time)
+                    continue
+                return f"Error processing query after {max_retries} attempts: {str(e)}"
+        return fixed_answer
+
+    @tool
+    def _analyze_video(self, url: str) -> str:
+        """Analyze video content using Gemini's video understanding capabilities."""
+        try:
+            # Validate URL
+            parsed_url = urlparse(url)
+            if not all([parsed_url.scheme, parsed_url.netloc]):
+                return "Please provide a valid video URL with http:// or https:// prefix."
+            
+            # Check if it's a YouTube URL
+            if 'youtube.com' not in url and 'youtu.be' not in url:
+                return "Only YouTube videos are supported at this time."
+
+            try:
+                # Configure yt-dlp with minimal extraction
+                ydl_opts = {
+                    'quiet': True,
+                    'no_warnings': True,
+                    'extract_flat': True,
+                    'no_playlist': True,
+                    'youtube_include_dash_manifest': False
+                }
+
+                with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+                    try:
+                        # Try basic info extraction
+                        info = ydl.extract_info(url, download=False, process=False)
+                        if not info:
+                            return "Could not extract video information."
+
+                        title = info.get('title', 'Unknown')
+                        description = info.get('description', '')
+                        
+                        # Create a detailed prompt with available metadata
+                        prompt = f"""Please analyze this YouTube video:
+Title: {title}
+URL: {url}
+Description: {description}
+Please provide a detailed analysis focusing on:
+1. Main topic and key points from the title and description
+2. Expected visual elements and scenes
+3. Overall message or purpose
+4. Target audience"""
+
+                        # Use the LLM with proper message format
+                        messages = [HumanMessage(content=prompt)]
+                        response = self.model.invoke(messages)
+                        return response.content if hasattr(response, 'content') else str(response)
+
+                    except Exception as e:
+                        if 'Sign in to confirm' in str(e):
+                            return "This video requires age verification or sign-in. Please provide a different video URL."
+                        return f"Error accessing video: {str(e)}"
+
+            except Exception as e:
+                return f"Error extracting video info: {str(e)}"
+
+        except Exception as e:
+            return f"Error analyzing video: {str(e)}"
+    
+    def __setup_agent__(self) -> AgentExecutor:
+        PREFIX = """
+                You are a general AI assistant that can use various tools to answer question. I will ask you a question. Report your thoughts, and finish your answer with the following template: 
+                FINAL ANSWER: [YOUR FINAL ANSWER].
+                YOUR FINAL ANSWER should be a number OR as few words as possible OR a comma separared list of numbers and/or strings.
+                If you are asked for a number, don't use comma to write your number neither use units such as $ or percent sign unless specified otherwise.
+                If you are asked for a string, don't use articles, neither abbreviations (eg. 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 put in the list is a number or a string.
+
+                NOTE:
+                - If Task ID is included in the question, remember to call the relevant read tools [ie. read_file, excel_read, csv_read, mp3_listen, image_caption]
+                - python_tool is called when the question mentions the term "Python" or any math calculation.
+        """
+        FORMAT_INSTRUCTIONS = """
+                To use a tool, use the following format:
+                Thought: Do I need to use a tool? Yes
+                Action: the action to take, should be one of [{tool_names}]
+                Action Input: the input to the action
+                Observation: the result of the action
+                When you have a response to say to the Human, or if you do not need to use a tool, you MUST use the format:
+                Thought: Do I need to use a tool? No
+                Final Answer: [your response here]
+                Begin! Remember to ALWAYS include 'Thought:', 'Action:', 'Action Input:', and 'Final Answer:' in your responses.
+        """
+        SUFFIX = """
+                Previous conversation history:
+                {chat_history}
+                New question: {input}
+                {agent_scratchpad}
+        """
+        agent = ConversationalAgent.from_llm_and_tools(
+            llm=self.model,
+            tools=self.tools,
+            prefix=PREFIX,
+            format_instructions=FORMAT_INSTRUCTIONS,
+            suffix=SUFFIX,
+            input_variables=["input", "chat_history", "agent_scratchpad", "tool_names"],
+            handle_parsing_errors=True
+        )
+        return AgentExecutor.from_agent_and_tools(
+            agent=agent,
+            tools=self.tools,
+            memory=self.memory,
+            max_iterations=30,
+            verbose=True,
+            handle_parsing_errors=True,
+            # return_only_outputs=True  # This ensures we only get the final output
+        )

app.py

@@ -0,0 +1,197 @@
+import os
+import gradio as gr
+import requests
+import inspect
+import pandas as pd
+
+# (Keep Constants as is)
+# --- Constants ---
+DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
+
+from gemini_agent import GEMINI_AGENT
+class BasicAgent:
+    def __init__(self):
+        self.agent = GEMINI_AGENT()
+
+    def __call__(self, task: dict) -> str:
+        try:
+            response = self.agent.run(task)
+            return response
+        except Exception as e:
+            print(f"Error is raised: {str(e)}")
+            return "Agent could not complete this task"
+
+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"
+
+    # 1. Instantiate Agent ( modify this part to create your agent)
+    try:
+        agent = BasicAgent()
+    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)
+
+    # 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:
+             print("Fetched questions list is empty.")
+             return "Fetched questions list is empty or invalid format.", None
+        print(f"Fetched {len(questions_data)} questions.")
+    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
+
+    # 3. Run your Agent
+    results_log = []
+    answers_payload = []
+    print(f"Running agent on {len(questions_data)} questions...")
+    for item in questions_data:
+        task_id = item.get("task_id")
+        question_text = item.get("question")
+        if not task_id or question_text is None:
+            print(f"Skipping item with missing task_id or question: {item}")
+            continue
+        try:
+            # submitted_answer = agent(question_text)
+            submitted_answer = agent(item) # dictionary that consists: task_id, question, Level, file_name
+            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:
+             print(f"Error running agent on task {task_id}: {e}")
+             results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": f"AGENT 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)
+
+    # 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()
+        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:
+        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("# Basic Agent Evaluation Runner")
+    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.
+        """
+    )
+
+    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(
+        fn=run_and_submit_all,
+        outputs=[status_output, results_table]
+    )
+
+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")
+    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)

gemini_agent.py

@@ -0,0 +1,131 @@
+import os
+import time
+from dotenv import load_dotenv
+from typing import TypedDict, Annotated, Optional
+
+from langgraph.prebuilt import ToolNode, tools_condition
+from langgraph.graph import StateGraph, START
+from langgraph.graph.message import add_messages
+
+from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage
+from langchain_google_genai import ChatGoogleGenerativeAI
+
+from tools import *
+
+load_dotenv()
+DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
+
+class AgentState(TypedDict):
+    """Agent state for the graph."""
+    input_file: Optional[str]
+    messages: Annotated[list[AnyMessage], add_messages]
+
+
+class GEMINI_AGENT:
+    def __init__(self):
+        self.llm = ChatGoogleGenerativeAI(
+            model="gemini-2.0-flash-lite",
+            temperature=0,
+            max_tokens=1024,
+            google_api_key=os.getenv("GEMINI_API_KEY"),
+        )
+        
+        self.tools = [
+            duckduck_websearch,
+            serper_websearch,
+            visit_webpage,
+            wiki_search,
+            youtube_viewer,
+            text_splitter,
+            read_file,
+            excel_read,
+            csv_read,
+            mp3_listen,
+            image_caption,
+            run_python,
+            multiply,
+            add,
+            subtract,
+            divide
+        ]
+
+        self.llm_with_tools = self.llm.bind_tools(self.tools)
+        self.app = self._graph_compile()
+
+    def _graph_compile(self):
+        builder = StateGraph(AgentState)
+        # Define nodes: these do the work
+        builder.add_node("assistant", self._assistant)
+        builder.add_node("tools", ToolNode(self.tools))
+        # Define edges: these determine how the control flow moves
+        builder.add_edge(START, "assistant")
+        builder.add_conditional_edges(
+            "assistant",
+            tools_condition,
+        )
+        builder.add_edge("tools", "assistant")
+        react_graph = builder.compile()
+        return react_graph
+
+    def _assistant(self, state: AgentState):
+        sys_msg = SystemMessage(
+            content=
+            """
+            You are a helpful assistant tasked with answering questions using a set of tools. When given a question, follow these steps:
+            1. Create a clear, step-by-step plan to solve the question.
+            2. If a tool is necessary, select the most appropriate tool based on its functionality. If one tool isn't working, use another with similar functionality.
+            3. Execute your plan and provide the response in the following format:
+
+            FINAL ANSWER: [YOUR FINAL ANSWER]
+
+            Your final answer should be:
+
+            - A number (without commas or units unless explicitly requested),
+            - A short string (avoid articles, abbreviations, and use plain text for digits unless otherwise specified),
+            - A comma-separated list (apply the formatting rules above for each element, with exactly one space after each comma).
+
+            Ensure that your answer is concise and follows the task instructions strictly. If the answer is more complex, break it down in a way that follows the format.
+            Begin your response with "FINAL ANSWER: " followed by the answer, and nothing else.
+            """
+        )
+
+        return {
+            "messages": [self.llm_with_tools.invoke([sys_msg] + state["messages"])],
+            "input_file": state["input_file"]
+        }
+
+    def extract_after_final_answer(self, text):
+        keyword = "FINAL ANSWER: "
+        index = text.find(keyword)
+        if index != -1:
+            return text[index + len(keyword):]
+        else:
+            return ""
+
+    def run(self, task: dict):
+        task_id, question, file_name = task["task_id"], task["question"], task["file_name"]
+        print(f"Agent received question (first 50 chars): {question[:50]}...")
+
+        if file_name == "" or file_name is None:
+            question_text = question
+        else:
+            question_text = f'{question} with TASK-ID: {task_id}'
+        messages = [HumanMessage(content=question_text)]
+
+        max_retries = 5
+        base_sleep = 1
+        for attempt in range(max_retries):
+            try:
+                response = self.app.invoke({"messages": messages, "input_file": None})
+                final_ans = self.extract_after_final_answer(response['messages'][-1].content)
+                time.sleep(60) # avoid rate limit
+                return final_ans
+            except Exception as e:
+                sleep_time = base_sleep * (attempt + 1)
+                if attempt < max_retries - 1:
+                    print(str(e))
+                    print(f"Attempt {attempt + 1} failed. Retrying in {sleep_time} seconds...")
+                    time.sleep(sleep_time)
+                    continue
+                return f"Error processing query after {max_retries} attempts: {str(e)}"
+        return "This is a default answer."

requirements.txt

@@ -0,0 +1,24 @@
+gradio
+requests
+python-dotenv
+langchain
+langchain-google-genai
+langchain-core
+langchain-community
+pandas
+langgraph
+langchain-anthropic
+pypdf
+youtube-transcript-api
+wikipedia
+duckduckgo-search
+transformers
+langchain-experimental
+unstructured
+openpyxl
+assemblyai
+langchain-groq
+torch
+yt-dlp
+beautifulsoup4
+google-genai

run.py

@@ -0,0 +1,203 @@
+import os
+import gradio as gr
+import requests
+import inspect
+import pandas as pd
+
+# (Keep Constants as is)
+# --- Constants ---
+DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
+
+# --- Basic Agent Definition ---
+# ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------
+# from smolagents import CodeAgent, DuckDuckGoSearchTool, OpenAIServerModel, VLLMModel, HfApiModel
+# class BasicAgent:
+#     def __init__(self):
+#         # model = OpenAIServerModel(model_id="gpt-4o")
+#         # model = VLLMModel(model_id="HuggingFaceTB/SmolLM-135M-Instruct")
+#         model = HfApiModel("Qwen/Qwen2.5-Coder-32B-Instruct")
+#         self.agent = CodeAgent(tools=[DuckDuckGoSearchTool()], model=model)
+#         print("BasicAgent initialized.")
+#     def __call__(self, question: str) -> str:
+#         print(f"Agent received question (first 50 chars): {question[:50]}...")
+#         fixed_answer = self.agent.run(question)
+#         # fixed_answer = "This is a default answer."
+#         print(f"Agent returning fixed answer: {fixed_answer}")
+#         return fixed_answer
+from agent import BasicAgent
+
+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"
+
+    # 1. Instantiate Agent ( modify this part to create your agent)
+    try:
+        agent = BasicAgent()
+    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)
+
+    # 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:
+             print("Fetched questions list is empty.")
+             return "Fetched questions list is empty or invalid format.", None
+        print(f"Fetched {len(questions_data)} questions.")
+    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
+
+    # 3. Run your Agent
+    results_log = []
+    answers_payload = []
+    print(f"Running agent on {len(questions_data)} questions...")
+    for item in questions_data:
+        task_id = item.get("task_id")
+        question_text = item.get("question")
+        if not task_id or question_text is None:
+            print(f"Skipping item with missing task_id or question: {item}")
+            continue
+        try:
+            submitted_answer = agent(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:
+             print(f"Error running agent on task {task_id}: {e}")
+             results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": f"AGENT 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)
+
+    # 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()
+        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:
+        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("# Basic Agent Evaluation Runner")
+    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.
+        """
+    )
+
+    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(
+        fn=run_and_submit_all,
+        outputs=[status_output, results_table]
+    )
+
+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")
+    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)

tools.py

@@ -0,0 +1,311 @@
+import os
+import re
+import requests
+import pandas as pd
+from typing import List
+from dotenv import load_dotenv
+
+from google import genai
+from google.genai import types
+
+from langchain_core.tools import tool
+from langchain.document_loaders import WebBaseLoader
+from langchain_experimental.tools import PythonREPLTool
+from langchain.text_splitter import CharacterTextSplitter
+from langchain_community.tools import DuckDuckGoSearchResults
+from langchain_community.retrievers import WikipediaRetriever
+from langchain_community.utilities import GoogleSerperAPIWrapper
+from langchain_community.document_loaders import ImageCaptionLoader, AssemblyAIAudioTranscriptLoader
+
+
+load_dotenv()
+DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
+
+
+def duckduck_websearch(query: str) -> str:
+    """
+    Performs a web search using the given query, downloads the content of two relevant web pages,
+    and returns their combined content as a raw string.
+
+    This is useful when the task requires analysis of web page content, such as retrieving poems, 
+    changelogs, or other textual resources.
+
+    Args:
+        query (str): The search query.
+
+    Returns:
+        str: The combined raw text content of the two retrieved web pages.
+    """
+    search_engine = DuckDuckGoSearchResults(output_format="list", num_results=2)
+    page_urls = [url["link"] for url in search_engine(query)]
+
+    loader = WebBaseLoader(web_paths=(page_urls))
+    docs = loader.load()
+
+    combined_text = "\n\n".join(doc.page_content[:15000] for doc in docs)
+
+    # Clean up excessive newlines, spaces and strip leading/trailing whitespace
+    cleaned_text = re.sub(r'\n{3,}', '\n\n', combined_text).strip()
+    cleaned_text = re.sub(r'[ \t]{6,}', ' ', cleaned_text)
+
+    # Strip leading/trailing whitespace
+    cleaned_text = cleaned_text.strip()
+    return cleaned_text
+
+
+def serper_websearch(query: str) -> str:
+    """
+    Performs a web search using the given query with SERPER Search Engine
+
+    Args:
+        query (str): The search query.
+    
+    Returns:
+        str: the search result
+    """
+    search = GoogleSerperAPIWrapper(serper_api_key=os.getenv("SERPER_API_KEY"))
+    results = search.run(query)
+    return results
+
+def visit_webpage(url: str) -> str:
+    """
+    Fetches raw HTML content of a web page.
+    
+    Args:
+        url: the webpage url
+    
+    Returns:
+        str: The combined raw text content of the webpage
+    """
+    try:
+        response = requests.get(url, timeout=5)
+        return response.text[:5000]
+    except Exception as e:
+        return f"[ERROR fetching {url}]: {str(e)}"
+
+def wiki_search(query: str) -> str:
+    """
+    Searches for a Wikipedia articles using the provided query and returns the content of the corresponding Wikipedia pages.
+
+    Args:
+        query (str): The search term to look up on Wikipedia.
+
+    Returns:
+        str: The text content of the Wikipedia articles related to the query.
+    """
+    retriever = WikipediaRetriever()
+    docs = retriever.invoke(query)
+    combined_text = "\n\n".join(doc.page_content for doc in docs)
+    return combined_text
+
+def youtube_viewer(youtube_url: str, question: str) -> str:
+    """
+    Analyzes a YouTube video from the provided URL and returns an answer 
+    to the given question based on the analysis results.
+
+    Args:
+        youtube_url (str): The URL of the YouTube video, in the format 
+            "https://www.youtube.com/...".
+        question (str): A question related to the content of the video.
+
+    Returns:
+        str: An answer to the question based on the video's content.
+    """
+    client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))
+    response = client.models.generate_content(
+        model='models/gemini-2.5-flash-preview-04-17',
+        contents=types.Content(
+            parts=[
+                types.Part(
+                    file_data=types.FileData(file_uri=youtube_url)
+                ),
+                types.Part(text=question)
+            ]
+        )
+    )
+    return response.text
+
+def text_splitter(text: str) -> List[str]:
+    """
+    Splits text into chunks using LangChain's CharacterTextSplitter.
+    
+    Args:
+        text: A string of text to split.
+    
+    Returns:
+        List[str]: a list of split text
+    """
+    splitter = CharacterTextSplitter(chunk_size=450, chunk_overlap=10)
+    return splitter.split_text(text)
+
+def read_file(task_id: str) -> str:
+    """
+    First download the file, then read its content
+    
+    Args:
+        dir: the task_id
+    
+    Returns:
+        str: the file content
+    """
+    file_url = f'{DEFAULT_API_URL}/files/{task_id}'
+    r = requests.get(file_url, timeout=15, allow_redirects=True)
+    with open('temp', "wb") as fp:
+        fp.write(r.content)
+    with open('temp') as f:
+        return f.read()
+
+def excel_read(task_id: str) -> str:
+    """
+    First download the excel file, then read its content
+    
+    Args:
+        dir: the task_id
+    
+    Returns:
+        str: the content of excel file
+    """
+    try:
+        file_url = f'{DEFAULT_API_URL}/files/{task_id}'
+        r = requests.get(file_url, timeout=15, allow_redirects=True)
+        with open('temp.xlsx', "wb") as fp:
+            fp.write(r.content)
+        # Read the Excel file
+        df = pd.read_excel('temp.xlsx')
+        # Run various analyses based on the query
+        result = (
+            f"Excel file loaded with {len(df)} rows and {len(df.columns)} columns.\n"
+        )
+        result += f"Columns: {', '.join(df.columns)}\n\n"
+        # Add summary statistics
+        result += "Summary statistics:\n"
+        result += str(df.describe())
+        return result
+    except Exception as e:
+        return f"Error analyzing Excel file: {str(e)}"
+   
+def csv_read(task_id: str) -> str:
+    """
+    First download the csv file, then read its content
+    
+    Args:
+        dir: the task_id
+    
+    Returns:
+        str: the content of csv file
+    """
+    try:
+        file_url = f'{DEFAULT_API_URL}/files/{task_id}'
+        r = requests.get(file_url, timeout=15, allow_redirects=True)
+        with open('temp.csv', "wb") as fp:
+            fp.write(r.content)
+        # Read the CSV file
+        df = pd.read_csv('temp.csv')
+        # Run various analyses based on the query
+        result = (
+            f"Excel file loaded with {len(df)} rows and {len(df.columns)} columns.\n"
+        )
+        result += f"Columns: {', '.join(df.columns)}\n\n"
+        # Add summary statistics
+        result += "Summary statistics:\n"
+        result += str(df.describe())
+        return result
+    except Exception as e:
+        return f"Error analyzing CSV file: {str(e)}"
+
+
+def mp3_listen(task_id: str) -> str:
+    """
+    First download the mp3 file, then listen to it
+    
+    Args:
+        dir: the task_id
+    
+    Returns:
+        str: the content of mp3 file
+    """
+    file_url = f'{DEFAULT_API_URL}/files/{task_id}'
+    r = requests.get(file_url, timeout=15, allow_redirects=True)
+    with open('temp.mp3', "wb") as fp:
+        fp.write(r.content)
+    loader = AssemblyAIAudioTranscriptLoader(file_path="temp.mp3", api_key=os.getenv("AssemblyAI_API_KEY"))
+    docs = loader.load()
+    contents = [doc.page_content for doc in docs]
+    return "\n".join(contents)
+    
+
+def image_caption(dir: str) -> str:
+    """
+    Understand the content of the provided image
+    
+    Args:
+        dir: the image url link
+    
+    Returns:
+        str: the image caption
+    """
+    loader = ImageCaptionLoader(images=[dir])
+    metadata = loader.load()
+    return metadata[0].page_content
+
+
+def run_python(code: str):
+    """ Run the given python code
+    
+    Args:
+        code: the python code
+    """
+    return PythonREPLTool().run(code)
+
+def multiply(a: float, b: float) -> float:
+    """
+    Multiply two numbers.
+    
+    Args:
+        a: first float
+        b: second float
+    
+    Returns:
+        float: the multiplication of a and b
+    """
+    return a * b
+
+def add(a: float, b: float) -> float:
+    """
+    Add two numbers.
+    
+    Args:
+        a: first float
+        b: second float
+    
+    Returns:
+        float: the sum of a and b
+    """
+    return a + b
+
+def subtract(a: float, b: float) -> float:
+    """
+    Subtract two numbers.
+    
+    Args:
+        a: first float
+        b: second float
+    
+    Returns:
+        float: the result after a subtracted by b
+    """
+    return a - b
+
+def divide(a: float, b: float) -> float:
+    """Divide two numbers.
+    
+    Args:
+        a: first float
+        b: second float
+    
+    Returns:
+        float: the result after a divided by b
+    """
+    if b == 0:
+        raise ValueError("Cannot divide by zero.")
+    return a / b