5. RAG - 실습 추가

• 실습 - 사내 문서 시스템

# week5_13_company_doc_system.py
from langchain_community.document_loaders import (
    TextLoader,
    CSVLoader,
    DirectoryLoader
)
from langchain_text_splitters import RecursiveCharacterTextSplitter
# from langchain_schema import Document
from langchain_core.documents import Document
import os
from datetime import datetime

class CompanyDocumentLoader:
    """사내 문서 통합 로더"""
    
    def __init__(self, base_dir="company_documents"):
        self.base_dir = base_dir
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50
        )
    
    def load_all_documents(self):
        """모든 문서 로드"""
        all_docs = []
        
        # 1. 정책 문서 (텍스트)
        policy_docs = self._load_policies()
        all_docs.extend(policy_docs)
        
        # 2. 직원 정보 (CSV)
        employee_docs = self._load_employees()
        all_docs.extend(employee_docs)
        
        # 3. 기타 문서
        other_docs = self._load_others()
        all_docs.extend(other_docs)
        
        return all_docs
    
    def _load_policies(self):
        """정책 문서 로드"""
        policy_dir = os.path.join(self.base_dir, "policies")
        if not os.path.exists(policy_dir):
            return []
        
        loader = DirectoryLoader(
            policy_dir,
            glob="**/*.txt",
            loader_cls=TextLoader,
            loader_kwargs={"encoding": "utf-8"}
        )
        docs = loader.load()
        
        # 메타데이터 추가
        for doc in docs:
            doc.metadata.update({
                "document_type": "policy",
                "department": "인사팀",
                "loaded_at": datetime.now().isoformat()
            })
        
        return docs
    
    def _load_employees(self):
        """직원 정보 로드"""
        csv_path = os.path.join(self.base_dir, "employees.csv")
        if not os.path.exists(csv_path):
            return []
        
        loader = CSVLoader(csv_path, encoding="utf-8")
        docs = loader.load()
        
        for doc in docs:
            doc.metadata.update({
                "document_type": "employee",
                "department": "인사팀"
            })
        
        return docs
    
    def _load_others(self):
        """기타 문서"""
        # 추가 문서 타입 처리
        return []
    
    def split_documents(self, documents):
        """문서 분할"""
        return self.splitter.split_documents(documents)
    
    def get_statistics(self, documents):
        """문서 통계"""
        stats = {
            "total_documents": len(documents),
            "by_type": {},
            "total_characters": sum(len(doc.page_content) for doc in documents)
        }
        
        for doc in documents:
            doc_type = doc.metadata.get("document_type", "unknown")
            stats["by_type"][doc_type] = stats["by_type"].get(doc_type, 0) + 1
        
        return stats

# 샘플 문서 구조 생성
def create_sample_structure():
    """샘플 문서 구조 생성"""
    os.makedirs("company_documents/policies", exist_ok=True)
    
    # 정책 문서
    policies = {
        "company_documents/policies/vacation.txt": """연차 휴가 정책

1. 기본 연차
- 입사 1년 미만: 월 1개
- 입사 1년 이상: 연 15개
- 3년 이상 근속 시 2년마다 1개 추가

2. 사용 규정
- 최소 1일 전 신청
- 부서장 승인 필수
- 연말 미사용 시 소멸""",
        
        "company_documents/policies/work_hours.txt": """근무 시간 규정

1. 정규 근무 시간
- 평일: 09:00 ~ 18:00
- 점심시간: 12:00 ~ 13:00

2. 재택근무
- 주 2회까지 가능
- 사전 승인 필수
- 업무 보고 의무"""
    }
    
    for path, content in policies.items():
        with open(path, "w", encoding="utf-8") as f:
            f.write(content)
    
    # 직원 정보
    import csv
    with open("company_documents/employees.csv", "w", encoding="utf-8", newline="") as f:
        writer = csv.writer(f)
        writer.writerows([
            ["이름", "부서", "직위", "이메일"],
            ["김철수", "개발팀", "팀장", "kim@parucnc.com"],
            ["이영희", "마케팅팀", "과장", "lee@parucnc.com"],
            ["박민수", "영업팀", "대리", "park@parucnc.com"]
        ])

