2026-03-19 강의 정리

LangChain의 RAG 콤포넌트 - 문서 로더 + 텍스트 분할

게시 2026/03/19

By kimseokryu

23 분읽는 시간

2026-03-19 강의 정리

오늘 강의 개요
LangChain Document 클래스 복습
파일 로딩 - 텍스트 파일 (.txt)
파일 로딩 - CSV 파일
파일 로딩 - JSON 파일
파일 형식별 비교 정리
Fine-tuning vs RAG 비교
코사인 유사도 기반 RAG 직접 구현
rag_with_langchain - LLM 연결
클래스(OOP) 기초 - Calculator, WordCounter
DocumentStore 클래스 - RAG를 OOP로 구현
collections.Counter 활용
핵심 개념 총정리
자주 나오는 실수 / 주의사항

1. 오늘 강의 개요

오늘은 실제 파일에서 문서를 로딩하고, 코사인 유사도 기반의 간단한 RAG를 직접 구현한다. 또한 OOP(클래스)로 문서 검색 시스템을 만든다.

오늘의 핵심 흐름

파일 (txt / csv / json)
        ↓ 로딩
Document 객체들 (page_content + metadata)
        ↓ 임베딩
벡터들 (1536차원)
        ↓ 코사인 유사도 검색
관련 문서 top-k개
        ↓ context 구성
LLM 답변 생성

오늘 만들 것

각 파일 형식별 로더 함수
retrieve_by_similarity() - 코사인 유사도 기반 검색
rag_with_langchain() - 검색 + LLM 답변
DocumentStore 클래스 - 위 기능을 OOP로 통합

2. LangChain Document 클래스 복습

Document 구조

  
from langchain_core.documents import Document

doc = Document(
    page_content = "문서의 실제 텍스트 내용",
    metadata     = {
        "source"    : "company_policy.txt",
        "category"  : "사내규정",
        "char_count": 500
    }
)

print(doc.page_content)           # 텍스트 내용
print(doc.metadata['source'])     # 파일명
print(doc.metadata['char_count']) # 문자 수

page_content : 임베딩되고 검색에 사용되는 실제 텍스트

metadata : 검색 결과에서 출처 표시, 필터링 등에 활용

왜 Document 클래스를 쓰나?

텍스트와 메타데이터를 함께 관리
LangChain의 모든 도구(벡터스토어, retriever 등)가 Document를 표준으로 사용
“어떤 파일의 어떤 부분에서 온 내용인지” 추적 가능

3. 파일 로딩 - 텍스트 파일 (.txt)

샘플 파일 생성

  
from pathlib import Path

SAMPLE_DIR = Path('sample_data')
SAMPLE_DIR.mkdir(exist_ok=True)

sample_texts = {
    'company_policy.txt': "주식회사 모두의연구소 사내 규정\n\n제1조 (목적)...",
    'ai_report.txt'     : "2024년 인공지능 산업 동향 보고서...",
    'product_manual.txt': "스마트 홈 허브 v3.0 사용자 매뉴얼...",
}

for filename, content in sample_texts.items():
    (SAMPLE_DIR / filename).write_text(content, encoding='utf-8')

텍스트 파일 로딩 함수

  
def load_text_files(directory):
    documents = []
    for fp in sorted(directory.glob('*.txt')):   # *.txt 파일 모두 찾기
        text = fp.read_text(encoding='utf-8')
        doc  = Document(
            page_content = text,
            metadata     = {
                "source"    : fp.name,
                "char_count": len(text),
            }
        )
        documents.append(doc)
    return documents

all_docs = load_text_files(Path('sample_data'))
print(f"로딩된 파일 수: {len(all_docs)}")

for doc in all_docs:
    print(f"  {doc.metadata['source']} ({doc.metadata['char_count']}자)")

Path.glob(‘*.txt’) : 디렉토리에서 .txt 파일 모두 찾기

sorted() : 파일명 알파벳 순 정렬 (일관성)

fp.read_text(encoding=’utf-8’) : 파일 내용 읽기

4. 파일 로딩 - CSV 파일

