feat: Add multiple splitting and chunking methods with KG/ontology support

- Add unified TextSplitter with method parameter support - Implement 20+ splitting methods (standard + KG/ontology) - Add entity-aware and relation-aware chunking for GraphRAG - Add graph-based, ontology-aware, and hierarchical chunking - Add configuration management and plugin registry - Add specialized KG chunkers (EntityAware, RelationAware, etc.) - Update pyproject.toml with optional dependencies - Maintain backward compatibility with existing chunkers
2026-08-29 04:26:20 +00:00 · 2025-11-14 23:03:41 +05:30
parent 0a1821d586
commit 3c31475fdc
7 changed files with 2542 additions and 3 deletions
@@ -163,6 +163,19 @@ models-huggingface = [
    "transformers>=4.20.0",
    "torch>=1.12.0"
 ]
+split-tiktoken = [
+    "tiktoken>=0.5.0"
+]
+split-community = [
+    "python-louvain>=0.16"
+]
+split-topic = [
+    "bertopic>=0.15.0",
+    "gensim>=4.3.0"
+]
+split-all = [
+    "semantica[split-tiktoken,split-community,split-topic]"
+]

 [project.urls]
 Homepage = "https://semantica.dev"
@@ -5,7 +5,30 @@ This module provides comprehensive document chunking and splitting capabilities
 for optimal processing and semantic analysis, enabling efficient handling of
 large documents through various chunking strategies.

+Supported Methods:
+    - Standard: recursive, token, sentence, paragraph, character, word, semantic_transformer, llm, huggingface, nltk
+    - KG/Ontology: entity_aware, relation_aware, graph_based, ontology_aware, hierarchical, community_detection, centrality_based, subgraph, topic_based
+
+Algorithms Used:
+    - Recursive Splitting: Separator hierarchy and greedy splitting
+    - Token Counting: BPE tokenization (tiktoken, transformers)
+    - Sentence Segmentation: NLTK, spaCy, regex-based
+    - Semantic Boundary Detection: Sentence transformer embeddings and similarity
+    - LLM-based Splitting: Prompt engineering for optimal split point detection
+    - Entity Boundary Detection: NER-based entity extraction and boundary preservation
+    - Triple Preservation: Graph-based triple integrity checking
+    - Graph Centrality Analysis: Degree, betweenness, closeness, eigenvector centrality
+    - Community Detection: Louvain algorithm, Leiden algorithm, modularity optimization
+
 Key Features:
+    - Multiple standard splitting methods
+    - KG/ontology/graph analytics-specific chunking methods
+    - Unified TextSplitter interface
+    - Entity-aware chunking for GraphRAG
+    - Relation-aware chunking for KG workflows
+    - Graph structure-based chunking
+    - Ontology concept-aware chunking
+    - Hierarchical multi-level chunking
    - Semantic-based chunking using NLP
    - Structure-aware chunking (headings, paragraphs, lists)
    - Sliding window chunking with overlap
@@ -14,21 +37,33 @@ Key Features:
    - Provenance tracking for data lineage

 Main Classes:
+    - TextSplitter: Unified text splitter with method parameter
    - SemanticChunker: Semantic-based chunking coordinator
    - StructuralChunker: Structure-aware chunking
    - SlidingWindowChunker: Fixed-size sliding window chunking
    - TableChunker: Table-specific chunking
+    - EntityAwareChunker: Entity boundary-preserving chunker
+    - RelationAwareChunker: Triple-preserving chunker
+    - GraphBasedChunker: Graph structure-based chunker
+    - OntologyAwareChunker: Ontology concept-based chunker
+    - HierarchicalChunker: Multi-level hierarchical chunker
    - ChunkValidator: Chunk quality validation
    - ProvenanceTracker: Chunk provenance tracking
    - Chunk: Chunk representation dataclass

 Example Usage:
+    >>> from semantica.split import TextSplitter
+    >>> splitter = TextSplitter(method="recursive", chunk_size=1000, chunk_overlap=200)
+    >>> chunks = splitter.split(text)
+    >>> 
+    >>> # Entity-aware for GraphRAG
+    >>> splitter = TextSplitter(method="entity_aware", ner_method="llm", chunk_size=1000)
+    >>> chunks = splitter.split(text)
+    >>> 
+    >>> # Using existing chunkers
    >>> from semantica.split import SemanticChunker
    >>> chunker = SemanticChunker(chunk_size=1000, chunk_overlap=200)
    >>> chunks = chunker.chunk(long_text)
