parser

langroid/parsing/parser.py

BaseParsingConfig

Bases: BaseSettings

Base class for document parsing configurations.

LLMPdfParserConfig

Bases: BaseSettings

Configuration for LLM-based parsing.

MarkerConfig

Bases: BaseSettings

Configuration for Markitdown-based parsing.

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

add_window_ids(chunks, n_neighbor_ids)

Give each chunk a distinct id and the ids of its neighbors.

Chunks may belong to multiple docs, but for each doc they appear consecutively, and all chunks of one doc share a metadata.id -- which is how they are grouped here, so a window never crosses a document.

Already-windowed chunks are left alone. Grouping by a shared metadata.id is what makes this non-idempotent: after one call every chunk has its own id, so a second call would see each chunk as a single-chunk document and collapse every window to a singleton, silently discarding the neighbors from the first call. A chunk is taken to be already windowed when it is alone under its id and that id is in its own window -- the exact state this function leaves behind, and one a group of siblings sharing an inherited id cannot be in.

The decision is per group, not per call, so mixing windowed chunks with fresh ones windows only the fresh ones.

Parameters:

Name	Type	Description	Default
chunks	List[Document]	chunks to window, modified in place.	required
n_neighbor_ids	int	how many neighbors to keep on each side.	required

Source code in langroid/parsing/parser.py

def add_window_ids(chunks: List[Document], n_neighbor_ids: int) -> None:
    """Give each chunk a distinct id and the ids of its neighbors.

    Chunks may belong to multiple docs, but for each doc they appear
    consecutively, and all chunks of one doc share a `metadata.id` -- which
    is how they are grouped here, so a window never crosses a document.

    Already-windowed chunks are left alone. Grouping by a shared
    `metadata.id` is what makes this non-idempotent: after one call every
    chunk has its own id, so a second call would see each chunk as a
    single-chunk document and collapse every window to a singleton,
    silently discarding the neighbors from the first call. A chunk is taken
    to be already windowed when it is alone under its id and that id is in
    its own window -- the exact state this function leaves behind, and one
    a group of siblings sharing an inherited id cannot be in.

    The decision is per group, not per call, so mixing windowed chunks with
    fresh ones windows only the fresh ones.

    Args:
        chunks: chunks to window, modified in place.
        n_neighbor_ids: how many neighbors to keep on each side.
    """
    # discard empty chunks
    chunks = [c for c in chunks if c.content.strip() != ""]
    if len(chunks) == 0:
        return

    # group by the original metadata.id
    # (each distinct orig_id refers to a different document)
    orig_id_to_chunks: Dict[str, List[Document]] = {}
    for c in chunks:
        orig_id_to_chunks.setdefault(c.metadata.id, []).append(c)

    k = n_neighbor_ids
    for orig_id, group in orig_id_to_chunks.items():
        if len(group) == 1 and orig_id in group[0].metadata.window_ids:
            # already windowed by an earlier call; re-running would replace
            # its window with a singleton rather than refine it. is_chunk is
            # still set, since this function has always set it on everything
            # it was handed.
            group[0].metadata.is_chunk = True
            continue
        # The original metadata.id is ignored since it is the same for all
        # chunks of a doc and is useless. We want a distinct id for each.
        ids = [ObjectRegistry.new_id() for _ in group]
        n = len(ids)
        window_ids = [ids[max(0, i - k) : min(n, i + k + 1)] for i in range(n)]
        for i, c in enumerate(group):
            c.metadata.window_ids = window_ids[i]
            c.metadata.id = ids[i]
            c.metadata.is_chunk = True