"""Text chunking via LangChain RecursiveCharacterTextSplitter. Replaces the custom recursive splitter with the standard LangChain implementation backed by tiktoken (``cl100k_base`` encoding). The splitter receives a list of LangChain ``Document`` objects (from the parsers) and returns a new list of ``Document`` objects with split content. Each output document inherits the source document's metadata. Compared to master branch: - ✅ Less code — delegates token counting to LangChain / tiktoken - ✅ Standard interface: ``split_documents(docs)`` is the LangChain pattern - ✅ Accurate token-based chunk sizing via tiktoken - ✅ Metadata automatically propagated from parent documents """ import logging import tiktoken from langchain_core.documents import Document from langchain_text_splitters import RecursiveCharacterTextSplitter from app.config import settings logger = logging.getLogger(__name__) # Shared tiktoken encoding used for both splitting and prompt token counting _ENCODING_NAME = "cl100k_base" def count_tokens(text: str) -> int: """Count the number of tiktoken tokens in ``text``. Used by ``app.generator.prompts`` to enforce context-window budgets. Args: text: Any plain-text string. Returns: Integer token count. Example:: n = count_tokens("The property is a two-storey terrace.") """ enc = tiktoken.get_encoding(_ENCODING_NAME) return len(enc.encode(text)) def build_splitter( chunk_size: int | None = None, chunk_overlap: int | None = None, ) -> RecursiveCharacterTextSplitter: """Create a tiktoken-backed :class:`RecursiveCharacterTextSplitter`. Args: chunk_size: Token count per chunk. Defaults to ``settings.chunk_size``. chunk_overlap: Token overlap between consecutive chunks. Defaults to ``settings.chunk_overlap``. Returns: Configured :class:`RecursiveCharacterTextSplitter`. Example:: splitter = build_splitter(chunk_size=500, chunk_overlap=75) chunks = splitter.split_documents(docs) """ return RecursiveCharacterTextSplitter.from_tiktoken_encoder( encoding_name=_ENCODING_NAME, chunk_size=chunk_size or settings.chunk_size, chunk_overlap=chunk_overlap or settings.chunk_overlap, separators=["\n\n", "\n", ". ", " ", ""], ) def split_documents( documents: list[Document], chunk_size: int | None = None, chunk_overlap: int | None = None, ) -> list[Document]: """Split ``documents`` into token-bounded chunks. Each output ``Document`` inherits the metadata of its parent and has its ``page_content`` set to the chunk text. No additional metadata is added here — the caller (pipeline) is responsible for injecting ``chunk_id``, ``doc_id``, ``tenant_id``, etc. Args: documents: Input LangChain :class:`Document` objects. chunk_size: Override for ``settings.chunk_size``. chunk_overlap: Override for ``settings.chunk_overlap``. Returns: List of chunk :class:`Document` objects. Example:: docs = parse_pdf(path) chunks = split_documents(docs) # Each chunk has page_content ≤ settings.chunk_size tokens """ splitter = build_splitter(chunk_size=chunk_size, chunk_overlap=chunk_overlap) chunks = splitter.split_documents(documents) logger.debug( "Split %d document(s) into %d chunks (size=%d, overlap=%d)", len(documents), len(chunks), chunk_size or settings.chunk_size, chunk_overlap or settings.chunk_overlap, ) return chunks