RAG 기반 챗봇 LLM 연동 & 벡터DB GCS 백업

SUIN2 2025. 7. 1. 15:13

RAG를 활용한 챗봇 개발 과정

흩어져있는 카테부 공지사항을 모아서 궁금한 부분에 대해 질문하면, 찾는 수고로움 없이 바로 알 수 있도록 하는 것이 우리 서비스의 목표이다. 이 메인 기능인 챗봇이 원하는 응답을 생성할 수 있도록, LLM 추론을 강화하기 위한 RAG 기술을 활용한다.

RAG: Retrieval-Augmented Generation의 약자로, 단순히 LLM에 질문을 던져 답을 생성하는 방식이 아니라, 외부 지식 소스(문서, DB, 위키 등)를 검색(Retrieve)하여 그 결과를 기반으로 텍스트를 생성(Generate)하는 구조

RAG를 활용하면 사내 지식과 같은, LLM이 기본적으로 학습하지 않아서 모르는 최신 외부 정보를 활용할 수 있고, 무엇보다 hallucination을 줄일 수 있기 때문에, 챗봇 시스템에서 거의 필수로 사용되고 있다. 이렇게 RAG를 도입하게 되면, 기본 구성이 사용자 질의와 관련된 문서를 검색하는 Retriever와 검색된 문서를 기반으로 응답을 생성하는 Generator(LLM)으로 구성된다.

작동 흐름

사용자 질의 입력: "휴가 신청하는 방법 알려줘."
질의 벡터화 후 관련 문서 검색 (Retriever)
검색된 문서들을 LLM 프롬프트로 전달
LLM이 문서 내용을 반영하여 응답 생성

LLM

가장 중요한 LLM 선택이다. chatgpt나 gemini와 같은 외부 LLM을 API 호출하여 사용하는 방식이 간단하고, 성능도 당연히 좋겠지만, 가장 적은 비용으로 최고의 성능을 내는게 이번 나의 개발 목표였기 때문에 로컬 LLM 탐색에 심혈을 기울였다.

모델 선정 기준은 아래와 같다.

기준 항목	설명
1. 모델 크기 대비 성능	7B~14B 범위 내에서 추론 속도와 정확도 간 균형이 우수한 모델을 선호
2. 양자화 가능 여부	GGUF, GPTQ 등으로 경량화 및 다양한 환경에서 배포 가능성 고려
3. 한국어 지원	한국어 instruction-following 성능 및 응답 자연스러움이 중요
4. 확장성	Streaming 출력, LangChain / RAG 구조 연동 가능 여부 등 통합성

위 기준을 바탕으로 후보로 뒀던 LLM은 KULLM3, Qwen2.5-14B, Qwen2.5-7B까지 세 개였다.

한정된 자원 안에서 우리가 사용하기로 한 GPU 메모리는 GCP L4였다. (개인 계정별 GCP 무료 크레딧 사용하기 !)

GCP L4에서 LLM과 임베딩 모델 두 개를 로드해야하기 때문에 가장 가벼운 사이즈인 Qwen2.5-7B를 선정하였다. 가벼운 사이즈만큼이나 응답 속도도 가장 빨랐다.

# qwen_loader.py

from langchain_community.llms import VLLM

# GCP VM에 올라간 Qwen2.5-7B-Instruct 모델을 vLLM으로 감싸 LangChain에서 사용할 수 있게 구성
llm = VLLM(
    model="Qwen/Qwen2.5-7B-Instruct",  # Hugging Face 또는 로컬 경로 가능 (GCP VM에 위치)
    trust_remote_code=True,
    max_tokens=512,
    temperature=0.3
)

모델을 추론할때 vLLM 서버를 사용한다. vLLM은 실제 LLM이 돌아가는 추론 서버로, 배치 처리 및 KV 캐시 등의 기능을 수행하여 모델 서빙에 있어 성능 최적화를 할 수 있다. (vLLM과 관련해서는 추후에 자세히 다룰 예정이다.)

이때, vLLM을 사용하기 위한 어댑터 역할로, 랭체인 서브파티모듈 중 하나로써 vllm을 사용하였다. 여기서 사용되는 vllm은 LangChain에서 vLLM 서버를 통해 LLM을 호출할 수 있게 해주는 래퍼 클래스이다.

한마디로, LangChain에서 vLLM 서버로 API 호출을 하여 사용할 수 있다. (다만, LangChain이 직접 서버 내부 최적화에 개입하는 것은 아니다.)

아무튼, llm = VLLM(...)을 통해 LangChain 내에서 사용할 LLM 객체를 정의하였다.

