open-webui-filters/thinking.py

"""
title: Collapsible Thought Filter
author: projectmoon
author_url: https://git.agnos.is/projectmoon/open-webui-filters
version: 0.2.0
license: AGPL-3.0+, MIT
required_open_webui_version: 0.3.32
"""

#########################################################
# OpenWebUI Filter that collapses model reasoning/thinking into a
# separate section in the reply.

# 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.
#########################################################

from typing import Optional, Dict, List
import re
from pydantic import BaseModel, Field

THOUGHT_ENCLOSURE = """
<details>
<summary>{{THOUGHT_TITLE}}</summary>
{{THOUGHTS}}
</details>
"""

DETAIL_DELETION_REGEX = r"</?details>[\s\S]*?</details>"

class Filter:
    class Valves(BaseModel):
        priority: int = Field(
            default=0, description="Priority level for the filter operations."
        )
        thought_title: str = Field(
            default="Thought Process",
            description="Title for the collapsible reasoning section."
        )
        thought_tag: str = Field(
            default="thinking",
            description="The XML tag for model thinking output."
        )
        output_tag: str = Field(
            default="output",
            description="The XML tag for model final output."
        )
        use_thoughts_as_context: bool = Field(
            default=False,
            description=("Include previous thought processes as context for the AI. "
                         "Disabled by default.")
        )
        pass

    def __init__(self):
        self.valves = self.Valves()

    def _create_thought_regex(self) -> str:
        tag = self.valves.thought_tag
        return f"<{tag}>(.*?)</{tag}>"

    def _create_output_regex(self) -> str:
        tag = self.valves.output_tag
        return f"<{tag}>(.*?)</{tag}>"

    def _create_thought_tag_deletion_regex(self) -> str:
        tag = self.valves.thought_tag
        return "</?{{THINK}}>[\s\S]*?</{{THINK}}>".replace("{{THINK}}", tag)

    def _create_output_tag_deletion_regex(self) -> str:
        tag = self.valves.output_tag
        return r"</?{{OUT}}>[\s\S]*?</{{OUT}}>".replace("{{OUT}}", tag)

    def _enclose_thoughts(self, messages: List[Dict[str, str]]) -> None:
        if not messages:
            return

        # collapsible thinking process section
        thought_regex = self._create_thought_regex()
        output_regex = self._create_output_regex()
        reply = messages[-1]["content"]

        thoughts = re.findall(thought_regex, reply, re.DOTALL)
        thoughts = "\n".join(thoughts).strip()

        output = re.findall(output_regex, reply, re.DOTALL)
        output = "\n".join(output).strip()

        print(thoughts)
        print(output)

        enclosure = THOUGHT_ENCLOSURE.replace("{{THOUGHT_TITLE}}", self.valves.thought_title)
        enclosure = enclosure.replace("{{THOUGHTS}}", thoughts).strip()

        # remove processed thinking and output tags.
        # some models do not close output tags properly.
        thought_tag_deletion_regex = self._create_thought_tag_deletion_regex()
        output_tag_deletion_regex = self._create_output_tag_deletion_regex()
        reply = re.sub(thought_tag_deletion_regex, "", reply, count=1)
        reply = re.sub(output_tag_deletion_regex, "", reply, count=1)
        reply = reply.replace(f"<{self.valves.output_tag}>", "", 1)
        reply = reply.replace(f"</{self.valves.output_tag}>", "", 1)

        # because some models do not close the output tag, we prefer
        # using the captured output via regex, but if that does not
        # work, we use whatever's left over as the output (which is
        # already set).
        if output is not None and len(output) > 0:
            reply = output

        # prevents empty thought process blocks when filter used with
        # malformed LLM output.
        if len(enclosure) > 0:
            reply = f"{enclosure}\n{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."""
        # <details> tags are created by the outlet filter for display
        # in OWUI.
        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", [])
            self._enclose_thoughts(original_messages)
            body["messages"] = original_messages
            return body
        except Exception as e:
            print(e)
            return body