LangChain으로 RAG 파이프라인 구축하기 — 문서 검색 AI 앱 만드는 완전 가이드

들어가며: LLM의 한계와 RAG의 등장

ChatGPT, Claude, Gemini 같은 대형언어모델(LLM)은 놀라운 성능을 보여주지만, 치명적인 약점이 있습니다. 바로 학습 데이터의 시간적 제한입니다. 모델이 학습한 데이터 이후의 정보는 알 수 없고, 기업 내부 문서나 개인 데이터도 접근하지 못합니다. 또한 "할루시네이션(hallucination)"이라 불리는 잘못된 정보를 마치 사실처럼 답변하는 문제도 있습니다.

이런 한계를 극복하기 위해 등장한 기술이 바로 RAG(Retrieval-Augmented Generation)입니다. RAG는 질문에 관련된 외부 문서를 먼저 검색한 뒤, 그 검색 결과를 LLM의 컨텍스트에 포함시켜 답변을 생성하는 방식입니다. 이렇게 하면 최신 정보도 반영할 수 있고, 기업 비공개 자료도 안전하게 활용할 수 있으며, 할루시네이션도 크게 줄일 수 있습니다.

이 글에서는 Python의 LangChain 프레임워크를 사용하여 RAG 파이프라인을 밑바닥부터 구축하는 방법을 단계별로 설명합니다. 최종적으로는 Streamlit으로 웹 UI까지 붙여서 실제로 동작하는 문서 검색 AI 앱을 완성할 것입니다.

RAG란 무엇인가: 개념 이해

RAG의 작동 원리를 이해하기 위해 간단한 예시를 생각해봅시다. 사용자가 "2025년 회사 실적은 어떻게 됐나?"라고 질문했다고 가정합니다.

일반적인 LLM의 답변: 학습 데이터에 2025년 정보가 없으므로 추측으로 답변하거나 "정보가 없습니다"라고 말합니다. (할루시네이션 위험)

RAG를 사용한 답변: 먼저 회사 실적 보고서 PDF에서 "2025년 실적" 관련 내용을 검색하고, 그 내용을 LLM에 전달합니다. LLM은 제공받은 문서를 기반으로 정확한 답변을 생성합니다.

RAG 파이프라인의 핵심 구성요소는 다음과 같습니다:

• 문서 로더(Document Loader): PDF, 텍스트, 웹 페이지 등 다양한 형식의 문서를 읽음
• 텍스트 분할(Text Splitter): 긴 문서를 LLM 토큰 제한에 맞춰 작은 청크(chunk)로 분할
• 임베딩(Embedding): 각 청크를 수치 벡터로 변환하여 의미 유사도 계산 가능하게 함
• 벡터 스토어(Vector Store): 임베딩된 벡터들을 저장하고 검색하는 데이터베이스
• 검색 및 생성(Retrieval & Generation): 질문과 유사한 청크를 벡터 스토어에서 검색하고, LLM이 그것을 바탕으로 답변 생성

이제 각 단계를 실제로 구현해보겠습니다.

개발 환경 설정 및 필수 라이브러리 설치

RAG 파이프라인 구축을 위해 먼저 필요한 Python 라이브러리들을 설치해야 합니다. Python 3.9 이상 환경을 가정합니다.

기본 라이브러리 설치:

pip install langchain langchain-community langchain-openai
pip install openai
pip install faiss-cpu
pip install pypdf
pip install python-dotenv
pip install streamlit

각 라이브러리의 역할:

• langchain: LangChain 프레임워크 핵심
• langchain-community: 문서 로더, 임베딩 등 커뮤니티 통합
• langchain-openai: OpenAI 모델 연동
• openai: OpenAI API 클라이언트
• faiss-cpu: CPU 기반 벡터 검색 라이브러리 (GPU를 원하면 faiss-gpu)
• pypdf: PDF 파일 읽기
• python-dotenv: 환경변수 관리 (.env 파일 사용)
• streamlit: 간단한 웹 UI 구축

설치 후 OpenAI API 키를 환경변수로 설정합니다. 프로젝트 루트에 .env 파일을 생성합니다:

OPENAI_API_KEY=sk-your-api-key-here