임베딩 모델

텍스트를 의미 기반의 벡터로 변환하는 모델이 필요하다. RAG에 활용될 공지사항 문서들을 벡터화화여 DB에 저장해야하는데, 이때 벡터화를 임베딩 모델이 해준다. 또한, 사용자 질의가 들어왔을때, 관련 문서를 검색하기 위해서는 사용자 질의도 벡터화를 해주어야 기존 문서와 사용자 질의 간의 검색이 진행될 수 있다. 이때, 벡터화를 하는 이유는 유사한 의미의 문장/단어를 벡터 공간상 가까운 거리로 표현하기 위해 사용한다.

💡 여기서 잠깐, 임베딩과 벡터가 같은 말인가 ??

임베딩은 데이터를 수치 공간으로 매핑하는 과정 또는 표현이고, 벡터는 그 결과물인 숫자 배열

임베딩(Embedding): 사람이 이해하는 복잡한 데이터를 기계가 이해 가능한 수치 공간으로 변환하는 것
벡터 (Vector): 단순한 숫자들의 나열 (보통 리스트 또는 배열 형태)

우리 서비스에서 사용한 임베딩 모델은 허깅페이스에서 가져올 수 있는 intfloat/multilingual-e5-base이다. 원래는 instruct에 특화된

intfloat/multilingual-e5-large-instruct을 사용하고자 했지만, 테스트 서버에 올릴때 OOM 에러가 발생하여, 다운그레이드를 해야했다.

# embedding_model.py

from langchain_community.embeddings import HuggingFaceEmbeddings

# RAG에서 사용할 다국어 임베딩 모델 정의
embedder = HuggingFaceEmbeddings(
    model_name="intfloat/multilingual-e5-base"
    )

벡터 검색

RAG가 사용자 질의와 관련성이 높은 문서를 검색하는 기술이라고 간단하게 표현할 수 있는데, 이때 어떤 기준으로 문서를 검색하느냐고 한다면, 대표적으로 유사도 기반, 키워드 기반, 필터 기반 등등이 있다.

우리는 벡터 검색 기술로, FaceBook에서 만든 벡터들간의 유사도를 계산하여 의미적으로 가까운 문서를 반환하는 Faiss 라이브러리를 선정하였다.

✨ FAISS

: 고차원 벡터 데이터에서 빠르고 효율적인 유사성 검색과 클러스터링을 수행하기 위해 개발된 라이브러리

일반적으로 벡터 검색은 벡터화된 데이터와 쿼리 사이에서 유사도를 계산하여 가장 유사한 데이터를 반환하는 방식으로 동작한다. 그러나 데이터가 커짐에 따라 유사도 계산의 연산량과 속도 측면에서 매우 비효율적일 수 있다.

이를 해결하기 위해 벡터 인덱싱 개념을 사용하여 대규모 벡터 데이터셋에도 빠른 검색을 가능하게 한다.

🩵 벡터 인덱싱(Vector Indexing)

: 대량의 벡터 데이터를 효율적으로 저장하고 검색할 수 있도록 구조화하는 기법

일반적으로 데이터베이스에서 인덱스를 사용하여 검색 속도를 높이는 것과 유사한 개념이다.

Faiss에서 사용하는 주요 벡터 인덱스 구조는 다음과 같다.

IndexFlatL2	모든 벡터 간 L2 거리(유클리디안 거리)를 정확하게 계산하여 최근접 이웃을 찾는 방식으로, 작은 데이터셋에서 정확한 검색이 필요한 경우 사용한다.
IndexFlatIP	모든 벡터 간 내적(Inner Product)을 정확히 계산하여 가장 유사한 벡터를 찾는 방식이다.
IndexIVFFlat	벡터 데이터를 여러 개의 클러스터로 나누고, 검색할 때 가장 관련 있는 클러스터에서만 검색하며, 대규모 데이터에서 검색 속도를 향상시키는 경우에 사용한다.
IndexHNSW	그래프 기반 탐색을 통해 유사한 벡터를 빠르게 찾는 방법으로, 추천 시스템에서 활용한다.
IndexPQ (Product Quantization)	벡터를 여러 개의 작은 부분으로 나누어 압축하고, 검색 시 근사값을 활용하여 속도를 향상시키며, 저장 공간을 줄이면서 성능을 유지할 수 있다.

Faiss의 유사도 계산은 기본적으로 유클리디안 거리 기반이며, IndexFlatL2로 설정하여 사용할 수 있다.

index = faiss.IndexFlatL2(vector_dimension)

