"""
title: Apriel‑1.5‑15B‑Thinker — Final+Think Filter
author: Bartosz Supczinski (
[email protected])
version: 1.0
"""
from typing import Optional, Dict, Any
import re
from pydantic import BaseModel, Field
class Filter:
class Valves(BaseModel):
priority: int = Field(default=-1000)
BEGIN: str = Field(default="[BEGIN FINAL RESPONSE]")
END: str = Field(default="[END FINAL RESPONSE]")
THINK_OPEN: str = Field(default="<think>\n")
THINK_CLOSE: str = Field(default="\n</think>\n")
THINK_HEADER: str = Field(default="")
PRESERVE_NEWLINES: bool = Field(default=False)
def __init__(self) -> None:
self.valves = self.Valves()
self.toggle = True
self._reset()
def _reset(self):
self._started = False
self._ended = False
self._carry = ""
self._think_open = False
def _longest_suffix_as_prefix(self, text: str, pattern: str) -> int:
if not text or not pattern:
return 0
max_k = min(len(text), len(pattern) - 1)
for k in range(max_k, 0, -1):
if text.endswith(pattern[:k]):
return k
return 0
def _is_prefix_of(self, s: str, pattern: str) -> bool:
return 0 < len(s) < len(pattern) and pattern.startswith(s)
def inlet(
self, body: Dict[str, Any], __user__: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
stops = body.get("stop", [])
if isinstance(stops, str):
stops = [stops]
for s in (self.valves.END, "<|end|>"):
if s not in stops:
stops.append(s)
body["stop"] = stops
return body
def _process_chunk(self, chunk: str) -> str:
s = self._carry + (chunk or "")
out = ""
if not self._started:
prefix = ""
if not self._think_open:
self._think_open = True
prefix = self.valves.THINK_OPEN + self.valves.THINK_HEADER + "\n"
i = s.find(self.valves.BEGIN)
if i == -1:
keep = self._longest_suffix_as_prefix(s, self.valves.BEGIN)
pre_chunk = s[:-keep] if keep else s
out += prefix + pre_chunk
self._carry = s[-keep:] if keep else ""
return out
pre = s[:i]
out += prefix + pre + self.valves.THINK_CLOSE
self._think_open = False
self._started = True
after = s[i + len(self.valves.BEGIN) :]
j = after.find(self.valves.END)
if j != -1:
out += after[:j]
self._ended = True
self._carry = ""
return out
keep_end = self._longest_suffix_as_prefix(after, self.valves.END)
self._carry = after[-keep_end:] if keep_end else ""
out += after[:-keep_end] if keep_end else after
return out
j = s.find(self.valves.END)
if j != -1:
out += s[:j]
self._ended = True
self._carry = ""
return out
keep = self._longest_suffix_as_prefix(s, self.valves.END)
self._carry = s[-keep:] if keep else ""
out += s[:-keep] if keep else s
return out
def stream(self, event: Dict[str, Any]) -> Dict[str, Any]:
if not getattr(self, "toggle", True):
return event
choices = event.get("choices") or []
for choice in choices:
delta = choice.get("delta", {})
if isinstance(delta.get("content"), str):
delta["content"] = self._process_chunk(delta["content"])
finished = choice.get("finish_reason") is not None
if finished:
if (not self._started) and self._carry:
if not self._is_prefix_of(self._carry, self.valves.BEGIN):
delta["content"] = (delta.get("content") or "") + self._carry
self._carry = ""
if self._think_open:
delta["content"] = (
delta.get("content") or ""
) + self.valves.THINK_CLOSE
self._think_open = False
if self._started and (not self._ended) and self._carry:
if not self._is_prefix_of(self._carry, self.valves.END):
delta["content"] = (delta.get("content") or "") + self._carry
self._carry = ""
data = event.get("data")
if isinstance(data, dict) and isinstance(data.get("content"), str):
data["content"] = self._process_chunk(data["content"])
return event
def outlet(
self, body: Dict[str, Any], __user__: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
msgs = body.get("messages", [])
if not msgs:
self._reset()
return body
for m in reversed(msgs):
if m.get("role") == "assistant" and isinstance(m.get("content"), str):
txt = m["content"]
if self.valves.BEGIN in txt:
pre, after = txt.split(self.valves.BEGIN, 1)
end_idx = after.find(self.valves.END)
final = after[:end_idx] if end_idx != -1 else after
txt = (
self.valves.THINK_OPEN
+ self.valves.THINK_HEADER
+ "\n"
+ pre
+ self.valves.THINK_CLOSE
+ final
)
txt = txt.replace(self.valves.BEGIN, "").replace(self.valves.END, "")
pref = self.valves.END[:-1]
txt = re.sub(rf"{re.escape(pref)}\s*$", "", txt)
if "<think>" in txt and "</think>" not in txt:
txt = txt + self.valves.THINK_CLOSE
m["content"] = txt
break
self._reset()
return body
