Update app.py

app.py

@@ -12,7 +12,7 @@ import traceback
 # --- 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
@@ -22,10 +22,10 @@ except ImportError:
 # --- Constants ---
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 
-# --- Agent Definition: A Professional ReAct Agent with Dynamic Examples ---
+# --- Agent Definition: A Smart Orchestrator + ReAct Agent ---
 class BasicAgent:
     def __init__(self):
-        print("Initializing ReAct Agent...")
+        print("Initializing Hybrid Agent...")
         try:
             self.llm = AzureChatOpenAI(
                 azure_endpoint="https://dsap.openai.azure.com/",
@@ -38,156 +38,111 @@ class BasicAgent:
             raise KeyError("CRITICAL: 'AZURE_API_KEY' secret is missing.")
         
         self.tools = {
-            "search": self.search,
-            "browse": self.browse,
-            "python": self.python,
-            "youtube_transcript": self.youtube_transcript,
+            "search_and_browse": self.search_and_browse,
+            "python_file_analyzer": self.python_file_analyzer,
         }
+        self.react_system_prompt = self._create_react_prompt()
         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. 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 = ""
-            if file_url.endswith(('.xlsx', '.csv')):
-                code_snippet = f"""
-import pandas as pd
-import requests
-import io
-url = '{file_url}'
-response = requests.get(url)
-df = pd.read_excel(io.BytesIO(response.content))
-print(df.to_string())
-"""
-            elif file_url.endswith('.py'):
-                 code_snippet = f"""
-import requests
-url = '{file_url}'
-response = requests.get(url)
-python_code_to_run = response.text
-print(python_code_to_run)
-"""
-            
-            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.
+    def _create_react_prompt(self) -> str:
+        """Creates the prompt for the ReAct loop (for web questions)."""
+        return """
+You are a helpful assistant that answers questions by searching the web.
 **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}
-
+1.  **Thought:** Analyze the user's question and decide what to search for.
+2.  **Action:** Use the `search_and_browse` tool.
+3.  **Action Input:** Provide a concise search query.
+4.  **Observation:** You will see the content of the top search results.
+5.  **Thought:** Analyze the search results. If you have enough information, provide the final answer. If not, refine your search and use the `search_and_browse` tool again.
+6.  **Final Answer:** Provide the final, direct answer to the user's question.
 Begin!
 """
 
     # --- Tool Definitions ---
-    def search(self, query: str) -> str:
-        """Searches the web with DuckDuckGo to find relevant URLs and information."""
+    def search_and_browse(self, query: str) -> str:
+        """Searches the web and browses the top 3 results to gather context."""
+        print(f"Tool: search_and_browse, 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}"
+                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 python_file_analyzer(self, file_url: str) -> str:
+        """
+        Downloads a file from a URL and analyzes its content using Python.
+        This tool is called directly by the orchestrator, not by the LLM.
+        """
+        print(f"Tool: python_file_analyzer, URL: {file_url}")
+        
+        # Handle non-downloadable file types first
+        if file_url.endswith(('.png', '.jpg', '.jpeg', '.gif')):
+            return "Limitation: I cannot analyze image content. Please describe the image."
+        if file_url.endswith(('.mp3', '.wav')):
+            return "Limitation: I cannot reliably transcribe audio files. Please provide a transcript."
 
-    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."""
+        # For downloadable files, use Python
         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}"
+            response = requests.get(file_url)
+            response.raise_for_status()
+            
+            if file_url.endswith('.xlsx'):
+                df = pd.read_excel(io.BytesIO(response.content))
+                return f"Successfully read the Excel file. Here is its content:\n\n{df.to_string()}"
+            elif file_url.endswith('.py'):
+                return f"Successfully read the Python file. Here is its content:\n\n{response.text}"
+            else:
+                return "Unsupported file type."
+        except Exception as e:
+            return f"Failed to download or process the file. Error: {e}"
 
-    # --- Main ReAct Loop ---
+    # --- Main Orchestrator Logic ---
     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)
-        
         question = task.get("question", "")
-        if file_url:
-            question += f"\nFile available at: {file_url}"
-
-        prompt = f"{system_prompt}\nQuestion: {question}\nThought:"
-        history = ""
-
-        for i in range(8):
-            full_prompt = prompt + history
-            llm_response = self.llm.invoke(full_prompt).content.strip()
-            history += f"\n{llm_response}"
-
-            final_answer_match = re.search(r"Final Answer:\s*(.*)", llm_response, re.DOTALL)
-            if final_answer_match:
-                return final_answer_match.group(1).strip()
-            
-            # *** THIS IS THE ONLY LINE THAT HAS BEEN CHANGED ***
-            # Corrected the typo from `ll.response` to `llm_response`
-            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}'."
-                history += f"\nObservation: {tool_result}\nThought:"
-            else:
-                return llm_response
+        print(f"\n--- New Task ---\nQuestion: {question[:150]}...")
+        
+        file_url = task.get("files", [None])[0]
 
-        return "Agent could not reach a final answer after 8 steps."
+        # STRATEGY 1: Deterministic File Handling (Orchestrator)
+        if file_url:
+            # The orchestrator calls the tool directly, removing LLM unreliability
+            context = self.python_file_analyzer(file_url)
+            final_prompt = f"Based ONLY on the following file content, provide a direct and concise answer to the user's question.\n\nFile Content:\n{context}\n\nUser Question:\n{question}"
+            return self.llm.invoke(final_prompt).content
+
+        # STRATEGY 2: Flexible Web Search (ReAct Loop)
+        else:
+            prompt = f"{self.react_system_prompt}\nQuestion: {question}\nThought:"
+            history = ""
+            for i in range(5): # Max 5 steps
+                full_prompt = prompt + history
+                llm_response = self.llm.invoke(full_prompt).content.strip()
+                
+                final_answer_match = re.search(r"Final Answer:\s*(.*)", llm_response, re.DOTALL)
+                if final_answer_match:
+                    return final_answer_match.group(1).strip()
+
+                # For web search, we assume the only tool is search_and_browse
+                action_match = re.search(r"Action Input:\s*(.*)", llm_response)
+                if action_match:
+                    query = action_match.group(1).strip()
+                    observation = self.search_and_browse(query)
+                    history += f"\n{llm_response}\nObservation: {observation}\nThought:"
+                else:
+                    return llm_response # Fallback
+            return "Agent could not reach a final answer after multiple web searches."
 
 # --- Your Original, Correct Submission and Gradio Code ---
 def run_and_submit_all(profile: gr.OAuthProfile | None):