Tokenizers

Tokenization utilities for text.

Provides methods to split text into regex-based or Unicode-aware tokens. Tokenization is used for alignment in resolver.py and for determining sentence boundaries for smaller context use cases. This module is not used for tokenization within the language model during inference.

BaseTokenizerError

Bases: BaseException

Base class for all tokenizer-related errors.

Source code in src/kibad_llm/extractors/chunking_utils/tokenizers.py

class BaseTokenizerError(BaseException):
    """Base class for all tokenizer-related errors."""

CharInterval dataclass

Represents a range of character positions in the original text.

Attributes:

Name	Type	Description
start_pos	int	The starting character index (inclusive).
end_pos	int	The ending character index (exclusive).

Source code in src/kibad_llm/extractors/chunking_utils/tokenizers.py

@dataclasses.dataclass(slots=True)
class CharInterval:
    """Represents a range of character positions in the original text.

    Attributes:
      start_pos: The starting character index (inclusive).
      end_pos: The ending character index (exclusive).
    """

    start_pos: int
    end_pos: int

InvalidTokenIntervalError

Bases: BaseTokenizerError

Error raised when a token interval is invalid or out of range.

Source code in src/kibad_llm/extractors/chunking_utils/tokenizers.py

class InvalidTokenIntervalError(BaseTokenizerError):
    """Error raised when a token interval is invalid or out of range."""

RegexTokenizer

Bases: Tokenizer

Regex-based tokenizer (default).

The RegexTokenizer is faster than UnicodeTokenizer for English text because it skips involved Unicode handling.

Source code in src/kibad_llm/extractors/chunking_utils/tokenizers.py

class RegexTokenizer(Tokenizer):
    """Regex-based tokenizer (default).

    The RegexTokenizer is faster than UnicodeTokenizer for English text because it
    skips involved Unicode handling.
    """

    # @debug_utils.debug_log_calls
    def tokenize(self, text: str) -> TokenizedText:
        """Splits text into tokens (words, digits, or punctuation).

        Each token is annotated with its character position and type. Tokens
        following a newline or carriage return have `first_token_after_newline`
        set to True.

        Args:
          text: The text to tokenize.

        Returns:
          A TokenizedText object containing all extracted tokens.
        """
        tokenized = TokenizedText(text=text)
        previous_end = 0
        for token_index, match in enumerate(_TOKEN_PATTERN.finditer(text)):
            start_pos, end_pos = match.span()
            matched_text = match.group()
            token = Token(
                index=token_index,
                char_interval=CharInterval(start_pos=start_pos, end_pos=end_pos),
                token_type=TokenType.WORD,
                first_token_after_newline=False,
            )
            if token_index > 0:
                # Optimization: Check gap without slicing.
                has_newline = text.find("\n", previous_end, start_pos) != -1
                if not has_newline:
                    has_newline = text.find("\r", previous_end, start_pos) != -1
                if has_newline:
                    token.first_token_after_newline = True
            if regex.fullmatch(_DIGITS_PATTERN, matched_text):
                token.token_type = TokenType.NUMBER
            elif _WORD_PATTERN.fullmatch(matched_text):
                token.token_type = TokenType.WORD
            else:
                token.token_type = TokenType.PUNCTUATION
            tokenized.tokens.append(token)
            previous_end = end_pos
        return tokenized

tokenize(text)

Splits text into tokens (words, digits, or punctuation).

Each token is annotated with its character position and type. Tokens following a newline or carriage return have first_token_after_newline set to True.

Parameters:

Name	Type	Description	Default
text	str	The text to tokenize.	required

Returns:

Type	Description
TokenizedText	A TokenizedText object containing all extracted tokens.

Source code in src/kibad_llm/extractors/chunking_utils/tokenizers.py

