Update app.py

app.py

@@ -1,35 +1,280 @@
 
 import os
-import inspect
 import gradio as gr
 import requests
+import inspect
 import pandas as pd
-from langchain_core.messages import HumanMessage
-from agent import build_graph
-
-
+from smolagents import tool, Tool, CodeAgent, DuckDuckGoSearchTool, HfApiModel, VisitWebpageTool, SpeechToTextTool, FinalAnswerTool
+from dotenv import load_dotenv
+import heapq
+from collections import Counter
+import re
+from io import BytesIO
+from youtube_transcript_api import YouTubeTranscriptApi
+from langchain_community.tools.tavily_search import TavilySearchResults
+from langchain_community.document_loaders import WikipediaLoader
+from langchain_community.utilities import WikipediaAPIWrapper
+from langchain_community.document_loaders import ArxivLoader
 
 # (Keep Constants as is)
 # --- Constants ---
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 
-# --- Basic Agent Definition ---
-# ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------
+#Load environment variables
+load_dotenv()
+
+
+import os
+import io
+import contextlib
+from transformers import pipeline, AutoTokenizer, AutoModelForCausalLM
+
+class MistralToolCallingAgentTool:
+    name = "mistral_tool_agent"
+    description = "Uses Mistral-7B-Instruct to answer questions using code or reasoning"
+
+    def __init__(self):
+        self.model_id = "mistralai/Mistral-7B-Instruct"
+        self.tokenizer = AutoTokenizer.from_pretrained(self.model_id)
+        self.model = AutoModelForCausalLM.from_pretrained(self.model_id, device_map="auto", torch_dtype="auto")
+        self.pipeline = pipeline(
+            "text-generation", model=self.model, tokenizer=self.tokenizer,
+            max_new_tokens=512, temperature=0.2
+        )
+
+    def _run_code(self, code: str) -> str:
+        buffer = io.StringIO()
+        try:
+            with contextlib.redirect_stdout(buffer):
+                exec(code, {})
+            return buffer.getvalue().strip()
+        except Exception as e:
+            return f"Error during execution: {e}"
+
+    def run(self, question: str) -> str:
+        prompt = f"""You are a helpful assistant. Use code to solve questions that involve calculations.
+If code is needed, return a block like <tool>code</tool>. End your answer with <final>answer</final>.
+
+Question: {question}
+Answer:"""
+        result = self.pipeline(prompt)[0]["generated_text"]
+
+        # Process result
+        if "<tool>" in result and "</tool>" in result:
+            code = result.split("<tool>")[1].split("</tool>")[0].strip()
+            output = self._run_code(code)
+            return f"Code result: {output}"
+
+        elif "<final>" in result and "</final>" in result:
+            final = result.split("<final>")[1].split("</final>")[0].strip()
+            return f"FINAL ANSWER: {final}"
+
+        return "Mistral agent could not determine how to respond."
+
+#from smolagents import Tool
+#from langchain_community.document_loaders import WikipediaLoader
+
+class WikiSearchTool(Tool):
+    name = "wiki_search"
+    description = "Search Wikipedia for a query and return up to 2 results."
+    inputs = {
+        "query": {"type": "string", "description": "The search term for Wikipedia."}
+    }
+    output_type = "string"
+
+    def forward(self, query: str) -> str:
+        search_docs = WikipediaLoader(query=query, load_max_docs=2).load()
+
+        formatted_search_docs = "\n\n---\n\n".join(
+            [
+                f'<Document source="{doc.metadata.get("source", "Wikipedia")}" page="{doc.metadata.get("page", "")}"/>\n{doc.page_content}\n</Document>'
+                for doc in search_docs
+            ]
+        )
+        return formatted_search_docs
+
+    
+
+
+class StringReverseTool(Tool):
+    name = "reverse_message"
+    description = "When you received a strange text, try to reverse it and perform action described in reversed message."
+    inputs = {
+        "message": {
+            "type": "string",
+            "description": "A message, which looks like strange and can be reversed to get actions to execute."
+        }
+    }
+    output_type = "string"
+
+    def __init__(self):
+        return
+
+    def forward(self, message: str):
+        return message[::-1]
+
+class KeywordsExtractorTool(Tool):
+    """Extracts top 5 keywords from a given text based on frequency."""
 
