Chris
Final 7.8.3
4d128ff
raw
history blame
32.2 kB
#!/usr/bin/env python3
"""
Web Search Tool for GAIA Agent System
Handles web searches using DuckDuckGo (primary), Tavily API (secondary), and Wikipedia (fallback)
"""
import re
import logging
import time
import os
from typing import Dict, List, Optional, Any
from urllib.parse import urlparse, urljoin
import requests
from bs4 import BeautifulSoup
from tools import BaseTool
logger = logging.getLogger(__name__)
class WebSearchResult:
"""Container for web search results"""
def __init__(self, title: str, url: str, snippet: str, content: str = "", source: str = ""):
self.title = title
self.url = url
self.snippet = snippet
self.content = content
self.source = source
def to_dict(self) -> Dict[str, str]:
return {
"title": self.title,
"url": self.url,
"snippet": self.snippet,
"content": self.content[:1500] + "..." if len(self.content) > 1500 else self.content,
"source": self.source
}
class WebSearchTool(BaseTool):
"""
Web search tool using DuckDuckGo (primary), Tavily API (secondary), and Wikipedia (fallback)
Provides multiple search engine options for reliability
"""
def __init__(self):
super().__init__("web_search")
# Configure requests session for web scraping
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
})
self.session.timeout = 10
# Initialize search engines
self.tavily_api_key = os.getenv("TAVILY_API_KEY")
self.use_tavily = self.tavily_api_key is not None
# Try to import DuckDuckGo
try:
from duckduckgo_search import DDGS
self.ddgs = DDGS()
self.use_duckduckgo = True
logger.info("✅ DuckDuckGo search initialized")
except ImportError:
logger.warning("⚠️ DuckDuckGo search not available - install duckduckgo-search package")
self.use_duckduckgo = False
# Try to import Wikipedia
try:
import wikipedia
self.wikipedia = wikipedia
self.use_wikipedia = True
logger.info("✅ Wikipedia search initialized")
except ImportError:
logger.warning("⚠️ Wikipedia search not available - install wikipedia package")
self.use_wikipedia = False
if self.use_tavily:
logger.info("✅ Tavily API key found - using as secondary search")
# Search engine priority: DuckDuckGo -> Tavily -> Wikipedia
search_engines = []
if self.use_duckduckgo:
search_engines.append("DuckDuckGo")
if self.use_tavily:
search_engines.append("Tavily")
if self.use_wikipedia:
search_engines.append("Wikipedia")
logger.info(f"🔍 Available search engines: {', '.join(search_engines)}")
def _execute_impl(self, input_data: Any, **kwargs) -> Dict[str, Any]:
"""
Execute web search operations based on input type
Args:
input_data: Can be:
- str: Search query or URL to extract content from
- dict: {"query": str, "action": str, "limit": int, "extract_content": bool}
"""
if isinstance(input_data, str):
# Handle both search queries and URLs
if self._is_url(input_data):
return self._extract_content_from_url(input_data)
else:
return self._search_web(input_data)
elif isinstance(input_data, dict):
query = input_data.get("query", "")
action = input_data.get("action", "search")
limit = input_data.get("limit", 5)
extract_content = input_data.get("extract_content", False)
if action == "search":
return self._search_web(query, limit, extract_content)
elif action == "extract":
return self._extract_content_from_url(query)
else:
raise ValueError(f"Unknown action: {action}")
else:
raise ValueError(f"Unsupported input type: {type(input_data)}")
def _is_url(self, text: str) -> bool:
"""Check if text is a URL"""
return bool(re.match(r'https?://', text))
def _extract_search_terms(self, question: str, max_length: int = 200) -> str:
"""
Extract focused search terms from a question
Intelligently builds search queries prioritizing key information
"""
import re
# Special handling for backwards text questions
if re.search(r'\.rewsna\b|etirw\b|dnatsrednu\b', question.lower()):
# This is backwards text - reverse it
words = question.split()
reversed_words = [word[::-1] for word in words]
reversed_question = ' '.join(reversed_words)
return self._extract_search_terms(reversed_question, max_length)
# Remove common question starters but keep meaningful content
clean_question = question
question_starters = [
r'^(what|who|when|where|why|how|which|whose)\s+',
r'\bis\s+the\s+',
r'\bare\s+the\s+',
r'\bwas\s+the\s+',
r'\bwere\s+the\s+',
r'\bdid\s+the\s+',
r'\bdo\s+the\s+',
r'\bcan\s+you\s+',
r'\bcould\s+you\s+',
r'\bplease\s+',
r'\btell\s+me\s+',
r'\bfind\s+',
r'\blist\s+',
]
for starter in question_starters:
clean_question = re.sub(starter, '', clean_question, flags=re.IGNORECASE)
# Extract key components in priority order
search_parts = []
# 1. Extract quoted phrases (highest priority)
quoted_phrases = re.findall(r'"([^"]+)"', question)
for phrase in quoted_phrases[:2]: # Max 2 quoted phrases
search_parts.append(phrase)
# 2. Extract proper nouns and names (high priority)
# Look for capitalized words that are likely names/places
proper_nouns = re.findall(r'\b[A-Z][a-z]+(?:\s+[A-Z][a-z]+)*\b', question)
# Filter out common words that might be capitalized
common_caps = {'The', 'This', 'That', 'These', 'Those', 'In', 'On', 'At', 'To', 'For', 'Of', 'With', 'By'}
meaningful_nouns = [noun for noun in proper_nouns if noun not in common_caps]
search_parts.extend(meaningful_nouns[:3]) # Max 3 proper nouns
# 3. Extract years and dates (medium priority)
years = re.findall(r'\b(19|20)\d{2}\b', question)
search_parts.extend(years[:2]) # Max 2 years
# 4. Extract specific important keywords based on question context
important_keywords = []
# Look for specific domains/topics
domain_keywords = {
'music': ['album', 'albums', 'song', 'songs', 'artist', 'band', 'music', 'released', 'published'],
'sports': ['player', 'team', 'game', 'match', 'season', 'championship', 'league'],
'science': ['research', 'study', 'paper', 'journal', 'scientist', 'experiment'],
'technology': ['software', 'program', 'code', 'website', 'application', 'system'],
'geography': ['country', 'city', 'place', 'location', 'region', 'area'],
'history': ['year', 'century', 'period', 'era', 'historical', 'ancient'],
'wikipedia': ['wikipedia', 'article', 'featured', 'promoted', 'nomination', 'nominated'],
'competition': ['competition', 'contest', 'award', 'winner', 'recipient', 'prize']
}
question_lower = question.lower()
for domain, keywords in domain_keywords.items():
for keyword in keywords:
if keyword in question_lower:
important_keywords.append(keyword)
# Add unique important keywords
unique_keywords = []
for keyword in important_keywords:
if keyword not in [part.lower() for part in search_parts]:
unique_keywords.append(keyword)
search_parts.extend(unique_keywords[:3]) # Max 3 domain keywords
# 5. Extract key content words (lower priority)
if len(search_parts) < 4: # Only if we need more terms
# Remove stop words and get meaningful content
stop_words = {
'the', 'a', 'an', 'and', 'or', 'but', 'in', 'on', 'at', 'to', 'for',
'of', 'with', 'by', 'from', 'up', 'about', 'into', 'through', 'during',
'before', 'after', 'above', 'below', 'between', 'among', 'this', 'that',
'these', 'those', 'i', 'me', 'my', 'we', 'our', 'you', 'your', 'he',
'him', 'his', 'she', 'her', 'it', 'its', 'they', 'them', 'their',
'is', 'are', 'was', 'were', 'be', 'been', 'being', 'have', 'has',
'had', 'do', 'does', 'did', 'will', 'would', 'could', 'should',
'may', 'might', 'must', 'can'
}
# Extract words, clean them, and filter
words = re.findall(r'\b\w+\b', clean_question.lower())
content_words = [w for w in words if w not in stop_words and len(w) > 2]
# Add important content words not already included
for word in content_words[:3]:
if word not in [part.lower() for part in search_parts]:
search_parts.append(word)
# Build the final search query
if search_parts:
search_query = ' '.join(search_parts)
else:
# Fallback: use first few meaningful words
words = question.split()[:6]
search_query = ' '.join(words)
# Clean up and ensure reasonable length
search_query = ' '.join(search_query.split()) # Remove extra whitespace
# Truncate at word boundary if too long
if len(search_query) > max_length:
search_query = search_query[:max_length].rsplit(' ', 1)[0]
# Ensure we have something to search for
if not search_query.strip():
search_query = question.split()[:3] # Use first 3 words as absolute fallback
search_query = ' '.join(search_query)
# Log for debugging
logger.info(f"📝 Extracted search terms: '{search_query}' from question: '{question[:100]}...'")
return search_query.strip()
def _search_web(self, query: str, limit: int = 5, extract_content: bool = False) -> Dict[str, Any]:
"""
Search the web using available search engines in priority order with improved search terms
"""
# Extract clean search terms from the query
search_query = self._extract_search_terms(query, max_length=200)
# Try DuckDuckGo first (most comprehensive for general web search)
if self.use_duckduckgo:
try:
ddg_result = self._search_with_duckduckgo(search_query, limit, extract_content)
if ddg_result.get('success') and ddg_result.get('count', 0) > 0:
return {
'success': True,
'found': True,
'results': [r.to_dict() if hasattr(r, 'to_dict') else r for r in ddg_result['results']],
'query': query,
'source': 'DuckDuckGo',
'total_found': ddg_result['count']
}
except Exception as e:
logger.warning(f"DuckDuckGo search failed, trying Tavily: {e}")
# Try Tavily if DuckDuckGo fails and API key is available
if self.use_tavily:
try:
tavily_result = self._search_with_tavily(search_query, limit, extract_content)
if tavily_result.get('success') and tavily_result.get('count', 0) > 0:
return {
'success': True,
'found': True,
'results': [r.to_dict() if hasattr(r, 'to_dict') else r for r in tavily_result['results']],
'query': query,
'source': 'Tavily',
'total_found': tavily_result['count']
}
except Exception as e:
logger.warning(f"Tavily search failed, trying Wikipedia: {e}")
# Fallback to Wikipedia search
if self.use_wikipedia:
try:
wiki_result = self._search_with_wikipedia(search_query, limit)
if wiki_result.get('success') and wiki_result.get('count', 0) > 0:
return {
'success': True,
'found': True,
'results': [r.to_dict() if hasattr(r, 'to_dict') else r for r in wiki_result['results']],
'query': query,
'source': 'Wikipedia',
'total_found': wiki_result['count']
}
except Exception as e:
logger.warning(f"Wikipedia search failed: {e}")
# No search engines available or all failed
logger.warning("All search engines failed, returning empty results")
return {
"query": query,
"found": False,
"success": False,
"message": "❌ All search engines failed or returned no results.",
"results": [],
"source": "none",
"total_found": 0
}
def _search_with_duckduckgo(self, query: str, limit: int = 5, extract_content: bool = False) -> Dict[str, Any]:
"""
Search using DuckDuckGo - primary search engine with improved error handling and rate limiting
"""
try:
logger.info(f"🦆 DuckDuckGo search for: {query}")
# Add small delay to avoid rate limiting
time.sleep(0.5)
# Use DuckDuckGo text search with retry logic
max_retries = 2
for attempt in range(max_retries):
try:
ddg_results = list(self.ddgs.text(query, max_results=min(limit, 10)))
break
except Exception as retry_error:
if attempt < max_retries - 1:
logger.warning(f"DuckDuckGo attempt {attempt + 1} failed, retrying in {2 ** attempt}s: {retry_error}")
time.sleep(2 ** attempt) # Exponential backoff
continue
else:
raise retry_error
if not ddg_results:
logger.warning("DuckDuckGo returned no results")
return self._search_with_fallback(query, limit)
# Process DuckDuckGo results
results = []
for result in ddg_results:
web_result = WebSearchResult(
title=result.get('title', 'No title'),
url=result.get('href', ''),
snippet=result.get('body', 'No description'),
source='DuckDuckGo'
)
results.append(web_result)
logger.info(f"✅ DuckDuckGo found {len(results)} results")
return {
'success': True,
'results': results,
'source': 'DuckDuckGo',
'query': query,
'count': len(results)
}
except Exception as e:
logger.warning(f"DuckDuckGo search failed: {str(e)}")
# Check if it's a rate limiting error and add longer delay
if "ratelimit" in str(e).lower() or "429" in str(e) or "202" in str(e):
logger.warning("Rate limiting detected, adding delay before fallback")
time.sleep(2.0)
return self._search_with_fallback(query, limit)
def _search_with_fallback(self, query: str, limit: int = 5) -> Dict[str, Any]:
"""Enhanced fallback search when DuckDuckGo fails"""
logger.info(f"🔄 Using fallback search engines for: {query}")
# Try Tavily API first if available
if hasattr(self, 'tavily') and self.tavily:
try:
logger.info("📡 Trying Tavily API search")
tavily_result = self.tavily.search(query, max_results=limit)
if tavily_result and 'results' in tavily_result:
results = []
for result in tavily_result['results'][:limit]:
web_result = WebSearchResult(
title=result.get('title', 'No title'),
url=result.get('url', ''),
snippet=result.get('content', 'No description'),
source='Tavily'
)
results.append(web_result)
if results:
logger.info(f"✅ Tavily found {len(results)} results")
return {
'success': True,
'results': results,
'source': 'Tavily',
'query': query,
'count': len(results)
}
except Exception as e:
logger.warning(f"Tavily search failed: {str(e)}")
# Fall back to Wikipedia search
logger.info("📚 Wikipedia search for: " + query)
try:
wiki_results = self._search_with_wikipedia(query, limit)
if wiki_results and wiki_results.get('success'):
logger.info(f"✅ Wikipedia found {wiki_results.get('count', 0)} results")
return wiki_results
except Exception as e:
logger.warning(f"Wikipedia fallback failed: {str(e)}")
# Final fallback - return empty but successful result to allow processing to continue
logger.warning("All search engines failed, returning empty results")
return {
'success': True,
'results': [],
'source': 'none',
'query': query,
'count': 0,
'note': 'All search engines failed'
}
def _search_with_tavily(self, query: str, limit: int = 5, extract_content: bool = False) -> Dict[str, Any]:
"""
Search using Tavily Search API - secondary search engine
"""
try:
logger.info(f"🔍 Tavily search for: {query}")
# Prepare Tavily API request
headers = {
"Content-Type": "application/json"
}
payload = {
"api_key": self.tavily_api_key,
"query": query,
"search_depth": "basic",
"include_answer": False,
"include_images": False,
"include_raw_content": extract_content,
"max_results": min(limit, 10)
}
# Make API request
response = self.session.post(
"https://api.tavily.com/search",
json=payload,
headers=headers,
timeout=15
)
response.raise_for_status()
tavily_data = response.json()
# Process Tavily results
results = []
tavily_results = tavily_data.get('results', [])
for result in tavily_results:
web_result = WebSearchResult(
title=result.get('title', 'No title'),
url=result.get('url', ''),
snippet=result.get('content', 'No description'),
content=result.get('raw_content', '') if extract_content else ''
)
results.append(web_result)
if results:
logger.info(f"✅ Tavily found {len(results)} results")
return {
'success': True,
'results': results,
'source': 'Tavily',
'query': query,
'count': len(results)
}
else:
logger.warning("Tavily returned no results")
# Fall back to Wikipedia
if self.use_wikipedia:
return self._search_with_wikipedia(query, limit)
except requests.exceptions.RequestException as e:
logger.error(f"Tavily API request failed: {e}")
except Exception as e:
logger.error(f"Tavily search error: {e}")
# Fall back to Wikipedia if Tavily fails
if self.use_wikipedia:
return self._search_with_wikipedia(query, limit)
return {
'success': False,
'results': [],
'source': 'Tavily',
'query': query,
'count': 0,
'note': 'Tavily search failed and no fallback available'
}
def _search_with_wikipedia(self, query: str, limit: int = 5) -> Dict[str, Any]:
"""
Search using Wikipedia - fallback search engine for factual information
"""
try:
logger.info(f"📚 Wikipedia search for: {query}")
self.wikipedia.set_lang("en")
# Clean up query for Wikipedia search and ensure it's not too long
search_terms = self._extract_search_terms(query, max_length=100) # Wikipedia has stricter limits
# Search Wikipedia pages
wiki_results = self.wikipedia.search(search_terms, results=min(limit * 2, 10))
if not wiki_results:
return {
'success': False,
'results': [],
'source': 'Wikipedia',
'query': query,
'count': 0,
'note': 'No Wikipedia articles found for this query'
}
results = []
processed = 0
for page_title in wiki_results:
if processed >= limit:
break
try:
page = self.wikipedia.page(page_title)
summary = page.summary[:300] + "..." if len(page.summary) > 300 else page.summary
web_result = WebSearchResult(
title=f"{page_title} (Wikipedia)",
url=page.url,
snippet=summary,
content=page.summary[:1000] + "..." if len(page.summary) > 1000 else page.summary
)
results.append(web_result)
processed += 1
except self.wikipedia.exceptions.DisambiguationError as e:
# Try the first suggestion from disambiguation
try:
if e.options:
page = self.wikipedia.page(e.options[0])
summary = page.summary[:300] + "..." if len(page.summary) > 300 else page.summary
web_result = WebSearchResult(
title=f"{e.options[0]} (Wikipedia)",
url=page.url,
snippet=summary,
content=page.summary[:1000] + "..." if len(page.summary) > 1000 else page.summary
)
results.append(web_result)
processed += 1
except:
continue
except self.wikipedia.exceptions.PageError:
# Page doesn't exist, skip
continue
except Exception as e:
# Other Wikipedia errors, skip this page
logger.warning(f"Wikipedia page error for '{page_title}': {e}")
continue
if results:
logger.info(f"✅ Wikipedia found {len(results)} results")
return {
'success': True,
'results': results,
'source': 'Wikipedia',
'query': query,
'count': len(results)
}
else:
return {
'success': False,
'results': [],
'source': 'Wikipedia',
'query': query,
'count': 0,
'note': 'No accessible Wikipedia articles found for this query'
}
except Exception as e:
logger.error(f"Wikipedia search failed: {e}")
return {
'success': False,
'results': [],
'source': 'Wikipedia',
'query': query,
'count': 0,
'note': f"Wikipedia search failed: {str(e)}"
}
def _extract_content_from_url(self, url: str) -> Dict[str, Any]:
"""
Extract readable content from a web page
"""
try:
logger.info(f"Extracting content from: {url}")
# Get page content
response = self.session.get(url)
response.raise_for_status()
# Parse with BeautifulSoup
soup = BeautifulSoup(response.content, 'html.parser')
# Remove script and style elements
for script in soup(["script", "style", "nav", "header", "footer", "aside"]):
script.decompose()
# Extract title
title = soup.find('title')
title_text = title.get_text().strip() if title else "No title"
# Extract main content
content = self._extract_main_content(soup)
# Extract metadata
meta_description = ""
meta_desc = soup.find('meta', attrs={'name': 'description'})
if meta_desc:
meta_description = meta_desc.get('content', '')
# Extract links
links = []
for link in soup.find_all('a', href=True)[:10]: # First 10 links
link_url = urljoin(url, link['href'])
link_text = link.get_text().strip()
if link_text and len(link_text) > 5: # Filter out short/empty links
links.append({"text": link_text, "url": link_url})
return {
"url": url,
"found": True,
"title": title_text,
"content": content,
"meta_description": meta_description,
"links": links,
"content_length": len(content),
"message": "Successfully extracted content from URL"
}
except requests.exceptions.RequestException as e:
return {
"url": url,
"found": False,
"message": f"Failed to fetch URL: {str(e)}",
"error_type": "network_error"
}
except Exception as e:
return {
"url": url,
"found": False,
"message": f"Failed to extract content: {str(e)}",
"error_type": "parsing_error"
}
def _extract_main_content(self, soup: BeautifulSoup) -> str:
"""
Extract main content from HTML using various strategies
"""
content_parts = []
# Strategy 1: Look for article/main tags
main_content = soup.find(['article', 'main'])
if main_content:
content_parts.append(main_content.get_text())
# Strategy 2: Look for content in common div classes
content_selectors = [
'div.content',
'div.article-content',
'div.post-content',
'div.entry-content',
'div.main-content',
'div#content',
'div.text'
]
for selector in content_selectors:
elements = soup.select(selector)
for element in elements:
content_parts.append(element.get_text())
# Strategy 3: Look for paragraphs in body
if not content_parts:
paragraphs = soup.find_all('p')
for p in paragraphs[:20]: # First 20 paragraphs
text = p.get_text().strip()
if len(text) > 50: # Filter out short paragraphs
content_parts.append(text)
# Clean and combine content
combined_content = '\n\n'.join(content_parts)
# Clean up whitespace and formatting
combined_content = re.sub(r'\n\s*\n', '\n\n', combined_content) # Multiple newlines
combined_content = re.sub(r' +', ' ', combined_content) # Multiple spaces
return combined_content.strip()[:5000] # Limit to 5000 characters
def test_web_search_tool():
"""Test the web search tool with various queries"""
tool = WebSearchTool()
# Test cases
test_cases = [
"Python programming tutorial",
"Mercedes Sosa studio albums 2000 2009",
"artificial intelligence recent developments",
"climate change latest research",
"https://en.wikipedia.org/wiki/Machine_learning"
]
print("🧪 Testing Web Search Tool...")
for i, test_case in enumerate(test_cases, 1):
print(f"\n--- Test {i}: {test_case} ---")
try:
result = tool.execute(test_case)
if result.success:
print(f"✅ Success: {result.result.get('message', 'No message')}")
search_engine = result.result.get('source', 'unknown')
print(f" Search engine: {search_engine}")
if result.result.get('found'):
if 'results' in result.result:
print(f" Found {len(result.result['results'])} results")
# Show first result details
if result.result['results']:
first_result = result.result['results'][0]
print(f" First result: {first_result.get('title', 'No title')}")
print(f" URL: {first_result.get('url', 'No URL')}")
elif 'content' in result.result:
print(f" Extracted {len(result.result['content'])} characters")
print(f" Title: {result.result.get('title', 'No title')}")
else:
print(f" Not found: {result.result.get('message', 'Unknown error')}")
else:
print(f"❌ Error: {result.error}")
print(f" Execution time: {result.execution_time:.2f}s")
except Exception as e:
print(f"❌ Exception: {str(e)}")
if __name__ == "__main__":
# Test when run directly
test_web_search_tool()