def tokenize(self, text: str) -> TokenizedText:
    """Splits text into tokens (words, digits, or punctuation).

    Each token is annotated with its character position and type. Tokens
    following a newline or carriage return have `first_token_after_newline`
    set to True.

    Args:
      text: The text to tokenize.

    Returns:
      A TokenizedText object containing all extracted tokens.
    """
    tokenized = TokenizedText(text=text)
    previous_end = 0
    for token_index, match in enumerate(_TOKEN_PATTERN.finditer(text)):
        start_pos, end_pos = match.span()
        matched_text = match.group()
        token = Token(
            index=token_index,
            char_interval=CharInterval(start_pos=start_pos, end_pos=end_pos),
            token_type=TokenType.WORD,
            first_token_after_newline=False,
        )
        if token_index > 0:
            # Optimization: Check gap without slicing.
            has_newline = text.find("\n", previous_end, start_pos) != -1
            if not has_newline:
                has_newline = text.find("\r", previous_end, start_pos) != -1
            if has_newline:
                token.first_token_after_newline = True
        if regex.fullmatch(_DIGITS_PATTERN, matched_text):
            token.token_type = TokenType.NUMBER
        elif _WORD_PATTERN.fullmatch(matched_text):
            token.token_type = TokenType.WORD
        else:
            token.token_type = TokenType.PUNCTUATION
        tokenized.tokens.append(token)
        previous_end = end_pos
    return tokenized

SentenceRangeError

Bases: BaseTokenizerError

Error raised when the start token index for a sentence is out of range.

Source code in src/kibad_llm/extractors/chunking_utils/tokenizers.py

class SentenceRangeError(BaseTokenizerError):
    """Error raised when the start token index for a sentence is out of range."""

Sentinel

Sentinel class for unique object identification.

Source code in src/kibad_llm/extractors/chunking_utils/tokenizers.py

class Sentinel:
    """Sentinel class for unique object identification."""

    def __init__(self, name: str):
        self.name = name

    def __repr__(self) -> str:
        return f"<{self.name}>"

Token dataclass

Represents a token extracted from text.

Each token is assigned an index and classified into a type (word, number, punctuation, or acronym). The token also records the range of characters (its CharInterval) that correspond to the substring from the original text. Additionally, it tracks whether it follows a newline.

Attributes:

Name	Type	Description
index	int	The position of the token in the sequence of tokens.
token_type	TokenType	The type of the token, as defined by TokenType.
char_interval	CharInterval	The character interval within the original text that this token spans.
first_token_after_newline	bool	True if the token immediately follows a newline or carriage return.

Source code in src/kibad_llm/extractors/chunking_utils/tokenizers.py

@dataclasses.dataclass(slots=True)
class Token:
    """Represents a token extracted from text.

    Each token is assigned an index and classified into a type (word, number,
    punctuation, or acronym). The token also records the range of characters
    (its CharInterval) that correspond to the substring from the original text.
    Additionally, it tracks whether it follows a newline.

    Attributes:
      index: The position of the token in the sequence of tokens.
      token_type: The type of the token, as defined by TokenType.
      char_interval: The character interval within the original text that this
        token spans.
      first_token_after_newline: True if the token immediately follows a newline
        or carriage return.
    """

    index: int
    token_type: TokenType
    char_interval: CharInterval = dataclasses.field(default_factory=lambda: CharInterval(0, 0))
    first_token_after_newline: bool = False

TokenInterval dataclass

Represents an interval over tokens in tokenized text.

The interval is defined by a start index (inclusive) and an end index (exclusive).

Attributes:

Name	Type	Description
start_index	int	The index of the first token in the interval.
end_index	int	The index one past the last token in the interval.

Source code in src/kibad_llm/extractors/chunking_utils/tokenizers.py

@dataclasses.dataclass(slots=True)
class TokenInterval:
    """Represents an interval over tokens in tokenized text.

    The interval is defined by a start index (inclusive) and an end index
    (exclusive).

    Attributes:
      start_index: The index of the first token in the interval.
      end_index: The index one past the last token in the interval.
    """

    start_index: int = 0
    end_index: int = 0

TokenType

Bases: IntEnum

Enumeration of token types produced during tokenization.

Attributes:

Name	Type	Description
WORD		Represents an alphabetical word token.
NUMBER		Represents a numeric token.
PUNCTUATION		Represents punctuation characters.

Source code in src/kibad_llm/extractors/chunking_utils/tokenizers.py