# 실행
create_sample_structure()

loader = CompanyDocumentLoader()
documents = loader.load_all_documents()

print("=== 문서 로드 완료 ===")
print(f"총 문서 수: {len(documents)}")

# 통계
stats = loader.get_statistics(documents)
print(f"\n=== 문서 통계 ===")
print(f"총 문서: {stats['total_documents']}개")
print(f"총 문자 수: {stats['total_characters']:,}자")
print(f"문서 타입별:")
for doc_type, count in stats["by_type"].items():
    print(f"  - {doc_type}: {count}개")

# 문서 분할
print("\n=== 문서 분할 ===")
split_docs = loader.split_documents(documents)
print(f"분할 후 청크 수: {len(split_docs)}")

# 샘플 출력
print("\n=== 청크 샘플 ===")
for i, doc in enumerate(split_docs[:3]):
    print(f"\n청크 {i+1}:")
    print(f"내용: {doc.page_content[:100]}...")
    print(f"메타데이터: {doc.metadata}")

결과 :

(llm_env) PS C:\dev\llm> & C:/dev/llm/llm_env/Scripts/python.exe c:/dev/llm/5_13_comp_doc_system.py
=== 문서 로드 완료 ===
총 문서 수: 5

=== 문서 통계 ===
총 문서: 5개
총 문자 수: 363자
문서 타입별:

• policy: 2개
• employee: 3개

=== 문서 분할 ===
분할 후 청크 수: 5

=== 청크 샘플 ===

청크 1:
내용: 연차 휴가 정책

1. 기본 연차

• 입사 1년 미만: 월 1개
• 입사 1년 이상: 연 15개
• 3년 이상 근속 시 2년마다 1개 추가

2. 사용 규정

• 최소 1일 전 신청...
  메타데이터: {'source': 'company_documents\policies\vacation.txt', 'document_type': 'policy', 'department': '인사팀', 'loaded_at': '2026-03-18T16:29:58.531264'}

청크 2:
내용: 근무 시간 규정

1. 정규 근무 시간

• 평일: 09:00 ~ 18:00
• 점심시간: 12:00 ~ 13:00

2. 재택근무

• 주 2회까지 가능
• 사전 승인 필수
• 업무...
  메타데이터: {'source': 'company_documents\policies\work_hours.txt', 'document_type': 'policy', 'department': '인사팀', 'loaded_at': '2026-03-18T16:29:58.531264'}

청크 3:
내용: 이름: 김철수
부서: 개발팀
직위: 팀장
이메일: kim@parucnc.com...
메타데이터: {'source': 'company_documents\employees.csv', 'row': 0, 'document_type': 'employee', 'department': '인사팀'}

• 실습 : 검색 가능한 문서 시스템

# week5_14_searchable_system.py
from langchain_community.document_loaders import TextLoader, DirectoryLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
import re

class SearchableDocumentSystem:
    """검색 가능한 문서 시스템"""
    
    def __init__(self):
        self.documents = []
        self.chunks = []
    
    def load_documents(self, directory):
        """문서 로드"""
        loader = DirectoryLoader(
            directory,
            glob="**/*.txt",
            loader_cls=TextLoader,
            loader_kwargs={"encoding": "utf-8"}
        )
        self.documents = loader.load()
        
        # 분할
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=300,
            chunk_overlap=50
        )
        self.chunks = splitter.split_documents(self.documents)
        
        return len(self.chunks)
    
    def keyword_search(self, keyword):
        """키워드 검색"""
        results = []
        for i, chunk in enumerate(self.chunks):
            if keyword.lower() in chunk.page_content.lower():
                results.append({
                    "chunk_id": i,
                    "content": chunk.page_content,
                    "metadata": chunk.metadata,
                    "relevance": chunk.page_content.lower().count(keyword.lower())
                })
        
        # 관련도순 정렬
        results.sort(key=lambda x: x["relevance"], reverse=True)
        return results
    
    def metadata_search(self, **filters):
        """메타데이터 검색"""
        results = []
        for i, chunk in enumerate(self.chunks):
            match = True
            for key, value in filters.items():
                if chunk.metadata.get(key) != value:
                    match = False
                    break
            if match:
                results.append({
                    "chunk_id": i,
                    "content": chunk.page_content,
                    "metadata": chunk.metadata
                })
        return results
    
    def get_document_by_source(self, source):
        """출처별 문서 조회"""
        results = []
        for chunk in self.chunks:
            if source in chunk.metadata.get("source", ""):
                results.append(chunk.page_content)
        return results

