from typing import Optional, Dict, List import re from pydantic import BaseModel, Field # Template for collapsible sections COLLAPSIBLE_SECTION = """ <details> <summary>{{TITLE}}</summary> {{CONTENT}} </details> """ # List of custom tags to process CUSTOM_TAGS = [ "imagine", "like-subjects", "associatedthoughts", "dialectical", "backwards-induction", "metacognition", "possible-secondary-solving-method", ] # Display names for tags TAG_NAMES = { "imagine": "Imagine", "like-subjects": "Like Subjects", "associatedthoughts": "Associated Thoughts", "dialectical": "Dialectical Analysis", "backwards-induction": "Backwards Induction", "metacognition": "Metacognition", "possible-secondary-solving-method": "Possible Secondary Solving Method", } 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.", ) def __init__(self): self.valves = self.Valves() def _extract_tag_content(self, text: str, tag: str) -> Optional[str]: """Extract content between custom tags.""" # Try with exact tag pattern = f"<{tag}>(.*?)</{tag}>" matches = re.findall(pattern, text, re.DOTALL) # If no match found and tag has a hyphen, try with alternative forms if not matches and "-" in tag: # Try with hyphen removed alt_tag = tag.replace("-", "") pattern = f"<{alt_tag}>(.*?)</{alt_tag}>" matches = re.findall(pattern, text, re.DOTALL) # If still no match, try with other variants if not matches and tag == "metacognition": pattern = f"<meta-cognition>(.*?)</meta-cognition>" matches = re.findall(pattern, text, re.DOTALL) if not matches and tag == "backwards-induction": pattern = f"<backwards induction>(.*?)</backwards induction>" matches = re.findall(pattern, text, re.DOTALL) return matches[0].strip() if matches else None def _extract_thought_content(self, text: str) -> Optional[str]: """Extract content between thought tags.""" pattern = f"{re.escape(self.valves.thought_start_tag)}(.*?){re.escape(self.valves.thought_end_tag)}" matches = re.findall(pattern, text, re.DOTALL) return matches[0].strip() if matches else None def _extract_solution_content(self, text: str) -> Optional[str]: """Extract content between solution tags.""" pattern = f"{re.escape(self.valves.solution_start_tag)}(.*?){re.escape(self.valves.solution_end_tag)}" matches = re.findall(pattern, text, re.DOTALL) return matches[0].strip() if matches else None def _get_solution(self, text: str) -> str: """Get the solution content from the text.""" # First try to extract from solution tags solution = self._extract_solution_content(text) if solution: return solution # If no solution tags, check if there's content outside of all tags # Create patterns for all tags to remove tag_patterns = [] for tag in CUSTOM_TAGS: tag_patterns.append(f"<{tag}>.*?</{tag}>") tag_patterns.append( f"{re.escape(self.valves.thought_start_tag)}.*?{re.escape(self.valves.thought_end_tag)}" ) tag_patterns.append( f"{re.escape(self.valves.solution_start_tag)}.*?{re.escape(self.valves.solution_end_tag)}" ) # Create combined pattern and remove all tagged content combined_pattern = "|".join(tag_patterns) remaining_text = re.sub(combined_pattern, "", text, flags=re.DOTALL).strip() # If there's text remaining, use it as the solution if remaining_text: return remaining_text # If still no solution, provide a default message return "No solution provided." def _create_collapsible_sections(self, text: str) -> str: """Create collapsible sections for all tags and thought content.""" sections = [] # Process custom tags for tag in CUSTOM_TAGS: content = self._extract_tag_content(text, tag) if content: title = TAG_NAMES.get(tag, tag.capitalize()) section = COLLAPSIBLE_SECTION.replace("{{TITLE}}", title).replace( "{{CONTENT}}", content ) sections.append(section) # Process thought content thought_content = self._extract_thought_content(text) if thought_content: section = COLLAPSIBLE_SECTION.replace( "{{TITLE}}", "Thought Process" ).replace("{{CONTENT}}", thought_content) sections.append(section) # Add solution solution_content = self._get_solution(text) sections.append(solution_content) return "\n".join(sections) def _has_tags(self, content: str) -> bool: """Check if the content has any of our tags.""" # Check for standard tags if ( self.valves.thought_start_tag in content or self.valves.solution_start_tag in content ): return True # Check for custom tags for tag in CUSTOM_TAGS: if f"<{tag}>" in content: return True # Check for hyphenated variants if "-" in tag and f"<{tag.replace('-', '')}>" in content: return True # Special check for meta-cognition if tag == "metacognition" and "<meta-cognition>" in content: return True # Special check for backwards induction if tag == "backwards-induction" and "<backwards induction>" in content: return True return False def inlet( self, body: Dict[str, any], __user__: Optional[Dict[str, any]] = None ) -> Dict[str, any]: """Process incoming messages.""" return body def outlet( self, body: Dict[str, any], __user__: Optional[Dict[str, any]] = None ) -> Dict[str, any]: """Process outgoing messages.""" try: if "messages" in body and body["messages"]: last_message = body["messages"][-1] if "content" in last_message and last_message["content"]: content = last_message["content"] # Check if content has any of our tags if self._has_tags(content): formatted_content = self._create_collapsible_sections(content) last_message["content"] = formatted_content except Exception as e: print(f"Error in outlet: {e}") return body