Update app.py

app.py

@@ -8,24 +8,27 @@ import re
 import io
 from contextlib import redirect_stdout
 import traceback
+import tempfile
 
 # --- Core Libraries ---
 try:
     from langchain_openai import AzureChatOpenAI
-    from ddgs import DDGS # Use the new, correct library
+    from ddgs import DDGS
     from bs4 import BeautifulSoup
     from youtube_transcript_api import YouTubeTranscriptApi
-    import openpyxl, librosa, soundfile as sf, numpy as np
+    import openpyxl, numpy as np
+    import whisper # The definitive audio transcription library
+    import ffmpeg
 except ImportError:
     raise ImportError("Required libraries are not installed. Check requirements.txt.")
 
 # --- Constants ---
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 
-# --- Agent Definition: A Professional ReAct Agent with Dynamic Examples ---
+# --- Agent Definition: The Specialist Architecture ---
 class BasicAgent:
     def __init__(self):
-        print("Initializing ReAct Agent...")
+        print("Initializing Specialist Agent...")
         try:
             self.llm = AzureChatOpenAI(
                 azure_endpoint="https://dsap.openai.azure.com/",
@@ -37,173 +40,91 @@ class BasicAgent:
         except KeyError:
             raise KeyError("CRITICAL: 'AZURE_API_KEY' secret is missing.")
         
+        # High-level specialist tools, not a long list of simple ones.
         self.tools = {
-            "search": self.search,
-            "browse": self.browse,
-            "python": self.python,
-            "youtube_transcript": self.youtube_transcript,
+            "web_search_specialist": self.web_search_specialist,
+            "file_analysis_specialist": self.file_analysis_specialist,
         }
+        self.whisper_model = whisper.load_model("base")
         print("Agent initialized.")
 
-    def _create_system_prompt(self, file_url=None) -> str:
-        """Creates the master prompt, dynamically injecting a file-handling example if a URL is provided."""
-        tool_docs = "\n".join([f"- {name}: {inspect.getdoc(func)}" for name, func in self.tools.items()])
-
-        web_search_example = """
-**Example: Web Search**
-Question: Who was the prime minister of the UK in 1999?
-Thought: I need to find out who was the prime minister of the UK in 1999. I will use the search tool.
-Action: search
-Action Input: prime minister of UK 1999
-Observation: [{{'title': 'Tony Blair - Wikipedia', 'href': 'https://en.wikipedia.org/wiki/Tony_Blair', ...}}]
-Thought: The search results point to Tony Blair. The first link looks promising. I will browse the Wikipedia page to confirm.
-Action: browse
-Action Input: https://en.wikipedia.org/wiki/Tony_Blair
-Observation: [Page content confirming Tony Blair was Prime Minister from 1997 to 2007]
-Thought: I have confirmed the answer from a reliable source.
-Final Answer: Tony Blair"""
-
-        file_analysis_example = ""
-        if file_url:
-            code_snippet = "# This is a placeholder, will be replaced by a specific file handler\n"
-            if file_url.endswith(('.xlsx', '.csv')):
-                code_snippet = f"""
-import pandas as pd
-import requests
-import io
-# The user's file is at this URL, which MUST be used.
-url = '{file_url}'
-response = requests.get(url)
-df = pd.read_excel(io.BytesIO(response.content))
-# Now, I must analyze the dataframe `df` to answer the question.
-# For example, to see the first few rows, I can print(df.head()).
-# To calculate total sales, I would use print(df['Sales'].sum()).
-print(df.to_string())
-"""
-            elif file_url.endswith('.py'):
-                 code_snippet = f"""
-import requests
-# The user's Python code file is at this URL, which MUST be used.
-url = '{file_url}'
-response = requests.get(url)
-python_code_to_run = response.text
-# Now, I must execute this code to find the output.
-# I will use another python action to run the code.
-print("Code downloaded. Ready for execution in the next step.")
-"""
-            
-            if code_snippet:
-                file_analysis_example = f"""
-**Example: File Analysis (Use this exact code pattern)**
-Question: Analyze the attached file. File available at: {file_url}
-Thought: The user has provided a file. I must use the `python` tool to download and analyze it using the exact URL from the question. The following code pattern is perfect for this. I will copy it exactly.
-Action: python
-Action Input:
-{code_snippet}
-Observation: [The output of the python script]
-Thought: I have analyzed the file content. Now I can answer the user's question based on the script's output.
-Final Answer: [Answer based on the script's output]"""
-
-        return f"""
-You are a helpful assistant that answers questions by thinking step-by-step and using the tools provided.
-
-**Process:**
-1.  **Thought:** Analyze the user's question and create a plan. If you see an example below that matches your plan, follow it exactly.
-2.  **Action:** Choose ONE tool from the list: {", ".join(self.tools.keys())}.
-3.  **Action Input:** Provide the input for the chosen tool. This can be multi-line.
-4.  **Observation:** After you use a tool, you will see its output.
-5.  Repeat this Thought/Action/Action Input/Observation cycle until you are certain you have the final answer.
-6.  **Thought:** Conclude that you have the final answer.
-7.  **Final Answer:** Provide the final, direct answer to the user's question.
-
-You have access to the following tools:
-{tool_docs}
-
-{web_search_example}
-{file_analysis_example}
-
-Begin!
-"""
-
-    # --- Tool Definitions ---
-    def search(self, query: str) -> str:
-        """Searches the web with DuckDuckGo to find relevant URLs and information."""
+    # --- Specialist Tool Definitions ---
+    def web_search_specialist(self, query: str) -> str:
+        """A specialist tool that searches the web and automatically browses the top 3 results."""
+        print(f"Tool: web_search_specialist, Query: {query}")
+        context = ""
         try:
             with DDGS() as ddgs:
-                return str([r for r in ddgs.text(query, max_results=4)])
-        except Exception as e: return f"Error during search: {e}"
-
-    def browse(self, url: str) -> str:
-        """Gets the full, clean text content of a single webpage URL."""
-        try:
-            response = requests.get(url, timeout=10, headers={'User-Agent': 'Mozilla/5.0'})
-            soup = BeautifulSoup(response.content, 'html.parser')
-            return ' '.join(soup.get_text().split())[:4000]
-        except Exception as e: return f"Error browsing {url}: {e}"
-
-    def python(self, code: str) -> str:
-        """Executes Python code to analyze data or files. Use `requests` to download files from URLs."""
-        code = code.strip().strip("`").replace("python\n", "").strip()
-        buffer = io.StringIO()
-        try:
-            safe_globals = {'pd': pd, 'np': np, 'requests': requests, 'io': io, 'librosa': librosa, 'sf': sf, 'openpyxl': openpyxl}
-            with redirect_stdout(buffer):
-                exec(code, safe_globals)
-            return f"Execution successful. Output:\n{buffer.getvalue()}"
-        except Exception as e: return f"Execution failed. Error:\n{traceback.format_exc()}"
-
-    def youtube_transcript(self, url: str) -> str:
-        """Fetches the full transcript of a YouTube video from its URL."""
-        try:
-            video_id = re.search(r"(?<=v=)[\w-]+", url).group(0)
-            return " ".join([item['text'] for item in YouTubeTranscriptApi.get_transcript(video_id)])
-        except Exception as e: return f"Error fetching transcript: {e}"
-
-    # --- Main ReAct Loop ---
-    def __call__(self, task: Dict[str, Any]) -> str:
-        file_url = task.get("files", [None])[0]
-        system_prompt = self._create_system_prompt(file_url=file_url)
+                results = [r for r in ddgs.text(query, max_results=3)]
+                if not results: return f"No results found for '{query}'."
+                
+                for result in results:
+                    try:
+                        url = result['href']
+                        response = requests.get(url, timeout=10, headers={'User-Agent': 'Mozilla/5.0'})
+                        soup = BeautifulSoup(response.content, 'html.parser')
+                        context += f"Source: {url}\nContent: {' '.join(soup.get_text().split())[:1500]}\n\n"
+                    except Exception as e:
+                        context += f"Could not browse {url}: {e}\n\n"
+            return context
+        except Exception as e:
+            return f"Error during search: {e}"
+
+    def file_analysis_specialist(self, file_url: str) -> str:
+        """A specialist tool that downloads and analyzes a file from a URL using deterministic Python."""
+        print(f"Tool: file_analysis_specialist, URL: {file_url}")
         
-        question = task.get("question", "")
-        if file_url:
-            question += f"\nFile available at: {file_url}"
-
-        # Initialize the history correctly for the ReAct loop
-        history = f"{system_prompt}\nQuestion: {question}\nThought:"
+        if any(file_url.endswith(ext) for ext in ['.png', '.jpg', '.jpeg', '.gif']):
+            return "Limitation: I cannot analyze image content. Please describe the image."
 
-        for i in range(8):
-            print(f"--- Step {i+1} ---")
+        try:
+            response = requests.get(file_url)
+            response.raise_for_status()
             
-            # The full prompt for this step is the entire history
-            llm_response = self.llm.invoke(history).content.strip()
+            if file_url.endswith('.xlsx'):
+                df = pd.read_excel(io.BytesIO(response.content))
+                return f"Successfully read the Excel file. Here is its full content:\n\n{df.to_string()}"
             
-            # Append the agent's reasoning to the history
-            history += f"\n{llm_response}"
+            elif file_url.endswith('.py'):
+                return f"Successfully read the Python file. Here is its content:\n\n{response.text}"
+
+            elif file_url.endswith(('.mp3', '.wav')):
+                 with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as tmp_audio_file:
+                    tmp_audio_file.write(response.content)
+                    tmp_audio_path = tmp_audio_file.name
+                 
+                 print(f"Transcribing audio file: {tmp_audio_path}")
+                 result = self.whisper_model.transcribe(tmp_audio_path, fp16=False)
+                 os.remove(tmp_audio_path)
+                 return f"Successfully transcribed the audio file. Here is the transcript:\n\n{result['text']}"
 
-            final_answer_match = re.search(r"Final Answer:\s*(.*)", llm_response, re.DOTALL)
-            if final_answer_match:
-                answer = final_answer_match.group(1).strip()
-                print(f"Final Answer Found: {answer}")
-                return answer
-            
-            action_match = re.search(r"Action:\s*(\w+)\s*Action Input:((.|\n)*)", llm_response)
-            if action_match:
-                tool_name = action_match.group(1).strip()
-                tool_input = action_match.group(2).strip(' \n"`')
-                if tool_name in self.tools:
-                    try:
-                        tool_result = self.tools[tool_name](tool_input)
-                    except Exception as e:
-                        tool_result = f"Error calling tool {tool_name}: {e}"
-                else:
-                    tool_result = f"Error: Unknown tool '{tool_name}'."
-                
-                # Append the observation to the history for the next step
-                history += f"\nObservation: {tool_result}\nThought:"
             else:
-                return llm_response
+                return "Unsupported file type."
+        except Exception as e:
+            return f"Failed to download or process the file. Error: {traceback.format_exc()}"
 
-        return "Agent could not reach a final answer after 8 steps."
+    # --- Main Orchestrator Logic ---
+    def __call__(self, task: Dict[str, Any]) -> str:
+        question = task.get("question", "")
+        print(f"\n--- New Task ---\nQuestion: {question[:150]}...")
+        
+        file_url = task.get("files", [None])[0]
+        context = ""
+
+        # The Orchestrator makes a simple, reliable decision.
+        if file_url:
+            context = self.file_analysis_specialist(file_url)
+        else:
+            context = self.web_search_specialist(query=question)
+        
+        # The LLM's only job is to summarize the context from the specialist tool.
+        final_prompt = f"Based ONLY on the following context, provide a direct and concise answer to the user's question. Do not use any other information. If the context is insufficient, say so.\n\nContext:\n{context}\n\nUser Question:\n{question}"
+        try:
+            final_answer = self.llm.invoke(final_prompt).content
+            print(f"Final Answer: {final_answer}")
+            return final_answer
+        except Exception as e:
+            return f"Error during final answer generation: {e}"
 
 # --- Your Original, Correct Submission and Gradio Code ---
 def run_and_submit_all(profile: gr.OAuthProfile | None):
@@ -235,7 +156,7 @@ def run_and_submit_all(profile: gr.OAuthProfile | None):
     submission_data = {"username": username, "agent_code": agent_code, "answers": answers_payload}
     
     try:
-        response = requests.post(f"{DEFAULT_API_URL}/submit", json=submission_data, timeout=60)
+        response = requests.post(f"{DEFAULT_API_URL}/submit", json=submission_data, timeout=90)
         response.raise_for_status()
         result_data = response.json()
         final_status = (f"Submission Successful! Score: {result_data.get('score', 'N/A')}%")