Python에서 이를 로드하는 코드:

import os
from dotenv import load_dotenv

load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")

문서 로딩: 다양한 형식 지원

RAG의 첫 번째 단계는 외부 문서를 읽어들이는 것입니다. LangChain은 PDF, 텍스트, 마크다운, Word, 웹 페이지 등 다양한 형식을 지원합니다.

PDF 문서 로딩:

from langchain_community.document_loaders import PyPDFLoader

# 단일 PDF 파일 로드
loader = PyPDFLoader("documents/report.pdf")
documents = loader.load()

# 로드된 문서 확인
print(f"총 {len(documents)}개 페이지 로드됨")
for doc in documents[:2]:
    print(f"내용 샘플: {doc.page_content[:200]}")
    print(f"메타데이터: {doc.metadata}")  # 파일명, 페이지 번호 등

일반 텍스트 파일 로딩:

from langchain_community.document_loaders import TextLoader

loader = TextLoader("documents/notes.txt", encoding="utf-8")
documents = loader.load()

여러 PDF 파일 한 번에 로드:

from langchain_community.document_loaders import DirectoryLoader

loader = DirectoryLoader(
    "documents/",  # 디렉토리 경로
    glob="**/*.pdf",  # 파일 패턴
    loader_cls=PyPDFLoader
)
documents = loader.load()
print(f"총 {len(documents)}개 페이지 로드됨")

웹 페이지 로딩:

from langchain_community.document_loaders import WebBaseLoader

loader = WebBaseLoader("https://example.com/article")
documents = loader.load()
print(f"웹 페이지 로드됨: {len(documents)}개 문서")

로드된 각 Document 객체는 page_content(실제 내용)와 metadata(메타데이터)로 구성됩니다. 메타데이터는 나중에 검색 결과의 출처를 추적할 때 유용합니다.

텍스트 분할: 최적의 청크 크기 설정

LLM은 토큰 제한이 있습니다(예: GPT-3.5는 4,096 토큰). 따라서 긴 문서를 그대로 LLM에 전달할 수 없습니다. 문서를 적절한 크기의 청크로 분할하는 것이 중요합니다.

기본 텍스트 분할:

from langchain.text_splitter import RecursiveCharacterTextSplitter

# 청크 설정
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,        # 청크 당 최대 문자 수
    chunk_overlap=200,      # 청크 간 겹치는 부분 (문맥 연속성 보장)
    separators=["\n\n", "\n", " ", ""]  # 분할 기준 우선순위
)

# 문서 분할
chunks = splitter.split_documents(documents)
print(f"총 {len(chunks)}개 청크로 분할됨")
print(f"첫 청크 크기: {len(chunks[0].page_content)} 문자")

청크 크기와 오버랩 설정 가이드:

• chunk_size=500: 짧은 Q&A 형식 문서에 적합. 정확성 높음
• chunk_size=1000: 일반적인 기술 문서, 블로그 글에 권장
• chunk_size=2000: 긴 형식 콘텐츠(보고서, 논문) 에 적합
• chunk_overlap은 보통 chunk_size의 10-20% 권장

의미 기반 분할 (Semantic Splitter):

최신 LangChain 버전에서는 의미 유사도 기반으로 더 똑똑하게 분할할 수도 있습니다:

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai.embeddings import OpenAIEmbeddings

embeddings = OpenAIEmbeddings()
semantic_splitter = SemanticChunker(embeddings)

chunks = semantic_splitter.split_documents(documents)

의미 기반 분할은 계산 비용이 높지만, 검색 품질이 더 좋습니다.

벡터 스토어 구축: FAISS 임베딩 저장 및 검색

이제 청크들을 벡터로 변환하고, FAISS라는 벡터 데이터베이스에 저장합니다. FAISS는 Meta에서 만든 오픈소스 벡터 검색 엔진으로, 대규모 데이터셋에서 빠른 유사도 검색을 지원합니다.

임베딩과 벡터 스토어 생성:

from langchain_openai.embeddings import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

# OpenAI 임베딩 모델 초기화
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# FAISS 벡터 스토어 생성 (청크 임베딩 + 저장)
vector_store = FAISS.from_documents(
    documents=chunks,
    embedding=embeddings
)

