Update app.py

app.py

@@ -16,6 +16,8 @@ from newspaper import Article
 from duckduckgo_search import DDGS
 from transformers import pipeline
 import logging
+import whisper
+from bs4 import BeautifulSoup
 
 
 from PIL import Image
@@ -33,103 +35,82 @@ summarizer = pipeline("summarization", model="facebook/bart-large-cnn")
 
 
 
-class SmartResearchAgent:
+
+
+class SmartAgentV2:
     def __init__(self):
-        logging.info("Initializing SmartResearchAgent")
+        self.qa_model = pipeline("text2text-generation", model="google/flan-t5-base")
         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")
+        self.whisper_model = whisper.load_model("base")
 
-    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 search_web(self, query):
+        with DDGS() as ddgs:
+            results = ddgs.text(query, max_results=3)
+            for r in results:
+                if "href" in r:
+                    return r["href"]
+        return "No results found."
 
-    def summarize(self, input_text_or_url: str) -> str:
+    def summarize_url(self, url):
         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)
+            article = Article(url)
+            article.download()
+            article.parse()
+            text = article.text
+            if not text.strip():
+                return "No content found."
+            summary = self.summarizer(text, max_length=150, 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."
+            return "Error summarizing."
 
-    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 generate_citation(self, url):
+        citation_id = hashlib.md5(url.encode()).hexdigest()[:6]
+        year = datetime.datetime.now().year
+        return f"@article{{cite{citation_id}, title={{Generated Citation}}, author={{Unknown}}, journal={{Online}}, year={{ {year} }}, url={{ {url} }} }}"
 
-    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 transcribe_audio(self, filepath):
+        result = self.whisper_model.transcribe(filepath)
+        return result["text"]
+
+    def sum_food_sales(self, filepath):
+        df = pd.read_excel(filepath)
+        food_df = df[df["Category"].str.lower() == "food"]
+        total = food_df["Sales"].sum()
+        return f"${total:.2f}"
 
-    def __call__(self, question: str, image: Image.Image = None) -> str:
-        logging.info(f"Received question: {question}")
-        q_lower = question.lower().strip()
+    def answer_fact(self, question):
+        return self.qa_model(question, max_length=100)[0]["generated_text"].strip()
 
+    def __call__(self, question: str, file=None):
+        q = 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:"):
+            if any(word in q for word in ["image", "chess", "diagram"]):
+                return "I'm a text-only agent and cannot interpret images."
+            if any(word in q for word in ["youtube", "video"]):
+                return "I'm unable to access or analyze video/audio from YouTube."
+            if file:
+                if filepath := getattr(file, "name", None):
+                    if filepath.endswith(".mp3"):
+                        transcript = self.transcribe_audio(filepath)
+                        return transcript
+                    elif filepath.endswith(".xlsx") or filepath.endswith(".xls"):
+                        return self.sum_food_sales(filepath)
+            if q.startswith("summarize:"):
+                url = question.split(":", 1)[1].strip()
+                return self.summarize_url(url)
+            elif q.startswith("generate citation:") or q.startswith("cite:"):
                 url = question.split(":", 1)[1].strip()
                 return self.generate_citation(url)
+            elif q.startswith("search:"):
+                query = question.split(":", 1)[1].strip()
+                return self.search_web(query)
             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."
+                return self.answer_fact(question)
         except Exception as e:
-            logging.exception("Unhandled error in agent call")
-            return f"Agent error: {e}"
-            
+            logging.error(f"Error: {e}")
+            return "An error occurred processing the question."
 
 
 def run_and_submit_all( profile: gr.OAuthProfile | None):