预处理

PreprocessReader

Bases: BaseReader

Preprocess 是一项 API 服务，可将任何类型的文档分割成用于语言模型任务的最佳文本块。Preprocess 将文档分割成尊重原始文档布局和语义的文本块。了解更多信息请访问 https://preprocess.co/。

参数

名称	类型	描述	默认值
api_key	str	[必需] Preprocess API 密钥。如果您还没有，请通过 [email protected] 申请。默认值: None	必需
file_path	str	[必需] 要预处理的文档路径 (转换并分割成块)。默认值: None	必需
table_output_format	str	文档中表格的输出格式。接受的值为 [text, markdown, html]。默认值: text	必需
repeat_table_header	bool	如果为 True，当表格跨多个块分割时，每个块将包含表格的行标题。默认值: False	必需
merge	bool	如果为 True，短块将与其他块合并以最大化块长度。默认值: False	必需
repeat_title	bool	如果为 True，每个块将包含父段落或部分的标题。默认值: False	必需
keep_header	bool	如果为 True，每页页眉的内容将包含在块中。默认值: True	必需
smart_header	bool	如果为 True，只有相关页眉会包含在块中，而无关信息将被移除。相关页眉指用作章节或段落标题的页眉。如果设为 False，则只考虑 keep_header 参数。如果 keep_header 为 False，则忽略此参数。默认值: True	必需
keep_footer	bool	如果为 True，每页页脚的内容将包含在块中。默认值: False	必需
image_text	bool	如果为 True，图像中包含的文本将添加到块中。默认值: False	必需

示例

>>> loader = PreprocessReader(api_key="your-api-key", file_path="valid/path/to/file")

源代码位于 llama-index-integrations/readers/llama-index-readers-preprocess/llama_index/readers/preprocess/base.py