class TokenType(enum.IntEnum):
    """Enumeration of token types produced during tokenization.

    Attributes:
      WORD: Represents an alphabetical word token.
      NUMBER: Represents a numeric token.
      PUNCTUATION: Represents punctuation characters.
    """

    WORD = 0
    NUMBER = 1
    PUNCTUATION = 2

TokenizedText dataclass

Holds the result of tokenizing a text string.

Attributes:

Name	Type	Description
text	str	The text that was tokenized. For UnicodeTokenizer, this is NOT normalized to NFC (to preserve indices).
tokens	list[Token]	A list of Token objects extracted from the text.

Source code in src/kibad_llm/extractors/chunking_utils/tokenizers.py

@dataclasses.dataclass
class TokenizedText:
    """Holds the result of tokenizing a text string.

    Attributes:
      text: The text that was tokenized. For UnicodeTokenizer, this is
        NOT normalized to NFC (to preserve indices).
      tokens: A list of Token objects extracted from the text.
    """

    text: str
    tokens: list[Token] = dataclasses.field(default_factory=list)

Tokenizer

Bases: ABC

Abstract base class for tokenizers.

Source code in src/kibad_llm/extractors/chunking_utils/tokenizers.py

class Tokenizer(abc.ABC):
    """Abstract base class for tokenizers."""

    @abc.abstractmethod
    def tokenize(self, text: str) -> TokenizedText:
        """Splits text into tokens.

        Args:
          text: The text to tokenize.

        Returns:
          A TokenizedText object.
        """

tokenize(text) abstractmethod

Splits text into tokens.

Parameters:

Name	Type	Description	Default
text	str	The text to tokenize.	required

Returns:

Type	Description
TokenizedText	A TokenizedText object.

Source code in src/kibad_llm/extractors/chunking_utils/tokenizers.py

@abc.abstractmethod
def tokenize(self, text: str) -> TokenizedText:
    """Splits text into tokens.

    Args:
      text: The text to tokenize.

    Returns:
      A TokenizedText object.
    """

UnicodeTokenizer

Bases: Tokenizer

Unicode-aware tokenizer for better non-English support.

