MrArray22/Final_Assignment_Template
0
1import os
2import tempfile
3import time
4import re
5import json
6from typing import List, Optional, Dict, Any
7from urllib.parse import urlparse
8import requests
9import yt_dlp
10from bs4 import BeautifulSoup
11from difflib import SequenceMatcher
12
13from langchain_core.messages import HumanMessage, SystemMessage
14from langchain_google_genai import ChatGoogleGenerativeAI
15from langchain_community.utilities import DuckDuckGoSearchAPIWrapper, WikipediaAPIWrapper
16from langchain.agents import Tool, AgentExecutor, ConversationalAgent, initialize_agent, AgentType
17from langchain.memory import ConversationBufferMemory
18from langchain.prompts import MessagesPlaceholder
19from langchain.tools import BaseTool, Tool, tool
20from google.generativeai.types import HarmCategory, HarmBlockThreshold
21from PIL import Image
22import google.generativeai as genai
23from pydantic import Field
24
25from smolagents import WikipediaSearchTool
26
27class SmolagentToolWrapper(BaseTool):
28 """Wrapper for smolagents tools to make them compatible with LangChain."""
29
30 wrapped_tool: object = Field(description="The wrapped smolagents tool")
31
32 def __init__(self, tool):
33 """Initialize the wrapper with a smolagents tool."""
34 super().__init__(
35 name=tool.name,
36 description=tool.description,
37 return_direct=False,
38 wrapped_tool=tool
39 )
40
41 def _run(self, query: str) -> str:
42 """Use the wrapped tool to execute the query."""
43 try:
44 # For WikipediaSearchTool
45 if hasattr(self.wrapped_tool, 'search'):
46 return self.wrapped_tool.search(query)
47 # For DuckDuckGoSearchTool and others
48 return self.wrapped_tool(query)
49 except Exception as e:
50 return f"Error using tool: {str(e)}"
51
52 def _arun(self, query: str) -> str:
53 """Async version - just calls sync version since smolagents tools don't support async."""
54 return self._run(query)
55
56class WebSearchTool:
57 def __init__(self):
58 self.last_request_time = 0
59 self.min_request_interval = 2.0 # Minimum time between requests in seconds
60 self.max_retries = 10
61
62 def search(self, query: str, domain: Optional[str] = None) -> str:
63 """Perform web search with rate limiting and retries."""
64 for attempt in range(self.max_retries):
65 # Implement rate limiting
66 current_time = time.time()
67 time_since_last = current_time - self.last_request_time
68 if time_since_last < self.min_request_interval:
69 time.sleep(self.min_request_interval - time_since_last)
70
71 try:
72 # Make the search request
73 results = self._do_search(query, domain)
74 self.last_request_time = time.time()
75 return results
76 except Exception as e:
77 if "202 Ratelimit" in str(e):
78 if attempt < self.max_retries - 1:
79 # Exponential backoff
80 wait_time = (2 ** attempt) * self.min_request_interval
81 time.sleep(wait_time)
82 continue
83 return f"Search failed after {self.max_retries} attempts: {str(e)}"
84
85 return "Search failed due to rate limiting"
86
87 def _do_search(self, query: str, domain: Optional[str] = None) -> str:
88 """Perform the actual search request."""
89 try:
90 # Construct search URL
91 base_url = "https://html.duckduckgo.com/html"
92 params = {"q": query}
93 if domain:
94 params["q"] += f" site:{domain}"
95
96 # Make request with increased timeout
97 response = requests.get(base_url, params=params, timeout=10)
98 response.raise_for_status()
99
100 if response.status_code == 202:
101 raise Exception("202 Ratelimit")
102
103 # Extract search results
104 results = []
105 soup = BeautifulSoup(response.text, 'html.parser')
106 for result in soup.find_all('div', {'class': 'result'}):
107 title = result.find('a', {'class': 'result__a'})
108 snippet = result.find('a', {'class': 'result__snippet'})
109 if title and snippet:
110 results.append({
111 'title': title.get_text(),
112 'snippet': snippet.get_text(),
113 'url': title.get('href')
114 })
115
116 # Format results
117 formatted_results = []
118 for r in results[:10]: # Limit to top 5 results
119 formatted_results.append(f"[{r['title']}]({r['url']})\n{r['snippet']}\n")
120
121 return "## Search Results\n\n" + "\n".join(formatted_results)
122
123 except requests.RequestException as e:
124 raise Exception(f"Search request failed: {str(e)}")
125
126def save_and_read_file(content: str, filename: Optional[str] = None) -> str:
127 """
128 Save content to a temporary file and return the path.
129 Useful for processing files from the GAIA API.
130
131 Args:
132 content: The content to save to the file
133 filename: Optional filename, will generate a random name if not provided
134
135 Returns:
136 Path to the saved file
137 """
138 temp_dir = tempfile.gettempdir()
139 if filename is None:
140 temp_file = tempfile.NamedTemporaryFile(delete=False)
141 filepath = temp_file.name
142 else:
143 filepath = os.path.join(temp_dir, filename)
144
145 # Write content to the file
146 with open(filepath, 'w') as f:
147 f.write(content)
148
149 return f"File saved to {filepath}. You can read this file to process its contents."
150
151
152def download_file_from_url(url: str, filename: Optional[str] = None) -> str:
153 """
154 Download a file from a URL and save it to a temporary location.
155
156 Args:
157 url: The URL to download from
158 filename: Optional filename, will generate one based on URL if not provided
159
160 Returns:
161 Path to the downloaded file
162 """
163 try:
164 # Parse URL to get filename if not provided
165 if not filename:
166 path = urlparse(url).path
167 filename = os.path.basename(path)
168 if not filename:
169 # Generate a random name if we couldn't extract one
170 import uuid
171 filename = f"downloaded_{uuid.uuid4().hex[:8]}"
172
173 # Create temporary file
174 temp_dir = tempfile.gettempdir()
175 filepath = os.path.join(temp_dir, filename)
176
177 # Download the file
178 response = requests.get(url, stream=True)
179 response.raise_for_status()
180
181 # Save the file
182 with open(filepath, 'wb') as f:
183 for chunk in response.iter_content(chunk_size=8192):
184 f.write(chunk)
185
186 return f"File downloaded to {filepath}. You can now process this file."
187 except Exception as e:
188 return f"Error downloading file: {str(e)}"
189
190
191def extract_text_from_image(image_path: str) -> str:
192 """
193 Extract text from an image using pytesseract (if available).
194
195 Args:
196 image_path: Path to the image file
197
198 Returns:
199 Extracted text or error message
200 """
201 try:
202 # Try to import pytesseract
203 import pytesseract
204 from PIL import Image
205
206 # Open the image
207 image = Image.open(image_path)
208
209 # Extract text
210 text = pytesseract.image_to_string(image)
211
212 return f"Extracted text from image:\n\n{text}"
213 except ImportError:
214 return "Error: pytesseract is not installed. Please install it with 'pip install pytesseract' and ensure Tesseract OCR is installed on your system."
215 except Exception as e:
216 return f"Error extracting text from image: {str(e)}"
217
218
219def analyze_csv_file(file_path: str, query: str) -> str:
220 """
221 Analyze a CSV file using pandas and answer a question about it.
222
223 Args:
224 file_path: Path to the CSV file
225 query: Question about the data
226
227 Returns:
228 Analysis result or error message
229 """
230 try:
231 import pandas as pd
232
233 # Read the CSV file
234 df = pd.read_csv(file_path)
235
236 # Run various analyses based on the query
237 result = f"CSV file loaded with {len(df)} rows and {len(df.columns)} columns.\n"
238 result += f"Columns: {', '.join(df.columns)}\n\n"
239
240 # Add summary statistics
241 result += "Summary statistics:\n"
242 result += str(df.describe())
243
244 return result
245 except ImportError:
246 return "Error: pandas is not installed. Please install it with 'pip install pandas'."
247 except Exception as e:
248 return f"Error analyzing CSV file: {str(e)}"
249
250@tool
251def analyze_excel_file(file_path: str, query: str) -> str:
252 """
253 Analyze an Excel file using pandas and answer a question about it.
254
255 Args:
256 file_path: Path to the Excel file
257 query: Question about the data
258
259 Returns:
260 Analysis result or error message
261 """
262 try:
263 import pandas as pd
264
265 # Read the Excel file
266 df = pd.read_excel(file_path)
267
268 # Run various analyses based on the query
269 result = f"Excel file loaded with {len(df)} rows and {len(df.columns)} columns.\n"
270 result += f"Columns: {', '.join(df.columns)}\n\n"
271
272 # Add summary statistics
273 result += "Summary statistics:\n"
274 result += str(df.describe())
275
276 return result
277 except ImportError:
278 return "Error: pandas and openpyxl are not installed. Please install them with 'pip install pandas openpyxl'."
279 except Exception as e:
280 return f"Error analyzing Excel file: {str(e)}"
281
282class GeminiAgent:
283 def __init__(self, api_key: str, model_name: str = "gemini-2.0-flash"):
284 # Suppress warnings
285 import warnings
286 warnings.filterwarnings("ignore", category=UserWarning)
287 warnings.filterwarnings("ignore", category=DeprecationWarning)
288 warnings.filterwarnings("ignore", message=".*will be deprecated.*")
289 warnings.filterwarnings("ignore", "LangChain.*")
290
291 self.api_key = api_key
292 self.model_name = model_name
293
294 # Configure Gemini
295 genai.configure(api_key=api_key)
296
297 # Initialize the LLM
298 self.llm = self._setup_llm()
299
300 # Setup tools
301 self.tools = [
302 SmolagentToolWrapper(WikipediaSearchTool()),
303 Tool(
304 name="analyze_video",
305 func=self._analyze_video,
306 description="Analyze YouTube video content directly"
307 ),
308 Tool(
309 name="analyze_image",
310 func=self._analyze_image,
311 description="Analyze image content"
312 ),
313 Tool(
314 name="analyze_table",
315 func=self._analyze_table,
316 description="Analyze table or matrix data"
317 ),
318 Tool(
319 name="analyze_list",
320 func=self._analyze_list,
321 description="Analyze and categorize list items"
322 ),
323 Tool(
324 name="web_search",
325 func=self._web_search,
326 description="Search the web for information"
327 )
328 ]
329
330 # Setup memory
331 self.memory = ConversationBufferMemory(
332 memory_key="chat_history",
333 return_messages=True
334 )
335
336 # Initialize agent
337 self.agent = self._setup_agent()
338
339
340 def run(self, query: str) -> str:
341 """Run the agent on a query with incremental retries."""
342 max_retries = 3
343 base_sleep = 1 # Start with 1 second sleep
344
345 for attempt in range(max_retries):
346 try:
347
348 # If no match found in answer bank, use the agent
349 response = self.agent.run(query)
350 return response
351
352 except Exception as e:
353 sleep_time = base_sleep * (attempt + 1) # Incremental sleep: 1s, 2s, 3s
354 if attempt < max_retries - 1:
355 print(f"Attempt {attempt + 1} failed. Retrying in {sleep_time} seconds...")
356 time.sleep(sleep_time)
357 continue
358 return f"Error processing query after {max_retries} attempts: {str(e)}"
359
360 print("Agent processed all queries!")
361
362 def _clean_response(self, response: str) -> str:
363 """Clean up the response from the agent."""
364 # Remove any tool invocation artifacts
365 cleaned = re.sub(r'> Entering new AgentExecutor chain...|> Finished chain.', '', response)
366 cleaned = re.sub(r'Thought:.*?Action:.*?Action Input:.*?Observation:.*?\n', '', cleaned, flags=re.DOTALL)
367 return cleaned.strip()
368
369 def run_interactive(self):
370 print("AI Assistant Ready! (Type 'exit' to quit)")
371
372 while True:
373 query = input("You: ").strip()
374 if query.lower() == 'exit':
375 print("Goodbye!")
376 break
377
378 print("Assistant:", self.run(query))
379
380 def _web_search(self, query: str, domain: Optional[str] = None) -> str:
381 """Perform web search with rate limiting and retries."""
382 try:
383 # Use DuckDuckGo API wrapper for more reliable results
384 search = DuckDuckGoSearchAPIWrapper(max_results=5)
385 results = search.run(f"{query} {f'site:{domain}' if domain else ''}")
386
387 if not results or results.strip() == "":
388 return "No search results found."
389
390 return results
391
392 except Exception as e:
393 return f"Search error: {str(e)}"
394
395 def _analyze_video(self, url: str) -> str:
396 """Analyze video content using Gemini's video understanding capabilities."""
397 try:
398 # Validate URL
399 parsed_url = urlparse(url)
400 if not all([parsed_url.scheme, parsed_url.netloc]):
401 return "Please provide a valid video URL with http:// or https:// prefix."
402
403 # Check if it's a YouTube URL
404 if 'youtube.com' not in url and 'youtu.be' not in url:
405 return "Only YouTube videos are supported at this time."
406
407 try:
408 # Configure yt-dlp with minimal extraction
409 ydl_opts = {
410 'quiet': True,
411 'no_warnings': True,
412 'extract_flat': True,
413 'no_playlist': True,
414 'youtube_include_dash_manifest': False
415 }
416
417 with yt_dlp.YoutubeDL(ydl_opts) as ydl:
418 try:
419 # Try basic info extraction
420 info = ydl.extract_info(url, download=False, process=False)
421 if not info:
422 return "Could not extract video information."
423
424 title = info.get('title', 'Unknown')
425 description = info.get('description', '')
426
427 # Create a detailed prompt with available metadata
428 prompt = f"""Please analyze this YouTube video:
429Title: {title}
430URL: {url}
431Description: {description}
432
433Please provide a detailed analysis focusing on:
4341. Main topic and key points from the title and description
4352. Expected visual elements and scenes
4363. Overall message or purpose
4374. Target audience"""
438
439 # Use the LLM with proper message format
440 messages = [HumanMessage(content=prompt)]
441 response = self.llm.invoke(messages)
442 return response.content if hasattr(response, 'content') else str(response)
443
444 except Exception as e:
445 if 'Sign in to confirm' in str(e):
446 return "This video requires age verification or sign-in. Please provide a different video URL."
447 return f"Error accessing video: {str(e)}"
448
449 except Exception as e:
450 return f"Error extracting video info: {str(e)}"
451
452 except Exception as e:
453 return f"Error analyzing video: {str(e)}"
454
455 def _analyze_table(self, table_data: str) -> str:
456 """Analyze table or matrix data."""
457 try:
458 if not table_data or not isinstance(table_data, str):
459 return "Please provide valid table data for analysis."
460
461 prompt = f"""Please analyze this table:
462
463{table_data}
464
465Provide a detailed analysis including:
4661. Structure and format
4672. Key patterns or relationships
4683. Notable findings
4694. Any mathematical properties (if applicable)"""
470
471 messages = [HumanMessage(content=prompt)]
472 response = self.llm.invoke(messages)
473 return response.content if hasattr(response, 'content') else str(response)
474
475 except Exception as e:
476 return f"Error analyzing table: {str(e)}"
477
478 def _analyze_image(self, image_data: str) -> str:
479 """Analyze image content."""
480 try:
481 if not image_data or not isinstance(image_data, str):
482 return "Please provide a valid image for analysis."
483
484 prompt = f"""Please analyze this image:
485
486{image_data}
487
488Focus on:
4891. Visual elements and objects
4902. Colors and composition
4913. Text or numbers (if present)
4924. Overall context and meaning"""
493
494 messages = [HumanMessage(content=prompt)]
495 response = self.llm.invoke(messages)
496 return response.content if hasattr(response, 'content') else str(response)
497
498 except Exception as e:
499 return f"Error analyzing image: {str(e)}"
500
501 def _analyze_list(self, list_data: str) -> str:
502 """Analyze and categorize list items."""
503 if not list_data:
504 return "No list data provided."
505 try:
506 items = [x.strip() for x in list_data.split(',')]
507 if not items:
508 return "Please provide a comma-separated list of items."
509 # Add list analysis logic here
510 return "Please provide the list items for analysis."
511 except Exception as e:
512 return f"Error analyzing list: {str(e)}"
513
514 def _setup_llm(self):
515 """Set up the language model."""
516 # Set up model with video capabilities
517 generation_config = {
518 "temperature": 0.0,
519 "max_output_tokens": 2000,
520 "candidate_count": 1,
521 }
522
523 safety_settings = {
524 HarmCategory.HARM_CATEGORY_HARASSMENT: HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
525 HarmCategory.HARM_CATEGORY_HATE_SPEECH: HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
526 HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
527 HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
528 }
529
530 return ChatGoogleGenerativeAI(
531 model="gemini-2.0-flash",
532 google_api_key=self.api_key,
533 temperature=0,
534 max_output_tokens=2000,
535 generation_config=generation_config,
536 safety_settings=safety_settings,
537 system_message=SystemMessage(content=(
538 "You are a precise AI assistant that helps users find information and analyze content. "
539 "You can directly understand and analyze YouTube videos, images, and other content. "
540 "When analyzing videos, focus on relevant details like dialogue, text, and key visual elements. "
541 "For lists, tables, and structured data, ensure proper formatting and organization. "
542 "If you need additional context, clearly explain what is needed."
543 ))
544 )
545
546 def _setup_agent(self) -> AgentExecutor:
547 """Set up the agent with tools and system message."""
548
549 # Define the system message template
550 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.
551
552TOOLS:
553------
554You have access to the following tools:"""
555
556 FORMAT_INSTRUCTIONS = """To use a tool, use the following format:
557
558Thought: Do I need to use a tool? Yes
559Action: the action to take, should be one of [{tool_names}]
560Action Input: the input to the action
561Observation: the result of the action
562
563When you have a response to say to the Human, or if you do not need to use a tool, you MUST use the format:
564
565Thought: Do I need to use a tool? No
566Final Answer: [your response here]
567
568Begin! Remember to ALWAYS include 'Thought:', 'Action:', 'Action Input:', and 'Final Answer:' in your responses."""
569
570 SUFFIX = """Previous conversation history:
571{chat_history}
572
573New question: {input}
574{agent_scratchpad}"""
575
576 # Create the base agent
577 agent = ConversationalAgent.from_llm_and_tools(
578 llm=self.llm,
579 tools=self.tools,
580 prefix=PREFIX,
581 format_instructions=FORMAT_INSTRUCTIONS,
582 suffix=SUFFIX,
583 input_variables=["input", "chat_history", "agent_scratchpad", "tool_names"],
584 handle_parsing_errors=True
585 )
586
587 # Initialize agent executor with custom output handling
588 return AgentExecutor.from_agent_and_tools(
589 agent=agent,
590 tools=self.tools,
591 memory=self.memory,
592 max_iterations=5,
593 verbose=True,
594 handle_parsing_errors=True,
595 return_only_outputs=True # This ensures we only get the final output
596 )
597
598@tool
599def analyze_csv_file(file_path: str, query: str) -> str:
600 """
601 Analyze a CSV file using pandas and answer a question about it.
602
603 Args:
604 file_path: Path to the CSV file
605 query: Question about the data
606
607 Returns:
608 Analysis result or error message
609 """
610 try:
611 import pandas as pd
612
613 # Read the CSV file
614 df = pd.read_csv(file_path)
615
616 # Run various analyses based on the query
617 result = f"CSV file loaded with {len(df)} rows and {len(df.columns)} columns.\n"
618 result += f"Columns: {', '.join(df.columns)}\n\n"
619
620 # Add summary statistics
621 result += "Summary statistics:\n"
622 result += str(df.describe())
623
624 return result
625 except ImportError:
626 return "Error: pandas is not installed. Please install it with 'pip install pandas'."
627 except Exception as e:
628 return f"Error analyzing CSV file: {str(e)}"
629
630@tool
631def analyze_excel_file(file_path: str, query: str) -> str:
632 """
633 Analyze an Excel file using pandas and answer a question about it.
634
635 Args:
636 file_path: Path to the Excel file
637 query: Question about the data
638
639 Returns:
640 Analysis result or error message
641 """
642 try:
643 import pandas as pd
644
645 # Read the Excel file
646 df = pd.read_excel(file_path)
647
648 # Run various analyses based on the query
649 result = f"Excel file loaded with {len(df)} rows and {len(df.columns)} columns.\n"
650 result += f"Columns: {', '.join(df.columns)}\n\n"
651
652 # Add summary statistics
653 result += "Summary statistics:\n"
654 result += str(df.describe())
655
656 return result
657 except ImportError:
658 return "Error: pandas and openpyxl are not installed. Please install them with 'pip install pandas openpyxl'."
659 except Exception as e:
660 return f"Error analyzing Excel file: {str(e)}"
661