print("벡터 스토어 생성 완료")
print(f"저장된 벡터 개수: {vector_store.index.ntotal}")

이 코드는 각 청크를 OpenAI의 text-embedding-3-small 모델로 임베딩한 후, FAISS 인덱스에 저장합니다. 처음 실행은 시간이 걸릴 수 있습니다.

저장된 벡터 스토어 로드 및 재사용:

벡터 스토어를 파일로 저장했다가 나중에 불러올 수 있습니다:

# 저장
vector_store.save_local("faiss_index")

# 나중에 로드
from langchain_community.vectorstores import FAISS
from langchain_openai.embeddings import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = FAISS.load_local(
    "faiss_index",
    embeddings,
    allow_dangerous_deserialization=True  # 로컬 파일 신뢰
)

유사도 검색 테스트:

# 쿼리 문자열로 유사 청크 검색
query = "2025년 회사 실적"
search_results = vector_store.similarity_search(query, k=3)

print(f"검색 결과 ({len(search_results)}개):")
for i, result in enumerate(search_results, 1):
    print(f"\n--- 결과 {i} ---")
    print(f"내용: {result.page_content[:300]}...")
    print(f"유사도 점수: {result.metadata}")  # FAISS 점수

또는 유사도 점수와 함께 반환할 수 있습니다:

search_results = vector_store.similarity_search_with_score(query, k=3)

for doc, score in search_results:
    print(f"유사도: {score:.3f}")
    print(f"내용: {doc.page_content[:200]}...")

k=3은 상위 3개 결과를 반환한다는 뜻입니다. 이 값을 조정하여 검색 결과 개수를 제어할 수 있습니다.

RAG 체인 완성: RetrievalQA와 LCEL

이제 검색과 생성을 연결하는 RAG 체인을 만듭니다. LangChain은 두 가지 방식을 제공합니다: 기존의 RetrievalQA와 최신의 LCEL(LangChain Expression Language)입니다.

방법 1: RetrievalQA (간단함):

from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI

# LLM 초기화
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.3)

# RetrievalQA 체인 생성
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",  # 검색 결과를 한 번에 컨텍스트에 포함
    retriever=vector_store.as_retriever(
        search_type="similarity",
        search_kwargs={"k": 3}  # 상위 3개 청크만 검색
    ),
    return_source_documents=True  # 검색 소스 반환
)

# 질문 실행
query = "LangChain RAG의 주요 구성요소는 무엇인가?"
result = qa_chain.invoke({"query": query})

print(f"질문: {query}")
print(f"답변: {result['result']}")
print(f"\n검색된 문서:")
for doc in result['source_documents']:
    print(f"- {doc.metadata}")
    print(f"  {doc.page_content[:150]}...")

방법 2: LCEL (더 강력함):

LCEL은 더 유연하고 사용자 정의가 가능합니다:

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser
from operator import itemgetter

# 1. 프롬프트 템플릿 정의
prompt_template = """다음 문서를 참고하여 질문에 답하세요.
문서에 답이 없으면 "문서에 관련 정보가 없습니다"라고 답하세요.

문서:
{context}

질문: {question}

답변:"""

prompt = ChatPromptTemplate.from_template(prompt_template)

# 2. LLM 초기화
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.3)

# 3. 검색기 생성
retriever = vector_store.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 3}
)

# 4. LCEL 체인 구성
rag_chain = (
    {"context": retriever | (lambda docs: "\n\n".join(d.page_content for d in docs)),
     "question": itemgetter("question")}
    | prompt
    | llm
    | StrOutputParser()
)

# 5. 실행
query = "LangChain RAG의 주요 구성요소는 무엇인가?"
answer = rag_chain.invoke({"question": query})
print(f"질문: {query}")
print(f"답변: {answer}")

chain_type 선택 가이드:

• stuff: 검색 결과를 모두 한 번에 프롬프트에 포함. 간단하고 빠르지만 토큰 제한에 취약
• map_reduce: 각 검색 결과를 먼저 요약한 후 최종 답변 생성. 비용 높음
• refine: 반복적으로 답변을 개선. 가장 정확하지만 느림

일반적으로 stuff가 가장 실용적입니다.

