포스팅 개요
1편에서 RAG의 개념과 아키텍처를 정리했습니다. 이번 편에서는 RAG를 구현할 때 많이 쓰는 프레임워크인 LangChain을 다룹니다.
LangChain은 LLM 기반 애플리케이션을 만들기 위한 프레임워크입니다. 문서 로딩, 텍스트 분할, 벡터화, 검색, 답변 생성까지 RAG에 필요한 구성 요소를 컴포넌트로 제공합니다.
1. LangChain이란
LangChain은 LLM 애플리케이션을 만들기 위한 Python/JavaScript 프레임워크입니다. RAG처럼 여러 단계가 이어지는 워크플로우를 구성할 때 특히 유용합니다.
(LangChain이 제공하는 구성 요소)
2. LangChain을 쓰는 이유
직접 구현하는 경우
# 문서 로딩 - 각각 다른 방법
pdf_text = extract_pdf("file.pdf")
word_text = extract_docx("file.docx")
web_text = scrape_webpage("url")
# 텍스트 분할 - 직접 로직 작성
def chunk_text(text, size=1000):
chunks = []
for i in range(0, len(text), size):
chunks.append(text[i:i+size])
return chunks
# 임베딩 - 각 모델마다 다른 방식
from sentence_transformers import SentenceTransformer
model = SentenceTransformer('model-name')
embeddings = model.encode(chunks)
# 벡터 DB - 직접 구축
import numpy as np
def similarity_search(query_vector, doc_vectors, top_k=5):
# 코사인 유사도 계산 로직
# 상위 k개 문서 반환 로직
pass
# LLM 호출 - API별로 다른 방식
import openai
response = openai.ChatCompletion.create(...)직접 구현하면 코드가 길어지고, 소스나 모델이 바뀔 때마다 연동 코드를 새로 작성해야 합니다.
LangChain으로 구현하는 경우
from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.llms import OpenAI
# 문서 로딩 - 통일된 방식
loader = PyPDFLoader("document.pdf")
documents = loader.load()
# 텍스트 분할 - 구분자 우선순위에 따라 재귀적으로 분할
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200
)
texts = text_splitter.split_documents(documents)
# 벡터화 & 저장
vectorstore = Chroma.from_documents(texts, OpenAIEmbeddings())
# 검색 & 답변 생성 - 체인으로 연결
from langchain.chains import RetrievalQA
qa = RetrievalQA.from_chain_type(
llm=OpenAI(),
retriever=vectorstore.as_retriever()
)
# 질문하기
result = qa.run("사용자 질문")같은 흐름을 훨씬 짧은 코드로 작성할 수 있습니다.
3. LangChain의 핵심 컴포넌트
LangChain은 RAG 구현에 필요한 요소를 컴포넌트 형태로 제공하고, 로더·임베딩 모델·벡터 DB·LLM을 같은 인터페이스로 교체할 수 있게 합니다.
3-1. Document Loaders: 문서 불러오기
Document Loader는 PDF, 웹페이지 등 여러 형태의 문서를 LangChain에서 다룰 수 있는 Document 객체로 변환합니다.
from langchain.document_loaders import (
PyPDFLoader, # PDF 파일
DirectoryLoader, # 폴더 전체
TextLoader, # 텍스트 파일
WebBaseLoader, # 웹페이지
CSVLoader # CSV 파일
)
# PDF 로딩
pdf_loader = PyPDFLoader("manual.pdf")
pdf_docs = pdf_loader.load()
# 폴더 전체 로딩
dir_loader = DirectoryLoader(
"./documents",
glob="**/*.txt",
loader_cls=TextLoader
)
text_docs = dir_loader.load()
# 웹페이지 로딩
web_loader = WebBaseLoader("https://example.com")
web_docs = web_loader.load()각 로더가 반환하는 Document 객체의 구조는 다음과 같습니다.
Document(
page_content="문서의 실제 내용",
metadata={
"source": "파일 경로 또는 URL",
"page": 0, # PDF의 경우 페이지 번호 (0부터 시작)
"title": "문서 제목"
}
)
(LangChain의 Document Loader)
3-2. Text Splitters: 텍스트 분할
큰 문서를 적절한 크기로 나누는 일은 검색 품질에 직접 영향을 줍니다. LangChain은 여러 분할 방식을 제공합니다.
from langchain.text_splitter import (
RecursiveCharacterTextSplitter, # 범용적으로 사용
CharacterTextSplitter, # 단순 문자 기준 분할
TokenTextSplitter, # 토큰 기준 분할
MarkdownTextSplitter # 마크다운 구조 기준 분할
)
# 가장 많이 사용되는 RecursiveCharacterTextSplitter
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, # 청크 크기
chunk_overlap=200, # 겹치는 부분
separators=["\n\n", "\n", " ", ""] # 분할 우선순위
)
chunks = text_splitter.split_documents(documents)chunk_overlap을 두는 이유
청크 경계에서 문맥이 잘리는 것을 막기 위해서입니다. 예를 들어 "파이썬은 프로그래밍 언어입니다. 파이썬의 특징은..."에서 첫 번째 청크가 첫 문장으로 끝나면, 두 번째 청크는 "파이썬의 특징은..."으로 시작해 앞선 맥락이 사라집니다.
overlap을 설정하면 앞 청크의 끝부분이 다음 청크에도 포함되므로 맥락이 이어집니다.
(텍스트 분할과 overlap)
3-3. Embeddings: 텍스트 벡터화
텍스트를 숫자 벡터로 변환합니다. 임베딩 모델이 달라도 같은 인터페이스로 사용할 수 있습니다.
from langchain.embeddings import (
OpenAIEmbeddings, # OpenAI 임베딩
HuggingFaceEmbeddings, # Hugging Face 모델들
OllamaEmbeddings # Ollama 로컬 모델
)
# OpenAI 임베딩 (유료)
openai_embeddings = OpenAIEmbeddings(
model="text-embedding-ada-002"
)
# 무료 한국어 임베딩 모델
hf_embeddings = HuggingFaceEmbeddings(
model_name="jhgan/ko-sroberta-multitask"
)
# 로컬 임베딩 (Ollama)
ollama_embeddings = OllamaEmbeddings(
model="nomic-embed-text"
)
# 사용 방법은 모두 동일
vector = hf_embeddings.embed_query("안녕하세요")
doc_vectors = hf_embeddings.embed_documents(["문서1", "문서2"])3-4. Vector Stores: 벡터 데이터베이스
임베딩된 벡터를 저장하고 유사도 검색을 수행하는 컴포넌트입니다.
from langchain.vectorstores import (
Chroma, # 로컬용 벡터 DB
Pinecone, # 클라우드 벡터 DB
FAISS, # Meta의 유사도 검색 라이브러리
Weaviate # 오픈소스 벡터 DB
)
# Chroma 사용 (가장 간단)
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
persist_directory="./chroma_db" # 로컬 저장
)
# 검색 수행
similar_docs = vectorstore.similarity_search(
"파이썬 설치 방법",
k=5 # 상위 5개 문서 반환
)
# 점수와 함께 반환 (Chroma는 거리 값이라 작을수록 유사)
similar_docs_with_score = vectorstore.similarity_search_with_score(
"파이썬 설치 방법",
k=5
)
(벡터 데이터베이스의 유사도 검색)
3-5. LLMs: 대화형 언어 모델
답변 생성을 담당하는 언어 모델 컴포넌트입니다.
from langchain.llms import (
OpenAI, # OpenAI GPT 모델들
Ollama, # 로컬 LLM
HuggingFacePipeline # Hugging Face 모델들
)
from langchain.chat_models import (
ChatOpenAI, # GPT-3.5, GPT-4
ChatOllama # Ollama 채팅 모델
)
# OpenAI 사용
openai_llm = ChatOpenAI(
model_name="gpt-3.5-turbo",
temperature=0.7
)
# Ollama 로컬 모델 사용
ollama_llm = ChatOllama(
model="llama3.1:8b",
temperature=0.7
)
# 직접 호출
response = openai_llm.predict("안녕하세요, 파이썬에 대해 설명해주세요.")4. 코드로 보는 LangChain 활용
앞에서 정리한 컴포넌트를 조합해 간단한 RAG 시스템을 구성합니다.
4-1. 기본 RAG 파이프라인
from langchain.document_loaders import DirectoryLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import Chroma
from langchain.llms import Ollama
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
# 1단계: 문서 로딩
print("📁 문서 로딩 중...")
loader = DirectoryLoader(
"./documents",
glob="**/*.txt",
loader_cls=TextLoader
)
documents = loader.load()
print(f"로딩된 문서 수: {len(documents)}개")
# 2단계: 텍스트 분할
print("✂️ 텍스트 분할 중...")
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
separators=["\n\n", "\n", ".", "!", "?", ",", " ", ""]
)
splits = text_splitter.split_documents(documents)
print(f"분할된 청크 수: {len(splits)}개")
# 3단계: 임베딩 & 벡터 저장
print("🔢 벡터화 중...")
embeddings = HuggingFaceEmbeddings(
model_name="jhgan/ko-sroberta-multitask"
)
vectorstore = Chroma.from_documents(
documents=splits,
embedding=embeddings,
persist_directory="./chroma_db"
)
print("벡터 DB 구축 완료!")
# 4단계: LLM 설정
print("🤖 LLM 설정 중...")
llm = Ollama(model="llama3.1:8b")
# 5단계: RAG 체인 구성
retriever = vectorstore.as_retriever(
search_kwargs={"k": 3} # 상위 3개 문서 검색
)
# 커스텀 프롬프트 설정
prompt_template = """다음 문서들을 참고하여 질문에 답변해주세요:
{context}
질문: {question}
답변: 위 문서들을 바탕으로 정확하고 자세한 답변을 제공해주세요.
문서에 명시되지 않은 내용은 추측하지 마시고, "제공된 문서에서 해당 정보를 찾을 수 없습니다"라고 말씀해주세요.
"""
PROMPT = PromptTemplate(
template=prompt_template,
input_variables=["context", "question"]
)
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=retriever,
chain_type_kwargs={"prompt": PROMPT},
return_source_documents=True
)
# 6단계: 질문하기
def ask_question(question):
print(f"\n🔍 질문: {question}")
print("💭 답변 생성 중...")
result = qa_chain({"query": question})
print(f"📝 답변: {result['result']}")
print(f"\n📚 참고 문서:")
for i, doc in enumerate(result['source_documents'], 1):
print(f"{i}. {doc.metadata.get('source', '알 수 없음')}")
print(f" 내용 미리보기: {doc.page_content[:100]}...")
# 사용 예시
ask_question("파이썬 설치 방법을 알려주세요")
ask_question("라이브러리를 설치하는 방법은 무엇인가요?")4-2. 고급 기능: 메모리와 대화 기록
메모리를 추가하면 이전 대화를 기억하는 RAG 시스템을 만들 수 있습니다.
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationalRetrievalChain
# 대화 기록을 저장할 메모리
memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True,
output_key="answer"
)
# 대화형 RAG 체인
conversational_chain = ConversationalRetrievalChain.from_llm(
llm=llm,
retriever=retriever,
memory=memory,
return_source_documents=True
)
# 연속적인 대화
def chat():
print("💬 대화형 RAG 시스템 시작! (종료하려면 'quit' 입력)")
while True:
question = input("\n사용자: ")
if question.lower() == 'quit':
break
result = conversational_chain({"question": question})
print(f"AI: {result['answer']}")
chat()4-3. 여러 문서 형식 처리
from langchain.document_loaders import (
DirectoryLoader,
PyPDFLoader,
Docx2txtLoader,
CSVLoader
)
def load_various_documents():
all_documents = []
# PDF 파일들
pdf_loader = DirectoryLoader(
"./documents",
glob="**/*.pdf",
loader_cls=PyPDFLoader
)
pdf_docs = pdf_loader.load()
all_documents.extend(pdf_docs)
# Word 문서들
docx_loader = DirectoryLoader(
"./documents",
glob="**/*.docx",
loader_cls=Docx2txtLoader
)
docx_docs = docx_loader.load()
all_documents.extend(docx_docs)
# CSV 파일들
csv_loader = DirectoryLoader(
"./documents",
glob="**/*.csv",
loader_cls=CSVLoader
)
csv_docs = csv_loader.load()
all_documents.extend(csv_docs)
return all_documents
# 사용
mixed_documents = load_various_documents()
print(f"총 {len(mixed_documents)}개 문서 로딩 완료")5. 체인(Chain)
체인(Chain) 은 여러 처리 단계를 하나로 연결하는 LangChain의 핵심 개념입니다.
5-1. 기본 체인
from langchain.chains import (
LLMChain, # 기본 LLM 체인
RetrievalQA, # 검색 기반 QA
StuffDocumentsChain, # 문서를 한 프롬프트에 모두 넣어 처리
MapReduceDocumentsChain # 문서별로 처리한 뒤 결과를 합침
)
# LLMChain: 가장 기본적인 체인
from langchain.prompts import PromptTemplate
prompt = PromptTemplate(
input_variables=["topic"],
template="{topic}에 대해 간단히 설명해주세요."
)
llm_chain = LLMChain(llm=llm, prompt=prompt)
result = llm_chain.run(topic="파이썬")5-2. 커스텀 체인 만들기
from langchain.chains.base import Chain
from typing import Dict, List
class CustomRAGChain(Chain):
"""커스텀 RAG 체인"""
vectorstore: object
llm: object
@property
def input_keys(self) -> List[str]:
return ["query"]
@property
def output_keys(self) -> List[str]:
return ["result", "sources"]
def _call(self, inputs: Dict[str, str]) -> Dict[str, str]:
query = inputs["query"]
# 1. 문서 검색
docs = self.vectorstore.similarity_search(query, k=3)
# 2. 컨텍스트 구성
context = "\n".join([doc.page_content for doc in docs])
# 3. 프롬프트 생성
prompt = f"""
다음 문서를 참고하여 질문에 답하세요:
{context}
질문: {query}
답변:
"""
# 4. LLM 호출
result = self.llm.predict(prompt)
return {
"result": result,
"sources": [doc.metadata for doc in docs]
}
# 사용
custom_chain = CustomRAGChain(
vectorstore=vectorstore,
llm=llm
)
result = custom_chain({"query": "파이썬이란?"})
(LangChain 체인의 연결 구조)
6. 활용 팁
6-1. 성능 최적화
# 1. 임베딩 캐싱
from langchain.embeddings import CacheBackedEmbeddings
from langchain.storage import LocalFileStore
fs = LocalFileStore("./cache/")
cached_embedder = CacheBackedEmbeddings.from_bytes_store(
underlying_embeddings=embeddings,
document_embedding_cache=fs
)
# 2. 벡터스토어 최적화
vectorstore = Chroma.from_documents(
documents=splits,
embedding=cached_embedder,
persist_directory="./chroma_db",
# 컬렉션 메타데이터 설정
collection_metadata={"hnsw:space": "cosine"}
)
# 3. 검색 설정 최적화
retriever = vectorstore.as_retriever(
search_type="mmr", # MMR (Maximal Marginal Relevance)
search_kwargs={
"k": 5,
"lambda_mult": 0.25 # 다양성 vs 관련성 균형
}
)6-2. 에러 처리와 로깅
import logging
from langchain.callbacks import get_openai_callback
# 로깅 설정
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def safe_rag_query(query: str):
try:
with get_openai_callback() as cb:
result = qa_chain({"query": query})
logger.info(f"토큰 사용량: {cb.total_tokens}")
logger.info(f"비용: ${cb.total_cost:.4f}")
return result
except Exception as e:
logger.error(f"RAG 처리 중 오류 발생: {e}")
return {"result": "죄송합니다. 처리 중 오류가 발생했습니다."}
# 사용
result = safe_rag_query("파이썬이란 무엇인가요?")6-3. 메타데이터 활용
# 문서에 메타데이터 추가
documents_with_metadata = []
for doc in documents:
doc.metadata.update({
"document_type": "manual",
"language": "korean",
"created_date": "2025-01-01"
})
documents_with_metadata.append(doc)
# 메타데이터 기반 필터링
retriever = vectorstore.as_retriever(
search_kwargs={
"k": 5,
"filter": {"document_type": "manual"} # 매뉴얼만 검색
}
)마무리
2편에서는 LangChain의 개념과 RAG 구현에 쓰이는 컴포넌트를 정리했습니다.
LangChain을 쓰면 각 단계의 구현체를 요구사항에 맞게 교체하며 조합할 수 있고, 체인으로 전체 흐름을 한 번에 구성할 수 있습니다.
다음 3편에서는 RAG의 또 다른 핵심 요소인 벡터 데이터베이스의 개념과 동작 방식, 활용 방법을 정리합니다.