""" title: BDS1 Smart Search Xng main (with jina scraper) author: Lug, Scrapper part from Enhanced Web Scrape 0.1.2 by ekatiyar and christian-taillon author_url: https://brainedai.netlify.app description: An advanced search tool that uses multiple LLMs to perform intelligent web searches, URL selection, and content extraction. Returns only the most relevant information. required_open_webui_version: 0.6.0 requirements: aiohttp, asyncio, requests, beautifulsoup4 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, Callable from pydantic import BaseModel, Field from datetime import datetime import unicodedata import re import asyncio import aiohttp import requests import urllib.parse def extract_title(text: str) -> Optional[str]: match = re.search(r"Title: (.*)\n", text) return match.group(1).strip() if match else None def clean_urls(text: str) -> str: return re.sub(r"\((http[^)]+)\)", "", text) class EventEmitter: def __init__(self, event_emitter: Callable[[dict], Any] = None): self.event_emitter = event_emitter async def progress_update(self, description: str): await self.emit(description) async def error_update(self, description: str): await self.emit(description, "error", True) async def success_update(self, description: str): await self.emit(description, "success", True) async def emit( self, description: str = "Unknown State", status: str = "in_progress", done: bool = False, ): if self.event_emitter: await self.event_emitter( { "type": "status", "data": { "status": status, "description": description, "done": done, }, } ) class Tools: def __init__(self): self.valves = self.Valves() # Headers pour les requêtes web (pour SearXNG) 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://search.parchlinux.com", 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=6, description="Maximum number of URLs to scrape after LLM selection", ) # ========== Jina API Configuration ========== JINA_API_KEY: str = Field( default="", description="Optional Jina API key. Allows higher rate limit and advanced features. Leave empty for basic free tier.", ) DISABLE_JINA_CACHING: bool = Field( default=False, description="Bypass Jina Cache when scraping (requires API key)", ) CLEAN_JINA_CONTENT: bool = Field( default=True, description="Remove links and image urls from scraped content (reduces tokens)", ) # ========== Scraping Configuration ========== search_scrape_and_summarize_words_limit: int = Field( default=5500, description="Word limit for smart search function" ) scrape_and_summarize_words_limit: int = Field( default=5500, description="Word limit for summarize function" ) scrape_only_words_limit: int = Field( default=3000, description="Word limit for scrape only function" ) # ========== Configuration LLM ========== llm_model_1: str = Field( default="mistralai/mistral-nemo", description="LLM 1 Model for URL selection", ) llm_base_url_1: str = Field( default="https://openrouter.ai/api/v1", description="Base URL for LLM 1", ) llm_api_key_1: str = Field( default="", description="API key for LLM 1", ) llm_1_fallback_models: str = Field( default="meta-llama/llama-3.1-8b-instruct,deepseek/deepseek-r1-0528-qwen3-8b", description="Comma-separated list of fallback models for LLM 1.", ) llm_model_2: str = Field( default="qwen/qwen3-next-80b-a3b-instruct", description="Model for content summarization", ) llm_base_url_2: str = Field( default="https://openrouter.ai/api/v1", description="Base URL for LLM 2", ) llm_api_key_2: str = Field( default="", description="API key for LLM 2", ) llm_2_fallback_models: str = Field( default="meituan/longcat-flash-chat,qwen/qwen3-30b-a3b-thinking-2507", description="Comma-separated list of fallback models for LLM 2.", ) # ========== Paramètres LLM 2 ========== llm_2_temperature: float = Field( default=0.35, description="Temperature for LLM 2", ) llm_2_max_tokens: int = Field( default=6000, 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=True, description="Enable sorting by throughput and ignore suppliers specified below for LLM 2", ) llm_2_ignore_provider: str = Field( default="chutes/bf16,novita/bf16", description="Provider to ignore for LLM 2 when throughput sort is enabled (e.g., 'nebius/fast'). Leave empty to ignore nothing.", ) def get_current_time(self) -> str: """ Get the current date and time. :return: The current date and time. """ now = datetime.now() current_time = now.strftime("%I:%M:%S %p") current_date = now.strftime("%A, %B %d, %Y") return f"Current Date and Time = {current_date}, {current_time}" # ============ JINA SCRAPER ============ def _normalize_url(self, u: str) -> str: return urllib.parse.urldefrag((u or "").strip())[0] def _concurrency(self) -> int: """Determines concurrency based on the presence of a Jina API key""" key_present = bool(self.valves.JINA_API_KEY) return 5 if key_present else 2 async def _http_get( self, url: str, headers: Dict[str, str], timeout=(5, 30), attempts: int = 3 ) -> requests.Response: last_exc: Optional[Exception] = None for attempt in range(1, attempts + 1): try: resp: requests.Response = await asyncio.to_thread( requests.get, url, headers=headers, timeout=timeout ) if resp.status_code in (429, 503): retry_after = resp.headers.get("Retry-After") if retry_after: try: delay = float(retry_after) except ValueError: delay = 0.5 * (2 ** (attempt - 1)) + random.random() * 0.2 else: delay = 0.5 * (2 ** (attempt - 1)) + random.random() * 0.2 await asyncio.sleep(delay) last_exc = requests.HTTPError(f"HTTP {resp.status_code} for {url}") continue resp.raise_for_status() return resp except requests.RequestException as e: last_exc = e if attempt == attempts: break delay = 0.5 * (2 ** (attempt - 1)) + random.random() * 0.2 await asyncio.sleep(delay) raise last_exc def _build_jina_headers(self) -> Dict[str, str]: """ Build headers for Jina. Without an API key, Jina does NOT accept custom headers such as X-With-Generated-Alt or X-No-Cache. """ headers = {} # Si on a une clé API, on peut utiliser les headers avancés if self.valves.JINA_API_KEY: headers["Authorization"] = f"Bearer {self.valves.JINA_API_KEY}" headers["X-No-Cache"] = ( "true" if self.valves.DISABLE_JINA_CACHING else "false" ) headers["X-With-Generated-Alt"] = "true" # Sinon, on garde seulement un User-Agent basique (optionnel) return headers async def _scrape_one( self, url: str, __event_emitter__: Optional[Callable[[dict], Any]] = None ) -> Dict[str, Optional[str]]: """Scrape a single URL with Jina""" emitter = EventEmitter(__event_emitter__) url = self._normalize_url(url) await emitter.progress_update(f"Scraping {url}") jina_url = f"https://r.jina.ai/{url}" headers = self._build_jina_headers() try: response = await self._http_get(jina_url, headers=headers) content = response.text if self.valves.CLEAN_JINA_CONTENT: await emitter.progress_update("Cleaning content...") content = clean_urls(content) title = extract_title(content) await emitter.success_update( f"Successfully scraped {title if title else url}" ) return { "url": url, "title": title or None, "content": content, "error": None, } except Exception as e: error_message = f"Error scraping {url}: {str(e)}" await emitter.error_update(error_message) return { "url": url, "title": None, "content": None, "error": error_message, } async def _web_scrape( self, urls: List[str], __event_emitter__: Optional[Callable[[dict], Any]] = None, ) -> List[Dict[str, Optional[str]]]: """ Scrape multiple URLs in parallel with Jina. Returns a list of dictionaries. """ emitter = EventEmitter(__event_emitter__) limit = self._concurrency() sem = asyncio.Semaphore(limit) results: List[Optional[Dict[str, Optional[str]]]] = [None] * len(urls) async def run_one(i: int, raw_url: str): async with sem: res = await self._scrape_one(raw_url, __event_emitter__) results[i] = res await asyncio.gather(*(run_one(i, u) for i, u in enumerate(urls))) await emitter.success_update(f"Finished scraping {len(urls)} pages") return [r for r in results if r is not None] # ============ SEARXNG SEARCH ============ 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", } 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: 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 (NOUVEAU) 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.3 tokens = max_tokens if max_tokens is not None else 5000 headers = {"Content-Type": "application/json"} if api_key: headers["Authorization"] = f"Bearer {api_key}" payload = { "model": model, "messages": messages, "temperature": temp, "max_tokens": tokens, } # Ajouter les modèles de fallback si configurés fallback_models_str = ( self.valves.llm_1_fallback_models if model_num == 1 else self.valves.llm_2_fallback_models ) 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}") # Ajouter le mode reasoning si activé pour LLM 2 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)}" # ============ UTILITY FUNCTIONS ============ 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 # ============ FONCTION PRINCIPALE ============ async def main_xng_smart_search( self, query: str, query_2: Optional[str] = None, __event_emitter__: Optional[Callable[[dict], Any]] = None, ) -> str: """ Performs an intelligent web search using multiple steps: 1. Performs initial web searches using SearXNG (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 using Jina. 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éterminer quelle requête utiliser pour les LLMs llm_query = query_2 if query_2 and query_2.strip() else query # Étape 1: Exécution des recherches web initiales pour print(f"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 = [] 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"]) 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: {llm_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 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 normalized_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 a sélectionné {len(selected_urls)} URLs valides") except Exception as e: print(f"⚠️ Erreur lors de l'appel à 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 activé: utilisation des {self.valves.max_urls_to_scrape} premières URLs de recherche" ) 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 à scraper: {selected_urls}") # Étape 3: Scraping des URLs sélectionnées print("Step 3: Scrape the selected URLs with Jina...") scraped_content = await self._web_scrape(selected_urls, __event_emitter__) 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 for content_item in scraped_content: # Vérifier d'abord si l'erreur est None if content_item.get("error") is None and content_item.get("content"): content_text = content_item.get("content", "") # Vérifier qu'il y a suffisamment de contenu if len(content_text.strip()) > 100: # Minimum 100 caractères # Tronquer le contenu si nécessaire truncated_content = self._truncate_to_n_words( content_text, self.valves.search_scrape_and_summarize_words_limit, ) content_for_llm.append( f"Source: {content_item.get('title', 'Unknown')} ({content_item['url']})\n" f"Content: {truncated_content}\n" ) else: failed_scrapes += 1 print(f"⚠️ Content too short for {content_item['url']}") else: failed_scrapes += 1 error_msg = content_item.get("error", "Unknown error") print(f"❌ Scraping failed for {content_item['url']}: {error_msg}") 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, "scraping_errors": [ { "url": item["url"], "error": item.get("error", "Content too short"), } for item in scraped_content ], "suggestion": "Try different keywords, check if URLs are accessible, or add a Jina API key for better reliability.", } ) # É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, "answer": final_answer, "sources_used": len(content_for_llm), "urls_scraped": selected_urls, "llm1_used": llm1_success, "fallback_used": not llm1_success, "scraping_stats": { "successful": len(content_for_llm), "failed": failed_scrapes, }, "search_results_count": len(initial_search_results.get("results", [])), "status": "success", }, ensure_ascii=False, ) async def scrape_url( self, url: str, __event_emitter__: Optional[Callable[[dict], Any]] = None ) -> str: """ Scrape the content of a specific URL with Jina. """ if not url or not url.startswith(("http://", "https://")): return json.dumps({"error": "Invalid URL provided."}) print(f"Scrapping de l'URL : {url} avec Jina...") scraped_data = await self._scrape_one(url, __event_emitter__) if scraped_data.get("error"): return json.dumps( { "error": f"Could not scrape URL: {url}", "details": scraped_data["error"], } ) # Tronquer le contenu content = scraped_data.get("content", "") truncated_content = self._truncate_to_n_words( content, self.valves.scrape_only_words_limit ) return json.dumps( { "title": scraped_data.get("title", "No title"), "url": url, "content": truncated_content, "excerpt": self._generate_excerpt(content), }, ensure_ascii=False, ) async def summarize_url( self, url: str, keywords: str = "", query_2: str = "", __event_emitter__: Optional[Callable[[dict], Any]] = None, ) -> str: """ Scrapes the content of a specific URL with Jina and summarizes it based on the keywords provided (or query_2 if provided). """ if not url or not url.startswith(("http://", "https://")): return json.dumps({"error": "Invalid URL provided."}) # Déterminer quelle requête utiliser pour le LLM 2 llm_query = query_2.strip() if query_2.strip() else keywords print(f"URL scraping and summarization : {url} with keywords :'{llm_query}'...") scraped_data = await self._scrape_one(url, __event_emitter__) if scraped_data.get("error"): return json.dumps( { "error": f"Could not scrape URL: {url}", "details": scraped_data["error"], } ) page_content = scraped_data.get("content", "") if not page_content: return json.dumps( {"message": "No content was extracted from the URL to summarize."} ) # Tronquer le contenu truncated_content = self._truncate_to_n_words( page_content, self.valves.scrape_and_summarize_words_limit ) # Utilisation de LLM 2 pour résumer le contenu print("Using LLM 2 to summarize content...") summary_prompt = [ { "role": "system", "content": f"You are an expert summarization assistant. Given web content and specific keywords, summarize the most relevant information concisely and accurately. Focus on the information related to the keywords. If keywords are provided, emphasize them. If no keywords, provide a general summary. Always include the source URL. Maximize relevant information, minimize fluff.", }, { "role": "user", "content": f"URL: {url}\nKeywords: {llm_query if llm_query else 'None provided'}\n\nContent to summarize:\n{truncated_content}\n\nProvide a concise summary:", }, ] summary = await self._call_llm(summary_prompt, model_num=2) if "Error" in summary: return json.dumps( {"error": f"LLM 2 failed to summarize content: {summary}"} ) return json.dumps( { "url": url, "keywords": keywords, "llm_query_used": llm_query, "summary": summary, "status": "success", }, ensure_ascii=False, )