+    name = "keywords_extractor"
+    description = "This tool returns the 5 most frequent keywords occur in provided block of text."
+
+    inputs = {
+        "text": {
+            "type": "string",
+            "description": "Text to analyze for keywords.",
+        }
+    }
+    output_type = "string"
+
+    def forward(self, text: str) -> str:
+        try:
+            all_words = re.findall(r'\b\w+\b', text.lower())
+            conjunctions = {'a', 'and', 'of', 'is', 'in', 'to', 'the'}
+            filtered_words = []
+            for w in all_words:
+                if w not in conjunctions:
+                    filtered_words.push(w)
+            word_counts = Counter(filtered_words)
+            k = 5
+            return heapq.nlargest(k, word_counts.items(), key=lambda x: x[1])
+        except Exception as e:
+            return f"Error during extracting most common words: {e}"
+
+@tool
+def parse_excel_to_json(task_id: str) -> dict:
+    """
+    For a given task_id fetch and parse an Excel file and save parsed data in structured JSON file.
+    Args:
+        task_id: An task ID to fetch.
+        
+    Returns:
+        {
+            "task_id": str,
+            "sheets": {
+                "SheetName1": [ {col1: val1, col2: val2, ...}, ... ],
+                ...
+            },
+            "status": "Success" | "Error"
+        }
+    """
+    url = f"https://agents-course-unit4-scoring.hf.space/files/{task_id}"
+
+    try:
+        response = requests.get(url, timeout=100)
+        if response.status_code != 200:
+            return {"task_id": task_id, "sheets": {}, "status": f"{response.status_code} - Failed"}
+
+        xls_content = pd.ExcelFile(BytesIO(response.content))
+        json_sheets = {}
+
+        for sheet in xls_content.sheet_names:
+            df = xls_content.parse(sheet)
+            df = df.dropna(how="all")  
+            rows = df.head(20).to_dict(orient="records")
+            json_sheets[sheet] = rows
+
+        return {
+            "task_id": task_id,
+            "sheets": json_sheets,
+            "status": "Success"
+        }
+
+    except Exception as e:
+        return {
+            "task_id": task_id,
+            "sheets": {},
+            "status": f"Error in parsing Excel file: {str(e)}"
+        }
+
+
+
+class VideoTranscriptionTool(Tool):
+    """Fetch transcripts from YouTube videos"""
+    name = "transcript_video"
+    description = "Fetch text transcript from YouTube movies with optional timestamps"
+    inputs = {
+        "url": {"type": "string", "description": "YouTube video URL or ID"},
+        "include_timestamps": {"type": "boolean", "description": "If timestamps should be included in output", "nullable": True}
+    }
+    output_type = "string"
+
+    def forward(self, url: str, include_timestamps: bool = False) -> str:
+
+        if "youtube.com/watch" in url:
+            video_id = url.split("v=")[1].split("&")[0]
+        elif "youtu.be/" in url:
+            video_id = url.split("youtu.be/")[1].split("?")[0]
+        elif len(url.strip()) == 11:  # Direct ID
+            video_id = url.strip()
+        else:
+            return f"YouTube URL or ID: {url} is invalid!"
+
+        try:
+            transcription = YouTubeTranscriptApi.get_transcript(video_id)
+
+            if include_timestamps:
+                formatted_transcription = []
+                for part in transcription:
+                    timestamp = f"{int(part['start']//60)}:{int(part['start']%60):02d}"
+                    formatted_transcription.append(f"[{timestamp}] {part['text']}")
+                return "\n".join(formatted_transcription)
+            else:
+                return " ".join([part['text'] for part in transcription])
+
+        except Exception as e:
+            return f"Error in extracting YouTube transcript: {str(e)}"
 
 class BasicAgent:
-    """A langgraph agent."""
     def __init__(self):
-        print("BasicAgent initialized.")
-        self.graph = build_graph()
+        token = os.environ.get("HF_API_TOKEN")
+        model = HfApiModel(
+            temperature=0.1,
+            token=token
+        )
+
+        # Existing tools
+        search_tool = DuckDuckGoSearchTool()
+        wiki_search_tool = WikiSearchTool()
+        str_reverse_tool = StringReverseTool()
+        keywords_extract_tool = KeywordsExtractorTool()
+        speech_to_text_tool = SpeechToTextTool()
+        visit_webpage_tool = VisitWebpageTool()
+        final_answer_tool = FinalAnswerTool()
+        video_transcription_tool = VideoTranscriptionTool()
+
+        # ✅ New Mistral-based Tool
+        mistral_tool = MistralToolCallingAgentTool()
+
+        system_prompt = f"""
+You are my general AI assistant. Your task is to answer the question I asked.
+First, provide an explanation of your reasoning, step by step, to arrive at the answer.
+Then, return your final answer in a single line, formatted as follows: "FINAL ANSWER: [YOUR FINAL ANSWER]".
+[YOUR FINAL ANSWER] should be a number, a string, or a comma-separated list of numbers and/or strings, depending on the question.
+If the answer is a number, do not use commas or units (e.g., $, %) unless specified.
+If the answer is a string, do not use articles or abbreviations (e.g., for cities), and write digits in plain text unless specified.
+If the answer is a comma-separated list, apply the above rules for each element based on whether it is a number or a string.
+"""
+
+        self.agent = CodeAgent(
+            model=model,
+            tools=[
+                search_tool, wiki_search_tool, str_reverse_tool,
+                keywords_extract_tool, speech_to_text_tool,
+                visit_webpage_tool, final_answer_tool,
+                parse_excel_to_json, video_transcription_tool,
+                mistral_tool  # 🔧 Add here
+            ],
+            add_base_tools=True
+        )
+        self.agent.prompt_templates["system_prompt"] = self.agent.prompt_templates["system_prompt"] + system_prompt
 
     def __call__(self, question: str) -> str:
         print(f"Agent received question (first 50 chars): {question[:50]}...")
-        # Wrap the question in a HumanMessage from langchain_core
-        messages = [HumanMessage(content=question)]
-        response = self.graph.invoke(messages)
-        answer = response['messages'][-1].content
-        return answer[14:]
+        answer = self.agent.run(question)
+        print(f"Agent returning answer: {answer}")
+        return answer
+