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 dotenv import load_dotenv 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 langchain_openai import AzureChatOpenAI from google.generativeai.types import HarmCategory, HarmBlockThreshold from PIL import Image import google.generativeai as genai from pydantic import Field from smolagents import WikipediaSearchTool load_dotenv() openai_api_version="2024-02-01" azure_endpoint=os.getenv('azure_endpoint') openai_api_key=os.getenv('openai_api_key') openai_model_name ='gpt-4o-mini' 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 = AzureChatOpenAI( api_version=openai_api_version, azure_endpoint=azure_endpoint, azure_deployment=openai_model_name, api_key=openai_api_key, # type: ignore streaming=True, temperature=0.0, ) # 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)}"