diff --git a/artificium_thinking.py b/artificium_thinking.py new file mode 100644 index 0000000..240603c --- /dev/null +++ b/artificium_thinking.py @@ -0,0 +1,199 @@ +""" +title: Artificium Thought Filter +author: projectmoon +author_url: https://git.agnos.is/projectmoon/open-webui-filters +version: 0.1.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. This is specificially for the +# Artificium model, based on Llama 3.1. It outputs its thought +# processes broken by markdown horizontal rules. Usually, it outputs a +# basic thought process followed by a breakdown. GENERALLY, text below +# the last horizontal line is the final answer. +# +# 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 = """ +
+{{THOUGHT_TITLE}} +{{THOUGHTS}} + +--- + +
+""" + +DEFAULT_THOUGHT_PLAN = """ +Your task execution plan should be: + - Break down the problem into the smallest possible steps + - Come up with a plan to solve each step, then execute each step. + - Analyze each step for mistakes, correct any mistakes found in each step. + - Finally, output the solution based on your reasoning. + +In your reply, break your reasoning down into "Plan", "Execution", "Mistake Analysis", and "Final Output". These should be ## Markdown Headers. +""" + +DETAIL_DELETION_REGEX = r"[\s\S]*?" + +class Filter: + class Valves(BaseModel): + priority: int = Field( + default=0, description="Priority level for the filter operations." + ) + task_title: str = Field( + default="Task Discovery", + description="Title for the collapsible task reasoning section." + ) + breakdown_title: str = Field( + default="Thought Process", + description="Title for the collapsible reasoning breakdown section." + ) + 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}>(.*?)" + + def _create_thought_tag_deletion_regex(self) -> str: + tag = self.valves.thought_tag + return "[\s\S]*?".replace("{{THINK}}", tag) + + def _create_output_tag_deletion_regex(self) -> str: + tag = self.valves.output_tag + return r"[\s\S]*?".replace("{{OUT}}", tag) + + def _parse_reply(self, messages: List[Dict[str, str]]) -> dict: + reply = messages[-1]["content"] + pattern = r'((?<=\n---\n)|^)(.*?)(?=---\n|$)' + matches = re.findall(pattern, reply, flags=re.DOTALL) + sections = [match[1].strip() for match in matches if match[0] == ''] + sections = [section for section in sections if section] + print(f"[Artificium Filter] Parsed {len(sections)} section(s)") + + # a few different situations. + # 1. 3+ sections = initial thoughts, breakdown, final output. + # 2. 2 sections = thoughts, final output + # 3. 1 section or 0 sections = do nothing + if len(sections) >= 3: + return { + "initial": sections[0], + "breakdown": "\n\n---\n\n".join(sections[1:-1]), + "final": sections[-1] + } + elif len(sections) == 2: + return { + "initial": sections[0], + "breakdown": None, + "final": sections[1] + } + else: + return { + "initial": None, + "breakdown": None, + "final": reply + } + + def _enclose_thoughts(self, messages: List[Dict[str, str]]) -> None: + if not messages: + return + + parsed_reply = self._parse_reply(messages) + final_reply = "" + + if parsed_reply["initial"] is not None: + initial_thoughts = (THOUGHT_ENCLOSURE + .replace("{{THOUGHT_TITLE}}", self.valves.task_title) + .replace("{{THOUGHTS}}", parsed_reply["initial"])) + final_reply = initial_thoughts + + if parsed_reply["breakdown"] is not None: + breakdown_thoughts = (THOUGHT_ENCLOSURE + .replace("{{THOUGHT_TITLE}}", self.valves.breakdown_title) + .replace("{{THOUGHTS}}", parsed_reply["breakdown"])) + final_reply = f"{final_reply}\n{breakdown_thoughts}" + + if parsed_reply["final"] is not None: + output = parsed_reply["final"] + final_reply = f"{final_reply}\n{output}" + + final_reply = final_reply.strip() + if final_reply: + messages[-1]["content"] = final_reply + + def _handle_include_thoughts(self, messages: List[Dict[str, str]]) -> None: + """Remove
tags from input, if configured to do so.""" + #
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