class PreprocessReader(BaseReader):
    """
    Preprocess is an API service that splits any kind of document into optimal chunks of text for use in language model tasks.
    Preprocess splits the documents into chunks of text that respect the layout and semantics of the original document.
    Learn more at https://preprocess.co/.

    Args:
        api_key (str):
            [Required] The Preprocess API Key.
            If you don't have one yet, please request it at [email protected].
            Default: `None`

        file_path (str):
            [Required] The path to the document to be preprocessed (convertend and split into chunks).
            Default: `None`

        table_output_format (str):
            The output format for tables within the document.
            Accepted values are [text, markdown, html].
            Default: `text`

        repeat_table_header (bool):
            If `True`, when tables are split across multiple chunks, each chunk will include the table's row header.
            Default: `False`

        merge (bool):
            If `True`, short chunks will be merged with others to maximize chunk length.
            Default: `False`

        repeat_title (bool):
            If `True`, each chunk will include the title of the parent paragraph or section.
            Default: `False`

        keep_header (bool):
            If `True`, the content of each page's header will be included in the chunks.
            Default: `True`

        smart_header (bool):
            If `True`, only relevant headers will be included in the chunks, while irrelevant information will be removed.
            Relevant headers are those that serve as section or paragraph titles.
            If set to `False`, only the `keep_header` parameter will be considered. If `keep_header` is `False`, this parameter will be ignored.
            Default: `True`

        keep_footer (bool):
            If `True`, the content of each page's footer will be included in the chunks.
            Default: `False`

        image_text (bool):
            If `True`, the text contained in images will be added to the chunks.
            Default: `False`


    Examples:
        >>> loader = PreprocessReader(api_key="your-api-key", file_path="valid/path/to/file")

    """

    def __init__(self, api_key: str, *args, **kwargs):
        """Initialise with parameters."""
        try:
            from pypreprocess import Preprocess
        except ImportError:
            raise ImportError(
                "`pypreprocess` package not found, please run `pip install"
                " pypreprocess`"
            )

        if api_key is None or api_key == "":
            raise ValueError(
                "Please provide an api key to be used while doing the auth with the system."
            )
        _info = {}
        self._preprocess = Preprocess(api_key)
        self._filepath = None
        self._process_id = None

        for key, value in kwargs.items():
            if key == "filepath":
                self._filepath = value
                self._preprocess.set_filepath(value)

            if key == "process_id":
                self._process_id = value
                self._preprocess.set_process_id(value)

            elif key in ["table_output_format", "table_output"]:
                _info["table_output_format"] = value

            elif key in ["repeat_table_header", "table_header"]:
                _info["repeat_table_header"] = value

            elif key in [
                "merge",
                "repeat_title",
                "keep_header",
                "keep_footer",
                "smart_header",
                "image_text",
            ]:
                _info[key] = value

        if _info != {}:
            self._preprocess.set_info(_info)

        if self._filepath is None and self._process_id is None:
            raise ValueError(
                "Please provide either filepath or process_id to handle the resutls."
            )

        self._chunks = None

    def load_data(self, return_whole_document=False) -> List[Document]:
        """
        Load data from Preprocess.

        Args:
            return_whole_document (bool):
                Returning a list of one element, that element containing the full document.
                Default: `false`

        Examples:
            >>> documents = loader.load_data()
            >>> documents = loader.load_data(return_whole_document=True)

        Returns:
            List[Document]:
                A list of documents each document containing a chunk from the original document.

        """
        if self._chunks is None:
            if self._process_id is not None:
                self._get_data_by_process()
            elif self._filepath is not None:
                self._get_data_by_filepath()

            if self._chunks is not None:
                if return_whole_document is True:
                    return [
                        Document(
                            text=" ".join(self._chunks),
                            metadata={"filename": os.path.basename(self._filepath)},
                        )
                    ]
                else:
                    return [
                        Document(
                            text=chunk,
                            metadata={"filename": os.path.basename(self._filepath)},
                        )
                        for chunk in self._chunks
                    ]
            else:
                raise Exception(
                    "There is error happened during handling your file, please try again."
                )

        else:
            if return_whole_document is True:
                return [
                    Document(
                        text=" ".join(self._chunks),
                        metadata={"filename": os.path.basename(self._filepath)},
                    )
                ]
            else:
                return [
                    Document(
                        text=chunk,
                        metadata={"filename": os.path.basename(self._filepath)},
                    )
                    for chunk in self._chunks
                ]

    def get_process_id(self):
        """
        Get process's hash id from Preprocess.

        Examples:
            >>> process_id = loader.get_process_id()

        Returns:
            str:
                Process's hash id.

        """
        return self._process_id

    def get_nodes(self) -> List[TextNode]:
        """
        Get nodes from Preprocess's chunks.

        Examples:
            >>> nodes = loader.get_nodes()

        Returns:
            List[TextNode]:
                List of nodes, each node will contains a chunk from the original document.

        """
        if self._chunks is None:
            self.load_data()

        nodes = []
        for chunk in self._chunks:
            text = str(chunk)
            id = hashlib.md5(text.encode()).hexdigest()
            nodes.append(TextNode(text=text, id_=id))

        if len(nodes) > 1:
            nodes[0].relationships[NodeRelationship.NEXT] = RelatedNodeInfo(
                node_id=nodes[1].node_id,
                metadata={"filename": os.path.basename(self._filepath)},
            )
            for i in range(1, len(nodes) - 1):
                nodes[i].relationships[NodeRelationship.NEXT] = RelatedNodeInfo(
                    node_id=nodes[i + 1].node_id,
                    metadata={"filename": os.path.basename(self._filepath)},
                )
                nodes[i].relationships[NodeRelationship.PREVIOUS] = RelatedNodeInfo(
                    node_id=nodes[i - 1].node_id,
                    metadata={"filename": os.path.basename(self._filepath)},
                )

            nodes[-1].relationships[NodeRelationship.PREVIOUS] = RelatedNodeInfo(
                node_id=nodes[-2].node_id,
                metadata={"filename": os.path.basename(self._filepath)},
            )
        return nodes

    def _get_data_by_filepath(self) -> None:
        pp_response = self._preprocess.chunk()
        if pp_response.status == "OK" and pp_response.success is True:
            self._process_id = pp_response.data["process"]["id"]
            response = self._preprocess.wait()
            if response.status == "OK" and response.success is True:
                # self._filepath = response.data['info']['file']['name']
                self._chunks = response.data["chunks"]

    def _get_data_by_process(self) -> None:
        response = self._preprocess.wait()
        if response.status == "OK" and response.success is True:
            self._filepath = response.data["info"]["file"]["name"]
            self._chunks = response.data["chunks"]

load_data

load_data(return_whole_document=False) -> List[Document]

