Update app.py

app.py

@@ -12,7 +12,7 @@ import traceback
 # --- Core Libraries ---
 try:
     from langchain_openai import AzureChatOpenAI
-    from ddgs import DDGS
+    from ddgs import DDGS # Use the new, correct library
     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 Smart Orchestrator + ReAct Agent ---
+# --- Agent Definition: A Professional ReAct Agent with Dynamic Examples ---
 class BasicAgent:
     def __init__(self):
-        print("Initializing Hybrid Agent...")
+        print("Initializing ReAct Agent...")
         try:
             self.llm = AzureChatOpenAI(
                 azure_endpoint="https://dsap.openai.azure.com/",
@@ -38,111 +38,172 @@ class BasicAgent:
             raise KeyError("CRITICAL: 'AZURE_API_KEY' secret is missing.")
         
         self.tools = {
-            "search_and_browse": self.search_and_browse,
-            "python_file_analyzer": self.python_file_analyzer,
+            "search": self.search,
+            "browse": self.browse,
+            "python": self.python,
+            "youtube_transcript": self.youtube_transcript,
         }
-        self.react_system_prompt = self._create_react_prompt()
         print("Agent initialized.")
 
-    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.
+    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 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.
+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_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 = ""
+    def search(self, query: str) -> str:
+        """Searches the web with DuckDuckGo to find relevant URLs and information."""
         try:
             with DDGS() as ddgs:
-                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."
+                return str([r for r in ddgs.text(query, max_results=4)])
+        except Exception as e: return f"Error during search: {e}"
 
-        # For downloadable files, use Python
+    def browse(self, url: str) -> str:
+        """Gets the full, clean text content of a single webpage URL."""
         try:
-            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}"
+            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 Orchestrator Logic ---
+    # --- Main ReAct Loop ---
     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]
-
-        # STRATEGY 1: Deterministic File Handling (Orchestrator)
+        system_prompt = self._create_system_prompt(file_url=file_url)
+        
+        question = task.get("question", "")
         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:"
+            question += f"\nFile available at: {file_url}"
+
+        # Initialize the history correctly for the ReAct loop
+        history = f"{system_prompt}\nQuestion: {question}\nThought:"
+
+        for i in range(8):
+            print(f"--- Step {i+1} ---")
+            
+            # The full prompt for this step is the entire history
+            llm_response = self.llm.invoke(history).content.strip()
+            
+            # Append the agent's reasoning to the history
+            history += f"\n{llm_response}"
+
+            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:
-                    return llm_response # Fallback
-            return "Agent could not reach a final answer after multiple web searches."
+                    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 "Agent could not reach a final answer after 8 steps."
 
 # --- Your Original, Correct Submission and Gradio Code ---
 def run_and_submit_all(profile: gr.OAuthProfile | None):