# 테스트
system = SearchableDocumentSystem()

# 문서 로드
chunk_count = system.load_documents("company_documents/policies")
print(f"로드된 청크: {chunk_count}개\n")

# 키워드 검색
print("=== '연차' 키워드 검색 ===")
results = system.keyword_search("연차")
for i, result in enumerate(results[:3]):
    print(f"\n결과 {i+1} (관련도: {result['relevance']})")
    print(f"내용: {result['content'][:150]}...")
    print(f"출처: {result['metadata']['source']}")

# 메타데이터 검색
print("\n=== 문서 타입별 검색 ===")
policy_docs = system.metadata_search(document_type="policy")
print(f"정책 문서: {len(policy_docs)}개")

결과 :
(llmenv) PS C:\dev\llm> & C:/dev/llm/llm_env/Scripts/python.exe c:/dev/llm/5!4_searchable_system.py
로드된 청크: 2개

=== '연차' 키워드 검색 ===

결과 1 (관련도: 2)
내용: 연차 휴가 정책

1. 기본 연차

• 입사 1년 미만: 월 1개
• 입사 1년 이상: 연 15개
• 3년 이상 근속 시 2년마다 1개 추가

2. 사용 규정

• 최소 1일 전 신청
• 부서장 승인 필수
• 연말 미사용 시 소멸...
  출처: company_documents\policies\vacation.txt

=== 문서 타입별 검색 ===
정책 문서: 0개

• 핵심 개념 정리

컴포넌트	역할/권장 사용 사례	지원 파일 형식	특징/비고
TextLoader	텍스트 파일 로드	.txt, .md	단순 텍스트 문서 처리에 적합
PyPDFLoader	PDF 파일 로드	.pdf	보고서, 계약서 등 활용 가능
CSVLoader	CSV 파일 로드	.csv	구조화된 데이터 처리에 적합
DirectoryLoader	디렉토리 전체 로드	폴더 내 다양한 파일	대량 문서 처리에 유용
CharacterTextSplitter	문자 기반 분할	모든 텍스트	간단한 텍스트 분할에 적합
RecursiveCharacterTextSplitter	계층적 분할	모든 텍스트	복잡한 문서 처리에 권장

• 베스트 프랙티스

1. 적절한 청크 크기 선택

python
# 문서 타입별 권장 크기
chunk_sizes = {
    "단문 (FAQ)": 200,
    "일반 문서": 500,
    "긴 문서 (보고서)": 1000,
    "코드": 300
}

2. Overlap 설정

python
# 중복 비율: 청크 크기의 10-20%
chunk_size = 500
chunk_overlap = 50  # 10%

3. 메타데이터 표준화

python
standard_metadata = {
    "source": "파일 경로",
    "document_type": "문서 타입",
    "created_at": "생성 시각",
    "department": "담당 부서",
    "author": "작성자"
}

• FAQ
  Q: 한글 인코딩 에러가 나요.
  A: encoding="utf-8" 파라미터를 명시하세요.
  Q: 청크 크기를 어떻게 정해야 하나요?
  A: 문서 타입과 LLM 컨텍스트 크기를 고려하세요. 일반적으로 500자 정도가 적당합니다.
  Q: PDF에서 표나 이미지는 어떻게 처리하나요?
  A: PyPDFLoader는 텍스트만 추출합니다. 표는 UnstructuredPDFLoader를, 이미지는 별도 처리가 필요합니다.
  Q: 분할 후 문맥이 끊기는 문제는?
  A: chunk_overlap을 적절히 설정하고, RecursiveCharacterTextSplitter를 사용하세요.