CSV 파일 생성

  
import csv

csv_data = [
    {'이름': '김철수', '부서': '개발팀', '직급': '선임'},
    {'이름': '이영희', '부서': '기획팀', '직급': '매니저'},
    {'이름': '박지민', '부서': '개발팀', '직급': '주임'},
]

with open(SAMPLE_DIR / 'employees.csv', 'w', encoding='utf-8', newline='') as f:
    writer = csv.DictWriter(f, fieldnames=['이름', '부서', '직급'])
    writer.writeheader()
    writer.writerows(csv_data)

CSV 로딩 함수

  
def load_csv_file(file_path):
    documents = []
    with open(file_path, 'r', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        for row in reader:
            # 각 컬럼을 "키: 값 | 키: 값" 형식으로 변환
            content = ' | '.join(f'{k}: {v}' for k, v in row.items())
            doc = Document(
                page_content = content,
                metadata     = {"source": file_path.name}
            )
            documents.append(doc)
    return documents

csv_docs = load_csv_file(SAMPLE_DIR / 'employees.csv')
for doc in csv_docs:
    print(doc.page_content)

CSV 로딩 결과

  
이름: 김철수 | 부서: 개발팀 | 직급: 선임
이름: 이영희 | 부서: 기획팀 | 직급: 매니저
이름: 박지민 | 부서: 개발팀 | 직급: 주임

csv.DictReader : CSV를 딕셔너리 형태로 읽기 (첫 행 = 헤더)

’ ‘.join(…) : 키-값 쌍을 구분자로 이어 붙임

각 행(row) → 하나의 Document로 변환

5. 파일 로딩 - JSON 파일

JSON 파일 생성

  
import json

test_json = {
    'name'    : 'RAG 프로젝트',
    'version' : '1.0',
    'features': ['검색', '생성', '인덱싱'],
    'author'  : '모두의연구소'
}

with open('sample_data/test.json', 'w', encoding='utf-8') as f:
    json.dump(test_json, f, ensure_ascii=False, indent=2)

JSON 로딩 함수

  
def load_json_as_document(file_path):
    path = Path(file_path)
    with open(path, 'r', encoding='utf-8') as f:
        data = json.load(f)

    # JSON 전체를 들여쓰기 형식 문자열로 변환 → page_content
    content = json.dumps(data, ensure_ascii=False, indent=2)
    keys    = list(data.keys())   # 최상위 키 목록

    return Document(
        page_content = content,
        metadata     = {'source': path.name, 'keys': keys}
    )

doc = load_json_as_document('sample_data/test.json')
print(doc.page_content)
# {
#   "name": "RAG 프로젝트",
#   "version": "1.0",
#   "features": ["검색", "생성", "인덱싱"],
#   ...
# }
print(doc.metadata['keys'])   # ['name', 'version', 'features', 'author']

json.loads() : 문자열 → 딕셔너리 (파싱)

json.dumps(indent=2) : 딕셔너리 → 들여쓰기 있는 JSON 문자열 (보기 좋음)

metadata의 keys 필드 : 어떤 키가 있는지 빠르게 확인 가능

6. 파일 형식별 비교 정리

파일 형식	로딩 도구	page_content 구성	특징
.txt	fp.read_text()	파일 전체 텍스트	가장 단순
.csv	csv.DictReader	“키: 값 \| 키: 값” (행별)	각 행 → Document
.json	json.load()	json.dumps(indent=2)	전체 JSON 문자열

실제 RAG 프로젝트에서

PDF, Word 파일: PyPDF2, python-docx + LangChain 로더 사용
웹 크롤링: langchain_community.document_loaders.WebBaseLoader
오늘 실습은 기초 파일 형식으로 원리 이해

7. Fine-tuning vs RAG 비교

둘 다 LLM에 새로운 지식을 추가하는 방법

방식	설명	특징
Fine-tuning	GPT 모델 자체를 추가 학습	비용 높음, 지식이 모델에 내재화, 업데이트 어려움, 학습 데이터 필요
RAG	외부 문서를 검색해서 답변	저비용, 지식 업데이트 쉬움, 실시간 최신 정보 반영 가능

실무에서는 대부분 RAG를 선택 이유: 회사 내부 문서는 자주 업데이트됨 → Fine-tuning은 업데이트마다 재학습 필요

Fine-tuning은 특수 도메인 용어/말투 학습에 사용

RAG의 기본 원리 (Retrieval-Augmented Generation)

문서들을 벡터로 변환하여 저장 (인덱싱)
사용자 질문도 벡터로 변환
코사인 유사도로 관련 문서 검색
검색된 문서를 context로 LLM에 전달
LLM이 context 기반으로 답변 생성

8. 코사인 유사도 기반 RAG 직접 구현

지식 베이스 정의

  
from sklearn.metrics.pairwise import cosine_similarity
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model='text-embedding-3-small')