This tokenizer uses Unicode character properties (Unicode Standard Annex #29) via the regex library's \X pattern to correctly handle grapheme clusters like Emojis and Hangul.

Unlike some Unicode tokenizers, this class does NOT normalize text to NFC. This ensures that token indices exactly match the original input string.

Note: Grapheme clustering makes this tokenizer slower than RegexTokenizer.

Source code in src/kibad_llm/extractors/chunking_utils/tokenizers.py

class UnicodeTokenizer(Tokenizer):
    """Unicode-aware tokenizer for better non-English support.

    This tokenizer uses Unicode character properties (Unicode Standard Annex #29)
    via the `regex` library's `\\X` pattern to correctly handle grapheme clusters
    like Emojis and Hangul.


    Unlike some Unicode tokenizers, this class does NOT normalize text to NFC.
    This ensures that token indices exactly match the original input string.

    Note: Grapheme clustering makes this tokenizer slower than RegexTokenizer.
    """

    # @debug_utils.debug_log_calls
    def tokenize(self, text: str) -> TokenizedText:
        """Splits text into tokens using Unicode properties.

        Args:
          text: The text to tokenize.

        Returns:
          A TokenizedText object.
        """
        tokens: list[Token] = []

        current_start = 0
        current_type = None
        current_script: str | Sentinel | None = None
        previous_end = 0

        for match in regex.finditer(r"\X", text):
            grapheme = match.group()
            start, _ = match.span()

            # 1. Handle Whitespace
            if grapheme.isspace():
                if current_type is not None:
                    self._emit_token(
                        tokens,
                        text,
                        current_start,
                        start,
                        current_type,
                        previous_end,
                    )
                    previous_end = start
                    current_type = None
                    current_script = None
                # Keep `previous_end` to detect newlines within the whitespace gap.
                continue

            g_type = _classify_grapheme(grapheme)

            # 2. Determine if we should merge with the current token
            should_merge = False
            if current_type is not None:
                if current_type == g_type:
                    if current_type == TokenType.WORD:
                        # Script Check
                        first_char = grapheme[0]

                        # Fast path: Explicit NO_GROUP (CJK/Thai) never merges.
                        if current_script is _NO_GROUP_SCRIPT:
                            should_merge = False

                        # CJK and Non-Spaced scripts require fragmentation.
                        elif _CJK_PATTERN.match(first_char) or _NON_SPACED_PATTERN.match(
                            first_char
                        ):
                            should_merge = False

                        else:
                            g_script = _get_script_fast(first_char)
                            # Safety: Do not merge distinct unknown scripts.
                            if (
                                current_script == g_script
                                and current_script is not _UNKNOWN_SCRIPT
                            ):
                                should_merge = True

                    elif current_type == TokenType.NUMBER:
                        should_merge = True

                    elif current_type == TokenType.PUNCTUATION:
                        # Heuristic: Merge punctuation only if identical (e.g. "!!").
                        last_grapheme = text[current_start:start]
                        if last_grapheme == grapheme:
                            should_merge = True
                        elif len(last_grapheme) >= len(grapheme) and last_grapheme.endswith(
                            grapheme
                        ):
                            should_merge = True

            # 3. State Transition
            if should_merge:
                # Extend current token
                pass
            else:
                # Flush previous token if exists
                if current_type is not None:
                    self._emit_token(
                        tokens,
                        text,
                        current_start,
                        start,
                        current_type,
                        previous_end,
                    )
                    previous_end = start

                # Start new token
                current_start = start
                current_type = g_type

                # Determine script for the new token
                if current_type == TokenType.WORD:
                    c = grapheme[0]
                    if _CJK_PATTERN.match(c) or _NON_SPACED_PATTERN.match(c):
                        current_script = _NO_GROUP_SCRIPT
                    else:
                        current_script = _get_script_fast(c)
                else:
                    current_script = None

        # 4. Flush final token
        if current_type is not None:
            self._emit_token(
                tokens,
                text,
                current_start,
                len(text),
                current_type,
                previous_end,
            )

        return TokenizedText(text=text, tokens=tokens)

    def _emit_token(
        self,
        tokens: list[Token],
        text: str,
        start: int,
        end: int,
        token_type: TokenType,
        previous_end: int,
    ):
        """Helper to create and append a token."""
        token = Token(
            index=len(tokens),
            char_interval=CharInterval(start_pos=start, end_pos=end),
            token_type=token_type,
            first_token_after_newline=False,
        )

        # Check for newlines in the gap between the previous token and this one
        if start > previous_end:
            gap = text[previous_end:start]
            if "\n" in gap or "\r" in gap:
                token.first_token_after_newline = True

        tokens.append(token)

tokenize(text)

Splits text into tokens using Unicode properties.

Parameters:

Name	Type	Description	Default
text	str	The text to tokenize.	required

Returns:

Type	Description
TokenizedText	A TokenizedText object.

Source code in src/kibad_llm/extractors/chunking_utils/tokenizers.py

def tokenize(self, text: str) -> TokenizedText:
    """Splits text into tokens using Unicode properties.

    Args:
      text: The text to tokenize.

    Returns:
      A TokenizedText object.
    """
    tokens: list[Token] = []

    current_start = 0
    current_type = None
    current_script: str | Sentinel | None = None
    previous_end = 0

    for match in regex.finditer(r"\X", text):
        grapheme = match.group()
        start, _ = match.span()

        # 1. Handle Whitespace
        if grapheme.isspace():
            if current_type is not None:
                self._emit_token(
                    tokens,
                    text,
                    current_start,
                    start,
                    current_type,
                    previous_end,
                )
                previous_end = start
                current_type = None
                current_script = None
            # Keep `previous_end` to detect newlines within the whitespace gap.
            continue

        g_type = _classify_grapheme(grapheme)

        # 2. Determine if we should merge with the current token
        should_merge = False
        if current_type is not None:
            if current_type == g_type:
                if current_type == TokenType.WORD:
                    # Script Check
                    first_char = grapheme[0]

                    # Fast path: Explicit NO_GROUP (CJK/Thai) never merges.
                    if current_script is _NO_GROUP_SCRIPT:
                        should_merge = False

                    # CJK and Non-Spaced scripts require fragmentation.
                    elif _CJK_PATTERN.match(first_char) or _NON_SPACED_PATTERN.match(
                        first_char
                    ):
                        should_merge = False

                    else:
                        g_script = _get_script_fast(first_char)
                        # Safety: Do not merge distinct unknown scripts.
                        if (
                            current_script == g_script
                            and current_script is not _UNKNOWN_SCRIPT
                        ):
                            should_merge = True

                elif current_type == TokenType.NUMBER:
                    should_merge = True

                elif current_type == TokenType.PUNCTUATION:
                    # Heuristic: Merge punctuation only if identical (e.g. "!!").
                    last_grapheme = text[current_start:start]
                    if last_grapheme == grapheme:
                        should_merge = True
                    elif len(last_grapheme) >= len(grapheme) and last_grapheme.endswith(
                        grapheme
                    ):
                        should_merge = True

        # 3. State Transition
        if should_merge:
            # Extend current token
            pass
        else:
            # Flush previous token if exists
            if current_type is not None:
                self._emit_token(
                    tokens,
                    text,
                    current_start,
                    start,
                    current_type,
                    previous_end,
                )
                previous_end = start

            # Start new token
            current_start = start
            current_type = g_type

            # Determine script for the new token
            if current_type == TokenType.WORD:
                c = grapheme[0]
                if _CJK_PATTERN.match(c) or _NON_SPACED_PATTERN.match(c):
                    current_script = _NO_GROUP_SCRIPT
                else:
                    current_script = _get_script_fast(c)
            else:
                current_script = None

    # 4. Flush final token
    if current_type is not None:
        self._emit_token(
            tokens,
            text,
            current_start,
            len(text),
            current_type,
            previous_end,
        )

    return TokenizedText(text=text, tokens=tokens)

find_sentence_range(text, tokens, start_token_index, known_abbreviations=_KNOWN_ABBREVIATIONS)

Finds a 'sentence' interval from a given start index.

Sentence boundaries are defined by

• punctuation tokens in _END_OF_SENTENCE_PATTERN
• newline breaks followed by an uppercase letter
• not abbreviations in _KNOWN_ABBREVIATIONS (e.g., "Dr.")

This favors terminating a sentence prematurely over missing a sentence boundary, and will terminate a sentence early if the first line ends with new line and the second line begins with a capital letter.

Parameters:

Name	Type	Description	Default
text	str	The text to analyze.	required
tokens	Sequence[Token]	The tokens that make up text. Note: For UnicodeTokenizer, use normalized text.	required
start_token_index	int	The index of the token to start the sentence from.	required
known_abbreviations	Set[str]	A set of strings that are known abbreviations and should not be treated as sentence boundaries.	_KNOWN_ABBREVIATIONS

Returns:

Type	Description
TokenInterval	A TokenInterval representing the sentence range [start_token_index, end). If
TokenInterval	no sentence boundary is found, the end index will be the length of
TokenInterval	tokens.

Raises:

Type	Description
SentenceRangeError	If start_token_index is out of range.

Source code in src/kibad_llm/extractors/chunking_utils/tokenizers.py

def find_sentence_range(
    text: str,
    tokens: Sequence[Token],
    start_token_index: int,
    known_abbreviations: Set[str] = _KNOWN_ABBREVIATIONS,
) -> TokenInterval:
    """Finds a 'sentence' interval from a given start index.

    Sentence boundaries are defined by:
      - punctuation tokens in _END_OF_SENTENCE_PATTERN
      - newline breaks followed by an uppercase letter
      - not abbreviations in _KNOWN_ABBREVIATIONS (e.g., "Dr.")

    This favors terminating a sentence prematurely over missing a sentence
    boundary, and will terminate a sentence early if the first line ends with new
    line and the second line begins with a capital letter.

    Args:
      text: The text to analyze.
      tokens: The tokens that make up `text`.
        Note: For UnicodeTokenizer, use normalized text.
      start_token_index: The index of the token to start the sentence from.
      known_abbreviations: A set of strings that are known abbreviations and
        should not be treated as sentence boundaries.


    Returns:
      A TokenInterval representing the sentence range [start_token_index, end). If
      no sentence boundary is found, the end index will be the length of
      `tokens`.

    Raises:
      SentenceRangeError: If `start_token_index` is out of range.
    """
    if not tokens:
        return TokenInterval(0, 0)

    if start_token_index < 0 or start_token_index >= len(tokens):
        raise SentenceRangeError(
            f"start_token_index={start_token_index} out of range. " f"Total tokens: {len(tokens)}."
        )

    i = start_token_index
    while i < len(tokens):
        if tokens[i].token_type == TokenType.PUNCTUATION:
            if _is_end_of_sentence_token(text, tokens, i, known_abbreviations):
                end_index = i + 1
                # Consume any trailing closing punctuation (e.g. quotes, parens)
                while end_index < len(tokens):
                    next_token_text = text[
                        tokens[end_index]
                        .char_interval.start_pos : tokens[end_index]
                        .char_interval.end_pos
                    ]
                    if (
                        tokens[end_index].token_type == TokenType.PUNCTUATION
                        and next_token_text in _CLOSING_PUNCTUATION
                    ):
                        end_index += 1
                    else:
                        break
                return TokenInterval(start_index=start_token_index, end_index=end_index)
        if _is_sentence_break_after_newline(text, tokens, i):
            return TokenInterval(start_index=start_token_index, end_index=i + 1)
        i += 1

    return TokenInterval(start_index=start_token_index, end_index=len(tokens))

tokenize(text, tokenizer=_DEFAULT_TOKENIZER)

Splits text into tokens using the provided tokenizer (default: RegexTokenizer).

Parameters:

Name	Type	Description	Default
text	str	The text to tokenize.	required
tokenizer	Tokenizer	The tokenizer instance to use.	_DEFAULT_TOKENIZER

Returns:

Type	Description
TokenizedText	A TokenizedText object.

Source code in src/kibad_llm/extractors/chunking_utils/tokenizers.py

def tokenize(text: str, tokenizer: Tokenizer = _DEFAULT_TOKENIZER) -> TokenizedText:
    """Splits text into tokens using the provided tokenizer (default: RegexTokenizer).

    Args:
      text: The text to tokenize.
      tokenizer: The tokenizer instance to use.

    Returns:
      A TokenizedText object.
    """
    return tokenizer.tokenize(text)

tokens_text(tokenized_text, token_interval)

Reconstructs the substring of the original text spanning a given token interval.

Parameters:

Name	Type	Description	Default
tokenized_text	TokenizedText	A TokenizedText object containing token data.	required
token_interval	TokenInterval	The interval specifying the range [start_index, end_index) of tokens.	required

Returns:

Type	Description
str	The exact substring of the original text corresponding to the token
str	interval.

Raises:

Type	Description
InvalidTokenIntervalError	If the token_interval is invalid or out of range.

Source code in src/kibad_llm/extractors/chunking_utils/tokenizers.py

def tokens_text(
    tokenized_text: TokenizedText,
    token_interval: TokenInterval,
) -> str:
    """Reconstructs the substring of the original text spanning a given token interval.

    Args:
      tokenized_text: A TokenizedText object containing token data.
      token_interval: The interval specifying the range [start_index, end_index)
        of tokens.

    Returns:
      The exact substring of the original text corresponding to the token
      interval.

    Raises:
      InvalidTokenIntervalError: If the token_interval is invalid or out of range.
    """
    if token_interval.start_index == token_interval.end_index:
        return ""

    if (
        token_interval.start_index < 0
        or token_interval.end_index > len(tokenized_text.tokens)
        or token_interval.start_index > token_interval.end_index
    ):

        raise InvalidTokenIntervalError(
            f"Invalid token interval. start_index={token_interval.start_index}, "
            f"end_index={token_interval.end_index}, "
            f"total_tokens={len(tokenized_text.tokens)}."
        )

    start_token = tokenized_text.tokens[token_interval.start_index]
    end_token = tokenized_text.tokens[token_interval.end_index - 1]
    return tokenized_text.text[
        start_token.char_interval.start_pos : end_token.char_interval.end_pos
    ]