Skip to content

Megaparse

Megaparse#

MegaparseProcessor #

Bases: ProcessorBase

Megaparse processor for PDF files.

It can be used to parse PDF files and split them into chunks.

It comes from the megaparse library.

Installation#

pip install megaparse
Source code in core/quivr_core/processor/implementations/megaparse_processor.py
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
class MegaparseProcessor(ProcessorBase):
    """
    Megaparse processor for PDF files.

    It can be used to parse PDF files and split them into chunks.

    It comes from the megaparse library.

    ## Installation
    ```bash
    pip install megaparse
    ```

    """

    supported_extensions = [FileExtension.pdf]

    def __init__(
        self,
        splitter: TextSplitter | None = None,
        splitter_config: SplitterConfig = SplitterConfig(),
        megaparse_config: MegaparseConfig = MegaparseConfig(),
    ) -> None:
        self.loader_cls = MegaParse
        self.enc = tiktoken.get_encoding("cl100k_base")
        self.splitter_config = splitter_config
        self.megaparse_config = megaparse_config

        if splitter:
            self.text_splitter = splitter
        else:
            self.text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
                chunk_size=splitter_config.chunk_size,
                chunk_overlap=splitter_config.chunk_overlap,
            )

    @property
    def processor_metadata(self):
        return {
            "chunk_overlap": self.splitter_config.chunk_overlap,
        }

    async def process_file_inner(self, file: QuivrFile) -> list[Document]:
        mega_parse = MegaParse(file_path=file.path, config=self.megaparse_config)  # type: ignore
        document: Document = await mega_parse.aload()
        print("\n\n document: ", document.page_content)
        if len(document.page_content) > self.splitter_config.chunk_size:
            docs = self.text_splitter.split_documents([document])
            for doc in docs:
                # if "Production Fonts (maximum)" in doc.page_content:
                #    print('Doc: ', doc.page_content)
                doc.metadata = {"chunk_size": len(self.enc.encode(doc.page_content))}
            return docs
        return [document]