Update app.py

app.py

@@ -12,6 +12,9 @@ from huggingface_hub import InferenceClient
 import base64
 from PIL import Image
 import io
+import tempfile
+import urllib.parse
+from pathlib import Path
 
 # --- Constants ---
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
@@ -21,6 +24,96 @@ cached_answers = {}
 cached_questions = []
 processing_status = {"is_processing": False, "progress": 0, "total": 0}
 
+# --- File Download Utility ---
+def download_attachment(url: str, temp_dir: str) -> Optional[str]:
+    """
+    Download an attachment from URL to a temporary directory.
+    Returns the local file path if successful, None otherwise.
+    """
+    try:
+        response = requests.get(url, timeout=30)
+        response.raise_for_status()
+        
+        # Extract filename from URL or create one based on content type
+        parsed_url = urllib.parse.urlparse(url)
+        filename = os.path.basename(parsed_url.path)
+        
+        if not filename or '.' not in filename:
+            # Try to determine extension from content type
+            content_type = response.headers.get('content-type', '').lower()
+            if 'image' in content_type:
+                if 'jpeg' in content_type or 'jpg' in content_type:
+                    filename = f"attachment_{int(time.time())}.jpg"
+                elif 'png' in content_type:
+                    filename = f"attachment_{int(time.time())}.png"
+                else:
+                    filename = f"attachment_{int(time.time())}.img"
+            elif 'audio' in content_type:
+                if 'mp3' in content_type:
+                    filename = f"attachment_{int(time.time())}.mp3"
+                elif 'wav' in content_type:
+                    filename = f"attachment_{int(time.time())}.wav"
+                else:
+                    filename = f"attachment_{int(time.time())}.audio"
+            elif 'python' in content_type or 'text' in content_type:
+                filename = f"attachment_{int(time.time())}.py"
+            else:
+                filename = f"attachment_{int(time.time())}.file"
+        
+        file_path = os.path.join(temp_dir, filename)
+        
+        with open(file_path, 'wb') as f:
+            f.write(response.content)
+        
+        print(f"Downloaded attachment: {url} -> {file_path}")
+        return file_path
+        
+    except Exception as e:
+        print(f"Failed to download attachment {url}: {e}")
+        return None
+
+# --- Code Processing Tool ---
+class CodeAnalysisTool:
+    def __init__(self, model_name: str = "meta-llama/Llama-3.1-8B-Instruct"):
+        self.client = InferenceClient(model=model_name, provider="sambanova")
+        
+    def analyze_code(self, code_path: str) -> str:
+        """
+        Analyze Python code and return insights.
+        """
+        try:
+            with open(code_path, 'r', encoding='utf-8') as f:
+                code_content = f.read()
+            
+            # Limit code length for analysis
+            if len(code_content) > 5000:
+                code_content = code_content[:5000] + "\n... (truncated)"
+            
+            analysis_prompt = f"""Analyze this Python code and provide a concise summary of:
+1. What the code does (main functionality)
+2. Key functions/classes
+3. Any notable patterns or issues
+4. Input/output behavior if applicable
+
+Code:
+```python
+{code_content}
+```
+
+Provide a brief, focused analysis:"""
+
+            messages = [{"role": "user", "content": analysis_prompt}]
+            response = self.client.chat_completion(
+                messages=messages,
+                max_tokens=500,
+                temperature=0.3
+            )
+            
+            return response.choices[0].message.content.strip()
+            
+        except Exception as e:
+            return f"Code analysis failed: {e}"
+
 # --- Image Processing Tool ---
 class ImageAnalysisTool:
     def __init__(self, model_name: str = "microsoft/Florence-2-large"):
@@ -111,6 +204,7 @@ class IntelligentAgent:
         self.client = InferenceClient(model=model_name, provider="sambanova")
         self.image_tool = ImageAnalysisTool()
         self.audio_tool = AudioTranscriptionTool()
