search also in general

tools.py

@@ -143,60 +143,63 @@ text_processor_tool = Tool(
     description="Processes text for various operations like summarization, number extraction, date extraction. Specify operation as second parameter."
 )
 
-def enhanced_web_retrieval_tool_func(query: str) -> str:
-    """Enhanced web search with vector retrieval for deep content analysis."""
+def enhanced_web_retrieval_tool_func(query: str, backend: str = "bing") -> str:
+    """Enhanced web search with cascading fallback: Wikipedia first, then general web search."""
     try:
         print(f"🔍 Enhanced web retrieval for: {query}")
         
-        # Step 1: Get search results with URLs
-        search_results = get_search_urls(query)
-        if not search_results:
-            return "No search results found."
+        # Step 1: Try Wikipedia search first
+        print("📚 Searching Wikipedia...")
+        wikipedia_results = get_wikipedia_search_urls(query, backend)
         
-        # Step 2: Fetch and process webpage content
-        documents = []
-        for result in search_results[:4]:  # Top 4 results as requested
-            url = result.get('url', '')
-            title = result.get('title', 'No title')
-            
-            print(f"📄 Fetching content from: {title}")
-            content = fetch_webpage_content(url)
-            if content:
-                doc = Document(
-                    page_content=content,
-                    metadata={"source": url, "title": title}
-                )
-                documents.append(doc)
+        if has_sufficient_results(wikipedia_results):
+            print(f"✅ Found {len(wikipedia_results)} Wikipedia results")
+            documents = fetch_and_process_results(wikipedia_results, "Wikipedia")
+            if documents:
+                return search_documents_with_vector_store(documents, query, "Wikipedia")
+        
+        # Step 2: Fallback to general web search
+        print("🌐 Wikipedia results insufficient, searching general web...")
+        web_results = get_general_web_search_urls(query, backend)
         
-        if not documents:
-            return "Could not fetch content from any search results."
+        if web_results:
+            print(f"✅ Found {len(web_results)} general web results")
+            documents = fetch_and_process_results(web_results, "General Web")
+            if documents:
+                return search_documents_with_vector_store(documents, query, "General Web")
         
-        # Step 3: Create vector store and search
-        return search_documents_with_vector_store(documents, query)
+        return "No sufficient results found in Wikipedia or general web search."
         
     except Exception as e:
         return f"Enhanced web retrieval failed: {str(e)}"
 
-def get_search_urls(query: str) -> list:
-    """Get search results from English Wikipedia only using DDGS."""
+def get_wikipedia_search_urls(query: str, backend: str = "auto") -> list:
+    """Get search results from English Wikipedia using DDGS."""
     try:
         with DDGS() as ddgs:
             # Create Wikipedia-specific search queries