실전 앱 완성: Streamlit으로 웹 UI 붙이기

이제 만든 RAG 파이프라인을 Streamlit으로 감싸서 사용자 친화적인 웹 애플리케이션으로 만들겠습니다.

app.py 전체 코드:

import streamlit as st
import os
from dotenv import load_dotenv
from langchain_openai.embeddings import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI

# 환경설정
load_dotenv()

st.set_page_config(page_title="문서 검색 AI", layout="wide")
st.title("문서 검색 AI 앱")
st.markdown("LangChain RAG로 만든 AI 어시스턴트입니다. 문서를 업로드하고 질문하세요!")

# 세션 상태 초기화
if "qa_chain" not in st.session_state:
    st.session_state.qa_chain = None

# 사이드바: 문서 업로드
with st.sidebar:
    st.header("문서 업로드")
    uploaded_files = st.file_uploader(
        "PDF 파일을 선택하세요",
        type=["pdf"],
        accept_multiple_files=True
    )

    if st.button("문서 처리") and uploaded_files:
        with st.spinner("문서를 처리 중입니다..."):
            from langchain_community.document_loaders import PyPDFLoader
            from langchain.text_splitter import RecursiveCharacterTextSplitter
            import tempfile

            all_docs = []

            # 임시 파일로 저장 후 로드
            for uploaded_file in uploaded_files:
                with tempfile.NamedTemporaryFile(delete=False, suffix=".pdf") as tmp_file:
                    tmp_file.write(uploaded_file.getbuffer())
                    tmp_path = tmp_file.name

                loader = PyPDFLoader(tmp_path)
                docs = loader.load()
                all_docs.extend(docs)
                os.remove(tmp_path)

            # 텍스트 분할
            splitter = RecursiveCharacterTextSplitter(
                chunk_size=1000,
                chunk_overlap=200
            )
            chunks = splitter.split_documents(all_docs)

            # 벡터 스토어 생성
            embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
            vector_store = FAISS.from_documents(chunks, embeddings)

            # RAG 체인 생성
            llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.3)
            st.session_state.qa_chain = RetrievalQA.from_chain_type(
                llm=llm,
                chain_type="stuff",
                retriever=vector_store.as_retriever(search_kwargs={"k": 3}),
                return_source_documents=True
            )

            st.success(f"{len(chunks)}개 청크로 처리 완료!")

# 메인 영역: 질문 입력
if st.session_state.qa_chain:
    st.subheader("질문하기")

    query = st.text_input("문서에 대해 질문하세요:")

    if query:
        with st.spinner("답변을 생성 중입니다..."):
            result = st.session_state.qa_chain.invoke({"query": query})

            st.markdown("### 답변")
            st.write(result["result"])

            with st.expander("검색된 문서 보기"):
                for i, doc in enumerate(result["source_documents"], 1):
                    st.markdown(f"**문서 {i}** ({doc.metadata.get('source', 'Unknown')})")
                    st.text(doc.page_content[:500] + "...")
else:
    st.info("좌측 사이드바에서 PDF 파일을 업로드하세요.")

실행 방법:

streamlit run app.py

브라우저에서 http://localhost:8501을 열면 앱이 실행됩니다.

주요 기능:

• 여러 PDF 파일 동시 업로드
• 실시간 문서 처리
• 질문에 대한 즉각적인 답변
• 검색된 문서 소스 표시
• 세션 상태 관리로 문서 재처리 불필요

트러블슈팅: 토큰 초과, 검색 품질 개선, 비용 절감

문제 1: Token 초과 에러 (OpenAIError: tokens exceeds maximum)

원인: 검색된 문서가 너무 많거나 컨텍스트가 너무 길어서 LLM의 토큰 제한 초과

해결책:

• 검색 개수 줄이기: search_kwargs={"k": 1} 또는 k": 2}
• 청크 크기 줄이기: chunk_size=500
• 더 저렴한 모델 사용: gpt-4-turbo 대신 gpt-3.5-turbo

# 최적화된 설정
retriever = vector_store.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 2}  # 상위 2개만 검색
)

splitter = RecursiveCharacterTextSplitter(
    chunk_size=700,  # 더 작게
    chunk_overlap=150
)

