""" 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.3 """ 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.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." ) supported_file_extensions: list[str] = Field( default_factory=list, description="File extensions to be processed.", ) def __init__(self): self.valves = self.Valves() self.toggle = True self.icon = """data:image/svg+xml;base64,PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIGZpbGw9Im5vbmUiIHZpZXdCb3g9IjAgMCAyNCAyNCIgc3Ryb2tlPSJjdXJyZW50Q29sb3IiIHN0cm9rZS13aWR0aD0iMS41IiBzdHJva2UtbGluZWNhcD0icm91bmQiIHN0cm9rZS1saW5lam9pbj0icm91bmQiIGNsYXNzPSJzaXplLTYiPgogIDxwYXRoIGQ9Ik0zIDYuNWMyLjUtMS41IDYuNS0xLjUgOSAwVjE4Yy0yLjUtMS41LTYuNS0xLjUtOSAwVjYuNXpNMTIgNS41YzIuNS0xLjUgNi41LTEuNSA5IDBWMThjLTIuNS0xLjUtNi41LTEuNS05IDBWNWUuNSIvPgo8L3N2Zz4=""" def _clean_text(self, text: str) -> str: return ( TextCleaner(text) .replace_multiple_newlines() .replace_multiple_spaces() .replace_newlines_between_words() .get() ) def _extract_file_extension(self, file_name: str) -> str | None: return file_name.rpartition(".")[-1] if "." in file_name else None def _process_file_content(self, file_info: dict) -> tuple[str | None, 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, None # Check if file extension is in the list of extensions to process if self.valves.supported_file_extensions: file_extension = self._extract_file_extension(file_name) if file_extension not in self.valves.supported_file_extensions: return None, None cleaned_file_content = self._clean_text(file_content) formatted_file_context = ( "<file_context>\n" f" <file_name>{file_name}</file_name>\n" " <content>\n" f"{cleaned_file_content}\n" " </content>\n" "</file_context>" ) return formatted_file_context, file_name 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 = [] files_names = set() for file_info in files: file_content, file_name = self._process_file_content(file_info) if file_name is None or file_content is None or file_name in files_names: continue files_cleaned_contents.append(file_content) files_names.add(file_name) return files_cleaned_contents def _filter_out_files_with_content_field(self, files: list[dict]) -> list[dict]: """ Filter out files that have the 'content' field, which are the files that were directly imported into the chat. This function does not affect files in the Knowledge Base since they do not have the 'content' field. """ return [ file for file in files if not file.get("file", {}).get("data", {}).get("content") ] 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 from the "Knowledge" database) body["files"] = self._filter_out_files_with_content_field(body["files"]) body["metadata"]["files"] = self._filter_out_files_with_content_field( body["metadata"]["files"] ) return body