Update app.py

app.py

@@ -1,321 +1,86 @@
 import os
-from typing import Annotated, Optional, TypedDict
+from dotenv import load_dotenv
 import gradio as gr
-from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
-from langchain_openai import ChatOpenAI
-from langgraph.graph.message import add_messages
-from langgraph.graph import StateGraph, START
-from langgraph.prebuilt import tools_condition, ToolNode
 import requests
-import pandas as pd
-from langchain.tools import Tool
-from dotenv import load_dotenv
 
-from arxiv_searcher import ArxivSearcher
-from chess_algebraic_notation_retriever import ChessAlgebraicNotationMoveRetriever
-from excel_file_reader import ExcelFileReader
-from image_question_answer_tool import ImageQuestionAnswerTool
-from python_code_question_answer_tool import PythonCodeQuestionAnswerTool
-from tavily_searcher import TavilySearcher
-from transcriber import Transcriber
-from wikipedia_searcher import WikipediaSearcher
-from youtube_video_question_answer_tool import YoutubeVideoQuestionAnswerTool
+from typing import List, Dict, Union
+import pandas as pd
+import wikipediaapi
+import requests
+from bs4 import BeautifulSoup
+import urllib.parse
+import random
+from typing import List, Dict
 
 load_dotenv()
 
 # (Keep Constants as is)
 # --- Constants ---
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
-ASSOCIATED_FILE_ENDPOINT = f"{DEFAULT_API_URL}/files/"
 
-# --- Basic Agent Definition ---
-# ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------
-#search_tool = DuckDuckGoSearchRun()
-
-#search_tool = DuckDuckGoSearcherTool()
 
-def retrieve_task_file(task_id: str) -> Optional[bytes]:
-    """
-    Retrieve the task file for a given task ID.
-    """
-    try:
-        response = requests.get(ASSOCIATED_FILE_ENDPOINT + task_id, timeout=15)
-        response.raise_for_status()
-        if response.status_code != 200:
-            print(f"Error fetching file: {response.status_code}")
-            return None
-        #print(f"Fetched file: {response.content}")
-        return response.content
-    except requests.exceptions.RequestException as e:
-        print(f"Error fetching file: {e}")
-        return None
-    except Exception as e:
-        print(f"An unexpected error occurred fetching file: {e}")
-        return None
+# --- Basic Agent Definition ---
 
-def retrieve_next_chess_move_in_algebraic_notation(task_file_path: str, is_black_turn: bool) -> str:
-    """
-    Retrieve the next chess move in algebraic notation from an image path.
-    """
-    if task_file_path is None:
-        return "Error: Task file not found."
-    # Retrieve the next chess move in algebraic notation
-    next_chess_move = ChessAlgebraicNotationMoveRetriever().retrieve(task_file_path, is_black_turn)
-    return next_chess_move
-
-# Initialize the tool
-retrieve_next_chess_move_in_algebraic_notation_tool = Tool(
-    name="retrieve_next_chess_move_in_algebraic_notation",
-    func=retrieve_next_chess_move_in_algebraic_notation,
-    description="Retrieve the next chess move in algebraic notation from an image path."
-)
-
-def transcribe_audio(file_path: str) -> str:
-    if file_path is None:
-        return "Error: Audio path not found."
-    # Transcribe the audio
-    return Transcriber().transcribe(file_path)
-
-# Initialize the tool
-transcribe_audio_tool = Tool(
-    name="transcribe_audio",
-    func=transcribe_audio,
-    description="Transcribe the audio from an audio path."
-)
-
-# Initialize the tool
-answer_python_code_tool = PythonCodeQuestionAnswerTool()
-
-# Initialize the tool
-answer_image_question_tool = ImageQuestionAnswerTool()
-
-# Initialize the tool
-answer_youtube_video_question_tool = YoutubeVideoQuestionAnswerTool()
-
-'''def answer_youtube_video_question(youtube_video_url: str, question: str) -> str:
-    """
-    Answer the question based on the youtube video.
-    """
-    if youtube_video_url is None:
-        return "Error: Video not found."
-    # Download the video
-    video_path = YoutubeVideoDownloader().download_video(youtube_video_url)
-    # Answer the question
-    return VideoQuestionAnswer().answer(video_path, question)
-# Initialize the tool
-answer_youtube_video_question_tool = Tool(
-    name="answer_youtube_video_question",
-    func=answer_youtube_video_question,
-    description="Answer the question based on the youtube video."
-)'''
-
-def read_excel_file(file_path: str) -> str:
-    if file_path is None:
-        return "Error: File not found."
-    return ExcelFileReader().read_file(file_path)
-
-# Initialize the tool
-read_excel_file_tool = Tool(
-    name="read_excel_file",
-    func=read_excel_file,
-    description="Read the excel file."
-)
-
-# Initialize the tool
-wikipedia_search_tool = Tool(
-    name="wikipedia_search",
-    func=WikipediaSearcher().search,
-    description="Search Wikipedia for a given query."
-)
-
-# Initialize the tool
-arxiv_search_tool = Tool(
-    name="arxiv_search",
-    func=ArxivSearcher().search,
-    description="Search Arxiv for a given query."
-)
-
-tavily_search_tool = Tool(
-    name="tavily_search",
-    func=TavilySearcher().search,
-    description="Search the web for a given query."
-)
-
-def format_gaia_answer(answer: str) -> str:
-    llm = ChatOpenAI(model="o3-mini", openai_api_key=os.getenv("OPENAI_API_KEY"))
-    prompt = f"""
-    You are formatting answers for the GAIA benchmark, which requires responses to be concise and unambiguous.
-    Given the answer: {answer}
-    Return the answer in the correct GAIA format:
-    - If the answer is a single word or number, return it without any additional text or formatting.
-    - If the answer is a list, return a comma-separated list without any additional text or formatting.
-    - If the answer is a string, return it without any additional text or formatting.
-    Do not include any prefixes, dots, enumerations, explanations, or quotation marks.
-    Do not include any additional text or formatting.
-    """
-    response = llm.invoke(prompt)
-    # Delete double quotes
-    return response.content.strip().replace('"', '')
-
-class AgentState(TypedDict):
-    # The document provided
-    messages: Annotated[list[AnyMessage], add_messages]
-    file_path: Optional[str]
-    
 class BasicAgent:
     def __init__(self):