-            queries = [
+            wikipedia_queries = [
                 f"{query} site:en.wikipedia.org"
             ]
             
             search_results = []
             seen_urls = set()
             
-            for wiki_query in queries:
+            for wiki_query in wikipedia_queries:
                 try:
-                    results = list(ddgs.text(wiki_query, max_results=10, region="us-en", backend="bing", safesearch="on"))
+                    results = list(ddgs.text(
+                        wiki_query, 
+                        max_results=8, 
+                        region="us-en", 
+                        backend=backend, 
+                        safesearch="moderate"
+                    ))
                     
                     for result in results:
                         url = result.get('href', '')
                         
-                        if url not in seen_urls:
+                        # Only include Wikipedia URLs and avoid duplicates
+                        if 'en.wikipedia.org' in url and url not in seen_urls:
                             search_results.append({
                                 'url': url,
                                 'title': result.get('title', 'No title'),
@@ -204,15 +207,16 @@ def get_search_urls(query: str) -> list:
                             })
                             seen_urls.add(url)
                             
-                            # Limit to 4 unique Wikipedia pages
-                            if len(search_results) >= 4:
+                            # Limit to 6 unique Wikipedia pages
+                            if len(search_results) >= 6:
                                 break
                     
-                    if len(search_results) >= 4:
+                    if len(search_results) >= 6:
                         break
                         
-                except Exception:
-                    continue  # Try next query
+                except Exception as e:
+                    print(f"Wikipedia search attempt failed: {e}")
+                    continue
             
             return search_results
             
@@ -220,6 +224,106 @@ def get_search_urls(query: str) -> list:
         print(f"Wikipedia search URL retrieval failed: {e}")
         return []
 
+def get_general_web_search_urls(query: str, backend: str = "auto") -> list:
+    """Get search results from general web using DDGS."""
+    try:
+        with DDGS() as ddgs:
+            search_results = []
+            seen_urls = set()
+            
+            try:
+                # General web search without site restriction
+                results = list(ddgs.text(
+                    query, 
+                    max_results=8, 
+                    region="us-en", 
+                    backend=backend, 
+                    safesearch="moderate"
+                ))
+                
+                for result in results:
+                    url = result.get('href', '')
+                    
+                    # Avoid duplicates and filter out low-quality sources
+                    if url not in seen_urls and is_quality_source(url):
+                        search_results.append({
+                            'url': url,
+                            'title': result.get('title', 'No title'),
+                            'snippet': result.get('body', 'No content')
+                        })
+                        seen_urls.add(url)
+                        
+                        # Limit to 6 unique web pages
+                        if len(search_results) >= 6:
+                            break
+                            
+            except Exception as e:
+                print(f"General web search attempt failed: {e}")
+            
+            return search_results
+            
+    except Exception as e:
+        print(f"General web search URL retrieval failed: {e}")
+        return []
+
+def is_quality_source(url: str) -> bool:
+    """Filter out low-quality or problematic sources."""
+    low_quality_domains = [
+        'pinterest.com', 'instagram.com', 'facebook.com', 'twitter.com',
+        'tiktok.com', 'youtube.com', 'reddit.com'
+    ]
+    
+    for domain in low_quality_domains:
+        if domain in url.lower():
+            return False
+    
+    return True
+
+def has_sufficient_results(results: list) -> bool:
+    """Check if search results are sufficient to proceed."""
+    if not results:
+        return False
+    
+    # Check for minimum number of results
+    if len(results) < 2:
+        return False
+    
+    # Check if results have meaningful content
+    meaningful_results = 0
+    for result in results:
+        snippet = result.get('snippet', '')
+        title = result.get('title', '')
+        
+        # Consider result meaningful if it has substantial content
+        if len(snippet) > 50 or len(title) > 10:
+            meaningful_results += 1
+    
+    return meaningful_results >= 2
+
+def fetch_and_process_results(results: list, source_type: str) -> list:
+    """Fetch and process webpage content from search results."""
+    documents = []
+    
+    for result in results[:4]:  # Process top 4 results
+        url = result.get('url', '')
+        title = result.get('title', 'No title')
+        
+        print(f"📄 Fetching content from: {title}")
+        content = fetch_webpage_content(url)
+        
+        if content and len(content.strip()) > 100:  # Ensure meaningful content
+            doc = Document(
+                page_content=content,
+                metadata={
+                    "source": url, 
+                    "title": title,
+                    "source_type": source_type
+                }
+            )
+            documents.append(doc)
+    
+    return documents
+
 def fetch_webpage_content(url: str) -> str:
     """Fetch and extract clean text content from a webpage."""
     try:
@@ -251,7 +355,7 @@ def fetch_webpage_content(url: str) -> str:
         print(f"Failed to fetch content from {url}: {e}")
         return ""
 
-def search_documents_with_vector_store(documents: list, query: str) -> str:
+def search_documents_with_vector_store(documents: list, query: str, source_type: str = "Web") -> str:
     """Create vector store and search for relevant information."""
     try:
         # Split documents into chunks
@@ -273,14 +377,17 @@ def search_documents_with_vector_store(documents: list, query: str) -> str:
         # Search for relevant chunks with the original query
         relevant_docs = vectorstore.similarity_search(query, k=5)
         
-        # Format results
+        # Format results with source type indication
         results = []
+        results.append(f"🔍 Search Results from {source_type}:\n")
+        
         for i, doc in enumerate(relevant_docs, 1):
             source = doc.metadata.get('source', 'Unknown source')
             title = doc.metadata.get('title', 'No title')
-            content = doc.page_content[:5000]  # First 500 chars
+            source_type_meta = doc.metadata.get('source_type', source_type)
+            content = doc.page_content[:2000]  # Increased content length
             
-            results.append(f"Result {i} from {title}:\n{content}\nSource: {source}\n")
+            results.append(f"Result {i} ({source_type_meta}) - {title}:\n{content}\nSource: {source}\n")
         
         return "\n---\n".join(results)
         
@@ -290,7 +397,7 @@ def search_documents_with_vector_store(documents: list, query: str) -> str:
 web_search_tool = Tool(
     name="enhanced_web_retrieval",
     func=enhanced_web_retrieval_tool_func,
-    description="Enhanced Wikipedia-only search with vector retrieval. Fetches full content from English Wikipedia pages and uses semantic search to find relevant information. Use this for factual questions that need detailed Wikipedia content analysis."
+    description="Enhanced cascading web search with vector retrieval. First searches Wikipedia for reliable factual information, then falls back to general web search if insufficient results are found. Supports multiple search backends (auto, html, lite, bing) and uses semantic search to find relevant information. Ideal for comprehensive research on any topic."
 )
 
 # List of all tools for easy import