Delete gemini_agent.py

gemini_agent.py

@@ -1,660 +0,0 @@
-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)}"