1편: LangChain 기초 - 5분만에 RAG 만들기
안녕하세요!
"RAG를 만들어보고 싶은데 코드가 너무 복잡해 보여요." 이런 고민 하셨나요? LangChain은 바로 이 문제를 해결하기 위해 탄생했습니다. 복잡한 LLM 애플리케이션을 레고 블록처럼 조립할 수 있게 해주는 프레임워크죠.
이번 시리즈에서는 LangChain의 기초부터 고급 기능까지 실전 예제와 함께 완전히 정복해보겠습니다. 첫 편에서는 정말로 5분 안에 작동하는 RAG 시스템을 만들어볼게요.
LangChain이란?
LangChain은 LLM 기반 애플리케이션을 쉽게 개발할 수 있도록 도와주는 오픈소스 프레임워크입니다. 2022년 10월 출시 이후 급속도로 성장하여 현재 GitHub 스타 90k 이상을 기록하며 LLM 개발의 사실상 표준이 되었습니다.
LangChain의 핵심 가치는 세 가지입니다. 첫째, 모듈화된 컴포넌트로 복잡한 작업을 간단하게 구성할 수 있습니다. 둘째, 다양한 LLM, 벡터 스토어, 도구들을 통합된 인터페이스로 사용할 수 있습니다. 셋째, 프로덕션 레벨의 기능(에러 핸들링, 로깅, 스트리밍)을 기본 제공합니다.
환경 설정
먼저 필요한 패키지를 설치합니다.
pip install langchain langchain-openai langchain-community chromadbOpenAI API 키를 환경변수로 설정하세요.
export OPENAI_API_KEY='your-api-key-here'5분 RAG 실습: 첫 번째 방법
가장 간단한 방법부터 시작하겠습니다. 텍스트 파일 하나로 질문에 답하는 시스템을 만들어봅시다.
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
# Step 1: 문서 로딩
loader = TextLoader('company_handbook.txt', encoding='utf-8')
documents = loader.load()
# Step 2: 텍스트 분할
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50
)
splits = text_splitter.split_documents(documents)
# Step 3: 벡터 스토어 생성
vectorstore = Chroma.from_documents(
documents=splits,
embedding=OpenAIEmbeddings()
)
# Step 4: RAG 체인 생성
qa_chain = RetrievalQA.from_chain_type(
llm=ChatOpenAI(model="gpt-4o-mini", temperature=0),
chain_type="stuff",
retriever=vectorstore.as_retriever()
)
# Step 5: 질문하기
response = qa_chain.invoke({"query": "연차는 몇 일인가요?"})
print(response['result'])네, 이게 전부입니다! 불과 20줄 정도의 코드로 문서 기반 QA 시스템이 완성되었습니다.
코드 분석: 각 단계의 의미
-
Document Loader: LangChain은 다양한 문서 형식을 지원합니다. TextLoader 외에도 PDFLoader, CSVLoader, WebBaseLoader 등 50가지 이상의 로더가 있어서 거의 모든 형식의 데이터를 처리할 수 있습니다.
-
Text Splitter: RecursiveCharacterTextSplitter는 문서를 의미 있는 단위로 분할합니다. chunk_size는 각 조각의 크기, chunk_overlap은 조각 간 겹치는 부분입니다. 겹치는 부분을 두는 이유는 문맥이 잘리는 것을 방지하기 위해서입니다.
-
Vector Store: Chroma는 경량 벡터 데이터베이스입니다. 문서를 임베딩으로 변환하고 저장하는 역할을 합니다. 프로덕션에서는 Pinecone이나 Weaviate 같은 더 강력한 옵션도 있습니다.
-
Retriever: 벡터 스토어에서 관련 문서를 검색하는 역할입니다. 기본적으로 유사도가 높은 상위 4개 문서를 가져옵니다.
-
Chain: RetrievalQA는 검색된 문서를 LLM에 전달하고 답변을 생성하는 전체 파이프라인입니다.
더 나은 방법: LCEL 사용하기
LangChain Expression Language(LCEL)를 사용하면 더 명확하고 유연한 코드를 작성할 수 있습니다. 같은 기능을 LCEL로 구현해볼까요?
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
# 프롬프트 템플릿 정의
template = """다음 컨텍스트를 기반으로 질문에 답변하세요.
컨텍스트에 없는 정보라면 모른다고 답하세요.
컨텍스트: {context}
질문: {question}
답변:"""
prompt = ChatPromptTemplate.from_template(template)
# LCEL 체인 구성
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
rag_chain = (
{
"context": vectorstore.as_retriever() | format_docs,
"question": RunnablePassthrough()
}
| prompt
| ChatOpenAI(model="gpt-4o-mini", temperature=0)
| StrOutputParser()
)
# 사용하기
response = rag_chain.invoke("연차는 몇 일인가요?")
print(response)LCEL의 장점은 명확합니다. 파이프(|) 연산자로 각 단계가 시각적으로 연결되어 데이터 흐름을 쉽게 파악할 수 있고, 각 컴포넌트를 독립적으로 테스트하고 교체할 수 있으며, 스트리밍, 병렬 실행 등 고급 기능을 자동으로 지원합니다.
다양한 문서 형식 다루기
실무에서는 PDF, 웹페이지, API 등 다양한 소스의 데이터를 처리해야 합니다.
from langchain_community.document_loaders import (
PyPDFLoader,
WebBaseLoader,
DirectoryLoader
)
# PDF 로딩
pdf_loader = PyPDFLoader("report.pdf")
pdf_docs = pdf_loader.load()
# 웹페이지 로딩
web_loader = WebBaseLoader("https://example.com/article")
web_docs = web_loader.load()
# 디렉토리의 모든 문서 로딩
dir_loader = DirectoryLoader(
'./documents',
glob="**/*.txt",
loader_cls=TextLoader
)
dir_docs = dir_loader.load()
# 모든 문서 합치기
all_docs = pdf_docs + web_docs + dir_docs
# 이후 과정은 동일
splits = text_splitter.split_documents(all_docs)
vectorstore = Chroma.from_documents(splits, OpenAIEmbeddings())Retriever 커스터마이징
검색 품질을 높이기 위해 Retriever를 세밀하게 조정할 수 있습니다.
# 기본 retriever - 상위 4개 문서
basic_retriever = vectorstore.as_retriever()
# 검색 개수 조정
retriever_k = vectorstore.as_retriever(
search_kwargs={"k": 6}
)
# 유사도 임계값 설정 (0.8 이상만)
retriever_threshold = vectorstore.as_retriever(
search_type="similarity_score_threshold",
search_kwargs={"score_threshold": 0.8}
)
# MMR (Maximum Marginal Relevance) - 다양성 있는 결과
retriever_mmr = vectorstore.as_retriever(
search_type="mmr",
search_kwargs={"k": 6, "fetch_k": 20}
)MMR은 유사도가 높으면서도 서로 다른 정보를 담은 문서를 선택합니다. 같은 내용의 중복 문서를 피하고 싶을 때 유용합니다.
대화 기록 추가하기
단발성 질문이 아니라 대화를 이어가려면 어떻게 할까요?
from langchain.chains import create_history_aware_retriever
from langchain_core.prompts import MessagesPlaceholder
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.messages import HumanMessage, AIMessage
# 대화 기록을 고려한 검색 체인
contextualize_q_prompt = ChatPromptTemplate.from_messages([
("system", "채팅 기록과 최신 질문을 고려하여 독립적인 검색 쿼리를 생성하세요."),
MessagesPlaceholder("chat_history"),
("human", "{input}"),
])
history_aware_retriever = create_history_aware_retriever(
ChatOpenAI(model="gpt-4o-mini"),
vectorstore.as_retriever(),
contextualize_q_prompt
)
# 답변 생성 체인
qa_prompt = ChatPromptTemplate.from_messages([
("system", "다음 컨텍스트를 사용하여 질문에 답변하세요:\n\n{context}"),
MessagesPlaceholder("chat_history"),
("human", "{input}"),
])
question_answer_chain = create_stuff_documents_chain(
ChatOpenAI(model="gpt-4o-mini"),
qa_prompt
)
rag_chain = create_retrieval_chain(
history_aware_retriever,
question_answer_chain
)
# 대화하기
chat_history = []
# 첫 번째 질문
response = rag_chain.invoke({
"input": "연차는 몇 일인가요?",
"chat_history": chat_history
})
print(response['answer'])
# 대화 기록 업데이트
chat_history.append(HumanMessage(content="연차는 몇 일인가요?"))
chat_history.append(AIMessage(content=response['answer']))
# 후속 질문
response = rag_chain.invoke({
"input": "사용 기한은 어떻게 되나요?",
"chat_history": chat_history
})
print(response['answer'])이제 "그건 언제까지인가요?", "더 자세히 설명해주세요" 같은 맥락 의존적 질문도 정확하게 처리할 수 있습니다.
스트리밍으로 실시간 응답
긴 답변을 기다리는 대신 생성되는 대로 보여줄 수 있습니다.
# LCEL 체인은 자동으로 스트리밍 지원
for chunk in rag_chain.stream("회사의 복지 제도에 대해 알려주세요"):
print(chunk, end="", flush=True)단 한 줄 추가로 ChatGPT처럼 실시간 응답이 가능합니다!
소스 문서 확인하기
답변의 근거가 된 문서를 함께 보여주는 것이 좋습니다.
from langchain.chains import RetrievalQAWithSourcesChain
# 소스 추적 체인
qa_with_sources = RetrievalQAWithSourcesChain.from_chain_type(
llm=ChatOpenAI(model="gpt-4o-mini"),
chain_type="stuff",
retriever=vectorstore.as_retriever()
)
result = qa_with_sources.invoke({"question": "연차는 몇 일인가요?"})
print("답변:", result['answer'])
print("\n출처:")
for doc in result.get('source_documents', []):
print(f"- {doc.metadata.get('source', 'Unknown')}")
print(f" 내용 미리보기: {doc.page_content[:100]}...")실전 팁
-
청크 크기 선택: 일반적으로 500-1000 토큰이 적절하지만, 문서 특성에 따라 다릅니다. 기술 문서는 크게, 대화형 콘텐츠는 작게 나누는 것이 좋습니다.
-
임베딩 모델 선택: OpenAI의 text-embedding-3-large가 성능이 좋지만 비용이 부담된다면 text-embedding-3-small이나 오픈소스 모델도 고려해보세요.
-
벡터 스토어 영구 저장: Chroma는 기본적으로 메모리에 저장됩니다. persist_directory를 지정하면 디스크에 저장됩니다.
vectorstore = Chroma.from_documents(
documents=splits,
embedding=OpenAIEmbeddings(),
persist_directory="./chroma_db"
)- 에러 처리: 프로덕션에서는 반드시 예외 처리를 추가하세요.
try:
response = rag_chain.invoke(query)
except Exception as e:
print(f"오류 발생: {e}")
response = "죄송합니다. 답변을 생성할 수 없습니다."마치며
첫 RAG 시스템을 성공적으로 만들었습니다! 🥳
하지만 실무에서는 더 복잡한 상황들이 기다리고 있습니다. "여러 데이터베이스를 동시에 검색하려면?", "검색 실패 시 대안은?", "복잡한 질문은 어떻게 처리할까?"
2편 "LCEL로 복잡한 Chain 구성하기"에서는 조건부 실행, 병렬 처리, 폴백 전략 등 실전 필수 패턴들을 다룹니다.
다음 편에서 만나요!😚
