""" title: BDS1 Smart Search Xng (with BeautifulSoup scraper, don't run too many BeautifulSoup scrapers at same time, main version of this tool use Jina scraper) author: Lug, scraper part from Web search v0.2.1 by siwadon author_url: https://brainedai.netlify.app/en/tools/web-search/brained-search/ description: An advanced search tool that uses 2 LLMs to perform URL selection, and content extraction. Returns only the most relevant information. required_open_webui_version: 0.6.0 requirements: aiohttp, asyncio, openai, anthropic, requests, beautifulsoup4, lxml, python-dotenv version: 0.9.0 license: MIT """ import json import time import random from urllib.parse import quote_plus from typing import Optional, List, Dict, Any from pydantic import BaseModel, Field from bs4 import BeautifulSoup import unicodedata import re import asyncio import aiohttp class Tools: def __init__(self): self.valves = self.Valves() # Headers pour les requêtes web self.headers = { "User-Agent": "Mozilla/5.0 (Linux; Android 15; SM-S931U Build/AP3A.240905.015.A2; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/132.0.6834.163 Mobile Safari/537.36" } class Valves(BaseModel): # SearXNG Configuration searxng_url: str = Field( default="http://host.docker.internal:8080", description="SearXNG principal base URL", ) searxng_fallback_url: str = Field( default="https://search.buddyverse.net", description="SearXNG fallback 1 base URL", ) searxng_fallback2_url: str = Field( default="https://darmarit.org/searx", description="SearXNG fallback 2 base URL", ) request_timeout: int = Field(default=27, description="Request timeout") # LangSearch API Configuration langsearch_api_key: str = Field( default="", description="Your LangSearch API key (used as 3rd fallback)", ) # Configuration de recherche max_search_results: int = Field( default=20, description="Maximum number of search results to get initially" ) max_urls_to_scrape: int = Field( default=5, description="Maximum number of URLs to scrape after LLM selection for Smart function", ) # Scraping Configuration search_scrape_and_summarize_words_limit: int = Field( default=5000, description="Word limit for smart search function" ) # Scrapping retries max_retries: int = Field( default=1, description="Maximum number of scrapping retries for failed HTTP requests", ) retry_wait_seconds: int = Field( default=4, description="Seconds to wait before retrying failed HTTP requests", ) # Configuration LLM llm_base_url_1: str = Field( default="https://openrouter.ai/api/v1", description="Base URL for LLM 1", ) llm_model_1: str = Field( default="meta-llama/llama-3.1-8b-instruct", description="LLM 1 Model for URL selection", ) llm_api_key_1: str = Field( default="", description="API key for LLM 1", ) llm_1_fallback_models: str = Field( default="mistralai/mistral-nemo,deepseek/deepseek-r1-0528-qwen3-8b", description="Comma-separated list of fallback models for LLM 1.", ) llm_base_url_2: str = Field( default="https://openrouter.ai/api/v1", description="Base URL for LLM 2", ) llm_model_2: str = Field( default="qwen/qwen3-30b-a3b-instruct-2507", description="LLM 2 Model for content summarization", ) llm_api_key_2: str = Field( default="", description="API key for LLM 2", ) llm_2_fallback_models: str = Field( default="qwen/qwen3-30b-a3b-thinking-2507,meituan/longcat-flash-chat,qwen/qwen3-next-80b-a3b-instruct", description="Comma-separated list of fallback models for LLM 2.", ) # Paramètres séparés pour LLM 2 llm_2_temperature: float = Field( default=0.3, description="Temperature for LLM 2", ) llm_2_max_tokens: int = Field( default=5500, description="Max tokens for LLM 2", ) llm_2_enable_reasoning: bool = Field( default=False, description="Enable reasoning mode for LLM 2 (if supported by model)", ) # Provider routing pour LLM 2 (OpenRouter) llm_2_provider_order: str = Field( default="", description="List of providers in order of preference (e.g., 'chutes/bf16,nebius/fp8'). Empty to sort by price (default) or by throughput.", ) llm_2_enable_throughput_sort: bool = Field( default=False, description="Enable sorting by throughput and ignore suppliers specified below for LLM 2", ) llm_2_ignore_provider: str = Field( default="", description="Provider to ignore for LLM 2 when throughput sort is enabled (e.g., 'nebius/fast'). Leave empty to ignore nothing.", ) def _search_searxng( self, query: str, base_url: str, engines: str = None, categories: str = None, count: int = 20, freshness: str = "", lang: str = "en", ) -> str: """ Internal method for performing a SearXNG search on a specific URL """ try: searxng_base_url = base_url.rstrip("/") search_path = "search" params = [] params.append(f"q={quote_plus(query)}") params.append(f"engines={quote_plus(engines)}") if categories and categories.strip(): params.append(f"category_{categories.strip()}=on") params.append(f"language={quote_plus(lang)}") params.append(f"time_range={quote_plus(freshness)}") params.append("format=json") search_url = f"{searxng_base_url}/{search_path}?{'&'.join(params)}" headers = { "User-Agent": "Mozilla/5.0 (compatible; OpenWebUI-SecureSearch/1.0)", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "en;q=0.5", "Accept-Encoding": "gzip, deflate", "Connection": "keep-alive", } import requests response = requests.get( search_url, headers=headers, timeout=self.valves.request_timeout, verify=True, ) if response.status_code != 200: return None json_data = response.json() results = [] for i, result in enumerate(json_data.get("results", [])): if i >= count: break results.append( { "title": result.get("title", "No title"), "url": result.get("url", "#"), "content": result.get("content", ""), "engine": result.get("engine", "unknown"), } ) if not results: return None return json.dumps( { "query": query, "engines": engines or "all", "results": results, "count": len(results), "status": "success", "server": base_url, } ) except Exception as e: return None # ============ LANGSEARCH SEARCH (fallback 3) ============ def _search_langsearch( self, query: str, count: int = 10, freshness: str = "noLimit", ) -> str: """ Internal method for performing a LangSearch search. Used as the third fallback. """ if not self.valves.langsearch_api_key: return None try: import requests url = "https://api.langsearch.com/v1/web-search" headers = { "Authorization": f"Bearer {self.valves.langsearch_api_key}", "Content-Type": "application/json", } payload = { "query": query, "freshness": freshness, "summary": False, "count": count, } response = requests.post( url, headers=headers, json=payload, timeout=self.valves.request_timeout ) if response.status_code != 200: return None data = response.json() # Formater les résultats pour correspondre au format SearXNG results = [] for item in data.get("data", {}).get("webPages", {}).get("value", []): # Combiner snippet et summary pour le content results.append( { "title": item.get("name", "No title"), "url": item.get("url", "#"), "content": item.get("snippet", ""), "engine": "langsearch", } ) if not results: return None return json.dumps( { "query": query, "engines": "langsearch", "results": results, "count": len(results), "status": "success_fallback3", "server": "LangSearch API", } ) except Exception as e: return None async def _search_xng( self, query: str, engines: str = "brave,duckduckgo,google,startpage", categories: str = "general", count: int = 20, freshness: str = "", lang: str = "en", ) -> str: """ Primary web search using SearXNG with 3 fallbacks (SearXNG x2 and LangSearch) """ if not query or len(query.strip()) < 1: return json.dumps({"error": "Query must be non-empty"}) sanitized_query = query.strip()[:500] # Essayer le serveur principal result = self._search_searxng( sanitized_query, self.valves.searxng_url, engines, categories, count, freshness, lang, ) if result: return result # Fallback 1 result = self._search_searxng( sanitized_query, self.valves.searxng_fallback_url, engines, categories, count, freshness, lang, ) if result: parsed = json.loads(result) parsed["status"] = "success_fallback1" parsed["fallback_server"] = self.valves.searxng_fallback_url return json.dumps(parsed) # Fallback 2 result = self._search_searxng( sanitized_query, self.valves.searxng_fallback2_url, engines, categories, count, freshness, lang, ) if result: parsed = json.loads(result) parsed["status"] = "success_fallback2" parsed["fallback_server"] = self.valves.searxng_fallback2_url return json.dumps(parsed) # Fallback 3: LangSearch result = self._search_langsearch( sanitized_query, count=count, freshness="noLimit", # LangSearch utilise son propre système de freshness ) if result: return result # Tous les fallbacks ont échoué return json.dumps( { "error": "All search services failed (SearXNG primary, fallback1, fallback2, and LangSearch)", "query": sanitized_query, "primary_server": self.valves.searxng_url, "fallback_server1": self.valves.searxng_fallback_url, "fallback_server2": self.valves.searxng_fallback2_url, "fallback_server3": "LangSearch API", } ) # ============ LLM ============ async def _call_llm( self, messages: List[Dict], model_num: int = 1, temperature: Optional[float] = None, max_tokens: Optional[int] = None, ) -> str: """Call a specific LLM with the given messages""" base_url = ( self.valves.llm_base_url_1 if model_num == 1 else self.valves.llm_base_url_2 ) model = self.valves.llm_model_1 if model_num == 1 else self.valves.llm_model_2 api_key = ( self.valves.llm_api_key_1 if model_num == 1 else self.valves.llm_api_key_2 ) if model_num == 2: temp = ( temperature if temperature is not None else self.valves.llm_2_temperature ) tokens = ( max_tokens if max_tokens is not None else self.valves.llm_2_max_tokens ) else: temp = temperature if temperature is not None else 0.35 tokens = max_tokens if max_tokens is not None else 4500 headers = {"Content-Type": "application/json"} if api_key: headers["Authorization"] = f"Bearer {api_key}" payload = { "model": model, "messages": messages, "temperature": temp, "max_tokens": tokens, } fallback_models_str = ( self.valves.llm_1_fallback_models if model_num == 1 else self.valves.llm_2_fallback_models ) # Ajouter les modèles de fallback si configurés if fallback_models_str and fallback_models_str.strip(): # Construire la liste des modèles : [modèle principal] + [modèles fallback] fallback_list = [ m.strip() for m in fallback_models_str.split(",") if m.strip() ] models_array = [model] + fallback_list payload["models"] = models_array print(f"🔄 Fallback models configured for LLM {model_num}: {models_array}") if model_num == 2 and self.valves.llm_2_enable_reasoning: payload["reasoning"] = {"enabled": True} # Ajouter les paramètres de routing pour LLM 2 if model_num == 2: # Construire l'objet provider si nécessaire provider_config = {} # Ajouter l'ordre des providers si configuré if ( self.valves.llm_2_provider_order and self.valves.llm_2_provider_order.strip() ): providers_list = [ p.strip() for p in self.valves.llm_2_provider_order.split(",") ] provider_config["order"] = providers_list print(f"📋 Provider order configured: {providers_list}") # Ajouter le tri par throughput et ignorer un provider si activé if self.valves.llm_2_enable_throughput_sort: provider_config["sort"] = "throughput" # Ajouter le provider à ignorer si spécifié if ( self.valves.llm_2_ignore_provider and self.valves.llm_2_ignore_provider.strip() ): provider_config["ignore"] = ( self.valves.llm_2_ignore_provider.strip() ) print( f"⚙️ Throughput sort enabled, ignoring: {self.valves.llm_2_ignore_provider}" ) else: print(f"⚙️ Throughput sort enabled, no provider ignored") # Ajouter la configuration au payload si elle existe if provider_config: payload["provider"] = provider_config print(f"URL: {base_url.rstrip('/')}/chat/completions") print(f"Model: {model}") if model_num == 2 and "provider" in payload: print(f"Provider config: {payload['provider']}") try: async with aiohttp.ClientSession() as session: async with session.post( f"{base_url.rstrip('/')}/chat/completions", headers=headers, json=payload, timeout=80, ) as response: print(f"Status: {response.status}") response_text = await response.text() if response.status == 200: result = await response.json() return result["choices"][0]["message"]["content"] else: return f"Error: HTTP {response.status} - {response_text}" except Exception as e: return f"Error calling LLM: {str(e)}" # ============ SCRAPING ============ def _format_text(self, original_text): soup = BeautifulSoup(original_text, "html.parser") formatted_text = soup.get_text(separator=" ", strip=True) formatted_text = unicodedata.normalize("NFKC", formatted_text) formatted_text = re.sub(r"\\s+", " ", formatted_text) formatted_text = formatted_text.strip() return formatted_text def _truncate_to_n_words(self, text, token_limit): tokens = text.split() truncated_tokens = tokens[:token_limit] return " ".join(truncated_tokens) def _generate_excerpt(self, content, max_length=200): return content[:max_length] + "..." if len(content) > max_length else content async def _fetch_url_content( self, session, url: str, word_limit: Optional[int] = None ) -> Dict: """Auxiliary function for scraping a single URL""" try: async with session.get( url, headers=self.headers, timeout=aiohttp.ClientTimeout(total=120) ) as response: if response.status == 200: html_content = await response.text() soup = BeautifulSoup(html_content, "html.parser") page_title = soup.title.string if soup.title else "No title found" page_title = unicodedata.normalize("NFKC", page_title.strip()) content_site = self._format_text( soup.get_text(separator=" ", strip=True) ) actual_word_limit = ( word_limit if word_limit is not None else self.valves.search_scrape_and_summarize_words_limit ) truncated_content = self._truncate_to_n_words( content_site, actual_word_limit ) return { "title": page_title, "url": url, "content": truncated_content, "excerpt": self._generate_excerpt(content_site), } else: return { "url": url, "content": f"Failed to retrieve the page. HTTP {response.status}", } except Exception as e: return { "url": url, "content": f"Failed to retrieve the page. Error: {str(e)}", } async def _web_scrape( self, session, urls: List[str], word_limit: Optional[int] = None ) -> str: """ Scrape the content of multiple URLs in parallel using an existing aiohttp session. """ tasks = [self._fetch_url_content(session, url, word_limit) for url in urls] scraped_data = await asyncio.gather(*tasks) return json.dumps(scraped_data, ensure_ascii=False) # ============ FONCTION PRINCIPALE ============ async def xng_smart_search( self, query: str, query_2: Optional[str] = None, ) -> str: """ Performs an intelligent web search using multiple steps: 1. Performs initial web searches (uses query for web search). 2. Uses LLM 1 to select the most relevant URLs from the search results (uses query_2 if provided). 3. Scrapes the content of the selected URLs. 4. Uses LLM 2 to extract and summarize relevant information from the scraped content (uses query_2 if provided). 5. Returns the final, concise answer. """ if not query or len(query.strip()) < 1: return json.dumps({"error": "Query must be non-empty"}) # Détermine quelle requête utiliser pour les LLMs llm_query = query_2 if query_2 and query_2.strip() else query # Step 1: Perform initial web searches print(f"Étape 1: Step 1: Perform initial web searches for '{query}'...") initial_search_results_str = await self._search_xng( query, count=self.valves.max_search_results ) initial_search_results = json.loads(initial_search_results_str) if initial_search_results.get("error"): return json.dumps( {"error": f"Search failed: {initial_search_results['error']}"} ) # Format results for LLM 1 search_results_for_llm = [] all_search_urls = [] # Stocker toutes les URLs pour fallback for i, res in enumerate(initial_search_results.get("results", [])): search_results_for_llm.append( f"{i+1}. Title: {res['title']}\n URL: {res['url']}\n Content Snippet: {res['content']}\n" ) all_search_urls.append(res["url"]) # Ajouter l'URL if not search_results_for_llm: return json.dumps({"message": "No relevant search results found."}) # Étape 2: Utilisation de LLM 1 pour sélectionner les URLs pertinentes print("Step 2: Use LLM 1 to select relevant URLs...") selected_urls = [] llm1_success = False llm1_prompt = [ { "role": "system", "content": f"You are an intelligent web search assistant. Given a user query and a list of search results, select the {self.valves.max_urls_to_scrape} most relevant URLs that are likely to contain the answer to the query. **Provide ONLY the URLs, one per line, nothing else. No explanations, no numbering, just the raw URLs**. If fewer than {self.valves.max_urls_to_scrape} relevant URLs are found, return all relevant ones.", }, { "role": "user", "content": f"Query: {query}\n\nSearch Results:\n" + "\n".join(search_results_for_llm) + f"\n\nList ONLY the {self.valves.max_urls_to_scrape} most relevant URLs (one per line):", }, ] try: selected_urls_str = await self._call_llm(llm1_prompt, model_num=1) if "Error" not in selected_urls_str: # Meilleur parsing des URLs import re # Extraire toutes les URLs du texte retourné url_pattern = r'https?://[^\s<>"{}|\^`\[\]]+' potential_urls = re.findall(url_pattern, selected_urls_str) # Valider que les URLs sont dans les résultats de recherche for url in potential_urls: url_clean = url.strip().rstrip(",").rstrip("/") normalized_search_urls = [u.rstrip("/") for u in all_search_urls] if url_clean in all_search_urls: selected_urls.append(url_clean) if len(selected_urls) >= self.valves.max_urls_to_scrape: break if selected_urls: llm1_success = True print(f"✅ LLM1 has selected {len(selected_urls)} Valid URLs") except Exception as e: print(f"⚠️ Error during call to LLM1: {str(e)}") # Fallback - Utiliser les premières URLs de recherche si LLM1 échoue if not llm1_success or not selected_urls: print( f"⚠️ Fallback enabled: use of the first {self.valves.max_urls_to_scrape} search URLs " ) selected_urls = all_search_urls[: self.valves.max_urls_to_scrape] if not selected_urls: return json.dumps({"message": "No URLs available for scraping."}) print(f"URLs to scrape: {selected_urls}") # Étape 3: Scraping des URLs sélectionnées print("Step 3: Scrape the selected URLs...") async with aiohttp.ClientSession() as session: scraped_content_str = await self._web_scrape(session, selected_urls) scraped_content = json.loads(scraped_content_str) if not scraped_content: return json.dumps( {"message": "No content could be scraped from the selected URLs."} ) # Format scraped content for LLM 2 content_for_llm = [] failed_scrapes = 0 # Compter les échecs for content_item in scraped_content: content_text = content_item.get("content", "") if ( "Failed to retrieve" not in content_text and len(content_text.strip()) > 50 ): # Vérifier longueur minimale content_for_llm.append( f"Source: {content_item.get('title', 'Unknown')} ({content_item['url']})\\n" f"Content: {content_text}\n" ) else: failed_scrapes += 1 print( f"📊 Scraping statistics: {len(content_for_llm)} successes, {failed_scrapes} failures" ) if not content_for_llm: return json.dumps( { "error": "No valid content was extracted from the scraped pages.", "details": f"Attempted to scrape {len(selected_urls)} URLs, all failed.", "urls_attempted": selected_urls, "suggestion": "Try different keywords or check if the URLs are accessible.", } ) # Étape 4: Utilisation de LLM 2 pour extraire et résumer les informations pertinentes print("Step 4: Use LLM 2 to extract and summarize relevant information...") llm2_prompt = [ { "role": "system", "content": "You are a Precision Information Extractor and Fidelity Curator. Your expertise lies in identifying and isolating relevant information with absolute fidelity to the source material. You function as a high-precision filter. Your mission is to copy word for word all information that could enable another LLM to respond to the user's query. If the source is not in English, you must provide an accurate English translation that preserves the structure and meaning of the original without paraphrasing. Aesthetic presentation is secondary to information integrity. Synthesis is a last resort. Only condense or summarize information if a single, critical passage is excessively long and would otherwise exceed the token limit. In all other cases, present the information as-is.", }, { "role": "user", "content": f"Query: {llm_query}\n\nScraped Content:\n" + "\n".join(content_for_llm) + "\n\nProvide a comprehensive answer based on the above content:", }, ] final_answer = await self._call_llm(llm2_prompt, model_num=2) if "Error" in final_answer: return json.dumps( {"error": f"LLM 2 failed to process content: {final_answer}"} ) # Step 5: Return the final structured response return json.dumps( { "query": query, "llm_query_used": llm_query, # Indication de quelle requête a été utilisée pour les LLMs "answer": final_answer, "sources_used": len(content_for_llm), "urls_scraped": selected_urls, "llm1_used": llm1_success, # Indiquer si LLM1 a réussi "fallback_used": not llm1_success, # Indiquer si fallback utilisé "scraping_stats": { "successful": len(content_for_llm), "failed": failed_scrapes, }, "search_results_count": len(initial_search_results.get("results", [])), "status": "success", }, ensure_ascii=False, )