Create gemini_agent.py

gemini_agent.py

@@ -0,0 +1,660 @@
+import os
+import tempfile
+import time
+import re
+import json
+from typing import List, Optional, Dict, Any
+from urllib.parse import urlparse
+import requests
+import yt_dlp
+from bs4 import BeautifulSoup
+from difflib import SequenceMatcher
+
+from langchain_core.messages import HumanMessage, SystemMessage
+from langchain_google_genai import ChatGoogleGenerativeAI
+from langchain_community.utilities import DuckDuckGoSearchAPIWrapper, WikipediaAPIWrapper
+from langchain.agents import Tool, AgentExecutor, ConversationalAgent, initialize_agent, AgentType
+from langchain.memory import ConversationBufferMemory
+from langchain.prompts import MessagesPlaceholder
+from langchain.tools import BaseTool, Tool, tool
+from google.generativeai.types import HarmCategory, HarmBlockThreshold
+from PIL import Image
+import google.generativeai as genai
+from pydantic import Field
+
+from smolagents import WikipediaSearchTool
+
+class SmolagentToolWrapper(BaseTool):
+    """Wrapper for smolagents tools to make them compatible with LangChain."""
+    
+    wrapped_tool: object = Field(description="The wrapped smolagents tool")
+    
+    def __init__(self, tool):
+        """Initialize the wrapper with a smolagents tool."""
+        super().__init__(
+            name=tool.name,
+            description=tool.description,
+            return_direct=False,
+            wrapped_tool=tool
+        )
+
+    def _run(self, query: str) -> str:
+        """Use the wrapped tool to execute the query."""
+        try:
+            # For WikipediaSearchTool
+            if hasattr(self.wrapped_tool, 'search'):
+                return self.wrapped_tool.search(query)
+            # For DuckDuckGoSearchTool and others
+            return self.wrapped_tool(query)
+        except Exception as e:
+            return f"Error using tool: {str(e)}"
+    
+    def _arun(self, query: str) -> str:
+        """Async version - just calls sync version since smolagents tools don't support async."""
+        return self._run(query)
+
+class WebSearchTool:
+    def __init__(self):
+        self.last_request_time = 0
+        self.min_request_interval = 2.0  # Minimum time between requests in seconds
+        self.max_retries = 10
+
+    def search(self, query: str, domain: Optional[str] = None) -> str:
+        """Perform web search with rate limiting and retries."""
+        for attempt in range(self.max_retries):
+            # Implement rate limiting
+            current_time = time.time()
+            time_since_last = current_time - self.last_request_time
+            if time_since_last < self.min_request_interval:
+                time.sleep(self.min_request_interval - time_since_last)
+            
+            try:
+                # Make the search request
+                results = self._do_search(query, domain)
+                self.last_request_time = time.time()
+                return results
+            except Exception as e:
+                if "202 Ratelimit" in str(e):
+                    if attempt < self.max_retries - 1:
+                        # Exponential backoff
+                        wait_time = (2 ** attempt) * self.min_request_interval
+                        time.sleep(wait_time)
+                        continue
+                return f"Search failed after {self.max_retries} attempts: {str(e)}"
+        
+        return "Search failed due to rate limiting"
+
+    def _do_search(self, query: str, domain: Optional[str] = None) -> str:
+        """Perform the actual search request."""
+        try:
+            # Construct search URL
+            base_url = "https://html.duckduckgo.com/html"
+            params = {"q": query}
+            if domain:
+                params["q"] += f" site:{domain}"
+
+            # Make request with increased timeout
+            response = requests.get(base_url, params=params, timeout=10)
+            response.raise_for_status()
+
+            if response.status_code == 202:
+                raise Exception("202 Ratelimit")
+
+            # Extract search results
+            results = []
+            soup = BeautifulSoup(response.text, 'html.parser')
+            for result in soup.find_all('div', {'class': 'result'}):
+                title = result.find('a', {'class': 'result__a'})
+                snippet = result.find('a', {'class': 'result__snippet'})
+                if title and snippet:
+                    results.append({
+                        'title': title.get_text(),
+                        'snippet': snippet.get_text(),
+                        'url': title.get('href')
+                    })
+
+            # Format results
+            formatted_results = []
+            for r in results[:10]:  # Limit to top 5 results
+                formatted_results.append(f"[{r['title']}]({r['url']})\n{r['snippet']}\n")
+
+            return "## Search Results\n\n" + "\n".join(formatted_results)
+
+        except requests.RequestException as e:
+            raise Exception(f"Search request failed: {str(e)}")
+
+def save_and_read_file(content: str, filename: Optional[str] = None) -> str:
+    """
+    Save content to a temporary file and return the path.
+    Useful for processing files from the GAIA API.
+    
+    Args:
+        content: The content to save to the file
+        filename: Optional filename, will generate a random name if not provided
+        
+    Returns:
+        Path to the saved file
+    """
+    temp_dir = tempfile.gettempdir()
+    if filename is None:
+        temp_file = tempfile.NamedTemporaryFile(delete=False)
+        filepath = temp_file.name
+    else:
+        filepath = os.path.join(temp_dir, filename)
+    
+    # Write content to the file
+    with open(filepath, 'w') as f:
+        f.write(content)
+    
+    return f"File saved to {filepath}. You can read this file to process its contents."
+
+
+def download_file_from_url(url: str, filename: Optional[str] = None) -> str:
+    """
+    Download a file from a URL and save it to a temporary location.
+    
+    Args:
+        url: The URL to download from
+        filename: Optional filename, will generate one based on URL if not provided
+        
+    Returns:
+        Path to the downloaded file
+    """
+    try:
+        # Parse URL to get filename if not provided
+        if not filename:
+            path = urlparse(url).path
+            filename = os.path.basename(path)
+            if not filename:
+                # Generate a random name if we couldn't extract one
+                import uuid
+                filename = f"downloaded_{uuid.uuid4().hex[:8]}"
+        
+        # Create temporary file
+        temp_dir = tempfile.gettempdir()
+        filepath = os.path.join(temp_dir, filename)
+        
+        # Download the file
+        response = requests.get(url, stream=True)
+        response.raise_for_status()
+        
+        # Save the file
+        with open(filepath, 'wb') as f:
+            for chunk in response.iter_content(chunk_size=8192):
+                f.write(chunk)
+        
+        return f"File downloaded to {filepath}. You can now process this file."
+    except Exception as e:
+        return f"Error downloading file: {str(e)}"
+
+
+def extract_text_from_image(image_path: str) -> str:
+    """
+    Extract text from an image using pytesseract (if available).
+    
+    Args:
+        image_path: Path to the image file
+        
+    Returns:
+        Extracted text or error message
+    """
+    try:
+        # Try to import pytesseract
+        import pytesseract
+        from PIL import Image
+        
+        # Open the image
+        image = Image.open(image_path)
+        
+        # Extract text
+        text = pytesseract.image_to_string(image)
+        
+        return f"Extracted text from image:\n\n{text}"
+    except ImportError:
+        return "Error: pytesseract is not installed. Please install it with 'pip install pytesseract' and ensure Tesseract OCR is installed on your system."
+    except Exception as e:
+        return f"Error extracting text from image: {str(e)}"
+
+
+def analyze_csv_file(file_path: str, query: str) -> str:
+    """
+    Analyze a CSV file using pandas and answer a question about it.
+    
+    Args:
+        file_path: Path to the CSV file
+        query: Question about the data
+        
+    Returns:
+        Analysis result or error message
+    """
+    try:
+        import pandas as pd
+        
+        # Read the CSV file
+        df = pd.read_csv(file_path)
+        
+        # Run various analyses based on the query
+        result = f"CSV file loaded with {len(df)} rows and {len(df.columns)} columns.\n"
+        result += f"Columns: {', '.join(df.columns)}\n\n"
+        
+        # Add summary statistics
+        result += "Summary statistics:\n"
+        result += str(df.describe())
+        
+        return result
+    except ImportError:
+        return "Error: pandas is not installed. Please install it with 'pip install pandas'."
+    except Exception as e:
+        return f"Error analyzing CSV file: {str(e)}"
+
+@tool
+def analyze_excel_file(file_path: str, query: str) -> str:
+    """
+    Analyze an Excel file using pandas and answer a question about it.
+    
+    Args:
+        file_path: Path to the Excel file
+        query: Question about the data
+        
+    Returns:
+        Analysis result or error message
+    """
+    try:
+        import pandas as pd
+        
+        # Read the Excel file
+        df = pd.read_excel(file_path)
+        
+        # Run various analyses based on the query
+        result = f"Excel file loaded with {len(df)} rows and {len(df.columns)} columns.\n"
+        result += f"Columns: {', '.join(df.columns)}\n\n"
+        
+        # Add summary statistics
+        result += "Summary statistics:\n"
+        result += str(df.describe())
+        
+        return result
+    except ImportError:
+        return "Error: pandas and openpyxl are not installed. Please install them with 'pip install pandas openpyxl'."
+    except Exception as e:
+        return f"Error analyzing Excel file: {str(e)}"
+
+class GeminiAgent:
+    def __init__(self, api_key: str, model_name: str = "gemini-2.0-flash"):
+        # Suppress warnings
+        import warnings
+        warnings.filterwarnings("ignore", category=UserWarning)
+        warnings.filterwarnings("ignore", category=DeprecationWarning)
+        warnings.filterwarnings("ignore", message=".*will be deprecated.*")
+        warnings.filterwarnings("ignore", "LangChain.*")
+        
+        self.api_key = api_key
+        self.model_name = model_name
+        
+        # Configure Gemini
+        genai.configure(api_key=api_key)
+        
+        # Initialize the LLM
+        self.llm = self._setup_llm()
+        
+        # Setup tools
+        self.tools = [
+            SmolagentToolWrapper(WikipediaSearchTool()),
+            Tool(
+                name="analyze_video",
+                func=self._analyze_video,
+                description="Analyze YouTube video content directly"
+            ),
+            Tool(
+                name="analyze_image",
+                func=self._analyze_image,
+                description="Analyze image content"
+            ),
+            Tool(
+                name="analyze_table",
+                func=self._analyze_table,
+                description="Analyze table or matrix data"
+            ),
+            Tool(
+                name="analyze_list",
+                func=self._analyze_list,
+                description="Analyze and categorize list items"
+            ),
+            Tool(
+                name="web_search",
+                func=self._web_search,
+                description="Search the web for information"
+            )
+        ]
+        
+        # Setup memory
+        self.memory = ConversationBufferMemory(
+            memory_key="chat_history",
+            return_messages=True
+        )
+        
+        # Initialize agent
+        self.agent = self._setup_agent()
+       
+
+    def run(self, query: str) -> str:
+        """Run the agent on a query with incremental retries."""
+        max_retries = 3
+        base_sleep = 1  # Start with 1 second sleep
+        
+        for attempt in range(max_retries):
+            try:
+
+                # If no match found in answer bank, use the agent
+                response = self.agent.run(query)
+                return response
+
+            except Exception as e:
+                sleep_time = base_sleep * (attempt + 1)  # Incremental sleep: 1s, 2s, 3s
+                if attempt < max_retries - 1:
+                    print(f"Attempt {attempt + 1} failed. Retrying in {sleep_time} seconds...")
+                    time.sleep(sleep_time)
+                    continue
+                return f"Error processing query after {max_retries} attempts: {str(e)}"
+
+        print("Agent processed all queries!")
+
+    def _clean_response(self, response: str) -> str:
+        """Clean up the response from the agent."""
+        # Remove any tool invocation artifacts
+        cleaned = re.sub(r'> Entering new AgentExecutor chain...|> Finished chain.', '', response)
+        cleaned = re.sub(r'Thought:.*?Action:.*?Action Input:.*?Observation:.*?\n', '', cleaned, flags=re.DOTALL)
+        return cleaned.strip()
+
+    def run_interactive(self):
+        print("AI Assistant Ready! (Type 'exit' to quit)")
+        
+        while True:
+            query = input("You: ").strip()
+            if query.lower() == 'exit':
+                print("Goodbye!")
+                break
+            
+            print("Assistant:", self.run(query))
+
+    def _web_search(self, query: str, domain: Optional[str] = None) -> str:
+        """Perform web search with rate limiting and retries."""
+        try:
+            # Use DuckDuckGo API wrapper for more reliable results
+            search = DuckDuckGoSearchAPIWrapper(max_results=5)
+            results = search.run(f"{query} {f'site:{domain}' if domain else ''}")
+            
+            if not results or results.strip() == "":
+                return "No search results found."
+                
+            return results
+
+        except Exception as e:
+            return f"Search error: {str(e)}"
+
+    def _analyze_video(self, url: str) -> str:
+        """Analyze video content using Gemini's video understanding capabilities."""
+        try:
+            # Validate URL
+            parsed_url = urlparse(url)
+            if not all([parsed_url.scheme, parsed_url.netloc]):
+                return "Please provide a valid video URL with http:// or https:// prefix."
+            
+            # Check if it's a YouTube URL
+            if 'youtube.com' not in url and 'youtu.be' not in url:
+                return "Only YouTube videos are supported at this time."
+
+            try:
+                # Configure yt-dlp with minimal extraction
+                ydl_opts = {
+                    'quiet': True,
+                    'no_warnings': True,
+                    'extract_flat': True,
+                    'no_playlist': True,
+                    'youtube_include_dash_manifest': False
+                }
+
+                with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+                    try:
+                        # Try basic info extraction
+                        info = ydl.extract_info(url, download=False, process=False)
+                        if not info:
+                            return "Could not extract video information."
+
+                        title = info.get('title', 'Unknown')
+                        description = info.get('description', '')
+                        
+                        # Create a detailed prompt with available metadata
+                        prompt = f"""Please analyze this YouTube video:
+Title: {title}
+URL: {url}
+Description: {description}
+
+Please provide a detailed analysis focusing on:
+1. Main topic and key points from the title and description
+2. Expected visual elements and scenes
+3. Overall message or purpose
+4. Target audience"""
+
+                        # Use the LLM with proper message format
+                        messages = [HumanMessage(content=prompt)]
+                        response = self.llm.invoke(messages)
+                        return response.content if hasattr(response, 'content') else str(response)
+
+                    except Exception as e:
+                        if 'Sign in to confirm' in str(e):
+                            return "This video requires age verification or sign-in. Please provide a different video URL."
+                        return f"Error accessing video: {str(e)}"
+
+            except Exception as e:
+                return f"Error extracting video info: {str(e)}"
+
+        except Exception as e:
+            return f"Error analyzing video: {str(e)}"
+
+    def _analyze_table(self, table_data: str) -> str:
+        """Analyze table or matrix data."""
+        try:
+            if not table_data or not isinstance(table_data, str):
+                return "Please provide valid table data for analysis."
+
+            prompt = f"""Please analyze this table:
+
+{table_data}
+
+Provide a detailed analysis including:
+1. Structure and format
+2. Key patterns or relationships
+3. Notable findings
+4. Any mathematical properties (if applicable)"""
+
+            messages = [HumanMessage(content=prompt)]
+            response = self.llm.invoke(messages)
+            return response.content if hasattr(response, 'content') else str(response)
+
+        except Exception as e:
+            return f"Error analyzing table: {str(e)}"
+
+    def _analyze_image(self, image_data: str) -> str:
+        """Analyze image content."""
+        try:
+            if not image_data or not isinstance(image_data, str):
+                return "Please provide a valid image for analysis."
+
+            prompt = f"""Please analyze this image:
+
+{image_data}
+
+Focus on:
+1. Visual elements and objects
+2. Colors and composition
+3. Text or numbers (if present)
+4. Overall context and meaning"""
+
+            messages = [HumanMessage(content=prompt)]
+            response = self.llm.invoke(messages)
+            return response.content if hasattr(response, 'content') else str(response)
+
+        except Exception as e:
+            return f"Error analyzing image: {str(e)}"
+
+    def _analyze_list(self, list_data: str) -> str:
+        """Analyze and categorize list items."""
+        if not list_data:
+            return "No list data provided."
+        try:
+            items = [x.strip() for x in list_data.split(',')]
+            if not items:
+                return "Please provide a comma-separated list of items."
+            # Add list analysis logic here
+            return "Please provide the list items for analysis."
+        except Exception as e:
+            return f"Error analyzing list: {str(e)}"
+
+    def _setup_llm(self):
+        """Set up the language model."""
+        # Set up model with video capabilities
+        generation_config = {
+            "temperature": 0.0,
+            "max_output_tokens": 2000,
+            "candidate_count": 1,
+        }
+        
+        safety_settings = {
+            HarmCategory.HARM_CATEGORY_HARASSMENT: HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+            HarmCategory.HARM_CATEGORY_HATE_SPEECH: HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+            HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+            HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
+        }
+        
+        return ChatGoogleGenerativeAI(
+            model="gemini-2.0-flash",
+            google_api_key=self.api_key,
+            temperature=0,
+            max_output_tokens=2000,
+            generation_config=generation_config,
+            safety_settings=safety_settings,
+            system_message=SystemMessage(content=(
+                "You are a precise AI assistant that helps users find information and analyze content. "
+                "You can directly understand and analyze YouTube videos, images, and other content. "
+                "When analyzing videos, focus on relevant details like dialogue, text, and key visual elements. "
+                "For lists, tables, and structured data, ensure proper formatting and organization. "
+                "If you need additional context, clearly explain what is needed."
+            ))
+        )
+        
+    def _setup_agent(self) -> AgentExecutor:
+        """Set up the agent with tools and system message."""
+        
+        # Define the system message template
+        PREFIX = """You are a helpful AI assistant that can use various tools to answer questions and analyze content. You have access to tools for web search, Wikipedia lookup, and multimedia analysis.
+
+TOOLS:
+------
+You have access to the following tools:"""
+
+        FORMAT_INSTRUCTIONS = """To use a tool, use the following format:
+
+Thought: Do I need to use a tool? Yes
+Action: the action to take, should be one of [{tool_names}]
+Action Input: the input to the action
+Observation: the result of the action
+
+When you have a response to say to the Human, or if you do not need to use a tool, you MUST use the format:
+
+Thought: Do I need to use a tool? No
+Final Answer: [your response here]
+
+Begin! Remember to ALWAYS include 'Thought:', 'Action:', 'Action Input:', and 'Final Answer:' in your responses."""
+
+        SUFFIX = """Previous conversation history:
+{chat_history}
+
+New question: {input}
+{agent_scratchpad}"""
+
+        # Create the base agent
+        agent = ConversationalAgent.from_llm_and_tools(
+            llm=self.llm,
+            tools=self.tools,
+            prefix=PREFIX,
+            format_instructions=FORMAT_INSTRUCTIONS,
+            suffix=SUFFIX,
+            input_variables=["input", "chat_history", "agent_scratchpad", "tool_names"],
+            handle_parsing_errors=True
+        )
+
+        # Initialize agent executor with custom output handling
+        return AgentExecutor.from_agent_and_tools(
+            agent=agent,
+            tools=self.tools,
+            memory=self.memory,
+            max_iterations=5,
+            verbose=True,
+            handle_parsing_errors=True,
+            return_only_outputs=True  # This ensures we only get the final output
+        )
+
+@tool
+def analyze_csv_file(file_path: str, query: str) -> str:
+    """
+    Analyze a CSV file using pandas and answer a question about it.
+    
+    Args:
+        file_path: Path to the CSV file
+        query: Question about the data
+        
+    Returns:
+        Analysis result or error message
+    """
+    try:
+        import pandas as pd
+        
+        # Read the CSV file
+        df = pd.read_csv(file_path)
+        
+        # Run various analyses based on the query
+        result = f"CSV file loaded with {len(df)} rows and {len(df.columns)} columns.\n"
+        result += f"Columns: {', '.join(df.columns)}\n\n"
+        
+        # Add summary statistics
+        result += "Summary statistics:\n"
+        result += str(df.describe())
+        
+        return result
+    except ImportError:
+        return "Error: pandas is not installed. Please install it with 'pip install pandas'."
+    except Exception as e:
+        return f"Error analyzing CSV file: {str(e)}"
+
+@tool
+def analyze_excel_file(file_path: str, query: str) -> str:
+    """
+    Analyze an Excel file using pandas and answer a question about it.
+    
+    Args:
+        file_path: Path to the Excel file
+        query: Question about the data
+        
+    Returns:
+        Analysis result or error message
+    """
+    try:
+        import pandas as pd
+        
+        # Read the Excel file
+        df = pd.read_excel(file_path)
+        
+        # Run various analyses based on the query
+        result = f"Excel file loaded with {len(df)} rows and {len(df.columns)} columns.\n"
+        result += f"Columns: {', '.join(df.columns)}\n\n"
+        
+        # Add summary statistics
+        result += "Summary statistics:\n"
+        result += str(df.describe())
+        
+        return result
+    except ImportError:
+        return "Error: pandas and openpyxl are not installed. Please install them with 'pip install pandas openpyxl'."
+    except Exception as e:
+        return f"Error analyzing Excel file: {str(e)}"