""" title: Turn.Phi4.R1Distill.Thoughts author: Quazim0t0 author_url: https://huggingface.co/Quazim0t0 version: 1.0 license: AGPL-3.0+, MIT required_open_webui_version: 0.3.32 """ # Fixed bug where function had to be on globally. # Works properly now and shouldn't need anymore updates # This should work with R1 Distill models # Based on the Add or Delete Text Filter by anfi. # https://openwebui.com/f/anfi/add_or_delete_text # # Therefore, portions of this code are licensed under the MIT license. # The modifications made for "thought enclosure" etc are licensed # under the AGPL using the MIT's sublicensing clause. # # For those portions under the MIT license, the following applies: # # MIT License # # Copyright (c) 2024 anfi # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal # in the Software without restriction, including without limitation the rights # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in all # copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # SOFTWARE. ######################################################### # OpenWebUI Filter that collapses model reasoning/thinking into a # separate section in the reply. from typing import Optional, Dict, List import re from pydantic import BaseModel, Field THOUGHT_SOLUTION_ENCLOSURE = """ <details> <summary>Thought Process</summary> {{THOUGHTS}} </details> {{SOLUTION}} """ DETAIL_DELETION_REGEX = r"</?details>[\s\S]*?</details>" LATEX_BOX_REGEX = r"\\boxed\{\\text\{(.*?)\}\}" # Regex to match \boxed{\text{...}} class Filter: class Valves(BaseModel): priority: int = Field( default=0, description="Priority level for the filter operations." ) thought_start_tag: str = Field( default="<|begin_of_thought|>", description="The tag marking the beginning of model thinking output.", ) thought_end_tag: str = Field( default="<|end_of_thought|>", description="The tag marking the end of model thinking output.", ) solution_start_tag: str = Field( default="<|begin_of_solution|>", description="The tag marking the beginning of model final output.", ) solution_end_tag: str = Field( default="<|end_of_solution|>", description="The tag marking the end of model final output.", ) use_thoughts_as_context: bool = Field( default=False, description=( "Include previous thought processes as context for the AI. " "Disabled by default." ), ) def __init__(self): self.valves = self.Valves() def _create_thought_regex(self) -> str: start_tag = re.escape(self.valves.thought_start_tag) end_tag = re.escape(self.valves.thought_end_tag) return f"{start_tag}(.*?){end_tag}" def _create_solution_regex(self) -> str: start_tag = re.escape(self.valves.solution_start_tag) end_tag = re.escape(self.valves.solution_end_tag) return f"{start_tag}(.*?){end_tag}" def _clean_latex_box(self, text: str) -> str: """ Remove LaTeX-style \boxed{\text{...}} formatting from the text. :param text: The text to clean. :return: The cleaned text. """ return re.sub(LATEX_BOX_REGEX, r"\1", text) def _process_and_enclose(self, messages: List[Dict[str, str]]) -> None: if not messages: return thought_regex = self._create_thought_regex() solution_regex = self._create_solution_regex() reply = messages[-1]["content"] # Extract thoughts and solutions thoughts = re.findall(thought_regex, reply, re.DOTALL) solutions = re.findall(solution_regex, reply, re.DOTALL) # Combine thoughts and solutions into the enclosure thoughts_content = ( "\n".join(thoughts).strip() if thoughts else "No thought process provided." ) solutions_content = ( "\n".join(solutions).strip() if solutions else "No solution provided." ) # Clean up LaTeX-style boxes in the solution solutions_content = self._clean_latex_box(solutions_content) enclosure = ( THOUGHT_SOLUTION_ENCLOSURE.replace("{{THOUGHTS}}", thoughts_content) .replace("{{SOLUTION}}", solutions_content) .strip() ) # Ensure the enclosure is used if thoughts or solutions: messages[-1]["content"] = enclosure else: # If no thoughts or solutions exist, preserve the original reply messages[-1]["content"] = reply def _handle_include_thoughts(self, messages: List[Dict[str, str]]) -> None: """Remove <details> tags from input, if configured to do so.""" if self.valves.use_thoughts_as_context: return for message in messages: message["content"] = re.sub( DETAIL_DELETION_REGEX, "", message["content"], count=1 ) def inlet( self, body: Dict[str, any], __user__: Optional[Dict[str, any]] = None ) -> Dict[str, any]: try: original_messages: List[Dict[str, str]] = body.get("messages", []) self._handle_include_thoughts(original_messages) body["messages"] = original_messages return body except Exception as e: print(e) return body def outlet( self, body: Dict[str, any], __user__: Optional[Dict[str, any]] = None ) -> Dict[str, any]: try: original_messages: List[Dict[str, str]] = body.get("messages", []) # Process the messages self._process_and_enclose(original_messages) # Update the body with processed messages body["messages"] = original_messages return body except Exception as e: print(f"Error in outlet function: {e}") return body