[LangChain] Document Loaders

 LLM의 input은 String type으로 넣어줘야 한다. 이를 위해 랭체인에서 제공하는 document_loaders는 pdf, notion, ppt 등의 여러 형식을 불러오고 LLM에 제공할 수 있도록 전처리를 도와준다.

 

 기본적으로 document_loaders의 클래스들은 load 메소드를 사용해, 여러 형태의 데이터를 알맞게 parsing해서 Document 객체를 원소로 갖는 리스트를 반환한다. 이때, Document 객체는 page_content와 metadata를 클래스 변수로 갖는다.

 

랭체인의 document_loaders의 클래스들이 어떤 형태로 각 문서를 처리하고 값을 반환하는지 TextLoader와 CSVLoader의 소스코드를 살펴보자.

---

• TextLoader

class TextLoader(BaseLoader):
    def __init__(
        self,
        file_path: Union[str, Path],
        encoding: Optional[str] = None,
        autodetect_encoding: bool = False,
    ):
        """Initialize with file path."""
        self.file_path = file_path
        self.encoding = encoding
        self.autodetect_encoding = autodetect_encoding

    def lazy_load(self) -> Iterator[Document]:
        """Load from file path."""
        text = ""
        try:
            with open(self.file_path, encoding=self.encoding) as f:
                text = f.read()
        except UnicodeDecodeError as e:
            ## 에러처리 ##
            continue
        metadata = {"source": str(self.file_path)}
        yield Document(page_content=text, metadata=metadata)

 

 TextLoader의 lazy_load 메소드를 살펴보면, with 절로 파일 경로(file_path)에 접근해 text를 읽어오고, dictionary type의 metadata에 source키로 파일 경로를 주고 있다. 최종적으로는 Document type을 generator 형태로 반환한다.

 

TextLoader가 상속받은 BaseLoader에는 load 메소드가 구현되어 있는데 이 부분을 살펴보면 아래처럼 generator를 리스트 형태로 반환하는 것을 확인할 수 있다.

class BaseLoader(ABC):  

    def load(self) -> List[Document]:  
    """Load data into Document objects."""  
    return list(self.lazy_load())

• CSVLoader

class CSVLoader(BaseLoader):
    """Load a `CSV` file into a list of Documents.
    Output Example:
        .. code-block:: txt

            column1: value1
            column2: value2
            column3: value3
    """

    def __init__(
        self,
        file_path: Union[str, Path],
        source_column: Optional[str] = None,
        metadata_columns: Sequence[str] = (),
        csv_args: Optional[Dict] = None,
        encoding: Optional[str] = None,
        autodetect_encoding: bool = False,
    ):

        self.file_path = file_path
        self.source_column = source_column
        self.metadata_columns = metadata_columns
        self.encoding = encoding
        self.csv_args = csv_args or {}
        self.autodetect_encoding = autodetect_encoding

    def lazy_load(self) -> Iterator[Document]:
        try:
            with open(self.file_path, newline="", encoding=self.encoding) as csvfile:
                yield from self.__read_file(csvfile)
        except UnicodeDecodeError as e:
            if self.autodetect_encoding:
                detected_encodings = detect_file_encodings(self.file_path)
                for encoding in detected_encodings:
                    try:
                        with open(
                            self.file_path, newline="", encoding=encoding.encoding
                        ) as csvfile:
                            yield from self.__read_file(csvfile)
                            break
                    except UnicodeDecodeError:
                        continue
            else:
                raise RuntimeError(f"Error loading {self.file_path}") from e
        except Exception as e:
            raise RuntimeError(f"Error loading {self.file_path}") from e

    def __read_file(self, csvfile: TextIOWrapper) -> Iterator[Document]:
        csv_reader = csv.DictReader(csvfile, **self.csv_args)
        for i, row in enumerate(csv_reader):
            try:
                source = (
                    row[self.source_column]
                    if self.source_column is not None
                    else str(self.file_path)
                )
            except KeyError:
                raise ValueError(
                    f"Source column '{self.source_column}' not found in CSV file."
                )
            content = "\n".join(
                f"""{k.strip() if k is not None else k}: {v.strip()
                if isinstance(v, str) else ','.join(map(str.strip, v))
                if isinstance(v, list) else v}"""
                for k, v in row.items()
                if k not in self.metadata_columns
            )
            metadata = {"source": source, "row": i}
            for col in self.metadata_columns:
                try:
                    metadata[col] = row[col]
                except KeyError:
                    raise ValueError(f"Metadata column '{col}' not found in CSV file.")
            yield Document(page_content=content, metadata=metadata)

 

 lazy_load 메소드는 csv.DictReader로 CSV 파일을 읽고, 각 행을 순회하면서 Document 객체를 생성한다. 이때, CSV 파일의 한 행이 하나의 Document 객체가 되며, 각 행은 key-value 로 변환되어 Document의 page_content에 저장된다.

메타데이터의 경우 기본적으로 파일 경로를 사용하며, 만약 각 행에서 source_column이 지정된 경우 해당 컬럼의 값을 사용한다.