전세 사기 피해 대처 관련 데이터 베이스를 만들자

RAG에서 사용할 벡터 DB를 만드는 단계

1. 원본 문서 준비 : RAG에 사용할 PDF·Markdown·텍스트 등 지식의 원본 자료를 수집하는 단계
2. 텍스트 정제 & chunking : 문서를 불필요한 요소를 제거한 뒤 의미 있는 작은 텍스트 조각으로 나누는 단계
3. embedding 생성 : 각 텍스트 chunk를 의미를 담은 숫자 벡터로 변환하는 단계
4. 벡터 DB 생성 : embedding을 저장하고 유사도 검색을 수행할 벡터 전용 데이터베이스를 만드는 단계
5. 벡터 + 메타데이터 저장 : embedding을 저장하고 유사도 검색을 수행할 벡터 전용 데이터베이스를 만드는 단계

다행히 우리 서비스는 기존에 사기 유형 파악과 가이드라인을 제공하기 위한 문서들을 가지고 있다.
다만, Q&A 자료나 용어 설명등 필요하다고 생각하는 몇가지의 문서를 추가로 준비했다.

따라서 다음 단계는 문서를 쪼개는 과정인 청킹이다.
청킹을 하기 전, 청킹을 해야하는 이유와 좋은 청킹의 방식을 알아보자.

오늘도 여러 기술 블로그를 읽었는데, 이렇게 좋은 자료를 만들어주시는 선배개발자 분들께 다시 한번 감사의 표시를...

Chunking 은 무엇이고 왜 필요할까

Chunking 이란 긴 문서를 작은 조각인 Chunk로 나누는 작업이다.
1번 과정에서 준비한 긴 문서를 그대로 임베딩 하지 않고, 500자나 1000 단위로 쪼갠 후 넣어야 한다.

왜 이런 과정이 필요할까?

Chunking이 필요한 이유

1️⃣ 질문에 대해 정확한 정보를 받기 위해서

• 내용증명 설명 문서에는 용어의 뜻, 목적, 과정, 주의사항, 대상 의 정보가 있습니다.
• 질문자 A는 내용증명의 목적을 질문했고, 내용 증명 설명을 받았습니다.

받은 내용 증명 문서에는 필요한 정보인 "목적" 이외에도 너무 많은 정보가 존재한다.

물론 문서를 너무 작게 자른다면, 문서의 맥락이 사라져 정보가 부족해질 수 있다.

2️⃣ LLM의 입력 한계가 존재한다.

• 위의 상황에서 내용증명문서를 그대로 넣었다고 가정해보자.

LLM은 문자 길이에 제한이 존재해서, 질문에 실패할 수도 있고 토큰 방식으로 비용이 청구되기 때문에 엄청난 돈을 내게될 수 있다. 💦 또한 문서에서 가장 중요한 정보를 찾기 어렵거나 중요도가 희석되거나, 일부 정보는 사라지게 되는 문제가 발생할 수 있다.
정확도와 비용측면에서 모두 안좋다.

즉, 모든 정보를 주는 것이 아니라 질문에 필요한 정확한 정보를 주는 것이 중요하고, 그러기 위해서는 적절한 단위로 문서를 자르는 과정이 필요하다

관련 용어 정리

• Precision (정밀도) : "내가 찾았다고 한 것 중 진짜 맞는 비율"
  - 정답으로 예측한 것 중 실제 정답 / 정답이라고 예측한 전체

• Recall (재현율) : "실제 정답 중 내가 찾아낸 비율"
  - 내가 찾아낸 정답 / 실제 정답 전체

좋은 Chunking 전략

Chunking 에도 여러 전략이 있지만, 3가지 정도를 소개 하려고 한다.

✔️ 고정 길이로 자르기

청킹은 보통 500자나 1000자가 권장된다.
긴 길이의 문서를 일정한 단위로 자르되, 문맥 정보를 유지하기 위해 청크끼리 오버랩을 줄 수 있다.

문서의 양이 방대하다면, 고정길이로 자르는 것이 유용할 수 있다.
또한 고정 단위로 자르는 방식이 무조건 성능 저하를 일으키는 것은 아니다. 여러 실험에서는 고정 길이 방식이 Recall 88.1~89.5 %을 유지하며 안정적인 성능을 보이기도 했다고 한다.

✔️ 의미 단위로 자르기

텍스트에는 복잡한 의미가 담겨있다. 내부에 표나 리스트가 있기도 하고 의미적인 구조가 존재하기도 한다.
이런 맥락과 텍스트 문서의 특징을 살려 하나의 문서 조각이 하나의 주제를 다를 수 있도록 짜르는 것이다.

