""" title: Qwen 3 No Think author: G30 author_url: https://github.com/open-webui funding_url: https://github.com/open-webui version: 0.1.1 description: Prepends /no_think to the system prompt """ from pydantic import BaseModel from typing import Optional, List, Dict, Any, Callable, Awaitable import logging # Setup logger logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) # Set to DEBUG for more verbose output if not logger.handlers: handler = logging.StreamHandler() formatter = logging.Formatter( "%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) handler.setFormatter(formatter) logger.addHandler(handler) logger.propagate = False class Filter: """ A simple filter that, if its internal `self.toggle` is True, prepends '/no_think' to the system prompt. If `self.toggle` is False, it makes no changes to the prompt. This toggle is set in the code and not via UI Valves. """ class Valves(BaseModel): pass # No user-configurable valves as per the requested structure def __init__(self): self.valves = self.Valves() # This is the "toggle" state for the filter's action. # Set to True: prepends "/no_think". # Set to False: the filter does nothing regarding /no_think. # To change the behavior, modify this line and restart Open WebUI. self.toggle: bool = True self.icon = """data:image/svg+xml;base64,PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIGZpbGw9Im5vbmUiIHZpZXdCb3g9IjAgMCAyNCAyNCIgc3Ryb2tlLXdpZHRoPSIxLjUiIHN0cm9rZT0iY3VycmVudENvbG9yIiBjbGFzcz0ic2l6ZS02Ij4KICA8cGF0aCBzdHJva2UtbGluZWNhcD0icm91bmQiIHN0cm9rZS1saW5lam9pbj0icm91bmQiIGQ9Ik0xMiAxOHYtNS4yNW0wIDBhNi4wMSA2LjAxIDAgMCAwIDEuNS0uMTg5bS0xLjUuMTg5YTYuMDEgNi4wMSAwIDAgMS0xLjUtLjE4OW0zLjc1IDcuNDc4YTEyLjA2IDEyLjA2IDAgMCAxLTQuNSAwbTMuNzUgMi4zODNhMTQuNDA2IDE0LjQwNiAwIDAgMS0zIDBNMTQuMjUgMTh2LS4xOTJjMC0uOTgzLjY1OC0xLjgyMyAxLjUwOC0yLjMxNmE3LjUgNy41IDAgMSAwLTcuNTE3IDBjLjg1LjQ5MyAxLjUwOSAxLjMzMyAxLjUwOSAyLjMxNlYxOCIgLz4KPC9zdmc+Cg==""" active_action_for_log = ( "prepends /no_think" if self.toggle else "filter inactive, no changes made" ) logger.info( f"Qwen 3 no_think filter initialized. Default behavior: {active_action_for_log} (based on self.toggle = {self.toggle})" ) async def inlet( self, body: Dict[str, Any], __event_emitter__: Callable[[Dict[str, Any]], Awaitable[None]], __user__: Optional[Dict[str, Any]] = None, __model__: Optional[Dict[str, Any]] = None, __request__: Optional[Any] = None, ) -> Dict[str, Any]: # If self.toggle is False, the filter does nothing. if not self.toggle: logger.debug( f"Qwen 3 no_think: self.toggle is False. Passing through without modification." ) return body # If self.toggle is True, proceed to add /no_think command_to_add = "/no_think" logger.debug( f"Qwen 3 no_think: self.toggle is True. Attempting to add '{command_to_add}'." ) messages: List[Dict[str, str]] = body.get("messages", []) if not messages: logger.warning( "No messages found in body, skipping modification even though toggle is True." ) return body system_message_found_at_index = -1 for i, msg in enumerate(messages): if msg.get("role") == "system": system_message_found_at_index = i break if system_message_found_at_index != -1: logger.debug( f"Existing system message found at index {system_message_found_at_index}." ) current_content = messages[system_message_found_at_index].get("content", "") # Remove any existing /no_think command to avoid duplication. # We also remove /think as a precaution to prevent potential conflicts, # even though this filter doesn't add /think itself. cleaned_content = ( current_content.replace("/no_think", "").replace("/think", "").strip() ) new_content = f"{command_to_add} {cleaned_content}".strip() # If cleaned_content was empty (e.g., original system prompt was just "/no_think" or "/think"), # new_content will correctly be just the command_to_add. if not cleaned_content: new_content = command_to_add messages[system_message_found_at_index]["content"] = new_content logger.info(f"Modified system message to: '{new_content}'") else: logger.debug("No existing system message found. Creating a new one.") messages.insert(0, {"role": "system", "content": command_to_add}) logger.info(f"Added new system message: '{command_to_add}'") body["messages"] = messages return body async def outlet( self, body: Dict[str, Any], __event_emitter__: Callable[[Dict[str, Any]], Awaitable[None]], __user__: Optional[Dict[str, Any]] = None, __model__: Optional[Dict[str, Any]] = None, __request__: Optional[Any] = None, ) -> Dict[str, Any]: logger.debug("Qwen 3 no_think outlet called, passing through.") return body