""" title: Full Context Mode Filter author: yurif author_url: https://github.com/yurif-freitas description: Filter to add the full file contents to conversation context version: 0.1 """ import re from typing import Self from pydantic import BaseModel, Field class TextCleaner: """ A utility class for cleaning and normalizing text content. This class provides methods to clean text by removing excessive whitespace, normalizing line breaks, and improving text formatting for better processing. Methods: replace_multiple_newlines(): Replace 3+ consecutive newlines with double newlines replace_multiple_spaces(): Replace multiple whitespace with single spaces replace_newlines_between_words(): Fix broken words spanning across lines get(): Return the cleaned text with stripped whitespace Attributes: text (str): The text content to be cleaned. """ def __init__(self, text: str): self.toggle = True self.text = text def replace_multiple_newlines(self) -> Self: self.text = re.sub(r"\n{3,}", "\n\n", self.text) return self def replace_multiple_spaces(self) -> Self: self.text = re.sub(r"\s+", " ", self.text) return self def replace_newlines_between_words(self) -> Self: self.text = re.sub(r"(\w)\n(\w)", r"\1 \2", self.text) return self def get(self) -> str: return self.text.strip() class Filter: class Valves(BaseModel): priority: int = Field( default=0, description="Priority level for the filter operations." ) def __init__(self): self.valves = self.Valves() def _clean_text(self, text: str) -> str: return ( TextCleaner(text) .replace_multiple_newlines() .replace_multiple_spaces() .replace_newlines_between_words() .get() ) def _process_file_content(self, file_info: dict) -> str | None: file_data = file_info.get("file", {}) file_content = file_data.get("data", {}).get("content") file_name = file_data.get("filename", "") if not file_content: return None cleaned_file_content = self._clean_text(file_content) return ( "<file_context>\n" f" <file_name>{file_name}</file_name>\n" " <content>\n" f"{cleaned_file_content}\n" " </content>\n" "</file_context>" ) def _create_context_message(self, files_cleaned_contents: list[str]) -> dict: combined_file_content = f"\n\n{'-' * 50}\n\n".join(files_cleaned_contents) return { "role": "user", "content": (f"<context_files>\n{combined_file_content}\n</context_files>"), } def _process_files(self, files: list[dict]) -> list[str]: files_cleaned_contents = [] for file_info in files: if file_content := self._process_file_content(file_info): files_cleaned_contents.append(file_content) return files_cleaned_contents def inlet(self, body: dict, __user__: dict | None = None) -> dict: # Early return if there are no files to process if not (files := body.get("files")): return body # Process and clean file contents files_cleaned_contents = self._process_files(files) if not files_cleaned_contents: return body # Get messages from body messages = body.get("messages", []) if not messages: return body # Add file context to conversation as the second to last user message context_message = self._create_context_message(files_cleaned_contents) messages.insert(-1, context_message) # Remove file content from body to avoid duplication (except for files in the Knowledge Base) body["files"] = [ file for file in body["files"] if not file.get("file", {}).get("data", {}).get("content") ] return body