parsing

langroid/parsing/init.py

PdfParsingConfig

Bases: BaseParsingConfig

enable_configs(values) classmethod

Ensure correct config is set based on library selection.

Source code in langroid/parsing/parser.py

@model_validator(mode="before")
@classmethod
def enable_configs(cls, values: Dict[str, Any]) -> Dict[str, Any]:
    """Ensure correct config is set based on library selection."""
    library = values.get("library")

    if library == "llm-pdf-parser":
        values.setdefault("llm_parser_config", LLMPdfParserConfig())
    else:
        values["llm_parser_config"] = None

    if library == "marker":
        values.setdefault("marker_config", MarkerConfig())
    else:
        values["marker_config"] = None

    return values

ParsingConfig

Bases: BaseSettings

convert_chunk_size_to_int(v) classmethod

Convert chunk_size to int, maintaining backward compatibility with Pydantic V1.

Source code in langroid/parsing/parser.py

@field_validator("chunk_size", mode="before")
@classmethod
def convert_chunk_size_to_int(cls, v: Any) -> int:
    """Convert chunk_size to int, maintaining backward compatibility
    with Pydantic V1.
    """
    if isinstance(v, float):
        return int(v)
    return int(v)

Parser(config)

Source code in langroid/parsing/parser.py

def __init__(self, config: ParsingConfig):
    self.config = config
    try:
        self.tokenizer = tiktoken.encoding_for_model(config.token_encoding_model)
    except Exception:
        self.tokenizer = tiktoken.encoding_for_model("text-embedding-3-small")

add_window_ids(chunks)

Chunks may belong to multiple docs, but for each doc, they appear consecutively. Add window_ids in metadata.

Chunks this has already windowed are left as they are, so calling it twice does not replace their neighbors with a singleton. See the module-level add_window_ids for the details.

Source code in langroid/parsing/parser.py

def add_window_ids(self, chunks: List[Document]) -> None:
    """Chunks may belong to multiple docs, but for each doc,
    they appear consecutively. Add window_ids in metadata.

    Chunks this has already windowed are left as they are, so calling
    it twice does not replace their neighbors with a singleton. See
    the module-level `add_window_ids` for the details.
    """
    add_window_ids(chunks, self.config.n_neighbor_ids)

chunk_tokens(text)

Split a text into chunks of ~CHUNK_SIZE tokens, based on punctuation and newline boundaries. Adapted from https://github.com/openai/chatgpt-retrieval-plugin/blob/main/services/chunks.py

Parameters:

Name	Type	Description	Default
text	str	The text to split into chunks.	required

Returns:

Type	Description
List[str]	A list of text chunks, each of which is a string of tokens
List[str]	roughly self.config.chunk_size tokens long.
List[str]	A chunk may exceed that size to finish a UTF-8 character.

Source code in langroid/parsing/parser.py

def chunk_tokens(
    self,
    text: str,
) -> List[str]:
    """
    Split a text into chunks of ~CHUNK_SIZE tokens,
    based on punctuation and newline boundaries.
    Adapted from
    https://github.com/openai/chatgpt-retrieval-plugin/blob/main/services/chunks.py

    Args:
        text: The text to split into chunks.

    Returns:
        A list of text chunks, each of which is a string of tokens
        roughly self.config.chunk_size tokens long.
        A chunk may exceed that size to finish a UTF-8 character.
    """
    # Return an empty list if the text is empty or whitespace
    if not text or text.isspace():
        return []

    # Tokenize the text
    tokens = self.tokenizer.encode(text, disallowed_special=())

    # Initialize an empty list of chunks
    chunks = []

    # Initialize a counter for the number of chunks
    num_chunks = 0

    # Loop until all tokens are consumed
    while tokens and num_chunks < self.config.max_chunks:
        # Take the first chunk_size tokens as a chunk
        chunk = tokens[: self.config.chunk_size]

        # BPE tokens can end inside a UTF-8 character. Complete it before
        # decoding so replacement characters do not corrupt the content.
        while True:
            try:
                chunk_text = self.tokenizer.decode(chunk, errors="strict")
                break
            except UnicodeDecodeError:
                if len(chunk) == len(tokens):
                    raise
                chunk = tokens[: len(chunk) + 1]
        remaining_tokens = tokens[len(chunk) :]

        # Skip the chunk if it is empty or whitespace
        if not chunk_text or chunk_text.isspace():
            # Remove the tokens corresponding to the chunk text
            # from remaining tokens
            tokens = tokens[len(chunk) :]
            # Continue to the next iteration of the loop
            continue

        # Find the last period or punctuation mark in the chunk
        punctuation_matches = [
            (m.start(), m.group())
            for m in re.finditer(r"(?:[.!?][\s\n]|\n)", chunk_text)
        ]

        last_punctuation = max([pos for pos, _ in punctuation_matches] + [-1])

        # If there is a punctuation mark, and the last punctuation index is
        # after MIN_CHUNK_SIZE_CHARS
        if (
            last_punctuation != -1
            and last_punctuation > self.config.min_chunk_chars
        ):
            remaining_tokens = (
                self.tokenizer.encode(
                    chunk_text[last_punctuation + 1 :], disallowed_special=()
                )
                + remaining_tokens
            )
            # Truncate the chunk text at the punctuation mark
            chunk_text = chunk_text[: last_punctuation + 1]

        # Replace redundant (3 or more) newlines with 2 newlines to preser
        # paragraph separation!
        # But do NOT strip leading/trailing whitespace, to preserve formatting
        # (e.g. code blocks, or in case we want to stitch chunks back together)
        chunk_text_to_append = re.sub(r"\n{3,}", "\n\n", chunk_text)

        if len(chunk_text_to_append) > self.config.discard_chunk_chars:
            # Append the chunk text to the list of chunks
            chunks.append(chunk_text_to_append)

        # Advance by the original tokens, not a re-encoded prefix whose
        # BPE boundaries may differ. Retain any text after a punctuation cut.
        tokens = remaining_tokens

        # Increment the number of chunks
        num_chunks += 1

    # There may be remaining tokens, but we discard them
    # since we have already reached the maximum number of chunks

    return chunks