Method _chunk_by_words

sdk/python/feast/chunker.py:116–150

(
        self, text: str, source_id: str, source_column: str
    )

        return str(source)

    def _chunk_by_words(
        self, text: str, source_id: str, source_column: str
    ) -> list[dict]:
        words = text.split()
        chunks = []

        step = self.config.chunk_size - self.config.chunk_overlap
        if step <= 0:
            raise ValueError(
                f"chunk_overlap ({self.config.chunk_overlap}) must be less than "
                f"chunk_size ({self.config.chunk_size})"
            )
        chunk_index = 0

        for i in range(0, len(words), step):
            chunk_words = words[i : i + self.config.chunk_size]

            if len(chunk_words) < self.config.min_chunk_size:
                continue

            chunk_text = " ".join(chunk_words)
            if self.config.max_chunk_chars:
                chunk_text = chunk_text[: self.config.max_chunk_chars]

            chunks.append(
                {
                    "chunk_id": f"{source_id}_{chunk_index}",
                    "original_id": source_id,
                    source_column: chunk_text,
                    "chunk_index": chunk_index,
                }
            )
            chunk_index += 1

        return chunks