""" title: Chat Token Clipper + Rolling Summary author: BobbyLLM version: 0.5 """ from typing import Dict, List, Optional, Union from pydantic import BaseModel, Field import time import os # --- helper functions for first user/assistant messages ---------------------- 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 # --- helpers for rolling summary --------------------------------------------- SUMMARY_PREFIX = "[CHAT_SUMMARY] " # marker so we can recognize our own summary def _is_summary_message(m: Dict) -> bool: """Detect summary messages by a special prefix in the content.""" if m.get("role") not in ("system", "assistant"): return False content = m.get("content", "") if not isinstance(content, str): return False return content.startswith(SUMMARY_PREFIX) def _extract_plain_ua_text(messages: List[Dict]) -> str: """ Turn all user/assistant messages (excluding summaries) into a plain text log that we can 'summarize' cheaply. """ lines: List[str] = [] for m in messages: if m.get("role") not in ("user", "assistant"): continue if _is_summary_message(m): continue role = m.get("role") content = m.get("content", "") if not isinstance(content, str): content = str(content) lines.append(f"{role.upper()}: {content}") return "\n".join(lines) def _cheap_rolling_summary(existing_summary: str, new_text: str, max_words: int) -> str: """ VERY cheap 'summary': - Append new_text to existing_summary - Keep only the last `max_words` words overall """ combined = (existing_summary or "").strip() if combined: combined = combined + "\n\n" + new_text else: combined = new_text words = combined.split() if len(words) <= max_words: return " ".join(words) # Keep the newest information return " ".join(words[-max_words:]) def _get_existing_summary(messages: List[Dict]) -> Optional[Dict]: for m in messages: if _is_summary_message(m): return m return None def _update_summary_message( messages: List[Dict], enable_summary: bool, summary_every_n_user_msgs: int, summary_max_words: int, ) -> List[Dict]: """ Optionally create/update a single rolling summary message in the history. - Runs only if enable_summary is True and summary_every_n_user_msgs > 0 """ if not enable_summary or summary_every_n_user_msgs <= 0 or summary_max_words <= 0: return messages # Count user messages (excluding any summary messages) user_msgs = [ m for m in messages if m.get("role") == "user" and not _is_summary_message(m) ] user_count = len(user_msgs) if user_count < summary_every_n_user_msgs: return messages # Only update on every Nth user turn if user_count % summary_every_n_user_msgs != 0: return messages # Extract old summary if present existing_summary_msg = _get_existing_summary(messages) existing_summary_text = "" if existing_summary_msg is not None: content = existing_summary_msg.get("content", "") if isinstance(content, str) and content.startswith(SUMMARY_PREFIX): existing_summary_text = content[len(SUMMARY_PREFIX) :] # Build new "summary" text from UA messages ua_text = _extract_plain_ua_text(messages) if not ua_text.strip(): return messages new_summary_text = _cheap_rolling_summary( existing_summary_text, ua_text, max_words=summary_max_words, ) summary_content = SUMMARY_PREFIX + new_summary_text # Remove any old summary messages messages_wo_old_summary = [m for m in messages if not _is_summary_message(m)] # Insert updated summary as a system message near the top (after any real system msgs) system_messages = [m for m in messages_wo_old_summary if m.get("role") == "system"] other_messages = [m for m in messages_wo_old_summary if m.get("role") != "system"] summary_msg = { "role": "system", "content": summary_content, } new_messages: List[Dict] = [] new_messages.extend(system_messages) new_messages.append(summary_msg) new_messages.extend(other_messages) return new_messages # ----------------------------------------------------------------------------- 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." ), ) # --- rolling summary configuration --------------------------------- enable_summary: bool = Field( default=False, description="Enable rolling conversation summary messages.", ) summary_every_n_user_msgs: int = Field( default=8, description=( "Create/update the rolling summary every N user messages. " "Set to 0 to disable." ), ) summary_max_words: int = Field( default=160, description=("Maximum length (in words) of the rolling summary text."), ) # --- debug configuration ------------------------------------------- debug: bool = Field( default=False, description="Enable debug dump of RAW+FINAL messages to a text file.", ) debug_dir: str = Field( default="", description=( "Directory where debug logs are written. " "If empty, uses the server's current working directory." ), ) 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.", ) # --- user overrides for summary ------------------------------------ enable_summary: Optional[bool] = Field( default=None, description="Override: enable rolling conversation summary.", ) summary_every_n_user_msgs: Optional[int] = Field( default=None, description="Override: create/update the summary every N user messages.", ) summary_max_words: Optional[int] = Field( default=None, description="Override: maximum length (in words) of the summary.", ) # --- user overrides for debug -------------------------------------- debug: Optional[bool] = Field( default=None, description="Override: enable debug dump of RAW+FINAL messages.", ) debug_dir: Optional[str] = Field( default=None, description="Override: directory for debug logs.", ) def __init__(self): self.valves = self.Valves() self.user_valves = self.UserValves() # --- helper for debug dumps --------------------------------------------- def _debug_dump_request_to_file( self, raw_messages: List[Dict], final_messages: List[Dict], debug_dir: str, max_preview: int = 1000, ): """ Write RAW and FINAL messages for a single request into one text file. """ # Resolve base directory if debug_dir: base_dir = debug_dir else: base_dir = os.getcwd() try: os.makedirs(base_dir, exist_ok=True) except Exception as e: print( f"[Filter DEBUG] Failed to create debug_dir '{debug_dir}': {e}. " f"Falling back to current working directory." ) base_dir = os.getcwd() ts = time.strftime("%Y%m%d-%H%M%S", time.localtime()) filename = f"chat_filter_debug_{ts}.txt" path = os.path.join(base_dir, filename) def _write_section(f, label: str, messages: List[Dict]): f.write(f"{label}\n") f.write(f"Timestamp: {ts}\n") f.write(f"Message count: {len(messages)}\n") f.write("-" * 60 + "\n\n") for i, m in enumerate(messages): role = m.get("role") content = m.get("content", "") if not isinstance(content, str): content = str(content) preview = content[:max_preview] if len(content) > max_preview: preview += "... [truncated]" f.write(f"[{i:02d}] role={role}\n") f.write(preview) f.write("\n" + "-" * 60 + "\n\n") with open(path, "w", encoding="utf-8") as f: _write_section(f, "RAW MESSAGES (BEFORE FILTER)", raw_messages) _write_section(f, "FINAL MESSAGES (AFTER FILTER)", final_messages) print(f"[Filter DEBUG] Wrote debug log to: {path}") def inlet(self, body: dict, user: Optional[dict] = None) -> dict: messages: List[Dict] = body.get("messages", []) # Resolve effective valves (user override > global > defaults) user_v = getattr(user, "valves", None) if user is not None else None 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) # --- base valves ---------------------------------------------------- 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 # --- summary valves ------------------------------------------------- enable_summary = _get_user_val("enable_summary") if enable_summary is None: enable_summary = self.valves.enable_summary enable_summary = bool(enable_summary) summary_every_n_user_msgs = _get_user_val("summary_every_n_user_msgs") if summary_every_n_user_msgs is None: summary_every_n_user_msgs = self.valves.summary_every_n_user_msgs summary_every_n_user_msgs = int(summary_every_n_user_msgs or 0) summary_max_words = _get_user_val("summary_max_words") if summary_max_words is None: summary_max_words = self.valves.summary_max_words summary_max_words = int(summary_max_words or 0) # --- debug valves --------------------------------------------------- debug_enabled = _get_user_val("debug") if debug_enabled is None: debug_enabled = self.valves.debug debug_enabled = bool(debug_enabled) debug_dir = _get_user_val("debug_dir") if debug_dir is None: debug_dir = self.valves.debug_dir or "" # Keep a copy of RAW messages for debug raw_messages = list(messages) # shallow copy is fine if not messages: if debug_enabled: self._debug_dump_request_to_file(raw_messages, [], debug_dir=debug_dir) return body # --- Step 1: maybe update / insert rolling summary ------------------ messages = _update_summary_message( messages, enable_summary=enable_summary, summary_every_n_user_msgs=summary_every_n_user_msgs, summary_max_words=summary_max_words, ) # --- Step 2: clipping logic ---------------------------------------- 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 [] # If we are going to keep_first, remove that first_user/assistant from recent_ua if keep_first: if first_user in recent_ua: recent_ua = [m for m in recent_ua if m is not first_user] if first_assistant in recent_ua: recent_ua = [m for m in recent_ua if m is not first_assistant] new_messages: List[Dict] = [] # Always keep all system messages (including our summary if present) 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 and 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 new_messages = trimmed # SAFETY FALLBACK: never send an empty message list if we had UA content if not new_messages and ua_messages: last_user = None for m in reversed(ua_messages): if m.get("role") == "user": last_user = m break if last_user is not None: new_messages = [last_user] body["messages"] = new_messages # DEBUG: write RAW + FINAL for this request if debug_enabled: self._debug_dump_request_to_file( raw_messages=raw_messages, final_messages=new_messages, debug_dir=debug_dir, ) return body def outlet(self, body: dict, user: Optional[dict] = None) -> dict: return body