Add max_chunk_size parameter inside SemanticChunker

[kolhesamiksha]

Checked

• I searched existing ideas and did not find a similar one
• I added a very descriptive title
• I've clearly described the feature request and motivation for it

Feature request

Adding max_chunk_size Constraint to Semantic Chunking

Motivation

While implementing Semantic Chunking in my production code, I encountered cases where some chunks exceeded the input sequence length supported by my embedding model (e.g 90k chunks length). This led to truncation and inefficient processing.

To address this, I tried the code and introduced a max_chunk_size parameter to set an upper limit on chunk size, ensuring that all chunks remain within the model’s accepted sequence length.

I believe this would be a valuable addition to LangChain’s chunking utilities, allowing users to better control chunk sizes and avoid embedding model limitations

Proposal (If applicable)

Add max_chunk_size as a parameter

This function processes sentence groups while ensuring that each chunk stays within the max_chunk_size limit.

def _process_group(self, group, max_chunk_size):
    chunk_list = []
    combined_text = ""

    for d in group:
        sentence = d['sentence']
        if max_chunk_size is not None and len(combined_text) + len(sentence) + 1 > max_chunk_size:
            # Store the current chunk and start a new one
            if combined_text:
                chunk_list.append(combined_text.strip())
            combined_text = sentence  # Start a new chunk
        else:
            combined_text += " " + sentence if combined_text else sentence  # Add sentence to chunk

    if combined_text:
        chunk_list.append(combined_text.strip())  # Add last remaining chunk

    return chunk_list

def split_text(
        self,
        text: str,
    ) -> List[str]:
        # Splitting the essay (by default on '.', '?', and '!')
        single_sentences_list = self._get_single_sentences_list(text)

        # having len(single_sentences_list) == 1 would cause the following
        # np.percentile to fail.
        if len(single_sentences_list) == 1:
            return single_sentences_list
        # similarly, the following np.gradient would fail
        if (
            self.breakpoint_threshold_type == "gradient"
            and len(single_sentences_list) == 2
        ):
            return single_sentences_list
        distances, sentences = self._calculate_sentence_distances(single_sentences_list)
        if self.number_of_chunks is not None:
            breakpoint_distance_threshold = self._threshold_from_clusters(distances)
            breakpoint_array = distances
        else:
            (
                breakpoint_distance_threshold,
                breakpoint_array,
            ) = self._calculate_breakpoint_threshold(distances)

        indices_above_thresh = [
            i
            for i, x in enumerate(breakpoint_array)
            if x > breakpoint_distance_threshold
        ]

        chunks = []
        start_index = 0

        # Iterate through the breakpoints to slice the sentences
        for index in indices_above_thresh:
            # The end index is the current breakpoint
            end_index = index

            # Slice the sentence_dicts from the current start index to the end index
            group = sentences[start_index : end_index + 1]
            combined_text = " ".join([d["sentence"] for d in group])
            # If specified, merge together small chunks.
            if (
                self.max_chunk_size is not None
                and len(combined_text) > self.max_chunk_size       
            ):
                continue
            chunks.append(combined_text)

            # Update the start index for the next group
            start_index = index + 1

        # The last group, if any sentences remain
       if start_index < len(sentences):
          group = sentences[start_index:]  # Remaining sentences after last index
          remaining_chunks = self._process_group(group, max_chunk_size)  # Process and split remaining sentences

          for chunk in remaining_chunks:
              chunks.append(chunk)
        return chunks