왜 파싱이 RAG 품질의 병목인가 — 커스텀 Document Loader

from typing import Iterator

from langchain_core.document_loaders import BaseLoader
from langchain_core.documents import Document


class CustomDocumentLoader(BaseLoader):
    """파일을 한 줄씩 읽어오는 문서 로더의 예시입니다."""

    def __init__(self, file_path: str) -> None:
        self.file_path = file_path

    def lazy_load(self) -> Iterator[Document]:  # <-- 인자를 받지 않습니다
        """제너레이터로 문서를 하나씩 생성해 반환합니다."""
        with open(self.file_path, encoding="utf-8") as f:
            line_number = 0
            for line in f:
                yield Document(
                    page_content=line,
                    metadata={"line_number": line_number, "source": self.file_path},
                )
                line_number += 1

from typing import Any, Dict, List, Optional
import olefile
import zlib
import struct
import re
import unicodedata
from langchain.schema import Document
from langchain.document_loaders.base import BaseLoader


class HWPReader(BaseLoader):
    """HWP 파일의 내용을 읽어 Document로 변환합니다."""

    def __init__(self, file_path: str, *args: Any, **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)
        self.file_path = file_path
        self.extra_info = None
        self._initialize_constants()

    def _initialize_constants(self) -> None:
        self.FILE_HEADER_SECTION = "FileHeader"
        self.HWP_SUMMARY_SECTION = "\x05HwpSummaryInformation"
        self.SECTION_NAME_LENGTH = len("Section")
        self.BODYTEXT_SECTION = "BodyText"
        self.HWP_TEXT_TAGS = [67]

    def lazy_load(self) -> List[Document]:
        load_file = olefile.OleFileIO(self.file_path)
        file_dir = load_file.listdir()

        if not self._is_valid_hwp(file_dir):
            raise ValueError("유효하지 않은 HWP 파일입니다.")

        result_text = self._extract_text(load_file, file_dir)
        return [self._create_document(text=result_text, extra_info=self.extra_info)]

    def _is_valid_hwp(self, dirs: List[List[str]]) -> bool:
        """헤더 스트림 존재 여부로 위장 파일을 조기 차단합니다."""
        return [self.FILE_HEADER_SECTION] in dirs and [self.HWP_SUMMARY_SECTION] in dirs

    def _get_body_sections(self, dirs: List[List[str]]) -> List[str]:
        section_numbers = [
            int(d[1][self.SECTION_NAME_LENGTH :])
            for d in dirs
            if d[0] == self.BODYTEXT_SECTION
        ]
        return [
            f"{self.BODYTEXT_SECTION}/Section{num}" for num in sorted(section_numbers)
        ]

    def _create_document(
        self, text: str, extra_info: Optional[Dict] = None
    ) -> Document:
        return Document(page_content=text, metadata=extra_info or {})

    def _extract_text(
        self, load_file: olefile.OleFileIO, file_dir: List[List[str]]
    ) -> str:
        sections = self._get_body_sections(file_dir)
        return "\n".join(
            self._get_text_from_section(load_file, section) for section in sections
        )

    def _is_compressed(self, load_file: olefile.OleFileIO) -> bool:
        """헤더 37번째 바이트의 1비트로 압축 여부를 판정합니다."""
        with load_file.openstream(self.FILE_HEADER_SECTION) as header:
            header_data = header.read()
            return bool(header_data[36] & 1)

    def _get_text_from_section(self, load_file: olefile.OleFileIO, section: str) -> str:
        with load_file.openstream(section) as bodytext:
            data = bodytext.read()

        unpacked_data = (
            zlib.decompress(data, -15) if self._is_compressed(load_file) else data
        )

        text = []
        i = 0
        while i < len(unpacked_data):
            header, rec_type, rec_len = self._parse_record_header(
                unpacked_data[i : i + 4]
            )
            if rec_type in self.HWP_TEXT_TAGS:
                rec_data = unpacked_data[i + 4 : i + 4 + rec_len]
                text.append(rec_data.decode("utf-16"))
            i += 4 + rec_len

        text = "\n".join(text)
        text = self.remove_chinese_characters(text)
        text = self.remove_control_characters(text)
        return text

    @staticmethod
    def remove_chinese_characters(s: str):
        return re.sub(r"[\u4e00-\u9fff]+", "", s)

    @staticmethod
    def remove_control_characters(s):
        """유니코드 카테고리 C(제어문자)를 제거합니다."""
        return "".join(ch for ch in s if unicodedata.category(ch)[0] != "C")

    @staticmethod
    def _parse_record_header(header_bytes: bytes) -> tuple:
        header = struct.unpack_from("<I", header_bytes)[0]
        rec_type = header & 0x3FF
        rec_len = (header >> 20) & 0xFFF
        return header, rec_type, rec_len

from llama_parse import LlamaParse
from llama_index.core import SimpleDirectoryReader
import os

parser = LlamaParse(
    result_type="markdown",          # "markdown" 또는 "text"
    num_workers=4,                   # 여러 파일 처리 시 API 호출 분할 수
    verbose=True,
    language="ko",                   # 기본값 'en'
    skip_diagonal_text=True,         # 대각선 워터마크 텍스트 무시
    use_vendor_multimodal_model=True,
    vendor_multimodal_model_name="openai-gpt4o",
    vendor_multimodal_api_key=os.environ.get("OPENAI_API_KEY"),
)

file_extractor = {".pdf": parser}
documents = SimpleDirectoryReader(
    input_files=["data/report.pdf"], file_extractor=file_extractor
).load_data()

parsing_instruction = """This document is related to the Digital Government Transformation Initiative.
Be sure to parse tables and should be interpreted as text with detailed informations.
Images, Graphs, Diagrams should be interpreted as text with detailed descriptions."""

class LlamaParseLoader(BaseLoader):
    def __init__(self, file_paths: List[str], parsing_instructions="") -> None:
        parser = LlamaParse(
            result_type="markdown",
            num_workers=4,
            language="ko",
            invalidate_cache=True,
            skip_diagonal_text=True,
            use_vendor_multimodal_model=True,
            vendor_multimodal_model_name="openai-gpt4o",
            vendor_multimodal_api_key=os.environ.get("OPENAI_API_KEY"),
            parsing_instruction=parsing_instructions,
        )
        file_extractor = {".pdf": parser}
        self.document_reader = SimpleDirectoryReader(
            input_files=file_paths,
            file_extractor=file_extractor,
        )

    # 버전 A — 타입만 List
    def lazy_load(self) -> List[Document]:
        documents = self.document_reader.load_data()
        return [doc.to_langchain_format() for doc in documents]

    # 버전 B — 계약에 맞는 제너레이터
    def lazy_load(self) -> Iterator[Document]:
        for doc in self.document_reader.load_data():
            yield doc.to_langchain_format()