코사인 유사도로 산출하려면 IndexFlatIP를 활용하면 되는데, 이때 IP는 Inner Product(내적)을 의미한다. 이때, 코사인 유사도는 방향만 중요하고 크기는 무시해야하기 때문에, normalize도 추가한다.

index = faiss.IndexFlatIP(vector_dimension)
faiss.normalize_L2(vector_data)

작동 흐름

로컬형 FAISS 인덱스를 생성 → GCS에 업로드(백업) → 서버 기동 시 다운로드 및 메모리에 로드 → RAG 기반 검색 수행

Faiss는 자체적으로 DB 서버가 아닌, 로컬 인덱스 파일(.index)을 생성한다. 클라우드형 VDB가 아니기 때문에, 클라우드에 따로 백업을 해두고, 서버 기동 시 .index 파일을 다운로드하여 메모리에 로드하면, RAG 검색 때 메모리에 올라간 .index로 유사도 검색을 수행하게 된다.

벡터 스토어 생성

docs/ 폴더 내에 모든 .md 문서를 로드
문서들을 500자 단위로 분할
각 청크에 대해 임베딩 수행
FAISS 인덱스 생성 및 로컬 저장

# create_vectorstore.py

from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from app.core.embedding_model import embedder
import os

# 1. Markdown 문서 로드 (docs/ 폴더 내 .md 파일 대상)
loader = DirectoryLoader(
    path="docs",
    glob="**/*.md",
    loader_cls=TextLoader,
    use_multithreading=True,
)
docs = loader.load()

# 2. 문서 분할
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=100,
)
chunks = splitter.split_documents(docs)

# 3. 벡터스토어 생성
vectorstore = FAISS.from_documents(chunks, embedding=embedder)

# 4. 로컬 저장
INDEX_SAVE_PATH = "vector/faiss_index"
os.makedirs(INDEX_SAVE_PATH, exist_ok=True)
vectorstore.save_local(INDEX_SAVE_PATH)

print(f"✅ 벡터스토어 저장 완료: {INDEX_SAVE_PATH}")

이 파일에서 사용하는 FAISS는 검색 기술로써가 아닌, 벡터 저장소 생성용으로 사용된다.

우리는 LangChain community에서 지원하는 Faiss를 사용하였다. 기존 Faiss와 동일한 역할을 하지만, 랭체인 기반 Faiss는 고수준 문서 검색 인터페이스를 지원하기 때문에, 사용에 있어 더욱 편리하다. 하지만 항상 그렇듯이, 랭체인 커뮤니티에서 제공하는 고수준 기술들은 기존 순수 기술의 베이직만 가져갈 수 있고 더 섬세한 튜닝 작업은 못한다는게 아쉽다. (그래서 추후에 vllm도 랭체인 기반으로 사용하다가 순수 vllm으로 바꿨다 ..) 아무튼 벡터 검색 기술은 기본으로만 사용해도 (임베딩 모델에 따라 성능 차이가 날 것 같아서) 괜찮을 것 같아서 랭체인 인터페이스를 따라가기로 했다.

이를 위해 langchain faiss에서 제공하는 from_ducuments() 메서드를 사용할 수 있는데, 문서 리스트를 받아 각 문서를 임베딩한 후, 해당 벡터들로부터 FAISS 인덱스를 생성하는 클래스 메서드이다.

그리고 save_loacl()을 호출하면, 해당 디렉토리에는 index.faiss와 index.pkl 두 개의 파일이 생성된다. .faiss는 FAISS 벡터 인덱스 그 자체이고, .pkl은 LangChain 메타데이터로, 각 벡터에 대응되는 문서 내용 및 ID가 포함되어 있다.

벡터 인덱스 GCS 업로드

로컬에 저장된 faiss_index 폴더 내 파일들을 GCS 버킷에 업로드

# upload_to_gcs.py

from google.cloud import storage
import os

bucket_name = "choon-assistance-ai-bucket"
source_dir = "../vector/faiss_index"
destination_prefix = "vector/faiss_index"

client = storage.Client()
bucket = client.bucket(bucket_name)

# 업로드
for filename in os.listdir(source_dir):
    local_path = os.path.join(source_dir, filename)
    blob = bucket.blob(f"{destination_prefix}/{filename}")
    blob.upload_from_filename(local_path)
    print(f"⬆️ 업로드 완료: {blob.name}")

현재 우리 서버는 무료 크레딧을 사용하기 위해 GCP 서버를 사용하고 있고, 따라서 faiss.index를 GCS 버킷에 백업을 해두기로 했다. 앞서 말했듯이, FAISS는 서버형 DB가 아니기 때문에, 로컬 파일 기반 벡터 인덱스를 생성하고, 이를 백업해둘 공유 저장소가 필요하기 때문이다.

