LangChain에서 RAG 사용하기

여러가지 로더 소개

파일 시스템 로더: CSV, JSON, PDF, Text, Directory, MS Office …
웹 기반 로더: 웹 페이지, 여러 웹 페이지에 대한 재귀적 로드, 사이트맵 기반 웹 페이지 로드, 유튜브 비디오 트랜스크립트 …
클라우드 스토리지
데이터베이스
협업 도구, 소셜 미디어, 오디오 …

대부분의 로더는 문서와 함께 메타데이터를 제공합니다.

2. 문서를 변환하기

Data Transformer: 문서를 청킹하기

!pip install langchain-text-splitters==0.3.0

from langchain_text_splitters import CharacterTextSplitter

text_splitter = CharacterTextSplitter(
    chunk_size = 1000,
    chunk_overlap = 0
)

docs = text_splitter.split_documents(raw_docs)

단순 청킹 이외에도 다양한 변환 처리가 가능합니다:

HTML을 일반 텍스트로 변환
메타데이터 추출
문서 번역
사용자 질문과의 관련성을 높이기 위해 문서에서 Q&A 생성

3. 임베딩하기

from langchain_openai import OpenAIEmbeddings

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

query = "What is LangChain?"

vector = embeddings.embed_query(query)

4. Vector Store 구축

여기서는 Chroma를 사용합니다:

# pip install langchain-chroma==0.1.4

from langchain_chroma import Chroma

db = Chroma.from_documents(
    docs,
    embeddings
)

retreiver = db.as_retriever()

query = "What is LangChain?"

context = retreiver.invoke(query)
first_doc = context[0]
print(first_doc.metadata)
print(first_doc.page_content)

LCEL 활용하기

# LCEL을 활용한 RAG chain 구현

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_template(
    '''
    Answer the question based on the context provided.
    Context: {context}
    Question: {question}
    '''
)

model = ChatOpenAI(
    model = "gpt-4o-mini",
    temperature = 0
)

from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

chain.invoke({"question": "What is LangChain?"})

📚 추가로 알아볼 내용

1. 고급 Document Loaders

AsyncWebCrawler:

from langchain_community.document_loaders import AsyncWebCrawler
import asyncio

async def crawl_websites():
    loader = AsyncWebCrawler(urls=["https://example1.com", "https://example2.com"])
    docs = await loader.aload()
    return docs

# 비동기 처리로 속도 향상

UnstructuredFileLoader (다양한 파일 형식):

from langchain_community.document_loaders import UnstructuredFileLoader

# PDF, DOCX, PPTX, HTML, TXT 등 자동 처리
loader = UnstructuredFileLoader("mixed_documents/")
docs = loader.load()

DatabaseLoader:

from langchain_community.document_loaders import SQLDatabaseLoader

loader = SQLDatabaseLoader(
    "postgresql://user:pass@localhost/db",
    "SELECT content, metadata FROM documents WHERE active = true"
)
docs = loader.load()

2. 정교한 텍스트 분할 전략

RecursiveCharacterTextSplitter (권장):

from langchain_text_splitters import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,  # 겹침으로 컨텍스트 보존
    separators=["\n\n", "\n", " ", ""]  # 우선순위별 구분자
)

TokenTextSplitter (토큰 기반):

from langchain_text_splitters import TokenTextSplitter

# 정확한 토큰 수 기반 분할
splitter = TokenTextSplitter(
    encoding_name="gpt2",  # 또는 "cl100k_base" for GPT-4
    chunk_size=512,
    chunk_overlap=50
)

SemanticChunker (의미 기반 분할):

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

semantic_chunker = SemanticChunker(
    OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile"  # 의미 변화 지점에서 분할
)

3. 벡터 스토어 최적화

메타데이터 필터링:

# 문서 저장 시 메타데이터 추가
docs_with_metadata = [
    Document(
        page_content="내용",
        metadata={"source": "document1.pdf", "page": 1, "category": "technical"}
    )
]

db = Chroma.from_documents(docs_with_metadata, embeddings)

# 필터 검색
retriever = db.as_retriever(
    search_kwargs={"filter": {"category": "technical"}}
)

하이브리드 검색 설정:

# 유사도 + 키워드 검색 조합
retriever = db.as_retriever(
    search_type="mmr",  # Maximal Marginal Relevance
    search_kwargs={
        "k": 10,
        "lambda_mult": 0.7  # 다양성과 관련성 균형 조절
    }
)

4. 성능 최적화 전략

배치 임베딩:

# 한 번에 여러 문서 임베딩 (효율적)
embeddings = OpenAIEmbeddings()
texts = [doc.page_content for doc in docs]
vectors = embeddings.embed_documents(texts)  # 배치 처리

인덱스 최적화:

# FAISS 사용 (대용량 데이터에 적합)
from langchain_community.vectorstores import FAISS

db = FAISS.from_documents(docs, embeddings)
db.save_local("faiss_index")  # 인덱스 저장