문제 2: 검색 품질이 낮음 (관련 없는 문서 반환)

원인: 벡터 임베딩 모델이 질문과 문서의 의미를 제대로 매칭하지 못함

해결책:

• 더 정교한 임베딩 모델 사용: text-embedding-3-large
• 검색 방식 변경: similarity 대신 mmr (Maximum Marginal Relevance) 사용
• 청크 크기와 오버랩 조정

# MMR 검색 (중복을 제거하면서 관련성 높은 결과 반환)
retriever = vector_store.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 3,
        "fetch_k": 10  # 먼저 10개를 가져온 후 3개를 선택
    }
)

# 더 강력한 임베딩 모델
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")

문제 3: API 비용이 높음

원인: 매번 임베딩과 LLM 호출 반복

해결책:

• 로컬 임베딩 모델 사용: sentence-transformers
• 벡터 스토어를 저장했다가 재사용
• 더 저렴한 모델 조합: gpt-3.5-turbo + text-embedding-3-small

# 로컬 임베딩 모델 (무료)
from langchain_community.embeddings import HuggingFaceEmbeddings

embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
)

# 벡터 스토어 저장 및 재사용
vector_store.save_local("faiss_index")

# 다음 실행 시
vector_store = FAISS.load_local("faiss_index", embeddings)

문제 4: 답변이 과도하게 길거나 부정확함

원인: 프롬프트 설계 미흡, temperature 설정 부적절

해결책:

# 프롬프트 최적화
prompt_template = """다음 문서만을 참고하여 간결하게 답하세요.

문서:
{context}

질문: {question}

답변 (한 문단으로):"""

# Temperature 조정 (0.0~1.0)
# 0.0: 결정적, 일관성 높음 (팩트 기반)
# 0.5: 균형적
# 1.0: 창의적, 다양성 높음
llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    temperature=0.1  # 팩트 기반 답변에는 낮게
)

성능 측정:

import time

start = time.time()
result = qa_chain.invoke({"query": "테스트 질문"})
elapsed = time.time() - start

print(f"응답 시간: {elapsed:.2f}초")
print(f"검색 결과: {len(result['source_documents'])}개")
print(f"답변 길이: {len(result['result'])}자")

결론: RAG의 미래와 다음 단계

이 글을 통해 LangChain을 사용하여 완전한 RAG 파이프라인을 구축하는 방법을 배웠습니다. 문서 로딩에서 시작해 벡터 스토어, RAG 체인, 그리고 Streamlit 웹 앱까지 모든 단계를 다뤘습니다.

RAG는 단순한 기술이 아니라 LLM의 활용성을 획기적으로 높이는 패러다임입니다. 기업 문서, 개인 지식베이스, 최신 뉴스 등 어떤 데이터 소스든 RAG를 통해 AI의 답변을 더 정확하고 신뢰할 수 있게 만들 수 있습니다.

다음 단계로 고려할 수 있는 기술들:

• 멀티 쿼리 검색: 여러 관점의 쿼리를 생성해 검색 결과 개선
• 쿼리 확장: 원래 질문을 LLM으로 변형하여 더 나은 검색
• 메타데이터 필터링: 문서 날짜, 카테고리 등으로 검색 범위 제한
• 하이브리드 검색: BM25(키워드) + 벡터 검색 결합
• 로컬 LLM: Ollama, LLaMA 2 등으로 비용 제거
• 에이전트 RAG: 도구 연동으로 상호작용형 AI 어시스턴트

RAG는 빠르게 발전하는 분야입니다. 이 글에 소개된 기술과 라이브러리는 작성 시점(2026년 4월)을 기준으로 하며, LangChain과 관련 생태계의 업데이트에 따라 문법이나 기능이 변경될 수 있습니다. 항상 공식 문서를 참고하시기 바랍니다.

유용한 리소스:

• LangChain 공식 문서: https://python.langchain.com
• FAISS 문서: https://faiss.ai
• Streamlit 문서: https://docs.streamlit.io
• OpenAI API 가이드: https://platform.openai.com/docs

이 글에 소개된 서비스와 도구는 작성 시점 기준이며, 업데이트에 따라 변경될 수 있습니다.