knowledge_base = [
    {
        'id'     : 1,
        'content': '파이썬은 1991년 귀도 반 로섬이 만든 프로그래밍 언어입니다. '
                   '간결한 문법과 풍부한 라이브러리가 특징입니다.',
        'source' : 'programming_guide.txt'
    },
    {
        'id'     : 2,
        'content': 'RAG는 Retrieval-Augmented Generation의 약자로, '
                   'LLM이 외부 데이터를 참조하여 답변을 생성하는 기술입니다.',
        'source' : 'ai_glossary.txt'
    },
    {
        'id'     : 3,
        'content': '벡터 데이터베이스는 텍스트를 숫자 벡터로 변환하여 '
                   '유사도 기반 검색을 제공하는 데이터베이스입니다.',
        'source' : 'database_manual.txt'
    },
]

유사도 기반 검색 함수

  
def retrieve_by_similarity(query, documents, top_k=2):
    """코사인 유사도로 가장 관련 있는 문서 top_k개 반환"""
    contents  = [doc['content'] for doc in documents]

    query_vec = embeddings.embed_query(query)
    doc_vecs  = embeddings.embed_documents(contents)

    scores = cosine_similarity([query_vec], doc_vecs)[0]
    ranked = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)

    return [(doc, float(score)) for doc, score in ranked[:top_k]]

# 테스트
results = retrieve_by_similarity("RAG 기술에 대해 알려주세요", knowledge_base)
for doc, score in results:
    print(f"[{score:.4f}] (출처: {doc['source']}) {doc['content'][:50]}")
# [0.5590] (출처: ai_glossary.txt) RAG는 Retrieval-Augmented Generation...

데이터 흐름

query "RAG 기술에 대해..."
      ↓ embed_query()
query_vec [0.12, -0.34, ...]
      ↓ cosine_similarity()
doc_vecs와 유사도 계산: [0.56, 0.15, 0.23, ...]
      ↓ sorted(..., reverse=True)
[(RAG 문서, 0.56), (벡터DB 문서, 0.23), ...]  상위 2개 반환

9. rag_with_langchain - LLM 연결

검색 + LLM 답변 함수

  
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model='gpt-4o-mini')

def rag_with_langchain(query, documents):
    """검색된 문서를 context로 사용하여 LLM 답변 생성"""
    if not documents:
        return '참고할 문서가 없습니다'

    # 문서를 context 텍스트로 변환
    contexts = '\n'.join(f"- {doc}" for doc in documents)

    messages = [
        SystemMessage(content='제공된 문서를 기반으로 정확하게 답변해주세요. '
                              '문서에 없는 내용은 모른다고 하세요.'),
        HumanMessage(content=f'참고 문서:\n{contexts}\n\n질문: {query}')
    ]
    response = llm.invoke(messages)
    return response.content

# 통합 실행
query   = 'RAG가 무엇인가요?'
results = retrieve_by_similarity(query, knowledge_base, top_k=2)
top_docs = [doc['content'] for doc, _ in results]
answer  = rag_with_langchain(query, top_docs)
print(answer)

RAG 전체 흐름

사용자 질문
    ↓ retrieve_by_similarity()
관련 문서 top-k개 (cosine similarity 기반)
    ↓ context 구성
