RecursiveJsonSplitter Class — langchain Architecture

Source Code

libs/text-splitters/langchain_text_splitters/json.py lines 12–190

class RecursiveJsonSplitter:
    """Splits JSON data into smaller, structured chunks while preserving hierarchy.

    This class provides methods to split JSON data into smaller dictionaries or
    JSON-formatted strings based on configurable maximum and minimum chunk sizes.
    It supports nested JSON structures, optionally converts lists into dictionaries
    for better chunking, and allows the creation of document objects for further use.
    """

    max_chunk_size: int = 2000
    """The maximum size for each chunk."""

    min_chunk_size: int = 1800
    """The minimum size for each chunk, derived from `max_chunk_size` if not
    explicitly provided.
    """

    def __init__(
        self, max_chunk_size: int = 2000, min_chunk_size: int | None = None
    ) -> None:
        """Initialize the chunk size configuration for text processing.

        This constructor sets up the maximum and minimum chunk sizes, ensuring that
        the `min_chunk_size` defaults to a value slightly smaller than the
        `max_chunk_size` if not explicitly provided.

        Args:
            max_chunk_size: The maximum size for a chunk.
            min_chunk_size: The minimum size for a chunk.

                If `None`, defaults to the maximum chunk size minus 200, with a lower
                bound of 50.
        """
        super().__init__()
        self.max_chunk_size = max_chunk_size
        self.min_chunk_size = (
            min_chunk_size
            if min_chunk_size is not None
            else max(max_chunk_size - 200, 50)
        )

    @staticmethod
    def _json_size(data: dict[str, Any]) -> int:
        """Calculate the size of the serialized JSON object."""
        return len(json.dumps(data))

    @staticmethod
    def _set_nested_dict(
        d: dict[str, Any],
        path: list[str],
        value: Any,  # noqa: ANN401
    ) -> None:
        """Set a value in a nested dictionary based on the given path."""
        for key in path[:-1]:
            d = d.setdefault(key, {})
        d[path[-1]] = value

    def _list_to_dict_preprocessing(
        self,
        data: Any,  # noqa: ANN401
    ) -> Any:  # noqa: ANN401
        if isinstance(data, dict):
            # Process each key-value pair in the dictionary
            return {k: self._list_to_dict_preprocessing(v) for k, v in data.items()}
        if isinstance(data, list):
            # Convert the list to a dictionary with index-based keys
            return {
                str(i): self._list_to_dict_preprocessing(item)
                for i, item in enumerate(data)
            }
        # Base case: the item is neither a dict nor a list, so return it unchanged
        return data

    def _json_split(
        self,
        data: Any,  # noqa: ANN401
        current_path: list[str] | None = None,
        chunks: list[dict[str, Any]] | None = None,
    ) -> list[dict[str, Any]]:
        """Split json into maximum size dictionaries while preserving structure."""
        current_path = current_path or []