✔️ AI 한테 시키기

또한 아예 문서의 핵심 주제를 AI가 의미적 유사성을 분석해 만든 단위로 문서를 쪼개는 방법도 있다.
문장을 쪼갠 후 임베딩을 하는 것이 아니라, 임베딩한 후 문장간의 유사도 분석을 통해 문서를 쪼개는 것이다.

➕ 맥락을 지키는 오버랩

오버랩(OverRap)은 이전 chunk의 끝부분을 다음 chunk 시작에 포함시키는 기술이다.
예를 들어서 설명하자면 첫 청크의 한두문장을 두번째 청크의 처음으로 추가하는 것이다. 이런 방식은 문서의 문맥을 살리게 되어 Recall을 높일 수 있다.

OverRap의 비율은 도메인에 따라 달라질 수 있으며 보통 10~20%를 권장한다.

• 법률/ 의료 : 30~50%
• 기술 문서/API : 20~30%
• 일반 문서/뉴스 : 10~20%
• Q&A : 5~10%

➕ 문서마다 다를 수 있으니 기본적으로 10%에서 시작해 조정해나가는 방법이 좋다고 한다.

무조건 OverRap 비율을 높이는 것이 정확도 향상으로 이어지지는 않는다.!

1. 오버랩률이 높아지면 그만큼 더 많은 저장소 공간을 필요로 하고, 인덱싱의 비용증가로 이어진다.

2. 또한 재현율을 높이려 한 선택이지만 정밀도(Precision)를 낮출 수 있다.
   오버랩이 높으면 같은 내용이 여러 청크에 포함되며 상관없는 문서도 함께 검색 결과에 들어가게 된다.
   이는 정밀도가 낮아지게 만든다. ( 정밀도 = 정답으로 예측한 것 중 실제 정답 / 정답이라고 예측한 전체)

청킹에는 어떤 문서를 넣어야하는가?

과연 청킹은 어떻게 하는걸까?
다행스럽게도 👩‍💻 개발자 도비가 손으로 일일히 하는 것은 아니다.

코드로 문서를 읽어오고, 선택한 전략에 맞춰서 적절한 문서 조각으로 나눈다.
이렇게 만들어진 문서 조각(청크)를 사람이 한번 더 검수하는 과정으로 이루어진다.

추천되는 문서 형태 : .md

이 문서를 읽고 자르는 주체는 컴퓨터이다. 컴퓨터 입장에서 해석하기 좋고, 불필요한 정보가 없는 문서 형태가 좋다.

word나 html은 컴퓨터가 이해하기에는 불필요한 정보가 많고, 일반 텍스트는 의미를 파악하기에는 정보가 부족하다. 마크다운은 표나 제목을 표시할 수 있어 의미 단위로 자르기에 간편하다.

우리 서비스는 어떤 전략으로 Chunking 을 할까?

전세 사기 피해 대처 가이드라인의 도메인 특성

• 법률 용어가 많다.
• 절차가 존재한다.
• 상황별 대처방법이 따로 존재한다. (절차가 여러 갈래가 된다.)
• 법률 조항은 중간에 짤리면 안된다.

문서의 내용이 어렵다고 판단했고, 용어 사전과 Q&A 자료를 함께 준비했다!

문서 구조 선택 : 마크다운