[참고 문서]\n- 문서1\n- 문서2
    ↓ rag_with_langchain()
LLM (문서 기반 답변 생성)
    ↓
최종 답변 (출처 명시)

10. 클래스(OOP) 기초 - Calculator, WordCounter

클래스 개념

  
# class : 변수(속성)와 메서드를 묶는 단위 (설계도)
# __init__ : 인스턴스 생성 시 자동 호출되는 초기화 메서드
# self : 인스턴스 자기 자신을 가리키는 참조

class Calculator:
    def __init__(self):
        self.history = []    # 인스턴스 변수

    def add(self, a, b):
        result = a + b
        self.history.append(f'{a} + {b} = {result}')
        return result

    def subtract(self, a, b):
        result = a - b
        self.history.append(f'{a} - {b} = {result}')
        return result

    def get_history(self):
        return self.history

사용 예시

  
calc = Calculator()       # 인스턴스 생성 (설계도 → 실체)
calc.add(1, 2)            # 3
calc.subtract(10, 3)      # 7
calc.get_history()        # ['1 + 2 = 3', '10 - 3 = 7']

calc2 = Calculator()      # 새 인스턴스 (별개의 history)
calc2.get_history()       # []  ← calc와 독립적!

인스턴스를 여러 개 만들면 각각 독립적인 데이터(history) 가짐

WordCounter 클래스

  
from collections import Counter

class WordCounter:
    def __init__(self):
        self.texts = []

    def add_text(self, text):
        self.texts.append(text)

    def count_words(self):
        """전체 단어 수 합계"""
        return sum(len(t.split()) for t in self.texts)

    def most_common(self, n):
        """가장 많이 등장한 n개 단어"""
        all_words = []
        for t in self.texts:
            all_words.extend(t.split())
        return Counter(all_words).most_common(n)

# 사용 예시
wc = WordCounter()
wc.add_text("AI가 세상을 바꾸고 있습니다 AI는 미래 기술입니다")
wc.add_text("AI와 머신러닝이 산업을 변화시키고 있습니다")
print(wc.count_words())    # 14
print(wc.most_common(3))   # [('AI가', 1), ('AI는', 1), ('AI와', 1)]

11. DocumentStore 클래스 - RAG를 OOP로 구현

설계

add() : 문서 추가
count() : 문서 수 확인
search() : 키워드 문자열 검색 (단순)
retrieve() : 임베딩 유사도 기반 검색
generate() : 검색 + LLM 답변 생성

전체 코드

  
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
from sklearn.metrics.pairwise import cosine_similarity

class DocumentStore:
    def __init__(self):
        self.documents = []    # 저장된 문서 리스트
        self.next_id   = 1     # 자동 증가 ID
        self.embeddings = OpenAIEmbeddings(model='text-embedding-3-small')
        self.llm        = ChatOpenAI(model='gpt-4o-mini')

    def add(self, content, source='unknown'):
        """문서 추가"""
        self.documents.append({
            'id'     : self.next_id,
            'content': content,
            'source' : source
        })
        self.next_id += 1

    def count(self):
        """저장된 문서 수 반환"""
        return len(self.documents)

    def search(self, keyword):
        """키워드 포함 문서 반환 (단순 문자열 검색)"""
        return [doc for doc in self.documents if keyword in doc['content']]

    def retrieve(self, query, top_k=3):
        """임베딩 유사도 기반 상위 k개 문서 반환"""
        if not self.documents:
            return []

        contents  = [doc['content'] for doc in self.documents]
        query_vec = self.embeddings.embed_query(query)
        doc_vecs  = self.embeddings.embed_documents(contents)

        scores = cosine_similarity([query_vec], doc_vecs)[0]
        ranked = sorted(zip(self.documents, scores), key=lambda x: x[1], reverse=True)
        return ranked[:top_k]

    def generate(self, query):
        """검색된 문서를 context로 사용하여 LLM 답변 생성"""
        retrieved = self.retrieve(query)
        if not retrieved:
            return "관련 문서를 찾을 수 없습니다."

        contexts = '\n'.join(
            f"[문서 {d[0]['id']}, 출처: {d[0]['source']}] {d[0]['content']}"
            for d in retrieved
        )
        messages = [
            SystemMessage(content='제공된 문서를 기반으로 정확하게 답변해주세요. '
                                   '출처를 명시해주세요.'),
            HumanMessage(content=f'참고 문서:\n{contexts}\n\n질문: {query}')
        ]
        return self.llm.invoke(messages).content

