""" title: Chat Token Clipper author: BobbyLLM version: 0.3 """ from typing import Dict, List, Optional, Union from pydantic import BaseModel, Field def _get_first_user_message( messages: List[Dict[str, Union[str, dict]]], ) -> Optional[Dict]: for m in messages: if m.get("role") == "user": return m return None def _get_first_assistant_message( messages: List[Dict[str, Union[str, dict]]], ) -> Optional[Dict]: for m in messages: if m.get("role") == "assistant": return m return None class Filter: class Valves(BaseModel): priority: int = Field( default=0, description="Priority level for the filter operations.", ) n_last_messages: int = Field( default=2, description="Number of last user/assistant message PAIRS to keep.", ) keep_first: bool = Field( default=True, description="Always keep the first user+assistant message pair, if present.", ) max_chars: int = Field( default=4000, description="Approximate max total characters for non-system messages " "(later messages kept, earlier dropped). 0 disables.", ) class UserValves(BaseModel): n_last_messages: Optional[int] = Field( default=None, description="Override: number of last user/assistant message PAIRS to keep.", ) keep_first: Optional[bool] = Field( default=None, description="Override: always keep the first user+assistant pair.", ) max_chars: Optional[int] = Field( default=None, description="Override: approximate max total characters for non-system messages.", ) def __init__(self): self.valves = self.Valves() self.user_valves = self.UserValves() def inlet(self, body: dict, __user__: Optional[dict] = None) -> dict: messages: List[Dict] = body.get("messages", []) if not messages: return body # Resolve effective valves (user override > global > defaults) user_v = getattr(__user__, "valves", None) if __user__ is not None else None # user_v may be a dict or a Pydantic model depending on OWUI internals def _get_user_val(name: str): if user_v is None: return None if isinstance(user_v, dict): return user_v.get(name) return getattr(user_v, name, None) n_last_pairs = _get_user_val("n_last_messages") if n_last_pairs is None: n_last_pairs = self.valves.n_last_messages n_last_pairs = max(0, int(n_last_pairs)) keep_first = _get_user_val("keep_first") if keep_first is None: keep_first = self.valves.keep_first keep_first = bool(keep_first) max_chars = _get_user_val("max_chars") if max_chars is None: max_chars = self.valves.max_chars max_chars = int(max_chars) if max_chars is not None else 0 if max_chars < 0: max_chars = 0 # Separate system vs user/assistant messages system_messages = [m for m in messages if m.get("role") == "system"] ua_messages = [m for m in messages if m.get("role") in ["user", "assistant"]] # First user/assistant pair (optional keep) first_user = _get_first_user_message(ua_messages) first_assistant = _get_first_assistant_message(ua_messages) # Collect last N pairs (2 messages per pair) n_to_keep = 2 * n_last_pairs if n_last_pairs > 0 else len(ua_messages) recent_ua = ua_messages[-n_to_keep:] if ua_messages else [] new_messages: List[Dict] = [] # Always keep all system messages (usually just one) new_messages.extend(system_messages) # Optionally keep the very first user+assistant messages if keep_first: if first_user is not None: new_messages.append(first_user) if first_assistant is not None: # avoid duplicating if first assistant is already in first_user position if first_assistant is not first_user: new_messages.append(first_assistant) # Ensure we don't start the UA section with assistant-only spam if recent_ua: # If first two are both user, drop the older one if ( recent_ua[0]["role"] == "user" and len(recent_ua) > 1 and recent_ua[1]["role"] == "user" ): recent_ua.pop(0) # If first is assistant, drop leading assistants while recent_ua and recent_ua[0]["role"] == "assistant": recent_ua.pop(0) new_messages.extend(recent_ua) # Optional: approximate max_chars cap for non-system messages if max_chars > 0: trimmed: List[Dict] = [] used_chars = 0 for m in new_messages: if m.get("role") == "system": trimmed.append(m) continue content = m.get("content", "") if not isinstance(content, str): content_str = str(content) else: content_str = content c_len = len(content_str) if used_chars + c_len <= max_chars: trimmed.append(m) used_chars += c_len # else: drop this older message; later ones are closer to the question new_messages = trimmed body["messages"] = new_messages return body def outlet(self, body: dict, __user__: Optional[dict] = None) -> dict: return body