""" Title: Auto Better Websearch Tool Description: Sélecteur Automatique de recherche web pour l'outil web_search Author: TRUC Yoann ! Require my better websearch tool to work with ! """ from pydantic import BaseModel, Field from typing import Callable, Awaitable, Any, Optional import re import logging from open_webui.apps.webui.models.users import Users from open_webui.apps.webui.models.tools import Tools from open_webui.utils.misc import get_last_user_message log = logging.getLogger(__name__) class Filter: class Valves(BaseModel): status: bool = Field( default=True, description="Active ou désactive le filtre automatique" ) match_threshold: int = Field( default=1, description="Nombre minimum de correspondances requises" ) symbols: list = Field( default=["?"], description="Symboles déclenchant la recherche web" ) web_search_keywords: list = Field( default=[ "chercher", "recherches", "trouver", "qui", "quoi", "quel", "quelle", "quels", "quelles", "comment", "pourquoi", "où", "quand", "combien", "est-ce que", "qu'est-ce", ], description="Mots-clés déclenchant la recherche web", ) def __init__(self): self.valves = self.Valves() async def should_trigger_web_search(self, message: str) -> tuple[bool, str]: """Vérifie si le message devrait déclencher une recherche web.""" message_lower = message.lower() # Vérifie la présence de symboles de question for symbol in self.valves.symbols: if symbol in message: return True, f"Symbole détecté : {symbol}" # Vérifie la présence de mots-clés de recherche for keyword in self.valves.web_search_keywords: if keyword in message_lower: return True, f"Mot-clé détecté : {keyword}" return False, "Aucun déclencheur détecté" async def inlet( self, body: dict, __event_emitter__: Callable[[Any], Awaitable[None]], __user__: Optional[dict] = None, __model__: Optional[dict] = None, ) -> dict: try: # Étape 1: Vérification du statut await __event_emitter__( { "type": "status", "data": { "description": "Vérification du statut du filtre...", "done": False, }, } ) if not self.valves.status: await __event_emitter__( { "type": "status", "data": { "description": "Filtre désactivé. Passage direct.", "done": True, }, } ) return body # Étape 2: Extraction du message messages = body["messages"] user_message = get_last_user_message(messages) log.info(f"Processing message: {user_message}") await __event_emitter__( { "type": "status", "data": { "description": "Analyse du message utilisateur...", "done": False, }, } ) # Étape 3: Vérification de la disponibilité de web_search available_tool_ids = ( __model__.get("info", {}).get("meta", {}).get("toolIds", []) ) await __event_emitter__( { "type": "status", "data": { "description": "Vérification de la disponibilité de l'outil web_search...", "done": False, }, } ) if "web_search" not in available_tool_ids: await __event_emitter__( { "type": "status", "data": { "description": "L'outil web_search n'est pas disponible pour ce modèle", "done": True, }, } ) log.warning("web_search tool not available for this model") return body # Étape 4: Analyse du besoin de recherche should_search, reason = await self.should_trigger_web_search(user_message) await __event_emitter__( { "type": "status", "data": { "description": f"Analyse du besoin de recherche : {reason}", "done": False, }, } ) if should_search: log.info(f"Web search trigger detected: {reason}") # Étape 5: Activation de la recherche web await __event_emitter__( { "type": "status", "data": { "description": "Préparation de l'outil de recherche web...", "done": False, }, } ) # Ajoute web_search aux outils à utiliser tool_ids = body.get("tool_ids", []) if "web_search" not in tool_ids: tool_ids.append("web_search") body["tool_ids"] = tool_ids await __event_emitter__( { "type": "status", "data": { "description": "Recherche web activée et prête à l'emploi", "done": True, }, } ) log.info("Web search tool added successfully") else: await __event_emitter__( { "type": "status", "data": { "description": "Aucun besoin de recherche web détecté", "done": True, }, } ) log.debug("No web search needed") return body except Exception as e: error_msg = f"Erreur lors du traitement : {str(e)}" log.error(error_msg) await __event_emitter__( { "type": "status", "data": { "description": error_msg, "done": True, }, } ) return body # Configuration du logging logging.basicConfig(level=logging.INFO)