RAG의 지식 베이스(Knowledge Base) 구축 완벽 가이드 📚

안녕하세요! 오늘은 RAG 시스템의 핵심인 지식 베이스 구축 방법에 대해 자세히 알아보겠습니다.

지식 베이스란? 🤔

지식 베이스는 RAG 시스템의 '두뇌' 역할을 하는 정보 저장소입니다:

• LLM이 참고할 수 있는 신뢰할 수 있는 정보들의 집합
• 구조화된 형태로 저장된 데이터 모음

지식 베이스 구축 단계 🚀

1. 데이터 수집 (Data Collection)

def collect_data():
    documents = []

    # 1. 파일 시스템에서 수집
    documents.extend(collect_from_files())

    # 2. 데이터베이스에서 수집
    documents.extend(collect_from_database())

    # 3. API를 통한 수집
    documents.extend(collect_from_api())

    return documents

주요 데이터 소스

• PDF 문서
• 워드 문서
• 마크다운 파일
• HTML 페이지
• 데이터베이스 레코드
• API 응답

2. 데이터 전처리 (Preprocessing) 🧹

def preprocess_documents(documents):
    processed_docs = []

    for doc in documents:
        # 1. 텍스트 추출
        text = extract_text(doc)

        # 2. 텍스트 정제
        clean_text = clean_text(text)

        # 3. 중복 제거
        if not is_duplicate(clean_text):
            processed_docs.append(clean_text)

    return processed_docs

텍스트 정제 작업

def clean_text(text):
    # 특수문자 제거
    text = remove_special_chars(text)

    # 공백 정규화
    text = normalize_whitespace(text)

    # HTML 태그 제거
    text = remove_html_tags(text)

    # 불필요한 줄바꿈 제거
    text = normalize_linebreaks(text)

    return text

3. 청크 생성 (Chunking) ✂️

def create_chunks(text, chunk_size=512, overlap=50):
    chunks = []

    # 1. 문장 단위로 분리
    sentences = split_into_sentences(text)

    # 2. 청크 생성
    current_chunk = []
    current_size = 0

    for sentence in sentences:
        sentence_size = len(sentence)

        if current_size + sentence_size > chunk_size:
            # 청크 저장
            chunks.append(" ".join(current_chunk))
            # 오버랩을 위해 마지막 일부 문장 유지
            current_chunk = current_chunk[-overlap:]
            current_size = sum(len(s) for s in current_chunk)

        current_chunk.append(sentence)
        current_size += sentence_size

    return chunks

4. 메타데이터 추가 📝

def add_metadata(chunk, source_doc):
    return {
        "text": chunk,
        "metadata": {
            "source": source_doc.filename,
            "page": source_doc.page_number,
            "created_at": source_doc.creation_date,
            "last_modified": source_doc.last_modified,
            "author": source_doc.author,
            "category": source_doc.category
        }
    }

5. 임베딩 생성 (Embedding Generation) 🔄

from sentence_transformers import SentenceTransformer

def create_embeddings(chunks):
    # 1. 임베딩 모델 초기화
    model = SentenceTransformer('all-MiniLM-L6-v2')

    # 2. 임베딩 생성
    embeddings = model.encode(chunks, 
                            show_progress_bar=True,
                            batch_size=32)

    return embeddings

최적화 전략 💡

1. 청크 사이즈 최적화

def optimize_chunk_size(text):
    # 문서 특성에 따른 동적 청크 사이즈 설정
    if is_technical_document(text):
        return 512  # 기술 문서는 작은 청크
    elif is_narrative_document(text):
        return 1024  # 이야기체는 큰 청크
    else:
        return 768  # 기본값

2. 중복 데이터 처리

def remove_duplicates(chunks):
    seen = set()
    unique_chunks = []

    for chunk in chunks:
        # 텍스트 정규화
        normalized = normalize_text(chunk)

        # 유사도 검사
        if not is_similar_to_existing(normalized, seen):
            seen.add(normalized)
            unique_chunks.append(chunk)

    return unique_chunks

3. 품질 관리

def check_chunk_quality(chunk):
    return {
        "length_check": len(chunk) >= 100,
        "content_check": contains_meaningful_content(chunk),
        "format_check": is_properly_formatted(chunk)
    }

저장 및 인덱싱 💾

1. 벡터 데이터베이스 저장

import chromadb

def store_in_vectordb(chunks, embeddings):
    client = chromadb.Client()
    collection = client.create_collection("knowledge_base")

    # 데이터 저장
    collection.add(
        embeddings=embeddings,
        documents=chunks,
        ids=[f"doc_{i}" for i in range(len(chunks))]
    )

    return collection

2. 메타데이터 인덱싱

def create_metadata_index(chunks_with_metadata):
    index = {
        "by_source": {},
        "by_category": {},
        "by_author": {}
    }

    for chunk in chunks_with_metadata:
        # 소스별 인덱싱
        index["by_source"].setdefault(
            chunk["metadata"]["source"], []
        ).append(chunk)

        # 카테고리별 인덱싱
        index["by_category"].setdefault(
            chunk["metadata"]["category"], []
        ).append(chunk)

        # 작성자별 인덱싱
        index["by_author"].setdefault(
            chunk["metadata"]["author"], []
        ).append(chunk)

    return index

데이터 유지보수 🔧

1. 정기적인 업데이트

def update_knowledge_base(collection, new_documents):
    # 1. 새로운 문서 처리
    chunks = preprocess_and_chunk(new_documents)
    embeddings = create_embeddings(chunks)

    # 2. 기존 데이터와 병합
    collection.add(
        embeddings=embeddings,
        documents=chunks,
        ids=[f"new_doc_{i}" for i in range(len(chunks))]
    )

2. 품질 모니터링

def monitor_knowledge_base(collection):
    # 1. 임베딩 품질 검사
    embedding_quality = check_embedding_quality(collection)

    # 2. 데이터 최신성 검사
    freshness = check_data_freshness(collection)

    # 3. 검색 성능 측정
    search_performance = measure_search_performance(collection)

    return {
        "quality_score": embedding_quality,
        "freshness_score": freshness,
        "search_score": search_performance
    }

마치며 🎁

지식 베이스 구축은 RAG 시스템의 성능을 좌우하는 핵심 요소입니다. 데이터의 품질과 관리에 충분한 시간과 노력을 투자하면, 더 정확하고 신뢰성 있는 AI 시스템을 구축할 수 있습니다!

---

추가 궁금하신 점이 있다면 댓글로 남겨주세요! 😊