feat: add default spliter

Author: CaralHsi

src/memos/mem_reader/read_multi_modal/file_content_parser.py

@@ -16,7 +16,7 @@
 from memos.types.openai_chat_completion_types import File
 
 from .base import BaseMessageParser, _derive_key
-from .utils import file_parser, text_splitter
+from .utils import get_parser, get_text_splitter
 
 
 logger = get_logger(__name__)
@@ -110,7 +110,7 @@ def __init__(
 
     def _split_text(self, text: str) -> list[str]:
         """
-        Split text into chunks using langchain text splitter from utils.
+        Split text into chunks using text splitter from utils.
 
         Args:
             text: Text to split
@@ -121,12 +121,13 @@ def _split_text(self, text: str) -> list[str]:
         if not text or not text.strip():
             return []
 
-        if not text_splitter:
+        splitter = get_text_splitter()
+        if not splitter:
             # If text splitter is not available, return text as single chunk
             return [text] if text.strip() else []
 
         try:
-            chunks = text_splitter.split_text(text)
+            chunks = splitter.split_text(text)
             logger.debug(f"[FileContentParser] Split text into {len(chunks)} chunks")
             return chunks
         except Exception as e:
@@ -178,7 +179,8 @@ def _parse_file(self, file_info: dict[str, Any]) -> str:
         Returns:
             Parsed text content
         """
-        if not file_parser:
+        parser = self.parser or get_parser()
+        if not parser:
             logger.warning("[FileContentParser] Parser not available")
             return ""
 
@@ -191,7 +193,7 @@ def _parse_file(self, file_info: dict[str, Any]) -> str:
 
         try:
             if os.path.exists(file_path):
-                parsed_text = file_parser.parse(file_path)
+                parsed_text = parser.parse(file_path)
                 return parsed_text
             else:
                 logger.warning(f"[FileContentParser] File not found: {file_path}")
@@ -375,7 +377,10 @@ def parse_fine(
         file_data = file_info.get("file_data", "")
         file_id = file_info.get("file_id", "")
         filename = file_info.get("filename", "")
-        if not file_parser:
+
+        # Use parser from utils
+        parser = self.parser or get_parser()
+        if not parser:
             logger.warning("[FileContentParser] Parser not available")
             return []
 
@@ -392,8 +397,7 @@ def parse_fine(
                         parsed_text, temp_file_path = self._handle_url(url_str, filename)
                         if temp_file_path:
                             try:
-                                # Use parser from utils (singleton)
-                                parser = self.parser or file_parser
+                                # Use parser from utils
                                 if parser:
                                     parsed_text = parser.parse(temp_file_path)
                                 else:

src/memos/mem_reader/read_multi_modal/utils.py

@@ -52,6 +52,49 @@
 DEFAULT_CHUNK_SIZE = int(os.getenv("FILE_PARSER_CHUNK_SIZE", "1000"))
 DEFAULT_CHUNK_OVERLAP = int(os.getenv("FILE_PARSER_CHUNK_OVERLAP", "200"))
 
+
+def _simple_split_text(text: str, chunk_size: int, chunk_overlap: int) -> list[str]:
+    """
+    Simple text splitter as fallback when langchain is not available.
+
+    Args:
+        text: Text to split
+        chunk_size: Maximum size of chunks
+        chunk_overlap: Overlap between chunks
+
+    Returns:
+        List of text chunks
+    """
+    if not text or len(text) <= chunk_size:
+        return [text] if text.strip() else []
+
+    chunks = []
+    start = 0
+    text_len = len(text)
+
+    while start < text_len:
+        # Calculate end position
+        end = min(start + chunk_size, text_len)
+
+        # If not the last chunk, try to break at a good position
+        if end < text_len:
+            # Try to break at newline, sentence end, or space
+            for separator in ["\n\n", "\n", "。", "！", "？", ". ", "! ", "? ", " "]:
+                last_sep = text.rfind(separator, start, end)
+                if last_sep != -1:
+                    end = last_sep + len(separator)
+                    break
+
+        chunk = text[start:end].strip()
+        if chunk:
+            chunks.append(chunk)
+
+        # Move start position with overlap
+        start = max(start + 1, end - chunk_overlap)
+
+    return chunks
+
+
 # Initialize parser instance
 file_parser = None
 try:
@@ -64,6 +107,8 @@
 
 # Initialize text splitter instance
 text_splitter = None
+_use_simple_splitter = False
+
 try:
     try:
         from langchain.text_splitter import RecursiveCharacterTextSplitter
@@ -82,18 +127,66 @@
         separators=["\n\n", "\n", "。", "！", "？", ". ", "! ", "? ", " ", ""],
     )
     logger.debug(
-        f"[FileContentParser] Initialized text splitter with chunk_size={DEFAULT_CHUNK_SIZE}, "
+        f"[FileContentParser] Initialized langchain text splitter with chunk_size={DEFAULT_CHUNK_SIZE}, "
         f"chunk_overlap={DEFAULT_CHUNK_OVERLAP}"
     )
 except ImportError as e:
     logger.warning(
-        f"[FileContentParser] langchain not available, text splitting will be disabled: {e}. "
+        f"[FileContentParser] langchain not available, using simple text splitter as fallback: {e}. "
         "Install with: pip install langchain or pip install langchain-text-splitters"
     )
     text_splitter = None
+    _use_simple_splitter = True
 except Exception as e:
-    logger.error(f"[FileContentParser] Failed to initialize text splitter: {e}")
+    logger.error(
+        f"[FileContentParser] Failed to initialize text splitter: {e}, using simple splitter as fallback"
+    )
     text_splitter = None
+    _use_simple_splitter = True
+
+
+def get_parser() -> Any:
+    """
+    Get parser instance.
+
+    Returns:
+        Parser instance (from ParserFactory) or None if not available
+    """
+    return file_parser
+
+
+def get_text_splitter(chunk_size: int | None = None, chunk_overlap: int | None = None) -> Any:
+    """
+    Get text splitter instance or a callable that uses simple splitter.
+
+    Args:
+        chunk_size: Maximum size of chunks when splitting text (used for simple splitter fallback)
+        chunk_overlap: Overlap between chunks when splitting text (used for simple splitter fallback)
+
+    Returns:
+        Text splitter instance (RecursiveCharacterTextSplitter) or a callable wrapper for simple splitter
+    """
+    if text_splitter is not None:
+        return text_splitter
+
+    # Return a callable wrapper that uses simple splitter
+    if _use_simple_splitter:
+        actual_chunk_size = chunk_size or DEFAULT_CHUNK_SIZE
+        actual_chunk_overlap = chunk_overlap or DEFAULT_CHUNK_OVERLAP
+
+        class SimpleTextSplitter:
+            """Simple text splitter wrapper."""
+
+            def __init__(self, chunk_size: int, chunk_overlap: int):
+                self.chunk_size = chunk_size
+                self.chunk_overlap = chunk_overlap
+
+            def split_text(self, text: str) -> list[str]:
+                return _simple_split_text(text, self.chunk_size, self.chunk_overlap)
+
+        return SimpleTextSplitter(actual_chunk_size, actual_chunk_overlap)
+
+    return None
 
 
 def extract_role(message: dict[str, Any]) -> str: