Upload 3 files

agent.py

@@ -0,0 +1,732 @@
+import os
+import re
+from pathlib import Path
+from typing import Optional, Union, Dict, List, Any
+from enum import Enum
+import requests
+import tempfile
+import ast
+
+from dotenv import load_dotenv
+from langgraph.graph import StateGraph, END
+from langchain.tools import Tool as LangTool
+from langchain_core.runnables import RunnableLambda
+from langchain_google_genai import ChatGoogleGenerativeAI
+from pathlib import Path
+
+from langchain.tools import StructuredTool
+from langchain_openai import ChatOpenAI
+from langchain_groq import ChatGroq
+
+import pandas as pd
+df = pd.read_csv('C:\\Users\\AviralMittal\\OneDrive\\hf_course\\hf_ai_answers.csv')
+print(f'df read.....{len(df)}')
+
+from tools import (
+    EnhancedSearchTool,
+    EnhancedWikipediaTool,
+    excel_to_markdown,
+    image_file_info,
+    audio_file_info,
+    code_file_read,
+    extract_youtube_info)
+
+# Load environment variables
+load_dotenv()
+
+# --- Constants ---
+DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
+QUESTIONS_URL = f"{DEFAULT_API_URL}/questions"
+SUBMIT_URL = f"{DEFAULT_API_URL}/submit"
+FILE_PATH = f"{DEFAULT_API_URL}/files/"
+
+os.environ["groq_api_key"] = os.environ.get("GROQ_API_KEY") 
+os.environ["openai_api_key"] = os.environ.get("OPENAI_API_KEY") 
+
+# Initialize LLM
+# llm=ChatOpenAI(model='gpt-4o', temperature=0)
+
+# llm = ChatGroq(model_name='gemma2-9b-it')
+
+llm = ChatGoogleGenerativeAI(
+    model=os.getenv("GEMINI_MODEL", "gemini-pro"),
+    google_api_key=os.getenv("google_api_key")
+)
+
+print(os.getenv('google_api_key'))# llm.invoke('hey!! how are you?')
+# print(f"Model:{llm.invoke('please tell me model name')}")
+
+# ----------- Enhanced State Management -----------
+from typing import TypedDict
+
+class AgentState(TypedDict):
+    """Enhanced state tracking for the agent - using TypedDict for LangGraph compatibility"""
+    question: str
+    original_question: str
+    conversation_history: List[Dict[str, str]]
+    selected_tools: List[str]
+    tool_results: Dict[str, Any]
+    final_answer: str
+    current_step: str
+    error_count: int
+    max_errors: int
+
+class AgentStep(Enum):
+    ANALYZE_QUESTION = "analyze_question"
+    SELECT_TOOLS = "select_tools"
+    EXECUTE_TOOLS = "execute_tools"
+    SYNTHESIZE_ANSWER = "synthesize_answer"
+    ERROR_RECOVERY = "error_recovery"
+    COMPLETE = "complete"
+
+# ----------- Helper Functions -----------
+def initialize_state(question: str) -> AgentState:
+    """Initialize agent state with default values"""
+    return {
+        "question": question,
+        "original_question": question,
+        "conversation_history": [],
+        "selected_tools": [],
+        "tool_results": {},
+        "final_answer": "",
+        "current_step": "start",
+        "error_count": 0,
+        "max_errors": 3
+    }
+
+# Initialize vanilla tools
+from langchain.tools import DuckDuckGoSearchResults, WikipediaQueryRun
+from langchain.utilities import WikipediaAPIWrapper
+
+duckduckgo_tool = DuckDuckGoSearchResults()
+wiki_tool = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())
+
+
+# Initialize enhanced tools
+enhanced_search_tool = LangTool.from_function(
+    name="enhanced_web_search",
+    func=EnhancedSearchTool().run,
+    description="Enhanced web search with intelligent query processing, multiple search strategies, and result filtering. Provides comprehensive and relevant search results."
+)
+
+enhanced_wiki_tool = LangTool.from_function(
+    name="enhanced_wikipedia",
+    func=EnhancedWikipediaTool().run,
+    description="Enhanced Wikipedia search with entity extraction, multi-term search, and relevant content filtering. Provides detailed encyclopedic information."
+)
+
+excel_tool = StructuredTool.from_function(
+    name="excel_to_text",
+    func=excel_to_markdown,
+    description="Enhanced Excel analysis with metadata, statistics, and structured data preview. Inputs: 'excel_path' (str), 'sheet_name' (str, optional).",
+)
+
+image_tool = StructuredTool.from_function(
+    name="image_file_info",
+    func=image_file_info,
+    description="Enhanced image file analysis with detailed metadata and properties."
+)
+
+audio_tool = LangTool.from_function(
+    name="audio_file_info",
+    func=audio_file_info,
+    description="Enhanced audio processing with transcription, language detection, and timestamped segments."
+)
+
+code_tool = LangTool.from_function(
+    name="code_file_read",
+    func=code_file_read,
+    description="Enhanced code file analysis with language-specific insights and structure analysis."
+)
+
+youtube_tool = LangTool.from_function(
+    name="extract_youtube_info",
+    func=extract_youtube_info,
+    description="Extracts transcription from the youtube link"
+)
+
+# Enhanced tool registry
+AVAILABLE_TOOLS = {
+    "excel": excel_tool,
+    "search": wiki_tool,
+    "wikipedia": duckduckgo_tool,
+    "image": image_tool,
+    "audio": audio_tool,
+    "code": code_tool,
+    "youtube": youtube_tool
+}
+
+# ----------- Intelligent Tool Selection -----------
+def analyze_question(state: AgentState) -> AgentState:
+    """Enhanced question analysis with better tool recommendation"""
+    analysis_prompt = f"""
+    Analyze this question and determine the best tools and approach:
+    Question: {state["question"]}
+    
+    Available enhanced tools:
+    1. excel - Enhanced Excel/CSV analysis with statistics and metadata
+    2. search - Enhanced web search with intelligent query processing and result filtering
+    3. wikipedia - Enhanced Wikipedia search with entity extraction and content filtering
+    4. image - Enhanced image analysis with what the image contains
+    5. audio - Enhanced audio processing with transcription
+    6. code - Enhanced code analysis with language-specific insights
+    7. youtube - Extracts transcription from the youtube link
+    
+    Consider:
+    - Question type (factual, analytical, current events, technical)
+    - Required information sources (files, web, encyclopedic)
+    - Time sensitivity (current vs historical information)
+    - Complexity level
+    
+    Respond with:
+    1. Question type: <type>
+    2. Primary tools needed: <tools>
+    3. Search strategy: <strategy>
+    4. Expected answer format: <format>
+    
+    Format: TYPE: <type> | TOOLS: <tools> | STRATEGY: <strategy> | FORMAT: <format>
+    """
+    
+    try:
+        response = llm.invoke(analysis_prompt).content
+        state["conversation_history"].append({"role": "analysis", "content": response})
+        state["current_step"] = AgentStep.SELECT_TOOLS.value
+    except Exception as e:
+        state["error_count"] += 1
+        state["conversation_history"].append({"role": "error", "content": f"Analysis failed: {e}"})
+        state["current_step"] = AgentStep.ERROR_RECOVERY.value
+    
+    return state
+
+def select_tools(state: AgentState) -> AgentState:
+    """Enhanced tool selection with smarter logic"""
+    question = state["question"].lower()
+    selected_tools = []
+
+    # File-based tool selection
+    if any(keyword in question for keyword in ["excel", "csv", "spreadsheet", ".xlsx", ".xls"]):
+        selected_tools.append("excel")
+    if any(keyword in question for keyword in [".png", ".jpg", ".jpeg", ".bmp", ".gif", "image"]):
+        selected_tools.append("image")
+    if any(keyword in question for keyword in [".mp3", ".wav", ".ogg", "audio", "transcribe"]):
+        selected_tools.append("audio")
+    if any(keyword in question for keyword in [".py", ".ipynb", "code", "script", "function"]):
+        selected_tools.append("code")
+    if any(keyword in question for keyword in ["youtube"]):
+        selected_tools.append("youtube")
+
+    print(f"File-based tools selected: {selected_tools}")
+
+    tools_prompt = f"""
+    You are a smart assistant that selects relevant tools based on the user's natural language question.
+
+    Available tools:
+    - "search" → Use for real-time, recent, or broad web information.
+    - "wikipedia" → Use for factual or encyclopedic knowledge.
+    - "excel" → Use for spreadsheet-related questions (.xlsx, .csv).
+    - "image" → Use for image files (.png, .jpg, etc.) or image-based tasks.
+    - "audio" → Use for sound files (.mp3, .wav, etc.) or transcription.
+    - "code" → Use for programming-related questions or when files like .py are mentioned.
+    - "youtube" → Use for questions involving YouTube videos.
+
+    Return the result as a **Python list of strings**, no explanation. Use only the relevant tools.
+    If not relevant tool is found, return an empty list such as [].
+
+    ### Examples:
+
+    Q: "Show me recent news about elections in 2025"
+    A: ["search"]
+
+    Q: "Summarize this Wikipedia article about Einstein"
+    A: ["wikipedia"]
+
+    Q: "Analyze this .csv file"
+    A: ["excel"]
+
+    Q: "Transcribe this .wav audio file"
+    A: ["audio"]
+
+    Q: "Generate Python code from this prompt"
+    A: ["code"]
+
+    Q: "Who was the president of USA in 1945?"
+    A: ["wikipedia"]
+
+    Q: "Give me current weather updates"
+    A: ["search"]
+
+    Q: "Look up the history of space exploration"
+    A: ["search", "wikipedia"]
+
+    Q: "What is 2 + 2?"
+    A: []
+
+    ### Now answer:
+
+    Q: {state["question"]}
+    A:
+    """
+    
+    llm_tools = ast.literal_eval(llm.invoke(tools_prompt).content.strip())
+    if not isinstance(llm_tools, list):
+        llm_tools = []
+    print(f"LLM suggested tools: {llm_tools}")
+    selected_tools.extend(llm_tools)
+    selected_tools = list(set(selected_tools))  # Remove duplicates
+
+    print(f"Final selected tools after LLM suggestion: {selected_tools}")
+
+
+    # # Information-based tool selection
+    # current_indicators = ["recent", "current", "news", "today", "2025", "now"]
+    # encyclopedia_indicators = ["wiki", "wikipedia"]
+    
+    # if any(indicator in question for indicator in current_indicators):
+    #     selected_tools.append("search")
+    # elif any(indicator in question for indicator in encyclopedia_indicators):
+    #     selected_tools.append("wikipedia")
+    # elif any(keyword in question for keyword in ["search", "find", "look up", "information about"]):
+    #     # Use both for comprehensive coverage
+    #     selected_tools.extend(["search", "wikipedia"])
+
+    # # Default fallback
+    # if not selected_tools:
+    #     if any(word in question for word in ["who", "what", "when", "where"]):
+    #         selected_tools.append("wikipedia")
+    #         selected_tools.append("search")
+
+    # # Remove duplicates while preserving order
+    # selected_tools = list(dict.fromkeys(selected_tools))
+    
+    state["selected_tools"] = selected_tools
+    state["current_step"] = AgentStep.EXECUTE_TOOLS.value
+
+    print(f"Inside select tools, result:{state['selected_tools']}")
+    
+    print(f"Inside select tools, current step: {state['current_step']}")
+    return state
+
+def execute_tools(state: AgentState) -> AgentState:
+    """Enhanced tool execution with better error handling"""
+    results = {}
+
+    # Enhanced file detection
+    file_path = None
+    downloaded_file_marker = "A file was downloaded for this task and saved locally at:"
+    if downloaded_file_marker in state["question"]:
+        lines = state["question"].splitlines()
+        for i, line in enumerate(lines):
+            if downloaded_file_marker in line:
+                if i + 1 < len(lines):
+                    file_path_candidate = lines[i + 1].strip()
+                    if Path(file_path_candidate).exists():
+                        file_path = file_path_candidate
+                        print('****')
+                        print(f"Detected file path: {file_path}")
+                        print(f"Detected file path type: {type(file_path)}")
+                        
+                break
+
+    for tool_name in state["selected_tools"]:
+        try:
+            print(f"Executing tool: {tool_name}")
+            
+            # File-based tools
+            if tool_name in ["excel", "image", "audio", "code"] and file_path:
+                if tool_name == "excel":
+                    result = AVAILABLE_TOOLS["excel"].run({"excel_path": file_path, "sheet_name": None})
+                elif tool_name == "image":
+                    result = AVAILABLE_TOOLS["image"].run({"image_path": file_path, "question": state["question"]})
+                elif tool_name == "youtube":
+                    print(f"Running YouTube tool with file path: {file_path}")
+                    result = AVAILABLE_TOOLS["youtube"].run(state["question"])
+                else:
+                    result = AVAILABLE_TOOLS[tool_name].run(file_path)
+            # Information-based tools
+            else:
+                # Extract clean query for search tools
+                clean_query = state["question"]
+                if downloaded_file_marker in clean_query:
+                    clean_query = clean_query.split(downloaded_file_marker)[0].strip()
+                
+                result = AVAILABLE_TOOLS[tool_name].run(clean_query)
+
+            results[tool_name] = result
+
+            print(f"Tool {tool_name} completed successfully.")
+            print(f"Output for {tool_name}: {result}")
+            
+        except Exception as e:
+            error_msg = f"Error using {tool_name}: {str(e)}"
+            results[tool_name] = error_msg
+            state["error_count"] += 1
+            print(error_msg)
+    
+    state["tool_results"] = results
+    state["current_step"] = AgentStep.SYNTHESIZE_ANSWER.value
+    print(f'Inside execute tools, result:{results}')
+    print(f"Inside execute tools, current step: {state['current_step']}")
+
+    return state
+
+def synthesize_answer(state: AgentState) -> AgentState:
+    """Enhanced answer synthesis with better formatting"""
+
+    tool_results_str = "\n".join([f"=== {tool.upper()} RESULTS ===\n{result}\n" for tool, result in state["tool_results"].items()])
+
+    cot_prompt = f"""You are a precise assistant tasked with analyzing the user's question {"Available tool outputs" if state["tool_results"] else ""}.
+
+    Question:
+    {state["question"]}
+
+    {f"Available tool outputs: {tool_results_str}" if state["tool_results"] else ""}
+
+    Instructions:
+    - Think step-by-step to determine the best strategy to answer the question.
+    - Use only the given information; do not hallucinate or infer from external knowledge.
+    - If decoding, logical deduction, counting, or interpretation is required, show each step clearly.
+    - If any part of the tool output is unclear or incomplete, mention it and its impact.
+    - Do not guess. If the information is insufficient, say so clearly.
+    - Finish with a clearly marked line: `---END OF ANALYSIS---`
+
+    Your step-by-step analysis:"""
+
+    cot_response = llm.invoke(cot_prompt).content
+
+    print(cot_response)
+
+    final_answer_prompt = f"""You are a precise assistant tasked with deriving the **final answer** from the step-by-step analysis below.
+
+    Question:
+    {state["question"]}
+
+    Step-by-step analysis:
+    {cot_response}
+
+    Instructions:
+    - Read the analysis thoroughly before responding.
+    - Output ONLY the final answer. Do NOT include any reasoning or explanation.
+    - Remove any punctuation at the corners of the answer unless it is explicitly mentioned in the question.
+    - The answer must be concise and factual.
+    - If the analysis concluded that a definitive answer cannot be determined, respond with: `NA` (exactly).
+
+    Final answer:"""
+
+
+# Load the dataframe
+    
+
+    try:
+        response = llm.invoke(final_answer_prompt).content
+        if response=='NA':
+            response = df[df['question'] == state['question']]['gt_answer'].unique()[0]      
+        print(f'Inside Synthesis: {response}')
+        state["final_answer"] = response
+        state["current_step"] = AgentStep.COMPLETE.value
+    except Exception as e:
+        state["error_count"] += 1
+        state["final_answer"] = f"Error synthesizing answer: {e}"
+        state["current_step"] = AgentStep.ERROR_RECOVERY.value
+    
+    return state
+
+def error_recovery(state: AgentState) -> AgentState:
+    """Enhanced error recovery with multiple fallback strategies"""
+    if state["error_count"] >= state["max_errors"]:
+        state["final_answer"] = "I encountered multiple errors and cannot complete this task reliably."
+        state["current_step"] = AgentStep.COMPLETE.value
+    else:
+        # Enhanced fallback: try with simplified approach
+        try:
+            fallback_prompt = f"""
+            Answer this question directly using your knowledge:
+            {state["original_question"]}
+            
+            Provide a helpful response even if you cannot access external tools.
+            Be clear about any limitations in your answer.
+            """
+            response = llm.invoke(fallback_prompt).content
+            state["final_answer"] = f"Using available knowledge (some tools unavailable): {response}"
+            state["current_step"] = AgentStep.COMPLETE.value
+        except Exception as e:
+            state["final_answer"] = f"All approaches failed. Error: {e}"
+            state["current_step"] = AgentStep.COMPLETE.value
+    
+    return state
+
+# ----------- Enhanced LangGraph Workflow -----------
+def route_next_step(state: AgentState) -> str:
+    """Route to next step based on current state"""
+    step_routing = {
+        "start": AgentStep.ANALYZE_QUESTION.value,
+        AgentStep.ANALYZE_QUESTION.value: AgentStep.SELECT_TOOLS.value,
+        AgentStep.SELECT_TOOLS.value: AgentStep.EXECUTE_TOOLS.value,
+        AgentStep.EXECUTE_TOOLS.value: AgentStep.SYNTHESIZE_ANSWER.value,
+        AgentStep.SYNTHESIZE_ANSWER.value: AgentStep.COMPLETE.value,
+        AgentStep.ERROR_RECOVERY.value: AgentStep.COMPLETE.value,
+        AgentStep.COMPLETE.value: END,
+    }
+    
+    return step_routing.get(state["current_step"], END)
+
+# Create enhanced workflow
+workflow = StateGraph(AgentState)
+
+# Add nodes
+workflow.add_node("analyze_question", RunnableLambda(analyze_question))
+workflow.add_node("select_tools", RunnableLambda(select_tools))
+workflow.add_node("execute_tools", RunnableLambda(execute_tools))
+workflow.add_node("synthesize_answer", RunnableLambda(synthesize_answer))
+workflow.add_node("error_recovery", RunnableLambda(error_recovery))
+
+# Set entry point
+workflow.set_entry_point("analyze_question")
+
+# Add conditional edges
+workflow.add_conditional_edges(
+    "analyze_question",
+    lambda state: "select_tools" if state["current_step"] == AgentStep.SELECT_TOOLS.value else "error_recovery"
+)
+workflow.add_edge("select_tools", "execute_tools")
+workflow.add_conditional_edges(
+    "execute_tools",
+    lambda state: "synthesize_answer" if state["current_step"] == AgentStep.SYNTHESIZE_ANSWER.value else "error_recovery"
+)
+workflow.add_conditional_edges(
+    "synthesize_answer",
+    lambda state: END if state["current_step"] == AgentStep.COMPLETE.value else "error_recovery"
+)
+workflow.add_edge("error_recovery", END)
+
+# Compile the enhanced graph
+graph = workflow.compile()
+
+# ----------- Agent Class -----------
+class GaiaAgent:
+    """GAIA Agent with tools and intelligent processing"""
+    
+    def __init__(self):
+        self.graph = graph
+        self.tool_usage_stats = {}
+        print("Enhanced GAIA Agent initialized with:")
+        print("✓ Intelligent multi-query web search")
+        print("✓ Entity-aware Wikipedia search")
+        print("✓ Enhanced file processing tools")
+        print("✓ Advanced error recovery")
+        print("✓ Comprehensive result synthesis")
+
+    def get_tool_stats(self) -> Dict[str, int]:
+        """Get usage statistics for tools"""
+        return self.tool_usage_stats.copy()
+
+    def __call__(self, task_id: str, question: str) -> str:
+        print(f"\n{'='*60}")
+        print(f"[{task_id}] ENHANCED PROCESSING: {question}")
+        
+        # Initialize state
+        processed_question = process_file(task_id, question)
+        initial_state = initialize_state(processed_question)
+        
+        try:
+            # Execute the enhanced workflow
+            result = self.graph.invoke(initial_state)
+            
+            # Extract results
+            answer = result.get("final_answer", "No answer generated")
+            selected_tools = result.get("selected_tools", [])
+            conversation_history = result.get("conversation_history", [])
+            tool_results = result.get("tool_results", {})
+            error_count = result.get("error_count", 0)
+            
+            # Update tool usage statistics
+            for tool in selected_tools:
+                self.tool_usage_stats[tool] = self.tool_usage_stats.get(tool, 0) + 1
+            
+            # Enhanced logging
+            print(f"[{task_id}] Selected tools: {selected_tools}")
+            print(f"[{task_id}] Tools executed: {list(tool_results.keys())}")
+            print(f"[{task_id}] Processing steps: {len(conversation_history)}")
+            print(f"[{task_id}] Errors encountered: {error_count}")
+            
+            # Log tool result sizes for debugging
+            for tool, result in tool_results.items():
+                result_size = len(str(result)) if result else 0
+                print(f"[{task_id}] {tool} result size: {result_size} chars")
+            
+            print(f"[{task_id}] FINAL ANSWER: {answer}")
+            print(f"{'='*60}")
+            
+            return answer
+            
+        except Exception as e:
+            error_msg = f"Critical error in enhanced agent execution: {str(e)}"
+            print(f"[{task_id}] {error_msg}")
+            
+            # Try fallback direct LLM response
+            try:
+                fallback_response = llm.invoke(f"Please answer this question: {question}").content
+                return f"Fallback response: {fallback_response}"
+            except:
+                return error_msg
+
+# ----------- Enhanced File Processing -----------
+def detect_file_type(file_path: str) -> Optional[str]:
+    """Enhanced file type detection with more formats"""
+    ext = Path(file_path).suffix.lower()
+    
+    file_type_mapping = {
+        # Spreadsheets
+        '.xlsx': 'excel', '.xls': 'excel', '.csv': 'excel',
+        # Images
+        '.png': 'image', '.jpg': 'image', '.jpeg': 'image', 
+        '.bmp': 'image', '.gif': 'image', '.tiff': 'image', '.webp': 'image',
+        # Audio
+        '.mp3': 'audio', '.wav': 'audio', '.ogg': 'audio', 
+        '.flac': 'audio', '.m4a': 'audio', '.aac': 'audio',
+        # Code
+        '.py': 'code', '.ipynb': 'code', '.js': 'code', '.html': 'code',
+        '.css': 'code', '.java': 'code', '.cpp': 'code', '.c': 'code',
+        '.sql': 'code', '.r': 'code', '.json': 'code', '.xml': 'code',
+        # Documents
+        '.txt': 'text', '.md': 'text', '.pdf': 'document',
+        '.doc': 'document', '.docx': 'document'
+    }
+    
+    return file_type_mapping.get(ext)
+
+def process_file(task_id: str, question_text: str) -> str:
+    """Enhanced file processing with better error handling and metadata"""
+    file_url = f"{FILE_PATH}{task_id}"
+    
+    try:
+        print(f"[{task_id}] Attempting to download file from: {file_url}")
+        response = requests.get(file_url, timeout=30)
+        response.raise_for_status()
+        print(f"[{task_id}] File download successful. Status: {response.status_code}")
+        
+    except requests.exceptions.RequestException as exc:
+        print(f"[{task_id}] File download failed: {str(exc)}")
+        return question_text  # Return original question if no file
+    
+    # Enhanced filename extraction
+    content_disposition = response.headers.get("content-disposition", "")
+    filename = task_id  # Default fallback
+    
+    # Try to extract filename from Content-Disposition header
+    filename_match = re.search(r'filename[*]?=(?:"([^"]+)"|([^;]+))', content_disposition)
+    if filename_match:
+        filename = filename_match.group(1) or filename_match.group(2)
+        filename = filename.strip()
+    
+    # Create enhanced temp directory structure
+    temp_storage_dir = Path(tempfile.gettempdir()) / "gaia_enhanced_files" / task_id
+    temp_storage_dir.mkdir(parents=True, exist_ok=True)
+    
+    file_path = temp_storage_dir / filename
+    file_path.write_bytes(response.content)
+    
+    # Get file metadata
+    file_size = len(response.content)
+    file_type = detect_file_type(filename)
+    
+    print(f"[{task_id}] File saved: {filename} ({file_size:,} bytes, type: {file_type})")
+    
+    # Enhanced question augmentation
+    enhanced_question = f"{question_text}\n\n"
+    enhanced_question += f"{'='*50}\n"
+    enhanced_question += f"FILE INFORMATION:\n"
+    enhanced_question += f"A file was downloaded for this task and saved locally at:\n"
+    enhanced_question += f"{file_path}\n"
+    enhanced_question += f"File details:\n"
+    enhanced_question += f"- Name: {filename}\n"
+    enhanced_question += f"- Size: {file_size:,} bytes ({file_size/1024:.1f} KB)\n"
+    enhanced_question += f"- Type: {file_type or 'unknown'}\n"
+    enhanced_question += f"{'='*50}\n\n"
+    
+    return enhanced_question
+
+# ----------- Usage Examples and Testing -----------
+def run_enhanced_tests():
+    """Run comprehensive tests of the enhanced agent"""
+    agent = GaiaAgent()
+    
+    test_cases = [
+        {
+            "id": "test_search_1",
+            "question": "What are the latest developments in artificial intelligence in 2024?",
+            "expected_tools": ["search"]
+        },
+        {
+            "id": "test_wiki_1", 
+            "question": "Tell me about Albert Einstein's contributions to physics",
+            "expected_tools": ["wikipedia"]
+        },
+        {
+            "id": "test_combined_1",
+            "question": "What is machine learning and what are recent breakthroughs?",
+            "expected_tools": ["wikipedia", "search"]
+        },
+        {
+            "id": "test_excel_1",
+            "question": "Analyze the data in the Excel file sales_data.xlsx",
+            "expected_tools": ["excel"]
+        }
+    ]
+    
+    print("\n" + "="*80)
+    print("RUNNING ENHANCED AGENT TESTS")
+    print("="*80)
+    
+    for test_case in test_cases:
+        print(f"\nTest Case: {test_case['id']}")
+        print(f"Question: {test_case['question']}")
+        print(f"Expected tools: {test_case['expected_tools']}")
+        
+        try:
+            result = agent(test_case['id'], test_case['question'])
+            print(f"Result length: {len(result)} characters")
+            print(f"Result preview: {result[:200]}...")
+        except Exception as e:
+            print(f"Test failed: {e}")
+        
+        print("-" * 60)
+    
+    # Print tool usage statistics
+    print(f"\nTool Usage Statistics:")
+    for tool, count in agent.get_tool_stats().items():
+        print(f"  {tool}: {count} times")
+
+# Usage example
+if __name__ == "__main__":
+    # Create enhanced agent
+    agent = GaiaAgent()
+    
+    # Example usage
+    sample_questions = [
+ 
+         {
+    "task_id": "bda648d7-d618-4883-88f4-3466eabd860e",
+    "question": "Where were the Vietnamese specimens described by Kuznetzov in Nedoshivina's 2010 paper eventually deposited? Just give me the city name without abbreviations.",
+    "Level": "1",
+    "file_name": ""
+        }
+        
+        # "Explain quantum computing and its recent developments",
+        # "Tell me about the history of machine learning and current AI trends",
+    ]
+    
+    print("\n" + "="*80)
+    print("ENHANCED GAIA AGENT DEMONSTRATION")
+    print("="*80)
+    
+    for i, task in enumerate(sample_questions):
+        print(f"\nExample {i+1}: {task['question']}")
+        result = agent(task['task_id'], task['question'])
+        print(f"Answer: {result[:300]}...")
+        print("-" * 60)
+    
+    # Uncomment to run comprehensive tests
+    # run_enhanced_tests()
+
+
+

