""" title: Context Length Filter author: Joey Kot version: 0.1 description: Filters messages to keep the system prompt and the last N non-system messages, optionally further truncating based on a maximum character count limit. """ from pydantic import BaseModel, Field, ValidationError from typing import Optional, List, Dict, Any import json class Filter: class Valves(BaseModel): priority: int = Field( default=0, description="Priority level for the filter operations." ) n_last_messages: int = Field( default=4, description="Number of last messages to retain.", gt=0 ) max_chars: int = Field( default=-1, description="Maximum characters to retain. (Remove messages one by one, at least one message will be kept.)", ) def __init__(self, config: Optional[Dict[str, Any]] = None): if config is None: config = {} try: self.valves = self.Valves(**config) except ValidationError as e: print( f"Warning: Invalid configuration provided for filter. Default values will be used. Error: {e}" ) self.valves = self.Valves() def _get_message_char_count(self, message: Dict[str, Any]) -> int: try: return len(json.dumps(message, ensure_ascii=False)) except (TypeError, OverflowError): content = message.get("content") if isinstance(content, str): return len(content) return 0 def _get_total_char_count(self, messages: List[Dict[str, Any]]) -> int: return sum(self._get_message_char_count(msg) for msg in messages) def inlet( self, body: Dict[str, Any], __user__: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: if not isinstance(body.get("messages"), list): return body messages: List[Dict[str, Any]] = body["messages"] n_to_keep: int = self.valves.n_last_messages max_chars: int = self.valves.max_chars if not messages: return body system_prompt: Optional[Dict[str, Any]] = None other_messages: List[Dict[str, Any]] = [] found_system = False for msg in messages: if not isinstance(msg, dict): continue if msg.get("role") == "system" and not found_system: system_prompt = msg found_system = True else: other_messages.append(msg) start_index = max(0, len(other_messages) - n_to_keep) truncated_others = other_messages[start_index:] final_messages: List[Dict[str, Any]] = [] if system_prompt: final_messages.append(system_prompt) final_messages.extend(truncated_others) if max_chars > 0: current_chars = self._get_total_char_count(final_messages) while current_chars > max_chars and len(truncated_others) > 0: removed_message = truncated_others.pop(0) final_messages = [] if system_prompt: final_messages.append(system_prompt) final_messages.extend(truncated_others) current_chars = self._get_total_char_count(final_messages) if current_chars > max_chars and system_prompt and len(final_messages) == 1: pass body["messages"] = final_messages return body