+        self.code_tool = CodeAnalysisTool(model_name)
         self.debug = debug
         if self.debug:
             print(f"IntelligentAgent initialized with model: {model_name}")
@@ -148,11 +242,108 @@ class IntelligentAgent:
                 print(f"Both chat completion and text generation failed: {e}")
             raise e
 
-    def _process_media_files(self, image_files: List[str] = None, audio_files: List[str] = None) -> str:
+    def _detect_and_download_attachments(self, question_data: dict) -> Tuple[List[str], List[str], List[str]]:
+        """
+        Detect and download attachments from question data.
+        Returns (image_files, audio_files, code_files)
+        """
+        image_files = []
+        audio_files = []
+        code_files = []
+        
+        # Create temporary directory for downloads
+        temp_dir = tempfile.mkdtemp(prefix="agent_attachments_")
+        
+        # Check for attachments in various fields
+        attachments = []
+        
+        # Common fields where attachments might be found
+        attachment_fields = ['attachments', 'files', 'media', 'resources']
+        
+        for field in attachment_fields:
+            if field in question_data:
+                field_data = question_data[field]
+                if isinstance(field_data, list):
+                    attachments.extend(field_data)
+                elif isinstance(field_data, str):
+                    attachments.append(field_data)
+        
+        # Also check if the question text contains URLs
+        question_text = question_data.get('question', '')
+        if 'http' in question_text:
+            import re
+            urls = re.findall(r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+', question_text)
+            attachments.extend(urls)
+        
+        # Download and categorize attachments
+        for attachment in attachments:
+            if isinstance(attachment, dict):
+                url = attachment.get('url') or attachment.get('link') or attachment.get('file_url')
+                file_type = attachment.get('type', '').lower()
+            else:
+                url = attachment
+                file_type = ''
+            
+            if not url:
+                continue
+                
+            # Download the file
+            file_path = download_attachment(url, temp_dir)
+            if not file_path:
+                continue
+            
+            # Categorize based on extension or type
+            file_ext = Path(file_path).suffix.lower()
+            
+            if file_type:
+                if 'image' in file_type or file_ext in ['.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp']:
+                    image_files.append(file_path)
+                elif 'audio' in file_type or file_ext in ['.mp3', '.wav', '.m4a', '.ogg', '.flac']:
+                    audio_files.append(file_path)
+                elif 'python' in file_type or 'code' in file_type or file_ext in ['.py', '.txt']:
+                    code_files.append(file_path)
+            else:
+                # Auto-detect based on extension
+                if file_ext in ['.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp']:
+                    image_files.append(file_path)
+                elif file_ext in ['.mp3', '.wav', '.m4a', '.ogg', '.flac']:
+                    audio_files.append(file_path)
+                elif file_ext in ['.py', '.txt']:
+                    code_files.append(file_path)
+        
+        if self.debug:
+            print(f"Downloaded attachments: {len(image_files)} images, {len(audio_files)} audio, {len(code_files)} code files")
+        
+        return image_files, audio_files, code_files
+
+    def _process_attachments(self, image_files: List[str] = None, audio_files: List[str] = None, code_files: List[str] = None) -> str:
         """
-        Process attached media files and return their content as text.
+        Process all types of attachments and return their content as text.
         """
-        media_content = []
+        attachment_content = []
+        
+        # Process code files
+        if code_files:
+            for code_file in code_files:
+                if code_file and os.path.exists(code_file):
+                    try:
+                        # First, include the raw code content (truncated)
+                        with open(code_file, 'r', encoding='utf-8') as f:
+                            code_content = f.read()
+                        
+                        if len(code_content) > 1000:
+                            code_preview = code_content[:1000] + "\n... (truncated)"
+                        else:
+                            code_preview = code_content
+                        
+                        attachment_content.append(f"Code File Content:\n```python\n{code_preview}\n```")
+                        
+                        # Then add analysis
+                        code_analysis = self.code_tool.analyze_code(code_file)
+                        attachment_content.append(f"Code Analysis: {code_analysis}")
+                        
+                    except Exception as e:
+                        attachment_content.append(f"Error processing code file {code_file}: {e}")
         
         # Process images
         if image_files:
@@ -161,15 +352,15 @@ class IntelligentAgent:
                     try:
                         # Analyze the image
                         image_description = self.image_tool.analyze_image(image_file)
-                        media_content.append(f"Image Analysis: {image_description}")
+                        attachment_content.append(f"Image Analysis: {image_description}")
                         
                         # Try to extract text from image
                         extracted_text = self.image_tool.extract_text_from_image(image_file)
                         if extracted_text and "No text found" not in extracted_text:
-                            media_content.append(f"Text from Image: {extracted_text}")
+                            attachment_content.append(f"Text from Image: {extracted_text}")
                             
                     except Exception as e:
-                        media_content.append(f"Error processing image {image_file}: {e}")
+                        attachment_content.append(f"Error processing image {image_file}: {e}")
         
         # Process audio files
         if audio_files:
@@ -178,16 +369,16 @@ class IntelligentAgent:
                     try:
                         # Transcribe the audio
                         transcription = self.audio_tool.transcribe_audio(audio_file)
-                        media_content.append(f"Audio Transcription: {transcription}")
+                        attachment_content.append(f"Audio Transcription: {transcription}")
                         
                     except Exception as e:
-                        media_content.append(f"Error processing audio {audio_file}: {e}")
+                        attachment_content.append(f"Error processing audio {audio_file}: {e}")
         
-        return "\n\n".join(media_content) if media_content else ""
+        return "\n\n".join(attachment_content) if attachment_content else ""
 
-    def _should_search(self, question: str, media_context: str = "") -> bool:
+    def _should_search(self, question: str, attachment_context: str = "") -> bool:
         """
-        Use LLM to determine if search is needed for the question, considering media context.
+        Use LLM to determine if search is needed for the question, considering attachment context.
         Returns True if search is recommended, False otherwise.
         """
         decision_prompt = f"""Analyze this question and decide if it requires real-time information, recent data, or specific facts that might not be in your training data.
@@ -208,18 +399,20 @@ SEARCH IS NOT NEEDED for:
 - Definitions of well-established concepts
 - How-to instructions for common tasks
 - Creative writing or opinion-based responses
-- Questions that can be answered from attached media content
+- Questions that can be answered from attached files (code, images, audio)
+- Code analysis, debugging, or explanation questions
+- Questions about uploaded content
 
 Question: "{question}"
 
-{f"Media Context Available: {media_context[:500]}..." if media_context else "No media context available."}
+{f"Attachment Context Available: {attachment_context[:500]}..." if attachment_context else "No attachment context available."}
 
 Respond with only "SEARCH" or "NO_SEARCH" followed by a brief reason (max 20 words).
 
 Example responses:
 - "SEARCH - Current weather data needed"
 - "NO_SEARCH - Mathematical concept, general knowledge sufficient"
-- "NO_SEARCH - Can be answered from attached image content"
+- "NO_SEARCH - Can be answered from attached code/image content"
 """
 
         try:
@@ -236,15 +429,15 @@ Example responses:
             
         except Exception as e:
             if self.debug:
-                print(f"Error in search decision: {e}, defaulting to search")
-            # Default to search if decision fails
-            return True
+                print(f"Error in search decision: {e}, defaulting to no search for attachment questions")
+            # Default to no search if decision fails and there are attachments
+            return len(attachment_context) == 0
 
-    def _answer_with_llm(self, question: str, media_context: str = "") -> str:
+    def _answer_with_llm(self, question: str, attachment_context: str = "") -> str:
         """
-        Generate answer using LLM without search, considering media context.
+        Generate answer using LLM without search, considering attachment context.
         """
-        context_section = f"\n\nMedia Context:\n{media_context}" if media_context else ""
+        context_section = f"\n\nAttachment Context:\n{attachment_context}" if attachment_context else ""
         
         answer_prompt = f"""You are a general AI assistant. I will ask you a question. YOUR ANSWER should be a number OR as few words as possible OR a comma separated 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 (e.g. 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 be put in the list is a number or a string. Do not add a dot after the numbers.
 
@@ -261,9 +454,9 @@ Answer:"""
         except Exception as e:
             return f"Sorry, I encountered an error generating the response: {e}"
 
-    def _answer_with_search(self, question: str, media_context: str = "") -> str:
+    def _answer_with_search(self, question: str, attachment_context: str = "") -> str:
         """
-        Generate answer using search results and LLM, considering media context.
+        Generate answer using search results and LLM, considering attachment context.
         """
         try:
             # Perform search
@@ -274,7 +467,7 @@ Answer:"""
                 print(f"Search results type: {type(search_results)}")
             
             if not search_results:
-                return "No search results found. Let me try to answer based on my knowledge:\n\n" + self._answer_with_llm(question, media_context)
+                return "No search results found. Let me try to answer based on my knowledge:\n\n" + self._answer_with_llm(question, attachment_context)
 
             # Format search results - handle different result formats
             if isinstance(search_results, str):
@@ -295,8 +488,8 @@ Answer:"""
                 
                 search_context = "\n\n".join(formatted_results)
 
-            # Generate answer using search context and media context
-            context_section = f"\n\nMedia Context:\n{media_context}" if media_context else ""
+            # Generate answer using search context and attachment context
+            context_section = f"\n\nAttachment Context:\n{attachment_context}" if attachment_context else ""
             
             answer_prompt = f"""You are a general AI assistant. I will ask you a question. Based on the search results and context section below, provide an answer to the question. If the search results don't fully answer the question, you can supplement with your general knowledge. Do not add dot if your answer is a number.
             Your ANSWER should be a number OR as few words as possible OR a comma separated 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 (e.g. 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 be put in the list is a number or a string.
@@ -336,11 +529,60 @@ Answer:"""
                     return "Search completed but no usable results found."
 
         except Exception as e:
-            return f"Search failed: {e}. Let me try to answer based on my knowledge:\n\n" + self._answer_with_llm(question, media_context)
+            return f"Search failed: {e}. Let me try to answer based on my knowledge:\n\n" + self._answer_with_llm(question, attachment_context)
+
+    def process_question_with_attachments(self, question_data: dict) -> str:
+        """
+        Process a question that may have attachments.
+        """
+        question_text = question_data.get('question', '')
+        
+        if self.debug:
+            print(f"Processing question with potential attachments: {question_text[:100]}...")
+
+        try:
+            # Detect and download attachments
+            image_files, audio_files, code_files = self._detect_and_download_attachments(question_data)
+            
+            # Process attachments to get context
+            attachment_context = self._process_attachments(image_files, audio_files, code_files)
+            
+            if self.debug and attachment_context:
+                print(f"Attachment context: {attachment_context[:200]}...")
+
+            # Decide whether to search
+            if self._should_search(question_text, attachment_context):
+                if self.debug:
+                    print("Using search-based approach")
+                answer = self._answer_with_search(question_text, attachment_context)
+            else:
+                if self.debug:
+                    print("Using LLM-only approach")
+                answer = self._answer_with_llm(question_text, attachment_context)
+
+            # Cleanup temporary files
+            if image_files or audio_files or code_files:
+                try:
+                    all_files = image_files + audio_files + code_files
+                    temp_dirs = set(os.path.dirname(f) for f in all_files)
+                    for temp_dir in temp_dirs:
+                        import shutil
+                        shutil.rmtree(temp_dir, ignore_errors=True)
+                except Exception as cleanup_error:
+                    if self.debug:
+                        print(f"Cleanup error: {cleanup_error}")
+
+        except Exception as e:
+            answer = f"Sorry, I encountered an error: {e}"
+
+        if self.debug:
+            print(f"Agent returning answer: {answer[:100]}...")
+        
+        return answer
 
     def __call__(self, question: str, image_files: List[str] = None, audio_files: List[str] = None) -> str:
         """
-        Main entry point - process media files, decide whether to search, and generate appropriate response.
+        Main entry point for manual testing - process media files and generate response.
         """
         if self.debug:
             print(f"Agent received question: {question}")
@@ -353,20 +595,20 @@ Answer:"""
 
         try:
             # Process media files first
-            media_context = self._process_media_files(image_files, audio_files)
+            attachment_context = self._process_attachments(image_files, audio_files, [])
             
-            if self.debug and media_context:
-                print(f"Media context: {media_context[:200]}...")
+            if self.debug and attachment_context:
+                print(f"Media context: {attachment_context[:200]}...")
 
             # Decide whether to search
-            if self._should_search(question, media_context):
+            if self._should_search(question, attachment_context):
                 if self.debug:
                     print("Using search-based approach")
-                answer = self._answer_with_search(question, media_context)
+                answer = self._answer_with_search(question, attachment_context)
             else:
                 if self.debug:
                     print("Using LLM-only approach")
-                answer = self._answer_with_llm(question, media_context)
+                answer = self._answer_with_llm(question, attachment_context)
 
         except Exception as e:
             answer = f"Sorry, I encountered an error: {e}"
@@ -399,13 +641,40 @@ def fetch_questions() -> Tuple[str, Optional[pd.DataFrame]]:
         # Create DataFrame for display
         display_data = []
         for item in questions_data:
+            # Check for attachments
+            has_attachments = False
+            attachment_info = ""
+            
+            # Check various fields for attachments
+            attachment_fields = ['attachments', 'files', 'media', 'resources']
+            for field in attachment_fields:
+                if field in item and item[field]:
+                    has_attachments = True
+                    if isinstance(item[field], list):
+                        attachment_info += f"{len(item[field])} {field}, "
+                    else:
+                        attachment_info += f"{field}, "
+            
+            # Check if question contains URLs
+            question_text = item.get("question", "")
+            if 'http' in question_text:
+                has_attachments = True
+                attachment_info += "URLs in text, "
+            
+            if attachment_info:
+                attachment_info = attachment_info.rstrip(", ")
+            
             display_data.append({
                 "Task ID": item.get("task_id", "Unknown"),
-                "Question": item.get("question", "")
+                "Question": question_text[:100] + "..." if len(question_text) > 100 else question_text,
+                "Has Attachments": "Yes" if has_attachments else "No",
+                "Attachment Info": attachment_info
             })
         
         df = pd.DataFrame(display_data)
-        status_msg = f"Successfully fetched {len(questions_data)} questions. Ready to generate answers."
+        
+        attachment_count = sum(1 for item in display_data if item["Has Attachments"] == "Yes")
+        status_msg = f"Successfully fetched {len(questions_data)} questions. {attachment_count} questions have attachments. Ready to generate answers."
         
         return status_msg, df
         
@@ -431,18 +700,19 @@ def generate_answers_async(model_name: str = "meta-llama/Llama-3.1-8B-Instruct",
         agent = IntelligentAgent(debug=True, model_name=model_name)
         cached_answers = {}
         
-        for i, item in enumerate(cached_questions):
+        for i, question_data in enumerate(cached_questions):
             if not processing_status["is_processing"]:  # Check if cancelled
                 break
                 
-            task_id = item.get("task_id")
-            question_text = item.get("question")
+            task_id = question_data.get("task_id")
+            question_text = question_data.get("question")
             
             if not task_id or question_text is None:
                 continue
                 
             try:
-                answer = agent(question_text)
+                # Use the new method that handles attachments
+                answer = agent.process_question_with_attachments(question_data)
                 cached_answers[task_id] = {
                     "question": question_text,
                     "answer": answer
@@ -487,6 +757,8 @@ def start_answer_generation(model_choice: str):
     thread.start()
     
     return f"Answer generation started using {model_choice}. Check progress."
+
+   
 def get_generation_progress():
     """
     Get the current progress of answer generation.