-        tools = [
-            tavily_search_tool, 
-            arxiv_search_tool,
-            wikipedia_search_tool, 
-            transcribe_audio_tool, 
-            answer_python_code_tool, 
-            answer_image_question_tool, 
-            answer_youtube_video_question_tool,
-            read_excel_file_tool
+        self.headers = {
+            'User-Agent': self._get_random_user_agent(),
+            'Accept-Language': 'en-US,en;q=0.5',
+        }
+    
+    def _get_random_user_agent(self) -> str:
+        browsers = [
+            'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
+            'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:89.0) Gecko/20100101 Firefox/89.0',
+            'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.1.1 Safari/605.1.15'
         ]
-        '''llm = ChatGoogleGenerativeAI(
-            model="gemini-2.0-flash", 
-            temperature=0.2,
-            api_key=os.getenv("GEMINI_API_KEY")
-        )'''
-        llm = ChatOpenAI(model="o3-mini", openai_api_key=os.getenv("OPENAI_API_KEY"))
-        self.llm_with_tools = llm.bind_tools(tools)
-        builder = StateGraph(AgentState)
-
-        # Define nodes: these do the work
-        builder.add_node("assistant", self.assistant)
-        builder.add_node("tools", ToolNode(tools))
-
-        # Define edges: these determine how the control flow moves
-        builder.add_edge(START, "assistant")
-        builder.add_conditional_edges(
-            "assistant",
-            # If the latest message requires a tool, route to tools
-            # Otherwise, provide a direct response
-            tools_condition,
-        )
-        builder.add_edge("tools", "assistant")
-        self.agent = builder.compile()
-
-        print("BasicAgent initialized.")
-
-    def assistant(self, state: AgentState):
-        # System message
-        textual_description_of_tools="""
-            tavily_search(query: str) -> str:
-                Search the web for a given query.
-                Args:
-                    query: Query to search the web for (string).
-                Returns:
-                    A single string containing the information found on the web.
-            arxiv_search(query: str) -> str:
-                Search Arxiv, that contains scientific papers, for a given query.
-                Args:
-                    query: Query to search Arxiv for (string).
-                Returns:
-                    A single string containing the answer to the question.
-            wikipedia_search(query: str) -> str:
-                Search Wikipedia for a given query.
-                Args:
-                    query: Query to search Wikipedia for (string).
-                Returns:
-                    A single string containing the answer to the question.
-            transcribe_audio(file_path: str) -> str:
-                Transcribe the audio from an audio path.
-                Args:
-                    file_path: File path of the audio file (string).
-                Returns:
-                    A single string containing the transcribed text from the audio.
-            
-            answer_python_code(file_path: str, question: str) -> str:
-                Answer the question based on the python code.
-                Args:
-                    file_path: File path of the python file (string).
-                    question: Question to answer (string).
-                Returns:
-                    A single string containing the answer to the question.
-            
-            answer_image_question(file_path: str, question: str) -> str:
-                Answer the question based on the image.
-                Args:
-                    file_path: File path of the image (string).
-                    question: Question to answer (string).
-                Returns:
-                    A single string containing the answer to the question.
-            
-            download_youtube_video(youtube_video_url: str) -> str:
-                Download the Youtube video into a local file based on the URL
-                Args:
-                    youtube_video_url: A youtube video url (string).
-                Returns:
-                    A single string containing the file path of the downloaded youtube video.
-            answer_youtube_video_question(file_path: str, question: str) -> str:
-                Answer the question based on file path of the downloaded youtube video
-                Args:
-                    file_path: File path of the downloaded youtube video (string).
-                    question: Question to answer (string).
-                Returns:
-                    A single string containing the answer to the question.
-            
-            read_excel_file(file_path: str) -> str:
-                Read the excel file.
-                Args:
-                    file_path: File path of the excel file (string).
-                Returns:
-                    A markdown formatted string containing the contents of the excel file.
-        """
-        file_path=state["file_path"]
-        prompt = f"""
-            You are a helpful assistant that can analyse images, videos, excel files and Python scripts and run computations with provided tools:
-            {textual_description_of_tools} 
-            You have access to the file path of the attached file in case it's informed. Currently the file path is: {file_path}
-            Be direct and specific. GAIA benchmark requires exact matching answers.
-            For example, if asked "What is the capital of France?", respond simply with "Paris".
-            Do not include any prefixes, dots, enumerations, explanations, or quotation marks.
-            Do not include any additional text or formatting.
-            If you are required a number, return a number, not the items.
-        """
-        sys_msg = SystemMessage(content=prompt)
+        return random.choice(browsers)
+
+    def search(self, query: str, num_results: int = 3) -> List[Dict]:
+        encoded_query = urllib.parse.quote_plus(query)
+        url = f"https://www.google.com/search?q={encoded_query}&num={num_results + 2}"
+        
+        try:
+            response = requests.get(url, headers=self.headers, timeout=10)
+            response.raise_for_status()
+            return self._parse_results(response.text, num_results)
+        except Exception as e:
+            print(f"Search failed: {str(e)}")
+            return []
 
-        return {
-            "messages": [self.llm_with_tools.invoke([sys_msg] + state["messages"], config={"configurable": {"file_path": state["file_path"]}})],
-            "file_path": state["file_path"]
-        }
-        '''return {
-            "messages": [self.llm_with_tools.invoke(
-                state["messages"],
-                config={"configurable": {"file_path": state["file_path"]}}  # Aquí pasas el task_id
-            )],
-            "file_path": state["file_path"]
-        }'''
     
-    def __call__(self, question: str, task_id: str, file_name: str) -> str:
-        print(f"######################### Agent received question (first 50 chars): {question[:50]}... with file_name: {file_name}")
-
-        # Get the file path
-        tmp_file_path = None
-        if file_name is not None and file_name != "":
-            file_content = retrieve_task_file(task_id)
-            if file_content is not None:
-                print(f"Saving file {file_name} to tmp folder")
-                tmp_file_path = f"tmp/{file_name}"
-                with open(tmp_file_path, "wb") as f:
-                    f.write(file_content)
-        # Show the file path
-        print(f"File path: {tmp_file_path}")
-
-        messages = self.agent.invoke({"messages": [HumanMessage(question)], "file_path": tmp_file_path})
-        # Show the messages
-        for m in messages['messages']:
-            m.pretty_print()
-        answer = messages["messages"][-1].content
-        answer = format_gaia_answer(answer)
-        print(f"######################### Agent returning answer: {answer}\n")
-        # Delete the file
-        if tmp_file_path is not None:
-            os.remove(tmp_file_path)
-        return answer
+
+    def _parse_results(self, html: str, max_results: int) -> List[Dict]:
+        soup = BeautifulSoup(html, 'html.parser')
+        results = []
+        
+        for i, result in enumerate(soup.select('.tF2Cxc, .g')[:max_results]):
+            title = result.select_one('h3, .LC20lb')
+            link = result.find('a')['href']
+            snippet = result.select_one('.IsZvec, .VwiC3b')
+            
+            if title and link:
+                results.append({
+                    'position': i + 1,
+                    'title': title.get_text(),
+                    'link': link if link.startswith('http') else f"https://www.google.com{link}",
+                    'snippet': snippet.get_text() if snippet else None
+                })
+        return results
+
+    def pretty_print(self, results: List[Dict]) -> str:
+        output = []
+        for res in results:
+            output.append(
+                f"{res['position']}. {res['title']}\n"
+                f"   🔗 {res['link']}\n"
+                f"   📝 {res['snippet'] or 'No description available'}\n"
+            )
+        return "\n".join(output)
+
+    def __call__(self, query: str) -> str:
+        """Added this to make the agent callable"""
+        return self.pretty_print(self.search(query))