# 로드 시 빠른 시작
db = FAISS.load_local("faiss_index", embeddings)

🔍 학습하면서 알게 된 점

1. 청킹 전략의 실제 영향

잘못된 청킹의 문제:

# 너무 작은 청크 - 컨텍스트 부족
small_splitter = CharacterTextSplitter(chunk_size=100, chunk_overlap=0)

# 너무 큰 청크 - 노이즈 증가
large_splitter = CharacterTextSplitter(chunk_size=5000, chunk_overlap=0)

# 최적화된 청킹
optimal_splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,      # 충분한 컨텍스트
    chunk_overlap=100,   # 20% 겹침으로 경계 정보 보존
    separators=["\n\n", "\n", ". ", " "]  # 자연스러운 구분점
)

청킹 품질 평가:

def evaluate_chunks(chunks):
    avg_length = sum(len(chunk.page_content) for chunk in chunks) / len(chunks)
    print(f"평균 청크 길이: {avg_length}")
    print(f"총 청크 수: {len(chunks)}")

    # 너무 짧거나 긴 청크 확인
    short_chunks = [c for c in chunks if len(c.page_content) < 200]
    long_chunks = [c for c in chunks if len(c.page_content) > 2000]

    print(f"짧은 청크 수: {len(short_chunks)}")
    print(f"긴 청크 수: {len(long_chunks)}")

2. 임베딩 모델 선택의 중요성

모델별 특성:

# OpenAI - 범용성 좋음, 비용 있음
openai_embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# Sentence Transformers - 무료, 한국어 지원 좋음
from langchain_community.embeddings import HuggingFaceEmbeddings
huggingface_embeddings = HuggingFaceEmbeddings(
    model_name="jhgan/ko-sroberta-multitask"
)

# 모델 성능 비교 필요
def compare_embeddings(text, embedding_models):
    for name, model in embedding_models.items():
        vector = model.embed_query(text)
        print(f"{name}: 벡터 차원 {len(vector)}")

3. 검색 품질 향상 기법

컨텍스트 압축:

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

# 관련없는 내용 제거
compressor = LLMChainExtractor.from_llm(llm)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=retriever
)

Self-Query 검색:

from langchain.retrievers.self_query.base import SelfQueryRetriever

# 자연어 질문을 메타데이터 필터로 변환
retriever = SelfQueryRetriever.from_llm(
    llm,
    vectorstore,
    document_content_description="기술 문서",
    metadata_field_info=[
        {"name": "source", "description": "문서 출처", "type": "string"},
        {"name": "date", "description": "작성 날짜", "type": "string"}
    ]
)

4. RAG 파이프라인 평가 및 디버깅

검색 결과 평가:

def debug_retrieval(question, retriever, k=3):
    docs = retriever.get_relevant_documents(question)

    print(f"질문: {question}")
    print(f"검색된 문서 수: {len(docs)}")

    for i, doc in enumerate(docs[:k]):
        print(f"\n문서 {i+1}:")
        print(f"내용: {doc.page_content[:200]}...")
        print(f"메타데이터: {doc.metadata}")

답변 품질 모니터링:

def evaluate_rag_response(question, context_docs, answer):
    # 컨텍스트 관련성 체크
    context_text = "\n".join([doc.page_content for doc in context_docs])

    relevance_prompt = f"""
    질문: {question}
    컨텍스트: {context_text}
    답변: {answer}

    답변이 컨텍스트를 기반으로 하고 있는지 1-10점으로 평가하세요.
    """

    # LLM으로 답변 품질 평가
    score = model.invoke(relevance_prompt)
    return score

5. 프로덕션 환경에서의 고려사항

벡터DB 선택 기준:

# 소규모 프로토타입 - Chroma
chroma_db = Chroma.from_documents(docs, embeddings)

# 중규모 서비스 - Pinecone
from langchain_pinecone import Pinecone
pinecone_db = Pinecone.from_documents(docs, embeddings, index_name="my-index")

# 대규모 엔터프라이즈 - Weaviate
from langchain_community.vectorstores import Weaviate
weaviate_db = Weaviate.from_documents(docs, embeddings)

캐싱 전략:

from functools import lru_cache

@lru_cache(maxsize=100)
def cached_retrieval(question: str):
    return retriever.get_relevant_documents(question)

# Redis 캐싱
import redis
import pickle

redis_client = redis.Redis()

def redis_cached_retrieval(question):
    cached = redis_client.get(f"rag:{question}")
    if cached:
        return pickle.loads(cached)

    result = retriever.get_relevant_documents(question)
    redis_client.setex(f"rag:{question}", 3600, pickle.dumps(result))
    return result

모니터링 메트릭:

검색 지연시간
답변 생성 시간
토큰 사용량
검색 정확도 (사용자 피드백 기반)
컨텍스트 활용률

RAG (Retrieval-Augmented Generation) 완전 정복