Repository: phidatahq/phidata
File: libs/agno/agno/knowledge/chunking/semantic.py (Python)

Code before:
from typing import List, Literal, Optional

from agno.knowledge.chunking.strategy import ChunkingStrategy
from agno.knowledge.document.base import Document
from agno.knowledge.embedder.base import Embedder
from agno.utils.log import log_info


def _get_chonkie_embedder_wrapper(embedder: Embedder):
    """Create a wrapper that adapts Agno Embedder to chonkie's BaseEmbeddings interface."""
    import numpy as np
    from chonkie.embeddings.base import BaseEmbeddings

    class _ChonkieEmbedderWrapper(BaseEmbeddings):
        """Wrapper to make Agno Embedders compatible with chonkie."""

        def __init__(self, agno_embedder: Embedder):
            super().__init__()
            self._embedder = agno_embedder

        def embed(self, text: str):
            embedding = self._embedder.get_embedding(text)  # type: ignore[attr-defined]
            return np.array(embedding)

        def get_tokenizer(self):
            """Return a simple token counter function."""
            return lambda text: len(text.split())

        @property
        def dimension(self) -> int:
            return getattr(self._embedder, "dimensions", 1536)

    return _ChonkieEmbedderWrapper(embedder)


class SemanticChunking(ChunkingStrategy):
    """Chunking strategy that splits text into semantic chunks using chonkie.

    Args:
        embedder: The embedder to use for generating embeddings.
        chunk_size: Maximum tokens allowed per chunk.
        similarity_threshold: Threshold for semantic similarity (0-1).
        similarity_window: Number of sentences to consider for similarity calculation.
        min_sentences_per_chunk: Minimum number of sentences per chunk.
        min_characters_per_sentence: Minimum number of characters per sentence.
        delimiters: Delimiters to use for sentence splitting.
        include_delimiters: Whether to include delimiter in prev/next sentence or None.
        skip_window: Number of groups to skip when merging (0=disabled).
        filter_window: Window length for the Savitzky-Golay filter.
        filter_polyorder: Polynomial order for the Savitzky-Golay filter.
        filter_tolerance: Tolerance for the Savitzky-Golay filter.

Review comment (suggestion):
We should allow users to provider their own  chonkie `BaseEmbeddings` class as well and use that directly. If they do, we should use that over an Agno embedder. 

Code after:
from typing import Any, Dict, List, Literal, Optional, Union

try:
    import numpy as np
except ImportError:
    raise ImportError("`numpy` not installed. Please install using `pip install numpy`")

try:
    from chonkie import SemanticChunker
    from chonkie.embeddings.base import BaseEmbeddings
except ImportError:
    raise ImportError(
        "`chonkie` is required for semantic chunking. "
        "Please install it using `pip install chonkie` to use SemanticChunking."
    )

from agno.knowledge.chunking.strategy import ChunkingStrategy
from agno.knowledge.document.base import Document
from agno.knowledge.embedder.base import Embedder
from agno.utils.log import log_info


def _get_chonkie_embedder_wrapper(embedder: Embedder):
    """Create a wrapper that adapts Agno Embedder to chonkie's BaseEmbeddings interface."""

    class _ChonkieEmbedderWrapper(BaseEmbeddings):
        """Wrapper to make Agno Embedders compatible with chonkie."""

        def __init__(self, agno_embedder: Embedder):
            super().__init__()
            self._embedder = agno_embedder

        def embed(self, text: str):
            embedding = self._embedder.get_embedding(text)  # type: ignore[attr-defined]
            return np.array(embedding, dtype=np.float32)

        def get_tokenizer(self):
            """Return a simple token counter function."""
            return lambda text: len(text.split())

        @property
        def dimension(self) -> int:
            return getattr(self._embedder, "dimensions")

    return _ChonkieEmbedderWrapper(embedder)


class SemanticChunking(ChunkingStrategy):
    """Chunking strategy that splits text into semantic chunks using chonkie.

    Args: