RAG 챗봇 구축 실전 — LangChain + ChromaDB + OpenAI로 나만의 문서 QA 봇 만들기

문서 → 청킹 → 임베딩 → 벡터 DB 저장

질문 → 임베딩 → 유사 문서 검색 → 프롬프트 구성 → LLM 답변

pip install langchain langchain-openai langchain-community \
            chromadb pypdf tiktoken streamlit python-dotenv

# .env
OPENAI_API_KEY=sk-proj-your-api-key-here

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
CHROMA_PERSIST_DIR = "./chroma_db"
COLLECTION_NAME = "my_documents"
CHUNK_SIZE = 1000
CHUNK_OVERLAP = 200
EMBEDDING_MODEL = "text-embedding-3-small"
LLM_MODEL = "gpt-4o-mini"

# document_loader.py
from langchain_community.document_loaders import (
    PyPDFLoader,
    DirectoryLoader,
    TextLoader,
)

def load_pdf(file_path: str):
    """단일 PDF 파일 로딩"""
    loader = PyPDFLoader(file_path)
    documents = loader.load()
    print(f"Loaded {len(documents)} pages from {file_path}")
    return documents

def load_directory(dir_path: str, glob: str = "**/*.pdf"):
    """디렉토리 내 모든 PDF 로딩"""
    loader = DirectoryLoader(
        dir_path,
        glob=glob,
        loader_cls=PyPDFLoader,
        show_progress=True,
    )
    documents = loader.load()
    print(f"Loaded {len(documents)} pages from {dir_path}")
    return documents

# 사용 예시
documents = load_directory("./docs")

# chunker.py
from langchain.text_splitter import RecursiveCharacterTextSplitter

def chunk_documents(documents, chunk_size=1000, chunk_overlap=200):
    """문서를 청크로 분할"""
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        length_function=len,
        separators=["\n\n", "\n", ".", " ", ""],
    )

    chunks = text_splitter.split_documents(documents)
    print(f"Split {len(documents)} documents into {len(chunks)} chunks")

    # 메타데이터에 청크 인덱스 추가
    for i, chunk in enumerate(chunks):
        chunk.metadata["chunk_index"] = i
        chunk.metadata["chunk_size"] = len(chunk.page_content)

    return chunks

chunks = chunk_documents(documents)

# vectorstore.py
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from config import CHROMA_PERSIST_DIR, COLLECTION_NAME, EMBEDDING_MODEL

def create_vectorstore(chunks):
    """청크를 임베딩하여 ChromaDB에 저장"""
    embeddings = OpenAIEmbeddings(model=EMBEDDING_MODEL)

    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=CHROMA_PERSIST_DIR,
        collection_name=COLLECTION_NAME,
    )

    print(f"Stored {len(chunks)} chunks in ChromaDB")
    return vectorstore

def load_vectorstore():
    """기존 ChromaDB 로드"""
    embeddings = OpenAIEmbeddings(model=EMBEDDING_MODEL)

    vectorstore = Chroma(
        persist_directory=CHROMA_PERSIST_DIR,
        collection_name=COLLECTION_NAME,
        embedding_function=embeddings,
    )

    count = vectorstore._collection.count()
    print(f"Loaded ChromaDB with {count} documents")
    return vectorstore

# retriever.py

def get_retriever(vectorstore, search_type="mmr", k=4):
    """벡터스토어에서 검색기 생성"""

    if search_type == "mmr":
        # MMR: 관련성 + 다양성 균형
        retriever = vectorstore.as_retriever(
            search_type="mmr",
            search_kwargs={
                "k": k,
                "fetch_k": 20,      # 후보 문서 수
                "lambda_mult": 0.7,  # 1.0=관련성, 0.0=다양성
            },
        )
    elif search_type == "similarity_score":
        # 유사도 임계값 기반
        retriever = vectorstore.as_retriever(
            search_type="similarity_score_threshold",
            search_kwargs={
                "score_threshold": 0.7,
                "k": k,
            },
        )
    else:
        # 기본 유사도 검색
        retriever = vectorstore.as_retriever(
            search_kwargs={"k": k},
        )

    return retriever

# rag_chain.py
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.runnable import RunnablePassthrough
from langchain.schema.output_parser import StrOutputParser
from config import LLM_MODEL

SYSTEM_TEMPLATE = """당신은 문서 기반 QA 어시스턴트입니다.
주어진 컨텍스트만을 사용하여 질문에 답변하세요.

규칙:
1. 컨텍스트에 없는 정보는 "해당 정보를 문서에서 찾을 수 없습니다"라고 답변하세요.
2. 답변은 한국어로 작성하세요.
3. 가능하면 구체적인 수치나 인용을 포함하세요.
4. 답변의 근거가 되는 문서를 언급하세요.

컨텍스트:
{context}
"""