사실 프로젝트를 기획하면서 노션에 정리를 해와서 수고롭지 않게 정리할 수 있었다.
마크다운 문법을 활용해서 제목 (#),중제목(##), 소제목(###)으로 정리했다.
출처가 중요하다 판단해, 문서의 하단에는 출처링크들을 함께 정리했다.

지금 구조가 확정은 아니고, 성능을 테스트 하면서 마크다운의 구조를 바꾸거나, 의미별로 잘 잘라지(?)게 다듬을 것이다.

청킹 전략 : 의미 단위 vs 고정 길이

맥락을 보존할 수 있도록 의미 단위 (##)로 자르고, 자른 청크의 길이가 500이 넘어갈 때는 500자 단위로 자르기로 했다.

당연히 이 방식도 이것도 확정은 아니고, 성능 테스트를 보면서 확정을 하도록 하겠다!

Chunking 전략을 (대략적으로) 정하자

문제 상황 : 아예 다른 답변을 준다.

상황 : 임차권 등기 명령을 신청했으나, 전세 사기 피해자와 관련된 대답을 줌

문제점 파악 : 문서의 핵심 주제와 작은 주제인 #, ## 옆에 있던 텍스트들이 빠지면서 검색 성능이 떨어지고 있었음

📑 계약 만료일이 지났음에도 집주인이 ~~ 

해결 : 우선 코드를 변경해서 헤더옆의 텍스트가 들어갈 수 있게 수정했다.

결과

📑 임차권등기명령 신청 임차권등기명령을 신청해야 하는 상황 ### 1. 임대차 기간 종료 후 보증금 미반환\n계약 만료

여러 자료를 찾아보면서 LLM 성능은 단계별로 조정하기 어렵다는 것을 알았다.
점진적으로 조절을 해야하긴하지만, 일단 대략적으로 정한 청킹 전략을 소개하겠다

의미 기반 청킹 & 길이 기준청킹 전력

1. #, ## 기반으로 나누기
   제목과, 대분류 텍스트는 제거하지 않고 content에 저장

2. 길이 기반으로 나누기
   1번의 결과가 기준(1000)자를 넘을 경우에는 텍스트 내부에서 큰 의미 단위로 자름: ( \n\n (문단) → \n (줄바꿈) → . (문장) → (단어) :

3. 청크를 관라히기 위한 chunk_id
   이후 테스트용 데이터 셋을 만들거나 관리에 사용할 chunk_id

   chunk_id = 파일명_헤더인덱스_서브인덱스(번호)

코드 전문

# chunking.py
# chunking.py
from langchain_text_splitters import (
    MarkdownHeaderTextSplitter,
    RecursiveCharacterTextSplitter
)

import os
import json
from pathlib import Path

global i


class MarkdownChunker:
    def __init__(self, chunk_size=800, chunk_overlap=80):
        """
        Args:
            chunk_size: 한 청크의 최대 크기 (글자 수)
            chunk_overlap: 청크 간 오버랩 크기 (10% = 80자)
        """
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap

        # 헤더 기반 분리기
        self.headers_to_split_on = [
            ("#", "제목"),
            ("##", "대분류"),
        ]

        self.markdown_splitter = MarkdownHeaderTextSplitter(
            headers_to_split_on=self.headers_to_split_on
        )

        # 크기 조정용 분리기
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=self.chunk_size,
            chunk_overlap=self.chunk_overlap,
            separators=["\n\n", "\n", ".", "!", "?", " ", ""],
            length_function=len,
        )

    # 제목과 부제목은 제외
    def chunk_file_1(self, file_path):
        """단일 마크다운 파일을 청킹"""
        print(f"처리 중: {file_path}")

        file_name = os.path.basename(file_path)

        with open(file_path, 'r', encoding='utf-8') as f:
            content = f.read()

        # 1단계: 헤더 단위로 먼저 분리
        try:
            md_chunks = self.markdown_splitter.split_text(content)
        except Exception as e:
            print(f"헤더 분리 실패, 텍스트 분리로 전환: {e}")
            md_chunks = [{"page_content": content, "metadata": {}}]

        # 2단계: 크기 조정
        final_chunks = []
        for i, chunk in enumerate(md_chunks):
            chunk_content = chunk.page_content if hasattr(chunk, 'page_content') else chunk
            chunk_metadata = chunk.metadata if hasattr(chunk, 'metadata') else {}

            base_id = f"{file_name}_{i}"

            # 청크가 너무 크면 재분할
            if len(chunk_content) > self.chunk_size:
                sub_chunks = self.text_splitter.split_text(chunk_content)
                for j, sub in enumerate(sub_chunks):
                    final_chunks.append({
                        'content': sub,
                        'metadata': {
                            'source': file_name,
                            'chunk_id': f"{base_id}_{j}",
                            **chunk_metadata
                        }
                    })
            else:
                final_chunks.append({
                    'content': chunk_content,
                    'metadata': {
                        'source': file_name,
                        'chunk_id': base_id,
                        **chunk_metadata
                    }
                })

        return final_chunks

    def chunk_file(self, file_path):
        """단일 마크다운 파일을 청킹하며 고유한 ID 부여"""
        print(f"처리 중: {file_path}")

        # 파일명을 ID의 일부로 사용하기 위해 추출
        file_name = os.path.basename(file_path)

        with open(file_path, 'r', encoding='utf-8') as f:
            content = f.read()

        try:
            md_chunks = self.markdown_splitter.split_text(content)
        except Exception as e:
            print(f"헤더 분리 실패: {e}")
            md_chunks = [{"page_content": content, "metadata": {}}]

        final_chunks = []
        for i, chunk in enumerate(md_chunks):
            chunk_content = chunk.page_content if hasattr(chunk, 'page_content') else chunk
            chunk_metadata = chunk.metadata if hasattr(chunk, 'metadata') else {}

            header_prefix = ""
            title = chunk_metadata.get('제목', '')
            category = chunk_metadata.get('대분류', '')

            if title or category:
                header_prefix = f"{title} {category} ".strip() + " "

            processed_content = header_prefix + chunk_content

            # 고유 ID 생성을 위한 베이스 ID (파일명 + 헤더순서)
            base_id = f"{file_name}_{i}"
            # base_id = f"{i}"

            if len(processed_content) > self.chunk_size:
                sub_chunks = self.text_splitter.split_text(processed_content)
                for j, sub in enumerate(sub_chunks):
                    final_chunks.append({
                        'content': sub,
                        'metadata': {
                            'source': file_name,
                            'chunk_id': f"{base_id}_{j}",  # 예: document.md_0_1
                            **chunk_metadata
                        }
                    })
            else:
                final_chunks.append({
                    'content': processed_content,
                    'metadata': {
                        'source': file_name,
                        'chunk_id': base_id,  # 예: document.md_0
                        **chunk_metadata
                    }
                })

        return final_chunks




    def chunk_directory(self, directory_path="문서"):
        """디렉토리 내 모든 .md 파일 청킹"""
        all_chunks = []

        # 모든 .md 파일 찾기
        base_path = Path(__file__).parent / directory_path
        print(base_path)

        if not base_path.exists():
            print(f"오류: {base_path} 경로를 찾을 수 없습니다.")
            return []

        md_files = list(base_path.glob("*.md"))

        print(f"\n총 {len(md_files)}개의 마크다운 파일을 찾았습니다.")

        for file_path in md_files:
            chunks = self.chunk_file(file_path)
            all_chunks.extend(chunks)
            print(f"  → {len(chunks)}개 청크 생성")

        print(f"\n총 {len(all_chunks)}개의 청크가 생성되었습니다.")
        return all_chunks

    def save_chunks(self, chunks, output_file="json/chunks.json"):
        """청크를 JSON 파일로 저장"""
        with open(output_file, 'w', encoding='utf-8') as f:
            json.dump(chunks, f, ensure_ascii=False, indent=2)
        print(f"\n청크가 {output_file}에 저장되었습니다.")

    def print_chunk_stats(self, chunks):
        """청크 통계 출력"""
        if not chunks:
            return

        chunk_lengths = [len(chunk['content']) for chunk in chunks]
        print("\n=== 청킹 통계 ===")
        print(f"총 청크 수: {len(chunks)}")
        print(f"평균 길이: {sum(chunk_lengths) / len(chunk_lengths):.0f}자")
        print(f"최소 길이: {min(chunk_lengths)}자")
        print(f"최대 길이: {max(chunk_lengths)}자")

        # 파일별 통계
        from collections import Counter
        sources = Counter([chunk['metadata']['source'] for chunk in chunks])
        print("\n파일별 청크 수:")
        for source, count in sources.items():
            print(f"  - {source}: {count}개")


def main():
    # 청커 초기화
    chunker = MarkdownChunker(
        chunk_size=1000,  # 최대 800자
        chunk_overlap=80  # 10% 오버랩
    )

    # 현재 디렉토리의 모든 .md 파일 청킹
    chunks = chunker.chunk_directory()

    # 통계 출력
    chunker.print_chunk_stats(chunks)

    # JSON 파일로 저장
    chunker.save_chunks(chunks, "json/chunks.json")

    # 첫 3개 청크 미리보기
    print("\n=== 첫 3개 청크 미리보기 ===")
    for i, chunk in enumerate(chunks[:3]):
        print(f"\n[청크 {i + 1}]")
        print(f"파일: {chunk['metadata']['source']}")
        print(f"메타데이터: {chunk['metadata']}")
        print(f"내용 (처음 200자):\n{chunk['content'][:200]}...")


if __name__ == "__main__":
    main()

다음 글

진짜로 성능 테스트를 다루자 !

유용한 글

RAG 시스템을 위한 문서 전처리 가이드: AI가 이해하기 쉬운 형태로 만들기
LLM 서비스 개발의 핵심 기술: RAG, VectorDB, LangChain 개념 정리
[Tech Series] kt cloud AI 검색 증강 생성(RAG) #3 : 청킹(Chunking) 전략과 최적화
청킹과 인덱싱(Chunking & Indexing)-셀렉트스타