-    >>> from semantica.split import StructuralChunker
-    >>> struct_chunker = StructuralChunker()
-    >>> chunks = struct_chunker.chunk(structured_document)

 Author: Semantica Contributors
 License: MIT
@@ -40,8 +75,39 @@ from .sliding_window_chunker import SlidingWindowChunker
 from .table_chunker import TableChunker
 from .chunk_validator import ChunkValidator
 from .provenance_tracker import ProvenanceTracker
+from .splitter import TextSplitter
+from .kg_chunkers import (
+    EntityAwareChunker,
+    RelationAwareChunker,
+    GraphBasedChunker,
+    OntologyAwareChunker,
+    HierarchicalChunker
+)
+from .methods import (
+    get_split_method,
+    list_available_methods,
+    split_recursive,
+    split_by_tokens,
+    split_by_sentences,
+    split_by_paragraphs,
+    split_by_characters,
+    split_by_words,
+    split_semantic_transformer,
+    split_llm,
+    split_entity_aware,
+    split_relation_aware,
+    split_graph_based,
+    split_ontology_aware,
+    split_hierarchical
+)
+from .config import SplitConfig, split_config
+from .registry import MethodRegistry, method_registry

 __all__ = [
+    # Unified splitter
+    "TextSplitter",
+    
+    # Existing chunkers
    "SemanticChunker",
    "Chunk",
    "StructuralChunker",
@@ -49,4 +115,34 @@ __all__ = [
    "TableChunker",
    "ChunkValidator",
    "ProvenanceTracker",
+    
+    # KG/Ontology chunkers
+    "EntityAwareChunker",
+    "RelationAwareChunker",
+    "GraphBasedChunker",
+    "OntologyAwareChunker",
+    "HierarchicalChunker",
+    
+    # Methods
+    "get_split_method",
+    "list_available_methods",
+    "split_recursive",
+    "split_by_tokens",
+    "split_by_sentences",
+    "split_by_paragraphs",
+    "split_by_characters",
+    "split_by_words",
+    "split_semantic_transformer",
+    "split_llm",
+    "split_entity_aware",
+    "split_relation_aware",
+    "split_graph_based",
+    "split_ontology_aware",
+    "split_hierarchical",
+    
+    # Config and Registry
+    "SplitConfig",
+    "split_config",
+    "MethodRegistry",
+    "method_registry",
 ]
@@ -0,0 +1,153 @@
+"""
+Configuration Management Module for Split
+
+This module provides centralized configuration management for text splitting and chunking,
+supporting multiple configuration sources including environment variables, config files,
+and programmatic configuration.
+
+Supported Configuration Sources:
+    - Environment variables: SPLIT_CHUNK_SIZE, SPLIT_CHUNK_OVERLAP, etc.
+    - Config files: YAML, JSON, TOML formats
+    - Programmatic: Python API for setting split configurations
+
+Algorithms Used:
+    - Environment Variable Parsing: OS-level environment variable access
+    - YAML Parsing: YAML parser for configuration file loading
+    - JSON Parsing: JSON parser for configuration file loading
+    - TOML Parsing: TOML parser for configuration file loading
+    - Fallback Chain: Priority-based configuration resolution
+    - Dictionary Merging: Deep merge algorithms for configuration updates
+
+Key Features:
+    - Environment variable support for split parameters
+    - Config file support (YAML, JSON, TOML formats)
+    - Programmatic configuration via Python API
+    - Method-specific configuration management
+    - Automatic fallback chain (config file -> environment -> defaults)
+    - Global config instance for easy access
+
+Main Classes:
+    - SplitConfig: Main configuration manager class for split module
+
+Example Usage:
+    >>> from semantica.split.config import split_config
+    >>> chunk_size = split_config.get("chunk_size", default=1000)
+    >>> split_config.set("chunk_size", 2000)
+    >>> method_config = split_config.get_method_config("recursive")
+
+Author: Semantica Contributors
+License: MIT
+"""
+
+import os
+from typing import Optional, Dict, Any
+from pathlib import Path
+
+from ..utils.logging import get_logger
+
+
+class SplitConfig:
+    """Configuration manager for split module - supports .env files, environment variables, and programmatic config."""
+    
+    def __init__(self, config_file: Optional[str] = None):
+        """Initialize configuration manager."""
+        self.logger = get_logger("split_config")
+        self._configs: Dict[str, Any] = {}
+        self._method_configs: Dict[str, Dict] = {}
+        self._load_config_file(config_file)
+        self._load_env_vars()
+    
+    def _load_config_file(self, config_file: Optional[str]):
+        """Load configuration from file."""
+        if config_file and Path(config_file).exists():
+            try:
+                # Support YAML, JSON, TOML
+                if config_file.endswith('.yaml') or config_file.endswith('.yml'):
+                    import yaml
+                    with open(config_file, 'r') as f:
+                        data = yaml.safe_load(f) or {}
+                        self._configs.update(data.get("split", {}))
+                        self._method_configs.update(data.get("split_methods", {}))
+                elif config_file.endswith('.json'):
+                    import json
+                    with open(config_file, 'r') as f:
+                        data = json.load(f) or {}
+                        self._configs.update(data.get("split", {}))
+                        self._method_configs.update(data.get("split_methods", {}))
+                elif config_file.endswith('.toml'):
+                    import toml
+                    with open(config_file, 'r') as f:
+                        data = toml.load(f) or {}
+                        if "split" in data:
+                            self._configs.update(data["split"])
+                        if "split_methods" in data:
+                            self._method_configs.update(data["split_methods"])
+            except Exception as e:
+                self.logger.warning(f"Failed to load config file {config_file}: {e}")
+    
+    def _load_env_vars(self):
+        """Load configuration from environment variables."""
+        # Common environment variable patterns for split module
+        env_mappings = {
+            "SPLIT_CHUNK_SIZE": ("chunk_size", int),
+            "SPLIT_CHUNK_OVERLAP": ("chunk_overlap", int),
+            "SPLIT_DEFAULT_METHOD": ("default_method", str),
+            "SPLIT_MAX_CHUNK_SIZE": ("max_chunk_size", int),
+            "SPLIT_MIN_CHUNK_SIZE": ("min_chunk_size", int),
+        }
+        
+        for env_key, (config_key, type_func) in env_mappings.items():
+            value = os.getenv(env_key)
+            if value:
+                try:
+                    self._configs[config_key] = type_func(value)
+                except (ValueError, TypeError):
+                    self.logger.warning(f"Failed to parse {env_key}={value}")
+    
+    def set(self, key: str, value: Any):
+        """Set configuration value programmatically."""
+        self._configs[key] = value
+    
+    def get(self, key: str, default: Any = None) -> Any:
+        """Get configuration value with fallback chain: config -> env -> default."""
+        # Check config first
+        if key in self._configs:
+            return self._configs[key]
+        
+        # Check environment variables
+        env_key = f"SPLIT_{key.upper()}"
+        value = os.getenv(env_key)
+        if value:
+            try:
+                # Try to convert to appropriate type
+                if isinstance(default, int):
+                    return int(value)
+                elif isinstance(default, float):
+                    return float(value)
+                elif isinstance(default, bool):
+                    return value.lower() in ("true", "1", "yes", "on")
+                return value
+            except (ValueError, TypeError):
+                pass
+        
+        return default
+    
+    def set_method_config(self, method: str, **config):
+        """Set method-specific configuration."""
+        self._method_configs[method] = config
+    
+    def get_method_config(self, method: str) -> Dict:
+        """Get method-specific configuration."""
+        return self._method_configs.get(method, {})
+    
+    def get_all(self) -> Dict[str, Any]:
+        """Get all configuration."""
+        return {
+            "config": self._configs.copy(),
+            "method_configs": self._method_configs.copy()
+        }
+
+
+# Global config instance
+split_config = SplitConfig()
+
@@ -0,0 +1,416 @@
+"""
+KG/Ontology-Specific Chunkers Module
+
+This module provides specialized chunkers for knowledge graph, ontology, and graph
+analytics workflows, designed to preserve graph structure and semantic relationships.
+
+Supported Chunkers:
+    - EntityAwareChunker: Preserves entity boundaries
+    - RelationAwareChunker: Preserves triple integrity
+    - GraphBasedChunker: Uses graph structure for chunking
+    - OntologyAwareChunker: Uses ontology concepts for chunking
+    - HierarchicalChunker: Multi-level hierarchical chunking
+
+Algorithms Used:
+    - Entity Boundary Detection: NER-based entity extraction and boundary preservation
+    - Triple Preservation: Graph-based triple integrity checking
+    - Graph Centrality Analysis: Degree, betweenness, closeness, eigenvector centrality
+    - Community Detection: Louvain algorithm, Leiden algorithm, modularity optimization
+    - Graph Connectivity: Connected components, shortest paths, bridge detection
+    - Ontology Hierarchy Traversal: Taxonomic structure traversal and concept grouping
+
+Key Features:
+    - Entity boundary preservation for GraphRAG
+    - Triple integrity preservation for KG workflows
+    - Graph structure-aware chunking
+    - Ontology concept-aware chunking
+    - Hierarchical multi-level chunking
+    - Integration with semantic extraction module
+
+Main Classes:
+    - EntityAwareChunker: Entity boundary-preserving chunker
+    - RelationAwareChunker: Triple-preserving chunker
+    - GraphBasedChunker: Graph structure-based chunker
+    - OntologyAwareChunker: Ontology concept-based chunker
+    - HierarchicalChunker: Multi-level hierarchical chunker
+
+Example Usage:
+    >>> from semantica.split.kg_chunkers import EntityAwareChunker
+    >>> chunker = EntityAwareChunker(chunk_size=1000, ner_method="llm")
+    >>> chunks = chunker.chunk(text)
+
+Author: Semantica Contributors
+License: MIT
+"""
+
+from typing import List, Dict, Any, Optional, Union
+from .semantic_chunker import Chunk
+from .methods import (
+    split_entity_aware,
+    split_relation_aware,
+    split_graph_based,
+    split_ontology_aware,
+    split_hierarchical
+)
+from ..utils.logging import get_logger
+from ..utils.progress_tracker import get_progress_tracker
+
+logger = get_logger("kg_chunkers")
+
+
+class EntityAwareChunker:
+    """
+    Entity boundary-preserving chunker for GraphRAG workflows.
+    
+    Ensures that entities and their associated information are kept together,
+    preserving the semantic integrity necessary for accurate graph-based retrieval.
+    """
+    
+    def __init__(
+        self,
+        chunk_size: int = 1000,
+        chunk_overlap: int = 200,
+        ner_method: str = "ml",
+        preserve_entities: bool = True,
+        **kwargs
+    ):
+        """
+        Initialize entity-aware chunker.
+        
+        Args:
+            chunk_size: Target chunk size
+            chunk_overlap: Overlap between chunks
+            ner_method: NER method to use ("pattern", "regex", "ml", "huggingface", "llm")
+            preserve_entities: Whether to preserve entity boundaries
+            **kwargs: Additional options for NER extractor
+        """
+        self.chunk_size = chunk_size
+        self.chunk_overlap = chunk_overlap
+        self.ner_method = ner_method
+        self.preserve_entities = preserve_entities
+        self.options = kwargs
+        self.logger = get_logger("entity_aware_chunker")
+        self.progress_tracker = get_progress_tracker()
+    
+    def chunk(self, text: str, **options) -> List[Chunk]:
+        """
+        Chunk text preserving entity boundaries.
+        
+        Args:
+            text: Input text
+            **options: Additional options
+            
+        Returns:
+            List of chunks
+        """
+        tracking_id = self.progress_tracker.start_tracking(
+            module="split",
+            submodule="EntityAwareChunker",
+            message="Chunking text with entity awareness"
+        )
+        
+        try:
+            merged_options = {**self.options, **options}
+            chunks = split_entity_aware(
+                text,
+                chunk_size=self.chunk_size,
+                ner_method=self.ner_method,
+                preserve_entities=self.preserve_entities,
+                **merged_options
+            )
+            
+            self.progress_tracker.stop_tracking(
+                tracking_id,
+                status="completed",
+                message=f"Created {len(chunks)} entity-aware chunks"
+            )
+            return chunks
+            
+        except Exception as e:
+            self.progress_tracker.stop_tracking(tracking_id, status="failed", message=str(e))
+            raise
+
+
+class RelationAwareChunker:
+    """
+    Triple-preserving chunker for KG workflows.
+    
+    Ensures that relation triples (subject-predicate-object) are preserved within
+    the same chunk, preventing the loss of relational context.
+    """
+    
+    def __init__(
+        self,
+        chunk_size: int = 1000,
+        chunk_overlap: int = 200,
+        relation_method: str = "ml",
+        preserve_triples: bool = True,
+        **kwargs
+    ):
+        """
+        Initialize relation-aware chunker.
+        
+        Args:
+            chunk_size: Target chunk size
+            chunk_overlap: Overlap between chunks
+            relation_method: Relation extraction method
+            preserve_triples: Whether to preserve triple integrity
+            **kwargs: Additional options for relation extractor
+        """
+        self.chunk_size = chunk_size
+        self.chunk_overlap = chunk_overlap
+        self.relation_method = relation_method
+        self.preserve_triples = preserve_triples
+        self.options = kwargs
+        self.logger = get_logger("relation_aware_chunker")
+        self.progress_tracker = get_progress_tracker()
+    
+    def chunk(self, text: str, **options) -> List[Chunk]:
+        """
+        Chunk text preserving triple integrity.
+        
+        Args:
+            text: Input text
+            **options: Additional options
+            
+        Returns:
+            List of chunks
+        """
+        tracking_id = self.progress_tracker.start_tracking(
+            module="split",
+            submodule="RelationAwareChunker",
+            message="Chunking text with relation awareness"
+        )
+        
+        try:
+            merged_options = {**self.options, **options}
+            chunks = split_relation_aware(
+                text,
+                chunk_size=self.chunk_size,
+                relation_method=self.relation_method,
+                preserve_triples=self.preserve_triples,
+                **merged_options
+            )
+            
+            self.progress_tracker.stop_tracking(
+                tracking_id,
+                status="completed",
+                message=f"Created {len(chunks)} relation-aware chunks"
+            )
+            return chunks
+            
+        except Exception as e:
+            self.progress_tracker.stop_tracking(tracking_id, status="failed", message=str(e))
+            raise
+
+
+class GraphBasedChunker:
+    """
+    Graph structure-based chunker using centrality or communities.
+    
+    Uses graph analysis (centrality measures, community detection) to determine
+    optimal chunk boundaries based on the underlying knowledge graph structure.
+    """
+    
+    def __init__(
+        self,
+        chunk_size: int = 1000,
+        strategy: str = "community",
+        algorithm: str = "louvain",
+        **kwargs
+    ):
+        """
+        Initialize graph-based chunker.
+        
+        Args:
+            chunk_size: Target chunk size
+            strategy: Strategy ("community", "centrality")
+            algorithm: Algorithm name ("louvain", "leiden", "betweenness", etc.)
+            **kwargs: Additional options
+        """
+        self.chunk_size = chunk_size
+        self.strategy = strategy
+        self.algorithm = algorithm
+        self.options = kwargs
+        self.logger = get_logger("graph_based_chunker")
+        self.progress_tracker = get_progress_tracker()
+    
+    def chunk(self, text: str, **options) -> List[Chunk]:
+        """
+        Chunk text using graph structure.
+        
+        Args:
+            text: Input text
+            **options: Additional options
+            
+        Returns:
+            List of chunks
+        """
+        tracking_id = self.progress_tracker.start_tracking(
+            module="split",
+            submodule="GraphBasedChunker",
+            message="Chunking text using graph structure"
+        )
+        
+        try:
+            merged_options = {**self.options, **options}
+            chunks = split_graph_based(
+                text,
+                chunk_size=self.chunk_size,
+                strategy=self.strategy,
+                algorithm=self.algorithm,
+                **merged_options
+            )
+            
+            self.progress_tracker.stop_tracking(
+                tracking_id,
+                status="completed",
+                message=f"Created {len(chunks)} graph-based chunks"
+            )
+            return chunks
+            
+        except Exception as e:
+            self.progress_tracker.stop_tracking(tracking_id, status="failed", message=str(e))
+            raise
+
+
+class OntologyAwareChunker:
+    """
+    Ontology concept and hierarchy-based chunker.
+    
+    Chunks text based on ontology concepts, hierarchies, and taxonomic structures,
+    ensuring alignment with domain-specific concepts and terminologies.
+    """
+    
+    def __init__(
+        self,
+        chunk_size: int = 1000,
+        ontology_uri: Optional[str] = None,
+        preserve_concepts: bool = True,
+        **kwargs
+    ):
+        """
+        Initialize ontology-aware chunker.
+        
+        Args:
+            chunk_size: Target chunk size
+            ontology_uri: Ontology URI (optional)
+            preserve_concepts: Whether to preserve concept boundaries
+            **kwargs: Additional options
+        """
+        self.chunk_size = chunk_size
+        self.ontology_uri = ontology_uri
+        self.preserve_concepts = preserve_concepts
+        self.options = kwargs
+        self.logger = get_logger("ontology_aware_chunker")
+        self.progress_tracker = get_progress_tracker()
+    
+    def chunk(self, text: str, **options) -> List[Chunk]:
+        """
+        Chunk text using ontology concepts.
+        
+        Args:
+            text: Input text
+            **options: Additional options
+            
+        Returns:
+            List of chunks
+        """
+        tracking_id = self.progress_tracker.start_tracking(
+            module="split",
+            submodule="OntologyAwareChunker",
+            message="Chunking text using ontology concepts"
+        )
+        
+        try:
+            merged_options = {**self.options, **options}
+            chunks = split_ontology_aware(
+                text,
+                chunk_size=self.chunk_size,
+                ontology_uri=self.ontology_uri,
+                preserve_concepts=self.preserve_concepts,
+                **merged_options
+            )
+            
+            self.progress_tracker.stop_tracking(
+                tracking_id,
+                status="completed",
+                message=f"Created {len(chunks)} ontology-aware chunks"
+            )
+            return chunks
+            
+        except Exception as e:
+            self.progress_tracker.stop_tracking(tracking_id, status="failed", message=str(e))
+            raise
+
+
+class HierarchicalChunker:
+    """
+    Multi-level hierarchical chunker.
+    
+    Creates multiple layers of chunks, from fine-grained (sentences) to coarse-grained
+    (sections), allowing for retrieval at various levels of granularity.
+    """
+    
+    def __init__(
+        self,
+        levels: List[str] = ["section", "paragraph", "sentence"],
+        chunk_sizes: Optional[List[int]] = None,
+        **kwargs
+    ):
+        """
+        Initialize hierarchical chunker.
+        
+        Args:
+            levels: Hierarchy levels (e.g., ["section", "paragraph", "sentence"])
+            chunk_sizes: Chunk sizes for each level
+            **kwargs: Additional options
+        """
+        self.levels = levels
+        self.chunk_sizes = chunk_sizes or [2000, 1000, 500]
+        self.options = kwargs
+        self.logger = get_logger("hierarchical_chunker")
+        self.progress_tracker = get_progress_tracker()
+    
+    def chunk(self, text: str, **options) -> List[Chunk]:
+        """
+        Chunk text hierarchically.
+        
+        Args:
+            text: Input text
+            **options: Additional options
+            
+        Returns:
+            List of chunks with hierarchical metadata
+        """
+        tracking_id = self.progress_tracker.start_tracking(
+            module="split",
+            submodule="HierarchicalChunker",
+            message="Chunking text hierarchically"
+        )
+        
+        try:
+            merged_options = {**self.options, **options}
+            chunks = split_hierarchical(
+                text,
+                levels=self.levels,
+                chunk_sizes=self.chunk_sizes,
+                **merged_options
+            )
+            
+            # Add hierarchical metadata
+            for chunk in chunks:
+                chunk.metadata["hierarchical"] = True
+                chunk.metadata["levels"] = self.levels
+            
+            self.progress_tracker.stop_tracking(
+                tracking_id,
+                status="completed",
+                message=f"Created {len(chunks)} hierarchical chunks"
+            )
+            return chunks
+            
+        except Exception as e:
+            self.progress_tracker.stop_tracking(tracking_id, status="failed", message=str(e))
+            raise
+
@@ -0,0 +1,123 @@
+"""
+Plugin Registry Module for Split
+
+This module provides a plugin registry system for registering custom splitting methods,
+enabling extensibility and community contributions to the text splitting toolkit.
+
+Supported Registration Types:
+    - Method Registry: Register custom splitting methods for:
+        * "split": Text splitting methods
+
+Algorithms Used:
+    - Registry Pattern: Dictionary-based registration and lookup
+    - Dynamic Registration: Runtime function registration
+    - Type Checking: Type validation for registered components
+    - Lookup Algorithms: Hash-based O(1) lookup for methods
+    - Task-based Organization: Hierarchical organization by task type
+
+Key Features:
+    - Method registry for custom splitting methods
+    - Task-based method organization (split)
+    - Dynamic registration and unregistration
+    - Easy discovery of available methods
+    - Support for community-contributed extensions
+
+Main Classes:
+    - MethodRegistry: Registry for custom splitting methods
+
+Global Instances:
+    - method_registry: Global method registry instance
+
+Example Usage:
+    >>> from semantica.split.registry import method_registry
+    >>> method_registry.register("split", "custom_method", custom_split_function)
+    >>> available = method_registry.list_all("split")
+
+Author: Semantica Contributors
+License: MIT
+"""
+
+from typing import Dict, Callable, Any, List, Optional
+
+
+class MethodRegistry:
+    """Registry for custom splitting methods."""
+    
+    _methods: Dict[str, Dict[str, Callable]] = {
+        "split": {},
+    }
+    
+    @classmethod
+    def register(cls, task: str, name: str, method_func: Callable):
+        """
+        Register a custom splitting method.
+        
+        Args:
+            task: Task type ("split")
+            name: Method name
+            method_func: Method function
+        """
+        if task not in cls._methods:
+            cls._methods[task] = {}
+        cls._methods[task][name] = method_func
+    
+    @classmethod
+    def get(cls, task: str, name: str) -> Optional[Callable]:
+        """
+        Get method by task and name.
+        
+        Args:
+            task: Task type ("split")
+            name: Method name
+            
+        Returns:
+            Method function or None
+        """
+        return cls._methods.get(task, {}).get(name)
+    
+    @classmethod
+    def list_all(cls, task: Optional[str] = None) -> Dict[str, List[str]]:
+        """
+        List all registered methods.
+        
+        Args:
+            task: Optional task type to filter by
+            
+        Returns:
+            Dictionary mapping task types to method names
+        """
+        if task:
+            return {task: list(cls._methods.get(task, {}).keys())}
+        return {t: list(m.keys()) for t, m in cls._methods.items()}
+    
+    @classmethod
+    def unregister(cls, task: str, name: str):
+        """
+        Unregister a method.
+        
+        Args:
+            task: Task type ("split")
+            name: Method name
+        """
+        if task in cls._methods and name in cls._methods[task]:
+            del cls._methods[task][name]
+    
+    @classmethod
+    def clear(cls, task: Optional[str] = None):
+        """
+        Clear all registered methods for a task or all tasks.
+        
+        Args:
+            task: Optional task type to clear (clears all if None)
+        """
+        if task:
+            if task in cls._methods:
+                cls._methods[task].clear()
+        else:
+            for task_dict in cls._methods.values():
+                task_dict.clear()
+
+
+# Global registry
+method_registry = MethodRegistry()
+
@@ -0,0 +1,216 @@
+"""
+Unified Text Splitter Module
+
+This module provides a unified interface for all text splitting and chunking methods,
+enabling easy switching between different splitting strategies with a consistent API.
+
+Supported Methods:
+    - All methods from methods.py (recursive, token, sentence, paragraph, etc.)
+    - All KG/ontology methods (entity_aware, relation_aware, graph_based, etc.)
+    - Integration with existing chunkers (SemanticChunker, StructuralChunker, etc.)
+
+Algorithms Used:
+    - Strategy Pattern: Method selection and delegation
+    - Factory Pattern: Unified creation of appropriate splitter
+    - Fallback Chain: Automatic fallback to alternative methods
+    - Adapter Pattern: Integration with existing chunker classes
+
+Key Features:
+    - Unified interface for all splitting methods
+    - Method parameter support (string or list for fallback chain)
+    - Integration with existing chunkers
+    - Backward compatibility with existing API
+    - Automatic method fallback
+    - Consistent Chunk output format
+
+Main Classes:
+    - TextSplitter: Unified text splitter with method parameter
+
+Example Usage:
+    >>> from semantica.split import TextSplitter
+    >>> splitter = TextSplitter(method="recursive", chunk_size=1000, chunk_overlap=200)
+    >>> chunks = splitter.split(text)
+    >>> 
+    >>> # Entity-aware for GraphRAG
+    >>> splitter = TextSplitter(method="entity_aware", ner_method="llm", chunk_size=1000)
+    >>> chunks = splitter.split(text)
+
+Author: Semantica Contributors
+License: MIT
+"""
+
+from typing import List, Optional, Union, Dict, Any
+from .semantic_chunker import Chunk
+from .methods import get_split_method, list_available_methods
+from .config import split_config
+from ..utils.logging import get_logger
+from ..utils.exceptions import ProcessingError
+
+logger = get_logger("text_splitter")
+
+
+class TextSplitter:
+    """
+    Unified text splitter with support for multiple splitting methods.
+    
+    This class provides a single interface for all splitting methods, allowing
+    easy switching between different strategies while maintaining backward
+    compatibility with existing chunkers.
+    """
+    
+    def __init__(
+        self,
+        method: Union[str, List[str]] = "recursive",
+        chunk_size: int = 1000,
+        chunk_overlap: int = 200,
+        **kwargs
+    ):
+        """
+        Initialize text splitter.
+        
+        Args:
+            method: Splitting method name or list of methods for fallback chain
+            chunk_size: Target chunk size in characters
+            chunk_overlap: Overlap between chunks in characters
+            **kwargs: Additional method-specific options:
+                - ner_method: NER method for entity_aware splitting
+                - relation_method: Relation extraction method for relation_aware
+                - provider: LLM provider for llm-based methods
+                - model: Model name for LLM or transformer methods
+                - tokenizer: Tokenizer name for token-based splitting
+                - separators: Separator list for recursive splitting
+                - strategy: Strategy for graph_based splitting
+                - algorithm: Algorithm name for graph/community methods
+        """
+        self.logger = get_logger("text_splitter")
+        
+        # Normalize method parameter
+        if isinstance(method, str):
+            self.methods = [method]
+        else:
+            self.methods = method if isinstance(method, list) else ["recursive"]
+        
+        # Set default parameters
+        self.chunk_size = chunk_size or split_config.get("chunk_size", 1000)
+        self.chunk_overlap = chunk_overlap or split_config.get("chunk_overlap", 200)
+        
+        # Store additional options
+        self.options = kwargs
+        
+        # Load method-specific config
+        self._load_method_configs()
+    
+    def _load_method_configs(self):
+        """Load method-specific configurations from config."""
+        for method in self.methods:
+            method_config = split_config.get_method_config(method)
+            if method_config:
+                # Merge method config into options (method config takes precedence)
+                for key, value in method_config.items():
+                    if key not in self.options:
+                        self.options[key] = value
+    
+    def split(self, text: str, **override_options) -> List[Chunk]:
+        """
+        Split text into chunks using the specified method(s).
+        
+        Args:
+            text: Input text to split
+            **override_options: Options to override for this split call
+            
+        Returns:
+            List of Chunk objects
+            
+        Raises:
+            ProcessingError: If all methods fail
+        """
+        if not text:
+            return []
+        
+        # Merge override options
+        options = {**self.options, **override_options}
+        options["chunk_size"] = options.get("chunk_size", self.chunk_size)
+        options["chunk_overlap"] = options.get("chunk_overlap", self.chunk_overlap)
+        
+        # Try each method in fallback chain
+        last_error = None
+        for method_name in self.methods:
+            try:
+                self.logger.debug(f"Attempting to split using method: {method_name}")
+                
+                # Get method function
+                method_func = get_split_method(method_name)
+                if not method_func:
+                    self.logger.warning(f"Method '{method_name}' not found, trying next method")
+                    continue
+                
+                # Call method with text and options
+                chunks = method_func(text, **options)
+                
+                if chunks:
+                    self.logger.info(f"Successfully split text into {len(chunks)} chunks using method: {method_name}")
+                    return chunks
+                else:
+                    self.logger.warning(f"Method '{method_name}' returned no chunks, trying next method")
+                    
+            except Exception as e:
+                self.logger.warning(f"Method '{method_name}' failed: {e}, trying next method")
+                last_error = e
+                continue
+        
+        # All methods failed
+        error_msg = f"All splitting methods failed: {', '.join(self.methods)}"
+        if last_error:
+            error_msg += f". Last error: {last_error}"
+        
+        raise ProcessingError(error_msg)
+    
+    def split_batch(self, texts: List[str], **override_options) -> List[List[Chunk]]:
+        """
+        Split multiple texts into chunks.
+        
+        Args:
+            texts: List of input texts
+            **override_options: Options to override for this split call
+            
+        Returns:
+            List of chunk lists (one per input text)
+        """
+        results = []
+        for text in texts:
+            chunks = self.split(text, **override_options)
+            results.append(chunks)
+        return results
+    
+    def get_available_methods(self) -> List[str]:
+        """
+        Get list of available splitting methods.
+        
+        Returns:
+            List of method names
+        """
+        return list_available_methods()
+    
+    def set_method(self, method: Union[str, List[str]]):
+        """
+        Change the splitting method(s).
+        
+        Args:
+            method: Method name or list of methods for fallback chain
+        """
+        if isinstance(method, str):
+            self.methods = [method]
+        else:
+            self.methods = method if isinstance(method, list) else ["recursive"]
+        self._load_method_configs()
+    
+    def update_options(self, **options):
+        """
+        Update splitting options.
+        
+        Args:
+            **options: Options to update
+        """
+        self.options.update(options)
+        self._load_method_configs()
+