def format_docs(docs):
    """검색된 문서를 포맷팅"""
    formatted = []
    for i, doc in enumerate(docs, 1):
        source = doc.metadata.get("source", "unknown")
        page = doc.metadata.get("page", "?")
        formatted.append(
            f"[문서 {i}] (출처: {source}, 페이지: {page})\n{doc.page_content}"
        )
    return "\n\n---\n\n".join(formatted)

def create_rag_chain(retriever):
    """RAG 체인 생성"""
    llm = ChatOpenAI(
        model=LLM_MODEL,
        temperature=0,
        max_tokens=2000,
    )

    prompt = ChatPromptTemplate.from_messages([
        ("system", SYSTEM_TEMPLATE),
        ("human", "{question}"),
    ])

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

    return chain

# conversation.py
from langchain.memory import ConversationBufferWindowMemory
from langchain.chains import ConversationalRetrievalChain
from langchain_openai import ChatOpenAI
from config import LLM_MODEL

def create_conversational_chain(retriever):
    """대화 히스토리를 지원하는 RAG 체인"""
    llm = ChatOpenAI(model=LLM_MODEL, temperature=0)

    memory = ConversationBufferWindowMemory(
        memory_key="chat_history",
        return_messages=True,
        output_key="answer",
        k=5,  # 최근 5턴만 유지
    )

    chain = ConversationalRetrievalChain.from_llm(
        llm=llm,
        retriever=retriever,
        memory=memory,
        return_source_documents=True,
        verbose=False,
    )

    return chain

# 사용 예시
chain = create_conversational_chain(retriever)

result = chain.invoke({"question": "이 문서의 핵심 내용은 무엇인가요?"})
print(result["answer"])
print(f"\n참조 문서 {len(result['source_documents'])}건")

# app.py
import streamlit as st
from document_loader import load_pdf
from chunker import chunk_documents
from vectorstore import create_vectorstore, load_vectorstore
from retriever import get_retriever
from rag_chain import create_rag_chain

st.set_page_config(page_title="📚 문서 QA 챗봇", layout="wide")
st.title("📚 RAG 문서 QA 챗봇")

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

    if uploaded_files and st.button("🔄 문서 처리"):
        with st.spinner("문서 처리 중..."):
            all_chunks = []
            for file in uploaded_files:
                # 임시 파일 저장
                temp_path = f"/tmp/{file.name}"
                with open(temp_path, "wb") as f:
                    f.write(file.getbuffer())

                docs = load_pdf(temp_path)
                chunks = chunk_documents(docs)
                all_chunks.extend(chunks)

            vectorstore = create_vectorstore(all_chunks)
            st.session_state["vectorstore"] = vectorstore
            st.success(f"✅ {len(all_chunks)}개 청크 처리 완료!")

# 메인: 채팅 인터페이스
if "messages" not in st.session_state:
    st.session_state.messages = []

# 이전 메시지 표시
for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

# 사용자 입력
if prompt := st.chat_input("문서에 대해 질문하세요..."):
    st.session_state.messages.append({"role": "user", "content": prompt})

    with st.chat_message("user"):
        st.markdown(prompt)

    with st.chat_message("assistant"):
        if "vectorstore" not in st.session_state:
            try:
                st.session_state["vectorstore"] = load_vectorstore()
            except Exception:
                st.error("먼저 문서를 업로드해주세요.")
                st.stop()

        vectorstore = st.session_state["vectorstore"]
        retriever = get_retriever(vectorstore)
        chain = create_rag_chain(retriever)

        with st.spinner("답변 생성 중..."):
            response = chain.invoke(prompt)

        st.markdown(response)
        st.session_state.messages.append(
            {"role": "assistant", "content": response}
        )

# 실행
streamlit run app.py --server.port 8501

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

def create_hybrid_retriever(chunks, vectorstore, k=4):
    """BM25 + 벡터 검색 앙상블"""
    bm25_retriever = BM25Retriever.from_documents(chunks)
    bm25_retriever.k = k

    vector_retriever = vectorstore.as_retriever(search_kwargs={"k": k})

    ensemble = EnsembleRetriever(
        retrievers=[bm25_retriever, vector_retriever],
        weights=[0.4, 0.6],  # 벡터 검색에 가중치
    )

    return ensemble

from langchain.retrievers import ContextualCompressionRetriever
from langchain_community.document_compressors import CohereRerank

def create_reranked_retriever(vectorstore, k=4, top_n=3):
    """Cohere Reranker로 검색 결과 재정렬"""
    base_retriever = vectorstore.as_retriever(search_kwargs={"k": k * 3})

    compressor = CohereRerank(
        model="rerank-v3.5",
        top_n=top_n,
    )

    return ContextualCompressionRetriever(
        base_compressor=compressor,
        base_retriever=base_retriever,
    )

# 청크에 요약 메타데이터 추가
from langchain_openai import ChatOpenAI

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

for chunk in chunks:
    summary = llm.invoke(
        f"다음 텍스트를 한 문장으로 요약하세요:\n{chunk.page_content}"
    ).content
    chunk.metadata["summary"] = summary