""" title: GGUF Deepseek R1 Formatter author: FeathInkYuuki author_url: https://github.com/FeathInkYuuki version: 0.4.0 """ from typing import Optional, Dict, Any from pydantic import BaseModel, Field import re import time class Filter: class Valves(BaseModel): THINK_TAG: str = Field( default="<think>", description="思考开始标记，会被转换为折叠面板" ) END_TAG: str = Field( default="</think>", description="思考结束标记，会被转换为折叠面板闭合" ) SHOW_TIME_COST: bool = Field( default=True, description="是否在折叠面板标题显示耗时" ) FORCE_THINKING: bool = Field( default=True, description="当没有检测到思考标签时自动包裹内容" ) def __init__(self): self.valves = self.Valves() self.start_time = 0.0 self.force_activated = False # 预编译正则表达式 self.think_pattern = re.compile(re.escape(self.valves.THINK_TAG), re.IGNORECASE) self.end_pattern = re.compile(re.escape(self.valves.END_TAG), re.IGNORECASE) self.content_wrap_pattern = re.compile( r'(<details class="deepseek-think">\n<summary>.*?</summary>\n)(.*?)(\n</details>)', re.DOTALL, ) def inlet( self, body: Dict[str, Any], user: Optional[dict] = None ) -> Dict[str, Any]: self.start_time = time.time() return body async def outlet( self, body: Dict[str, Any], user: Optional[dict] = None ) -> Dict[str, Any]: if not body.get("messages"): return body last_msg = body["messages"][-1] content = last_msg.get("content", "") duration = time.time() - self.start_time # 强制思考模式优化 if self.valves.FORCE_THINKING: has_think = self.think_pattern.search(content) has_end = self.end_pattern.search(content) if not has_think: # 插入思考开始标签 content = f"{self.valves.THINK_TAG}\n{content}" self.force_activated = True # 自动补全结束标签 if not has_end: content += f"\n{self.valves.END_TAG}" formatted = self._convert_tags(content, duration) formatted = self._handle_unclosed_tags(formatted) last_msg["content"] = formatted return body def _convert_tags(self, content: str, duration: float) -> str: time_str = f" ({duration:.1f}s)" if self.valves.SHOW_TIME_COST else "" # 使用预编译正则进行替换 content = self.think_pattern.sub( f'<details class="deepseek-think">\n<summary>💭 深度思考{time_str}</summary>\n', content, ) content = self.end_pattern.sub("\n</details>", content) # 内容包裹优化（单次匹配） content = self.content_wrap_pattern.sub( lambda m: f"{m.group(1)}{self._format_content(m.group(2))}{m.group(3)}", content, count=1, # 根据约束条件只处理第一个匹配 ) return content def _format_content(self, text: str) -> str: """优化的引用格式化方法""" return text.replace("\n", "\n> ").rstrip("> ") def _handle_unclosed_tags(self, content: str) -> str: """线性复杂度未闭合标签处理""" open_count = content.count('<details class="deepseek-think">') close_count = content.count("</details>") if open_count > close_count: content += "\n</details>" * (open_count - close_count) return content def __call__(self, body: Dict[str, Any]) -> Dict[str, Any]: return self.outlet(body)