GCS(Google Cloud Storage): GCP(Google Cloud Platform)에서 제공하는 파일 저장소 서비스로, 클라우드 기반 폴더/디렉토리 시스템이자 버킷 기반 오브젝트 스토리지
버킷: GCS에서 파일을 넣는 최상위 컨테이너. 클라우드에서 사용되는 디렉토리 개념
오프젝트 스토리지: 파일(오브젝트) 단위로 저장하며 URL 기반 접근이 가능한 저장소 (DB나 구조화된 저장소 또한 다양하게 존재)

GCS → 로컬 다운로드 + 인덱스 로딩

GCS로부터 인덱스 파일 다운로드
LangChain의 FAISS.load_local()로 메모리에 로드

# vector_store.py

import os
from google.cloud import storage
from langchain_community.vectorstores import FAISS
from app.core.embedding_model import embedder

# 환경변수 또는 config.py에서 관리할 수도 있음
GCS_BUCKET = "choon-assistance-ai-bucket"
GCS_PREFIX = "vector/faiss_index"
LOCAL_INDEX_DIR = "/tmp/faiss_index"

def download_faiss_from_gcs():
    client = storage.Client()
    bucket = client.bucket(GCS_BUCKET)
    blobs = bucket.list_blobs(prefix=GCS_PREFIX)

    os.makedirs(LOCAL_INDEX_DIR, exist_ok=True)

    for blob in blobs:
        if blob.name.endswith("/"):  # 디렉토리 무시
            continue
        filename = os.path.basename(blob.name)
        blob.download_to_filename(os.path.join(LOCAL_INDEX_DIR, filename))

def load_vectorstore():
    download_faiss_from_gcs()
    return FAISS.load_local(LOCAL_INDEX_DIR, embeddings=embedder, allow_dangerous_deserialization=True)

load_local()은 로컬 디스크(tmp)에 저장된 인덱스 파일(.faiss, .pkl)을 읽어 다시 메모리에 로드하는 함수이다. load_vectorstore()를 통해 VectorStore 객체를 생성한다.

RAG 기반 질의 처리

load_vectorstore()로 retriever를 불러오기
사용자 질의를 통해 retriever로 유사 문서 검색
검색된 문서들의 내용을 context로 사용하여 프롬프트에 넣기
LangChain 기반 프롬프트 체인 실행
LLM이 RAG 방식으로 답변 생성

# llm_client.py

from langchain_core.output_parsers import StrOutputParser
from app.model.qwen2_5_loader import llm
from app.model.prompt_template import chat_prompt
from app.core.vector_store import load_vectorstore

# ✅ FAISS 기반 retriever 로드
faiss_vectorstore = load_vectorstore()
retriever = faiss_vectorstore.as_retriever()

# ✅ 사용자 질문 → RAG 기반 응답 생성
def get_chat_response(question: str) -> str:
    # 유사 문서 검색
    docs = retriever.get_relevant_documents(question)

    # context 조합 (텍스트만 추출)
    context = "\n\n".join([doc.page_content for doc in docs])

    # LangChain 체인 구성
    chain = chat_prompt | llm | StrOutputParser()

    # 실행
    return chain.invoke({
        "context": context,
        "question": question
    })

이제 as_retriever()를 사용하여 FAISS 객체를 LangChain의 Retriever 인터페이스로 변환한다. 이를 통해 get_relevant_documents() 메서드를 사용할 수 있고, 이 메서드는 사용자 질의를 임베딩하여 Faiss 인덱스에서 유사한 벡터들을 찾아서 연결된 문서 객체 리스트로 반환하는 역할을 한다.

응답 결과

포스트맨으로 테스트 결과, "휴가 신청 어떻게 해"라는 질문에 대해 적절한 문서를 찾아서 반환한 것을 볼 수 있다.

응답 결과를 자세히 봤을때, 내용이 그렇게 훌룡하다고 판단되지는 않는다. 기존 코랩에서 모델 실험 시에 꽤나 좋은 퀄리티의 응답이 나왔는데, 갑자기 확 떨어진 느낌을 받아서 추후에 디버깅을 해보니, 랭체인 문제였던걸로 나왔다. 관련해서 글을 따로 올릴 예정이다 ^!^

Reference

해당 내용에 대한 깃허브 주소: https://github.com/100-hours-a-week/20-real-ai/pull/30

https://wikidocs.net/234016

https://devocean.sk.com/blog/techBoardDetail.do?ID=165867&boardType=techBlog