Update app.py

app.py

@@ -18,6 +18,10 @@ from transformers import pipeline
 import logging
 
 
+from PIL import Image
+from transformers import BlipProcessor, BlipForConditionalGeneration
+
+
 # (Keep Constants as is)
 # --- Constants ---
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
@@ -36,11 +40,12 @@ class BasicAgent:
         print(f"Agent returning fixed answer: {fixed_answer}")
         return fixed_answer
 
-
 class SmartResearchAgent:
     def __init__(self):
         logging.info("Initializing SmartResearchAgent")
         self.summarizer = pipeline("summarization", model="facebook/bart-large-cnn")
+        self.blip_processor = BlipProcessor.from_pretrained("Salesforce/blip-image-captioning-base")
+        self.blip_model = BlipForConditionalGeneration.from_pretrained("Salesforce/blip-image-captioning-base")
 
     def search_web(self, query: str) -> str:
         try:
@@ -85,39 +90,147 @@ class SmartResearchAgent:
             logging.error(f"Citation error: {e}")
             return "Error during citation generation."
 
-    def __call__(self, question: str) -> str:
+    def caption_image(self, image_path: str) -> str:
+        try:
+            image = Image.open(image_path).convert("RGB")
+            inputs = self.blip_processor(image, return_tensors="pt")
+            out = self.blip_model.generate(**inputs)
+            caption = self.blip_processor.decode(out[0], skip_special_tokens=True)
+            return f"Image analysis: {caption}"
+        except Exception as e:
+            logging.error(f"Image processing error: {e}")
+            return "Unable to process the image."
+
+    def __call__(self, question: str, image: Image.Image = None) -> str:
         logging.info(f"Received question: {question}")
         q_lower = question.lower().strip()
 
         try:
+            # 🔍 Handle image-based chess questions
+            if image is not None:
+                logging.info("Image input detected")
+                temp_path = "/tmp/input_image.jpg"
+                image.save(temp_path)
+                return self.caption_image(temp_path)
+
+            # 🔍 Handle text-based tasks
             if q_lower.startswith("search:"):
                 query = question.split(":", 1)[1].strip()
-                result = self.search_web(query)
+                return self.search_web(query)
             elif q_lower.startswith("summarize:"):
                 target = question.split(":", 1)[1].strip()
-                result = self.summarize(target)
+                return self.summarize(target)
             elif q_lower.startswith("generate citation:") or q_lower.startswith("cite:"):
                 url = question.split(":", 1)[1].strip()
-                result = self.generate_citation(url)
+                return self.generate_citation(url)
             else:
-                # Default: search + summarize first link
+                # Default: search + summarize
                 search_result = self.search_web(question)
-                logging.debug(f"Search result:\n{search_result}")
                 first_url = next((line.split(": ", 1)[-1] for line in search_result.splitlines() if "http" in line), None)
                 if first_url:
                     summary = self.summarize(first_url)
-                    result = f"{summary}\n\nSource: {first_url}"
+                    return f"{summary}\n\nSource: {first_url}"
                 else:
-                    result = "Sorry, I couldn't find relevant information."
+                    return "Sorry, I couldn't find relevant information."
+        except Exception as e:
+            logging.exception("Unhandled error in agent call")
+            return f"Agent error: {e}"class SmartResearchAgent:
+    def __init__(self):
+        logging.info("Initializing SmartResearchAgent")
+        self.summarizer = pipeline("summarization", model="facebook/bart-large-cnn")
+        self.blip_processor = BlipProcessor.from_pretrained("Salesforce/blip-image-captioning-base")
+        self.blip_model = BlipForConditionalGeneration.from_pretrained("Salesforce/blip-image-captioning-base")
 
-            logging.info(f"Agent answer: {result}")
-            return result
+    def search_web(self, query: str) -> str:
+        try:
+            with DDGS() as ddgs:
+                results = ddgs.text(query, max_results=3)
+                top = [f"{i+1}. {r['title']}: {r['href']}" for i, r in enumerate(results)]
+                return "\n".join(top) if top else "No results found."
+        except Exception as e:
+            logging.error(f"Search error: {e}")
+            return "Error during web search."
 
+    def summarize(self, input_text_or_url: str) -> str:
+        try:
+            if input_text_or_url.startswith("http"):
+                article = Article(input_text_or_url)
+                article.download()
+                article.parse()
+                input_text_or_url = article.text
+            if not input_text_or_url.strip():
+                return "No content to summarize."
+            summary = self.summarizer(input_text_or_url, max_length=160, min_length=40, do_sample=False)
+            return summary[0]['summary_text'].strip()
+        except Exception as e:
+            logging.error(f"Summarization error: {e}")
+            return "Error during summarization."
+
+    def generate_citation(self, url: str) -> str:
+        try:
+            citation_id = hashlib.md5(url.encode()).hexdigest()[:6]
+            year = datetime.datetime.now().year
+            citation = (
+                f"@article{{cite{citation_id},\n"
+                f"  title={{Generated Reference}},\n"
+                f"  author={{Unknown}},\n"
+                f"  journal={{Online}},\n"
+                f"  year={{ {year} }},\n"
+                f"  url={{ {url} }}\n"
+                f"}}"
+            )
+            return citation
+        except Exception as e:
+            logging.error(f"Citation error: {e}")
+            return "Error during citation generation."
+
+    def caption_image(self, image_path: str) -> str:
+        try:
+            image = Image.open(image_path).convert("RGB")
+            inputs = self.blip_processor(image, return_tensors="pt")
+            out = self.blip_model.generate(**inputs)
+            caption = self.blip_processor.decode(out[0], skip_special_tokens=True)
+            return f"Image analysis: {caption}"
+        except Exception as e:
+            logging.error(f"Image processing error: {e}")
+            return "Unable to process the image."
+
+    def __call__(self, question: str, image: Image.Image = None) -> str:
+        logging.info(f"Received question: {question}")
+        q_lower = question.lower().strip()
+
+        try:
+            # 🔍 Handle image-based chess questions
+            if image is not None:
+                logging.info("Image input detected")
+                temp_path = "/tmp/input_image.jpg"
+                image.save(temp_path)
+                return self.caption_image(temp_path)
+
+            # 🔍 Handle text-based tasks
+            if q_lower.startswith("search:"):
+                query = question.split(":", 1)[1].strip()
+                return self.search_web(query)
+            elif q_lower.startswith("summarize:"):
+                target = question.split(":", 1)[1].strip()
+                return self.summarize(target)
+            elif q_lower.startswith("generate citation:") or q_lower.startswith("cite:"):
+                url = question.split(":", 1)[1].strip()
+                return self.generate_citation(url)
+            else:
+                # Default: search + summarize
+                search_result = self.search_web(question)
+                first_url = next((line.split(": ", 1)[-1] for line in search_result.splitlines() if "http" in line), None)
+                if first_url:
+                    summary = self.summarize(first_url)
+                    return f"{summary}\n\nSource: {first_url}"
+                else:
+                    return "Sorry, I couldn't find relevant information."
         except Exception as e:
             logging.exception("Unhandled error in agent call")
             return f"Agent error: {e}"
-            
 
+            
 
 def run_and_submit_all( profile: gr.OAuthProfile | None):
     """