Update app.py

app.py

@@ -22,6 +22,93 @@ DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 #Load environment variables
 load_dotenv()
 
+import io
+import contextlib
+import traceback
+from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
+from smolagents import Tool, CodeAgent, DuckDuckGoSearchTool, FinalAnswerTool, HfApiModel
+
+
+class CodeLlamaTool(Tool):
+    name = "code_llama_tool"
+    description = "Solves reasoning/code questions using Meta Code Llama 7B Instruct"
+
+    inputs = {
+        "question": {
+            "type": "string",
+            "description": "The question requiring code-based or reasoning-based solution"
+        }
+    }
+    output_type = "string"
+
+    def __init__(self):
+        self.model_id = "codellama/CodeLlama-7b-Instruct-hf"
+        token = os.getenv("HF_TOKEN")
+
+        self.tokenizer = AutoTokenizer.from_pretrained(self.model_id, token=token)
+        self.model = AutoModelForCausalLM.from_pretrained(
+            self.model_id,
+            device_map="auto",
+            torch_dtype="auto",
+            token=token
+        )
+        self.pipeline = pipeline(
+            "text-generation",
+            model=self.model,
+            tokenizer=self.tokenizer,
+            max_new_tokens=512,
+            temperature=0.2,
+            truncation=True
+        )
+
+    def forward(self, question: str) -> str:
+        prompt = f"""You are an AI that uses Python code to answer questions.
+Question: {question}
+Instructions:
+- If solving requires code, use a block like <tool>code</tool>.
+- Always end with <final>FINAL ANSWER</final> containing the final number or string.
+Example:
+Question: What is 5 * sqrt(36)?
+Answer:
+<tool>
+import math
+print(5 * math.sqrt(36))
+</tool>
+<final>30.0</final>
+Answer:"""
+
+        response = self.pipeline(prompt)[0]["generated_text"]
+        return self.parse_and_execute(response)
+
+    def parse_and_execute(self, response: str) -> str:
+        try:
+            # Extract and run code if exists
+            if "<tool>" in response and "</tool>" in response:
+                code = response.split("<tool>")[1].split("</tool>")[0].strip()
+                result = self._run_code(code)
+                return f"FINAL ANSWER (code output): {result}"
+
+            # Extract final result directly
+            elif "<final>" in response and "</final>" in response:
+                final = response.split("<final>")[1].split("</final>")[0].strip()
+                return f"FINAL ANSWER: {final}"
+
+            return f"Could not extract final answer.\n\n{response}"
+
+        except Exception as e:
+            return f"Error in parse_and_execute: {str(e)}\n\nFull response:\n{response}"
+
+    def _run_code(self, code: str) -> str:
+        buffer = io.StringIO()
+        try:
+            with contextlib.redirect_stdout(buffer):
+                exec(code, {})
+            return buffer.getvalue().strip()
+        except Exception:
+            return f"Error executing code:\n{traceback.format_exc()}"
+
+
+
 from duckduckgo_search import DDGS
 import wikipedia
 import arxiv
@@ -138,73 +225,242 @@ class FinalAnswerTool:
     def run(self, answer: str) -> str:
         return f"FINAL ANSWER: {answer}"
 
+#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:
     def __init__(self):
         token = os.environ.get("HF_API_TOKEN")
         model = HfApiModel(
-            temperature=0.0,  # Reduced for deterministic output
+            temperature=0.1,
             token=token
         )
 
-        # Curated toolset - remove redundant/conflicting tools
+        # 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 Llama Tool
+        code_llama_tool = CodeLlamaTool()
         arxiv_search_tool = ArxivSearchTool()
         doc_qa_tool = HuggingFaceDocumentQATool()
+        image_qa_tool = HuggingFaceImageQATool()
+        translation_tool = HuggingFaceTranslationTool()
         python_tool = PythonCodeExecutionTool()
-        final_answer_tool = FinalAnswerTool()
 