从 Preprocess 加载数据。

参数

名称	类型	描述	默认值
return_whole_document	bool	返回一个只包含一个元素的列表，该元素包含完整文档。默认值: false	False

示例

>>> documents = loader.load_data()
>>> documents = loader.load_data(return_whole_document=True)

返回

类型	描述
List[Document]	List[Document]: 一个文档列表，每个文档包含原始文档的一个块。

源代码位于 llama-index-integrations/readers/llama-index-readers-preprocess/llama_index/readers/preprocess/base.py

def load_data(self, return_whole_document=False) -> List[Document]:
    """
    Load data from Preprocess.

    Args:
        return_whole_document (bool):
            Returning a list of one element, that element containing the full document.
            Default: `false`

    Examples:
        >>> documents = loader.load_data()
        >>> documents = loader.load_data(return_whole_document=True)

    Returns:
        List[Document]:
            A list of documents each document containing a chunk from the original document.

    """
    if self._chunks is None:
        if self._process_id is not None:
            self._get_data_by_process()
        elif self._filepath is not None:
            self._get_data_by_filepath()

        if self._chunks is not None:
            if return_whole_document is True:
                return [
                    Document(
                        text=" ".join(self._chunks),
                        metadata={"filename": os.path.basename(self._filepath)},
                    )
                ]
            else:
                return [
                    Document(
                        text=chunk,
                        metadata={"filename": os.path.basename(self._filepath)},
                    )
                    for chunk in self._chunks
                ]
        else:
            raise Exception(
                "There is error happened during handling your file, please try again."
            )

    else:
        if return_whole_document is True:
            return [
                Document(
                    text=" ".join(self._chunks),
                    metadata={"filename": os.path.basename(self._filepath)},
                )
            ]
        else:
            return [
                Document(
                    text=chunk,
                    metadata={"filename": os.path.basename(self._filepath)},
                )
                for chunk in self._chunks
            ]

get_process_id

get_process_id()

从 Preprocess 获取进程的哈希 ID。

示例

>>> process_id = loader.get_process_id()

返回

名称	类型	描述
str		进程的哈希 ID。

源代码位于 llama-index-integrations/readers/llama-index-readers-preprocess/llama_index/readers/preprocess/base.py

def get_process_id(self):
    """
    Get process's hash id from Preprocess.

    Examples:
        >>> process_id = loader.get_process_id()

    Returns:
        str:
            Process's hash id.

    """
    return self._process_id

get_nodes

get_nodes() -> List[TextNode]

从 Preprocess 的块获取节点。

示例

>>> nodes = loader.get_nodes()

返回

类型	描述
List[TextNode]	List[TextNode]: 节点列表，每个节点包含原始文档的一个块。

源代码位于 llama-index-integrations/readers/llama-index-readers-preprocess/llama_index/readers/preprocess/base.py

def get_nodes(self) -> List[TextNode]:
    """
    Get nodes from Preprocess's chunks.

    Examples:
        >>> nodes = loader.get_nodes()

    Returns:
        List[TextNode]:
            List of nodes, each node will contains a chunk from the original document.

    """
    if self._chunks is None:
        self.load_data()

    nodes = []
    for chunk in self._chunks:
        text = str(chunk)
        id = hashlib.md5(text.encode()).hexdigest()
        nodes.append(TextNode(text=text, id_=id))

    if len(nodes) > 1:
        nodes[0].relationships[NodeRelationship.NEXT] = RelatedNodeInfo(
            node_id=nodes[1].node_id,
            metadata={"filename": os.path.basename(self._filepath)},
        )
        for i in range(1, len(nodes) - 1):
            nodes[i].relationships[NodeRelationship.NEXT] = RelatedNodeInfo(
                node_id=nodes[i + 1].node_id,
                metadata={"filename": os.path.basename(self._filepath)},
            )
            nodes[i].relationships[NodeRelationship.PREVIOUS] = RelatedNodeInfo(
                node_id=nodes[i - 1].node_id,
                metadata={"filename": os.path.basename(self._filepath)},
            )

        nodes[-1].relationships[NodeRelationship.PREVIOUS] = RelatedNodeInfo(
            node_id=nodes[-2].node_id,
            metadata={"filename": os.path.basename(self._filepath)},
        )
    return nodes