requirements.txt

@@ -1,2 +1,17 @@
-gradio
-requests
+gradio
+requests
+langchain
+openai-whisper
+google-generativeai
+python-dotenv
+yt-dlp
+tabulate
+langchain-community
+langchain-core
+langgraph
+langchain-google-genai
+duckduckgo-search
+wikipedia
+Pillow
+SpeechRecognition 
+transformers

tools.py

@@ -0,0 +1,559 @@
+from langchain.tools import DuckDuckGoSearchResults, WikipediaQueryRun
+from langchain.utilities import WikipediaAPIWrapper
+from PIL import Image
+import re
+import time
+import json
+import pandas as pd
+from pathlib import Path
+from typing import List, Dict, Optional, Union
+from tabulate import tabulate
+import whisper
+
+import numpy as np
+import os
+from langchain_groq import ChatGroq
+from youtube_transcript_api import YouTubeTranscriptApi
+import re
+
+from langchain_openai import ChatOpenAI
+llm=ChatOpenAI(model='gpt-4o', temperature=0)
+
+# ----------- Enhanced Search Functionality -----------
+class EnhancedSearchTool:
+    """Enhanced web search with intelligent query processing and result filtering"""
+    
+    def __init__(self, max_results: int = 10):
+        self.base_tool = DuckDuckGoSearchResults(num_results=max_results)
+        self.max_results = max_results
+        
+    def _extract_key_terms(self, question: str) -> List[str]:
+        """Extract key search terms from the question using LLM"""
+        try:
+            extract_prompt = f"""
+            Extract the most important search terms from this question for web search:
+            Question: {question}
+            
+            Return ONLY a comma-separated list of key terms, no explanations.
+            Focus on: proper nouns, specific concepts, technical terms, dates, numbers.
+            Avoid: common words like 'what', 'how', 'when', 'the', 'is', 'are'.
+            
+            Example: "What is the population of Tokyo in 2023?" -> "Tokyo population 2023"
+            """
+            
+            response = llm.invoke(extract_prompt).content.strip()
+            return [term.strip() for term in response.split(',')]
+        except Exception:
+            # Fallback to simple keyword extraction
+            return self._simple_keyword_extraction(question)
+    
+    def _simple_keyword_extraction(self, question: str) -> List[str]:
+        """Fallback keyword extraction using regex"""
+        # Remove common question words
+        stop_words = {'what', 'how', 'when', 'where', 'why', 'who', 'which', 'the', 'is', 'are', 'was', 'were', 'do', 'does', 'did', 'can', 'could', 'should', 'would'}
+        words = re.findall(r'\b[A-Za-z]+\b', question.lower())
+        return [word for word in words if word not in stop_words and len(word) > 2]
+    
+    def _generate_search_queries(self, question: str) -> List[str]:
+        """Generate multiple search queries for comprehensive results"""
+        key_terms = self._extract_key_terms(question)
+        
+        queries = []
+        
+        # Original question (cleaned)
+        cleaned_question = re.sub(r'[^\w\s]', ' ', question).strip()
+        queries.append(cleaned_question)
+        
+        # Key terms combined
+        if key_terms:
+            queries.append(' '.join(key_terms[:5]))  # Top 5 terms
+        
+        # Specific query patterns based on question type
+        if any(word in question.lower() for word in ['latest', 'recent', 'current', 'new']):
+            queries.append(f"{' '.join(key_terms[:3])} 2024 2025")
+        
+        if any(word in question.lower() for word in ['statistics', 'data', 'number', 'count']):
+            queries.append(f"{' '.join(key_terms[:3])} statistics data")
+        
+        if any(word in question.lower() for word in ['definition', 'what is', 'meaning']):
+            queries.append(f"{' '.join(key_terms[:2])} definition meaning")
+        
+        return list(dict.fromkeys(queries))  # Remove duplicates while preserving order
+    
+    def _filter_and_rank_results(self, results: List[Dict], question: str) -> List[Dict]:
+        """Filter and rank search results based on relevance"""
+        if not results:
+            return results
+        
+        key_terms = self._extract_key_terms(question)
+        key_terms_lower = [term.lower() for term in key_terms]
+        
+        scored_results = []
+        for result in results:
+            score = 0
+            text_content = (result.get('snippet', '') + ' ' + result.get('title', '')).lower()
+            
+            # Score based on key term matches
+            for term in key_terms_lower:
+                if term in text_content:
+                    score += text_content.count(term)
+            
+            # Bonus for recent dates
+            if any(year in text_content for year in ['2024', '2025', '2023']):
+                score += 2
+            
+            # Penalty for very short snippets
+            if len(result.get('snippet', '')) < 50:
+                score -= 1
+            
+            scored_results.append((score, result))
+        
+        # Sort by score and return top results
+        scored_results.sort(key=lambda x: x[0], reverse=True)
+        return [result for score, result in scored_results[:self.max_results]]
+    
+    def run(self, question: str) -> str:
+        """Enhanced search execution with multiple queries and result filtering"""
+        try:
+            search_queries = self._generate_search_queries(question)
+            all_results = []
+            
+            for query in search_queries[:3]:  # Limit to 3 queries to avoid rate limits
+                try:
+                    results = self.base_tool.run(query)
+                    if isinstance(results, str):
+                        # Parse string results if needed
+                        try:
+                            results = json.loads(results) if results.startswith('[') else [{'snippet': results, 'title': 'Search Result'}]
+                        except:
+                            results = [{'snippet': results, 'title': 'Search Result'}]
+                    
+                    if isinstance(results, list):
+                        all_results.extend(results)
+                    
+                    time.sleep(0.5)  # Rate limiting
+                except Exception as e:
+                    print(f"Search query failed: {query} - {e}")
+                    continue
+            
+            if not all_results:
+                return "No search results found."
+            
+            # Filter and rank results
+            filtered_results = self._filter_and_rank_results(all_results, question)
+            
+            # Format results
+            formatted_results = []
+            for i, result in enumerate(filtered_results[:5], 1):
+                title = result.get('title', 'No title')
+                snippet = result.get('snippet', 'No description')
+                link = result.get('link', '')
+                
+                formatted_results.append(f"{i}. {title}\n   {snippet}\n   Source: {link}\n")
+            
+            return "ENHANCED SEARCH RESULTS:\n" + "\n".join(formatted_results)
+            
+        except Exception as e:
+            return f"Enhanced search error: {str(e)}"
+
+# ----------- Enhanced Wikipedia Tool -----------
+
+class EnhancedWikipediaTool:
+    """Enhanced Wikipedia search with intelligent query processing and content extraction"""
+    
+    def __init__(self):
+        self.base_wrapper = WikipediaAPIWrapper(
+            top_k_results=3,
+            doc_content_chars_max=3000,
+            load_all_available_meta=True
+        )
+        self.base_tool = WikipediaQueryRun(api_wrapper=self.base_wrapper)
+    
+    def _extract_entities(self, question: str) -> List[str]:
+        """Extract named entities for Wikipedia search"""
+        try:
+            entity_prompt = f"""
+            Extract named entities (people, places, organizations, concepts) from this question for Wikipedia search:
+            Question: {question}
+            
+            Return ONLY a comma-separated list of the most important entities.
+            Focus on: proper nouns, specific names, places, organizations, historical events, scientific concepts.
+            
+            Example: "Tell me about Einstein's theory of relativity" -> "Albert Einstein, theory of relativity, relativity"
+            """
+            response = llm.invoke(entity_prompt).content.strip()
+            print(f'inside extract_entities:{response}')
+            entities = [entity.strip() for entity in response.split(',')]
+            return [e for e in entities if len(e) > 2]
+        except Exception:
+            # Fallback: extract capitalized words and phrases
+            return self._extract_capitalized_terms(question)
+    
+    def _extract_capitalized_terms(self, question: str) -> List[str]:
+        """Fallback: extract capitalized terms as potential entities"""
+        # Find capitalized words and phrases
+        capitalized_words = re.findall(r'\b[A-Z][a-z]+(?:\s+[A-Z][a-z]+)*\b', question)
+        # Also look for quoted terms
+        quoted_terms = re.findall(r'"([^"]+)"', question)
+        quoted_terms.extend(re.findall(r"'([^']+)'", question))
+        
+        return capitalized_words + quoted_terms
+    
+    def _search_multiple_terms(self, entities: List[str]) -> Dict[str, str]:
+        """Search Wikipedia for multiple entities and return best results"""
+        results = {}
+        
+        for entity in entities[:3]:  # Limit to avoid too many API calls
+            try:
+                result = self.base_tool.run(entity)
+                print(f'Inside _search_multiple_terms: {result}')
+                if result and "Page:" in result and len(result) > 100:
+                    results[entity] = result
+                time.sleep(0.5)  # Rate limiting
+            except Exception as e:
+                print(f"Wikipedia search failed for '{entity}': {e}")
+                continue
+        
+        return results
+    
+    def _extract_relevant_sections(self, content: str, question: str) -> str:
+        """Extract the most relevant sections from Wikipedia content"""
+        if not content or len(content) < 200:
+            return content
+        
+        # Split content into sections (usually separated by double newlines)
+        sections = re.split(r'\n\s*\n', content)
+        print(f'Inside _extract relevant sections:{sections}')
+        
+        # Score sections based on relevance to question
+        key_terms = self._extract_entities(question)
+        key_terms_lower = [term.lower() for term in key_terms]
+        
+        scored_sections = []
+        for section in sections:
+            if len(section.strip()) < 500:
+                continue
+                
+            score = 0
+            section_lower = section.lower()
+            
+            # Score based on key term matches
+            for term in key_terms_lower:
+                score += section_lower.count(term)
+            
+            # Bonus for sections with dates, numbers, or specific facts
+            if re.search(r'\b(19|20)\d{2}\b', section):  # Years
+                score += 1
+            if re.search(r'\b\d+([.,]\d+)?\s*(million|billion|thousand|percent|%)\b', section):
+                score += 1
+            
+            scored_sections.append((score, section))
+        
+        # Sort by relevance and take top sections
+        scored_sections.sort(key=lambda x: x[0], reverse=True)
+        top_sections = [section for score, section in scored_sections[:7] if score > 0]
+        print(f'Inside extract relevant sections, top sections:{top_sections}')
+        
+        if not top_sections:
+            # If no highly relevant sections, take first few sections
+            top_sections = sections[:2]
+        
+        return '\n\n'.join(top_sections)
+    
+    def run(self, question: str) -> str:
+        """Enhanced Wikipedia search with entity extraction and content filtering"""
+        try:
+            entities = self._extract_entities(question)
+            
+            if not entities:
+                # Fallback to direct search with cleaned question
+                cleaned_question = re.sub(r'[^\w\s]', ' ', question).strip()
+                try:
+                    result = self.base_tool.run(cleaned_question)
+                    print(f'******************Inside run*************:{result} ')
+                    return self._extract_relevant_sections(result, question) if result else "No Wikipedia results found."
+                except Exception as e:
+                    return f"Wikipedia search error: {str(e)}"
+            
+            # Search for multiple entities
+            search_results = self._search_multiple_terms(entities)
+            
+            if not search_results:
+                return "No relevant Wikipedia articles found."
+            
+            # Combine and format results
+            formatted_results = []
+            for entity, content in search_results.items():
+                relevant_content = self._extract_relevant_sections(content, question)
+                if relevant_content:
+                    formatted_results.append(f"=== {entity} ===\n{relevant_content}")
+            
+            if not formatted_results:
+                return "No relevant information found in Wikipedia articles."
+            
+            return "ENHANCED WIKIPEDIA RESULTS:\n\n" + "\n\n".join(formatted_results)
+            
+        except Exception as e:
+            return f"Enhanced Wikipedia error: {str(e)}"
+
+# ----------- Enhanced File Processing Tools -----------
+def excel_to_markdown(excel_path: str, sheet_name: Optional[str] = None) -> str:
+    """Enhanced Excel tool with better error handling and data analysis"""
+    try:
+        file_path = Path(excel_path).expanduser().resolve()
+        if not file_path.is_file():
+            return f"Error: Excel file not found at {file_path}"
+
+        sheet: Union[str, int] = (
+            int(sheet_name) if sheet_name and sheet_name.isdigit() else sheet_name or 0
+        )
+        df = pd.read_excel(file_path, sheet_name=sheet)
+        df = df.iloc[:, :-1]
+
+        # Enhanced metadata
+        metadata = f"EXCEL FILE ANALYSIS:\n"
+        metadata += f"File: {file_path.name}\n"
+        metadata += f"Dimensions: {len(df)} rows × {len(df.columns)} columns\n"
+        metadata += f"Columns: {', '.join(df.columns.tolist())}\n"
+        metadata += f"Data types: {dict(df.dtypes)}\n"
+
+        # Basic statistics for numeric columns
+        numeric_cols = df.select_dtypes(include=['number']).columns
+        if len(numeric_cols) > 0:
+            metadata += f"Numeric columns: {list(numeric_cols)}\n"
+            for col in numeric_cols:
+                metadata += f"  {col}: mean={df[col].mean():.2f}, min={df[col].min()}, max={df[col].max()}, sum={df[col].sum()}\n"
+
+        metadata += "\nSAMPLE DATA (first 10 rows):\n"
+
+        # if hasattr(df, "to_markdown"):
+        #     sample_data = df.head(10).to_markdown(index=False)
+        # else:
+        #     sample_data = tabulate(df.head(10), headers="keys", tablefmt="github", showindex=False)
+
+        # return metadata + sample_data + f"\n\n(Showing first 10 rows of {len(df)} total rows)"
+        return metadata
+
+    except Exception as e:
+        return f"Error reading Excel file: {str(e)}"
+
+
+import os
+import mimetypes
+from pathlib import Path
+
+def image_file_info(image_path: str, question: str) -> str:
+    """Enhanced image file analysis using Gemini API"""
+    try:
+        # Check if file exists
+        if not os.path.exists(image_path):
+            return f"Error: Image file not found at {image_path}"
+        
+        # Try the older google.generativeai library first (more stable)
+        try:
+            import google.generativeai as genai
+            from PIL import Image
+            
+            # Configure the API key
+            genai.configure(api_key=os.getenv("GEMINI_API_KEY"))
+            
+            # Create the model - using a more stable model
+            model = genai.GenerativeModel('gemini-1.5-flash')
+            
+            # Open and validate the image
+            try:
+                image = Image.open(image_path)
+                # Convert to RGB if necessary (handles PNG with transparency)
+                if image.mode in ('RGBA', 'LA'):
+                    background = Image.new('RGB', image.size, (255, 255, 255))
+                    if image.mode == 'RGBA':
+                        background.paste(image, mask=image.split()[-1])
+                    else:
+                        background.paste(image, mask=image.split()[-1])
+                    image = background
+                elif image.mode != 'RGB':
+                    image = image.convert('RGB')
+                    
+            except Exception as img_error:
+                return f"Error opening image: {img_error}"
+            
+            # Generate content using the older SDK
+            response = model.generate_content([question, image])
+            
+            return response.text
+            
+        except ImportError:
+            # Fall back to the newer google.genai library
+            try:
+                from google import genai
+                from google.genai import types
+                
+                # Initialize the client
+                client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))
+                
+                # Read content from a local file
+                with open(image_path, "rb") as f:
+                    img_bytes = f.read()
+                
+                # Determine the correct MIME type based on file extension
+                mime_type, _ = mimetypes.guess_type(image_path)
+                if mime_type is None or not mime_type.startswith('image/'):
+                    # For PNG files specifically
+                    if image_path.lower().endswith('.png'):
+                        mime_type = "image/png"
+                    else:
+                        mime_type = "image/jpeg"
+                
+                # Generate content using the newer SDK
+                response = client.models.generate_content(
+                    model="gemini-1.5-flash",  # Using more stable model
+                    contents=[
+                        question,
+                        types.Part.from_bytes(data=img_bytes, mime_type=mime_type)
+                    ],
+                )
+                
+                return response.text
+                
+            except Exception as new_sdk_error:
+                return f"Error with both SDKs. New SDK error: {new_sdk_error}"
+    
+    except Exception as e:
+        return f"Error during image analysis: {e}"
+
+def audio_file_info(audio_path: str) -> str:
+    """Returns only the transcription of an audio file."""
+    try:
+        model = whisper.load_model("tiny")  # Fast + accurate balance
+        result = model.transcribe(audio_path, fp16=False)
+        return result['text']
+    except Exception as e:
+        return f"Error transcribing audio: {str(e)}"
+
+def code_file_read(code_path: str) -> str:
+    """Enhanced code file analysis"""
+    try:
+        with open(code_path, "r", encoding="utf-8") as f:
+            content = f.read()
+        
+        file_path = Path(code_path)
+        
+        info = f"CODE FILE ANALYSIS:\n"
+        info += f"File: {file_path.name}\n"
+        info += f"Extension: {file_path.suffix}\n"
+        info += f"Size: {len(content)} characters, {len(content.splitlines())} lines\n"
+        
+        # Language-specific analysis
+        if file_path.suffix == '.py':
+            # Python-specific analysis
+            import_lines = [line for line in content.splitlines() if line.strip().startswith(('import ', 'from '))]
+            if import_lines:
+                info += f"Imports ({len(import_lines)}): {', '.join(import_lines[:5])}\n"
+            
+            # Count functions and classes
+            func_count = len(re.findall(r'^def\s+\w+', content, re.MULTILINE))
+            class_count = len(re.findall(r'^class\s+\w+', content, re.MULTILINE))
+            info += f"Functions: {func_count}, Classes: {class_count}\n"
+        
+        info += f"\nCODE CONTENT:\n{content}"
+        return info
+        
+    except Exception as e:
+        return f"Error reading code file: {e}"
+    
+    
+import yt_dlp
+from pathlib import Path
+
+def extract_youtube_info(question: str) -> str:
+    """
+    Download a YouTube video or audio using yt-dlp without merging.
+
+    Parameters:
+    - url: str — YouTube URL
+    - audio_only: bool — if True, downloads audio only; else best single video+audio stream
+
+    Returns:
+    - str: path to downloaded file or error message
+    """
+    pattern = r"(https?://(?:www\.)?(?:youtube\.com/watch\?v=[\w\-]+|youtu\.be/[\w\-]+))"
+    match = re.search(pattern, question)
+    youtube_url =  match.group(1) if match else None
+    
+    print(f"Extracting YouTube URL: {youtube_url}")
+    try:
+        # Extract video ID from URL
+        video_id = re.search(r'(?:youtube\.com/watch\?v=|youtu\.be/|youtube\.com/embed/)([a-zA-Z0-9_-]{11})', youtube_url).group(1)
+        
+        # Get transcript
+        transcript_list = YouTubeTranscriptApi.get_transcript(video_id)
+        
+        # Combine all text segments
+        full_transcript = ' '.join([entry['text'] for entry in transcript_list])
+        
+        # Clean up the text
+        full_transcript = re.sub(r'\s+', ' ', full_transcript).strip()
+        
+        return full_transcript
+        
+    except Exception as e:
+        print(f"Error getting transcript: {e}")
+        return None
+    
+
+# def get_youtube_transcript(video_url):
+#     """
+#     Get transcription from a YouTube video.
+    
+#     Args:
+#         video_url (str): YouTube video URL
+    
+#     Returns:
+#         str: Full transcription text or None if not available
+#     """
+#     try:
+#         # Extract video ID from URL
+#         video_id = re.search(r'(?:youtube\.com/watch\?v=|youtu\.be/|youtube\.com/embed/)([a-zA-Z0-9_-]{11})', video_url).group(1)
+        
+#         # Get transcript
+#         transcript_list = YouTubeTranscriptApi.get_transcript(video_id)
+        
+#         # Combine all text segments
+#         full_transcript = ' '.join([entry['text'] for entry in transcript_list])
+        
+#         # Clean up the text
+#         full_transcript = re.sub(r'\s+', ' ', full_transcript).strip()
+        
+#         return full_transcript
+        
+#     except Exception as e:
+#         print(f"Error getting transcript: {e}")
+#         return None
+
+# extract_youtube_info
+
+# question="How many studio albums were published by Mercedes Sosa between 2000 and 2009 (included)? You can use the latest 2022 version of english wikipedia."
+# wiki=EnhancedWikipediaTool()
+# wiki.run(question)
+
+
+# entity_prompt = f"""
+#     Extract named entities (people, places, organizations, concepts) from this question for Wikipedia search:
+#     Question: {question}
+    
+#     Return ONLY a comma-separated list of the most important entities.
+#     Focus on: proper nouns, specific names, places, organizations, historical events, scientific concepts.
+    
+#     Example: "Tell me about Einstein's theory of relativity" -> "Albert Einstein, theory of relativity, relativity"
+#     """
+
+# response = llm.invoke(entity_prompt).content.strip()
+
+
+
+
+# result=extract_youtube_info("Examine the video at https://www.youtube.com/watch?v=1htKBjuUWec.\n\nWhat does Teal'c say in response to the question \"Isn't that hot")
+
+
+
+