-        # Strategic tool selection
-        tools = [
-            search_tool, 
-            wiki_search_tool,
-            arxiv_search_tool,
-            doc_qa_tool,
-            python_tool,
-            final_answer_tool
-        ]
-
-        # Enhanced system prompt
-        system_prompt = """
-You are a precision question-answering AI. Follow this protocol:
-1. Analyze the question type: factual, computational, or multi-step
-2. Select the optimal tool:
-   - Use Search/Wiki/Arxiv for factual queries
-   - Use Python tool for calculations
-   - Use DocQA for document-based questions
-3. Execute necessary actions
-4. Verify answer matches question requirements
-5. Output FINAL ANSWER using this format:
-   "FINAL ANSWER: [EXACT_RESULT]"
-
-Answer rules:
-- Numbers: Plain format (e.g., 1000000)
-- Strings: No articles/abbreviations (e.g., "Paris" not "city of Paris")
-- Lists: Comma-separated (e.g., "red,blue,green")
-- Never include units ($, kg, etc.) unless explicitly required
-- For true/false: Use "true" or "false" lowercase
+        system_prompt = f"""
+You are my general AI assistant. Your primary goal is to answer the user's question accurately and concisely.
+
+Here's a detailed plan for answering:
+1. **Understand the Question:** Carefully parse the question to identify key entities, relationships, and the type of information requested.
+2. **Reasoning Steps (Chain-of-Thought):** Before attempting to answer, outline a step-by-step reasoning process. This helps in breaking down complex questions.
+3. **Tool Selection and Usage:** Based on your reasoning, select the most appropriate tool(s) to gather information or perform operations.
+   - Use `search_tool` (DuckDuckGoSearchTool) for general web searches.
+   - Use `wiki_search_tool` for encyclopedic knowledge.
+   - Use `arxiv_search_tool` for scientific papers.
+   - Use `visit_webpage_tool` to read content from URLs found via search.
+   - Use `doc_qa_tool` for answering questions about specific documents (if provided).
+   - Use `image_qa_tool` for questions about images.
+   - Use `translation_tool` for language translation.
+   - Use `python_tool` or `code_llama_tool` for code generation, execution, or complex calculations/data manipulation.
+   - Use `keywords_extract_tool` to identify important terms from text.
+   - Use `str_reverse_tool` for string manipulation if needed (less common for Q&A).
+   - Use `speech_to_text_tool` or `video_transcription_tool` if audio/video input is part of the question.
+   - Use `parse_excel_to_json` if the question involves data from Excel.
+4. **Information Synthesis:** Combine and process the information obtained from tools. Cross-reference if necessary to ensure accuracy.
+5. **Formulate Final Answer:** Construct the final answer according to the specified format.
+
+**Final Answer Format:**
+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 explicitly specified in the question.
+- If the answer is a string, do not use articles (a, an, the) or common abbreviations (e.g., "NY" for "New York") unless specified. 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.
+- If you cannot find a definitive answer, state "FINAL ANSWER: I don't know."
+
+Let's think step by step.
 """
+self.agent.prompt_templates["system_prompt"] = self.agent.prompt_templates["system_prompt"] + system_prompt
+
 
         self.agent = CodeAgent(
             model=model,
-            tools=tools,
-            add_base_tools=False  # Prevent tool conflicts
+            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,
+                arxiv_search_tool,
+                doc_qa_tool, image_qa_tool,
+                translation_tool, python_tool, 
+                code_llama_tool  # 🔧 Add here
+            ],
+            add_base_tools=True
         )
-        # Force strict prompt template
-        self.agent.prompt_templates["system_prompt"] = system_prompt
+        self.agent.prompt_templates["system_prompt"] = self.agent.prompt_templates["system_prompt"] + system_prompt
 
     def __call__(self, question: str) -> str:
-        print(f"Processing: {question[:50]}...")
-        try:
-            result = self.agent.run(question)
-            # Extract final answer using regex
-            import re
-            match = re.search(r"FINAL ANSWER:\s*(.+)", result, re.IGNORECASE)
-            return match.group(1).strip() if match else result
-        except Exception as e:
-            print(f"Error: {str(e)}")
-            return "Unable to determine answer"
-
+        print(f"Agent received question (first 50 chars): {question[:50]}...")
+        answer = self.agent.run(question)
+        print(f"Agent returning answer: {answer}")
+        return answer
 
 
 def run_and_submit_all( profile: gr.OAuthProfile | None):