사용 예시

  
store = DocumentStore()

# 문서 추가
store.add('파이썬은 데이터 분석에 좋다', 'guide.txt')
store.add('RAG는 검색증강 생성이다',     'glossary.txt')
store.add('LangChain은 LLM 프레임워크', 'langchain.txt')

print(store.count())              # 3

# 키워드 검색
results = store.search('파이썬')
print(results)   # [{'id': 1, 'content': '파이썬은...', 'source': 'guide.txt'}]

# 임베딩 유사도 검색
results = store.retrieve('RAG 기술이 뭔가요?', top_k=2)
for doc, score in results:
    print(f"[{score:.4f}] {doc['content']}")

# RAG 답변 생성
answer = store.generate('RAG에 대해 설명해주세요')
print(answer)
# RAG는 "검색증강 생성"의 약자로... [문서 2, 출처: glossary.txt]

OOP로 묶는 장점

여러 DocumentStore 인스턴스를 독립적으로 운영 가능 (사내 규정용, 제품 매뉴얼용, AI 보고서용 각각 별도 store)
add/search/retrieve/generate 인터페이스가 일관됨
내부 구현(임베딩 모델, LLM)을 변경해도 사용 코드 불변

12. collections.Counter 활용

Counter란?

  
from collections import Counter

# 리스트의 각 요소 빈도 계산
all_words = [1, 1, 1, 1, 1, 2, 2, 2, 2, 3, 3, 3, 4, 4, 5]
c = Counter(all_words)
print(c)
# Counter({1: 5, 2: 4, 3: 3, 4: 2, 5: 1})

# 가장 많이 등장한 n개
c.most_common(2)
# [(1, 5), (2, 4)]   ← (값, 빈도) 튜플

텍스트 단어 빈도 분석

  
texts     = ["AI가 세상을 바꾸고 AI가 발전한다", "AI는 미래 기술이다"]
all_words = []
for t in texts:
    all_words.extend(t.split())   # 단어 리스트로 합침

print(Counter(all_words).most_common(3))
# [('AI가', 2), ('세상을', 1), ('바꾸고', 1)]

Counter(iterable) : 각 요소의 빈도를 딕셔너리 형태로

.most_common(n) : 상위 n개 (값, 빈도) 튜플 리스트

FAG 시스템에서 자주 질문되는 키워드 분석 등에 활용

13. 핵심 개념 총정리

개념	핵심 내용
Document 클래스	page_content + metadata로 문서 표현
.txt 로딩	Path.glob(‘*.txt’) + read_text()
.csv 로딩	DictReader로 각 행을 Document로 변환
.json 로딩	json.load() + json.dumps(indent=2)
RAG 기본 구조	검색(cosine similarity) → context → LLM 답변
Fine-tuning vs RAG	학습 vs 검색, 비용/업데이트 용이성 차이
class	설계도, 속성(변수) + 메서드(함수)의 묶음
__init__	인스턴스 생성 시 자동 호출, 초기화
self	인스턴스 자기 자신을 참조
DocumentStore	OOP로 add/count/search/retrieve/generate 통합
Counter	iterable 요소 빈도 계산, most_common(n)

오늘 배운 것의 RAG 시스템 위치

.txt/.csv/.json 로딩  ← 오늘 배움
        ↓
Document 객체들       ← 오늘 배움
        ↓
임베딩 → 벡터         ← 17일
        ↓
cosine_similarity 검색 ← 오늘 배움
        ↓
LLM 답변 생성         ← 오늘 배움
        ↓
FAISS + LCEL 체인     ← 내일 (20일)

14. 자주 나오는 실수 / 주의사항

실수 1: CSV 파일 쓸 때 newline=’’ 빠뜨리기

  
# 빠뜨리면 Windows에서 각 행마다 빈 줄이 생김
with open('employees.csv', 'w', encoding='utf-8') as f:    # 잘못됨
with open('employees.csv', 'w', encoding='utf-8', newline='') as f:  # 올바름

실수 2: JSON dumps vs loads 혼동

  
# json.loads() : 문자열 → 딕셔너리 (parsing, 읽기)
data = json.loads('{"name": "RAG"}')

# json.dumps() : 딕셔너리 → 문자열 (serialization, 쓰기)
text = json.dumps({"name": "RAG"}, ensure_ascii=False)

# 파일 작업:
# json.load(file)  : 파일 → 딕셔너리
# json.dump(data, file) : 딕셔너리 → 파일

실수 3: init에서 self 빠뜨리기

  
class DocumentStore:
    def __init__(self):
        documents = []    # 틀림! 로컬 변수로 생성됨

    def add(self, content):
        documents.append(content)   # NameError! documents가 없음

# 올바른 코드
class DocumentStore:
    def __init__(self):
        self.documents = []   # self.로 인스턴스 변수 생성

    def add(self, content):
        self.documents.append(content)   # self.로 접근

실수 4: retrieve 결과의 구조 착각

  
# retrieve()는 [(doc_dict, score), ...] 반환
results = store.retrieve("RAG란?")

for item in results:
    print(item)          # (딕셔너리, 점수) 튜플

# 올바른 언패킹
for doc, score in results:
    print(f"score: {score:.4f}")
    print(f"content: {doc['content']}")

실수 5: glob 패턴 주의

  
# .txt 파일만 찾기
directory.glob('*.txt')    # 올바름 (현재 폴더)
directory.glob('**/*.txt') # 하위 폴더까지 재귀 탐색

# 파일 이름에 숫자 있을 때 정렬 주의
sorted(directory.glob('*.txt'))        # 문자열 정렬 (0, 1, 10, 2, ...)
sorted(directory.glob('*.txt'), key=lambda p: p.name)  # 이름 기준

끝

모두의연구소, AI/LLM 서비스 개발 과정

AI LLM

목차

1. 오늘 강의 개요

오늘의 핵심 흐름

오늘 만들 것

2. LangChain Document 클래스 복습

Document 구조

왜 Document 클래스를 쓰나?

3. 파일 로딩 - 텍스트 파일 (.txt)

샘플 파일 생성

텍스트 파일 로딩 함수

4. 파일 로딩 - CSV 파일

CSV 파일 생성

CSV 로딩 함수

CSV 로딩 결과

5. 파일 로딩 - JSON 파일

JSON 파일 생성

JSON 로딩 함수

6. 파일 형식별 비교 정리

실제 RAG 프로젝트에서

7. Fine-tuning vs RAG 비교

둘 다 LLM에 새로운 지식을 추가하는 방법

RAG의 기본 원리 (Retrieval-Augmented Generation)

8. 코사인 유사도 기반 RAG 직접 구현

지식 베이스 정의

유사도 기반 검색 함수

데이터 흐름

9. rag_with_langchain - LLM 연결

검색 + LLM 답변 함수

RAG 전체 흐름

10. 클래스(OOP) 기초 - Calculator, WordCounter

클래스 개념

사용 예시

WordCounter 클래스

11. DocumentStore 클래스 - RAG를 OOP로 구현

설계

전체 코드

사용 예시

OOP로 묶는 장점

12. collections.Counter 활용

Counter란?

텍스트 단어 빈도 분석

13. 핵심 개념 총정리

오늘 배운 것의 RAG 시스템 위치

14. 자주 나오는 실수 / 주의사항

실수 1: CSV 파일 쓸 때 newline=’’ 빠뜨리기

실수 2: JSON dumps vs loads 혼동

실수 3: __init__에서 self 빠뜨리기

실수 4: retrieve 결과의 구조 착각

실수 5: glob 패턴 주의

끝

인기 태그

실수 3: init에서 self 빠뜨리기