문서 인덱싱과 검색 — 벡터 스토어 구축 · Semantic Search 구현

2025.08.16 - [RAG] - RAG Bot 만들기: Retrieval 파트 —Vector DB, Vectorization

RAG Bot 만들기: Retrieval 파트 —Vector DB, Vectorization

 

이 글은 앞선 글(문서 로딩·전처리, 벡터 DB 개요)에서 만든 docs/chunks를 이어서 사용합니다. 목표는 인덱싱(Indexing) → 검색(Retrieval) → Semantic Search 체감 품질 개선을 실전 코드로 끝까지 구현하는 것입니다.

 

---

1. 개념 정리: 인덱싱과 검색의 역할 분담

• 인덱싱(Indexing): 전처리된 문서를 청킹하고 임베딩으로 벡터화하여 벡터 스토어에 저장하는 과정입니다.
• 검색(Retrieval): 사용자의 질문을 임베딩해서 가까운 벡터 k개를 찾아오는 단계입니다.
• Semantic Search: 키워드가 아닌 의미 유사도로 찾는 검색. 하이브리드(BM25+벡터), MMR, 메타데이터 필터, 리랭킹 등을 조합하여 품질을 끌어올립니다.

---

2. 인덱싱 파이프라인 설계

2-1. ID/메타데이터 전략

문서 추적성과 증분 업데이트를 위해 ID, source, version, locale 같은 메타데이터를 일관되게 저장하세요.

import hashlib
from langchain_core.documents import Document

# 예시: 해시 기반 ID 부여
indexed_docs = []
for i, d in enumerate(chunks):  # chunks: 이전 글에서 만든 Document 리스트
    key = hashlib.sha1((d.metadata.get("source", "") + d.page_content).encode("utf-8")).hexdigest()[:16]
    d.metadata.update({
        "id": key,
        "doc_type": d.metadata.get("doc_type", "kb"),
        "locale": d.metadata.get("locale", "ko"),
        "version": d.metadata.get("version", "v1"),
    })
    indexed_docs.append(Document(page_content=d.page_content, metadata=d.metadata))

2-2. 임베딩 모델 선택

• OpenAI text-embedding-3-small(1536D): 비용 대비 우수, 범용 추천
• (대안) HuggingFace bge-small, gte-small 등: 비용↓/온프레미스 가능

from langchain_openai import OpenAIEmbeddings
emb = OpenAIEmbeddings(model="text-embedding-3-small")  # OPENAI_API_KEY 필요

---

3. 벡터 스토어 구축 (FAISS · Chroma 예제)

3-1. FAISS (로컬)

from langchain_community.vectorstores import FAISS

faiss_store = FAISS.from_documents(indexed_docs, embedding=emb)
faiss_store.save_local("indexes/faiss_kb")

# 재시작/다른 프로세스에서 로드
faiss_store = FAISS.load_local("indexes/faiss_kb", emb, allow_dangerous_deserialization=True)

3-2. Chroma (로컬/서버)

from langchain_chroma import Chroma

chroma_store = Chroma(
    collection_name="kb",
    persist_directory="indexes/chroma_kb",
    embedding_function=emb,
)
# 신규 문서 업서트 (id를 직접 지정하면 중복/갱신 관리가 쉬움)
ids = [d.metadata["id"] for d in indexed_docs]
chroma_store.add_documents(indexed_docs, ids=ids)
chroma_store.persist()  # 디스크에 영속화

증분 업데이트 팁

• 문서 해시(ID) 비교로 변경분만 재임베딩/업서트
• 삭제된 원문은 tombstone(삭제 마크) 메타데이터를 넣어 필터링하거나, 스토어에서 삭제 API 사용

---

4. 검색(Semantic Search) 기본형

4-1. Top-k 유사도 검색

query = "체크인 시간과 노쇼 수수료가 궁금해"

# FAISS 예시
faiss_hits = faiss_store.similarity_search(query, k=4)
for i, d in enumerate(faiss_hits, 1):
    print(i, d.metadata.get("source"), d.page_content[:80])

# Chroma + 메타데이터 필터 (언어=ko)
ret = chroma_store.as_retriever(search_kwargs={"k": 4, "filter": {"locale": "ko"}})
for d in ret.invoke(query):
    print(d.metadata.get("locale"), d.metadata.get("source"))

4-2. MMR(Maximum Marginal Relevance)로 다양성 향상

문맥이 비슷한 청크만 몰리지 않게 중복을 줄이고 커버리지를 늘리는 방법입니다.

# MMR 검색 (FAISS)
mmr_retriever = faiss_store.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 6, "fetch_k": 20, "lambda_mult": 0.3}  # 다양성↑(0~1)
)
mmr_docs = mmr_retriever.invoke("환불 규정 요약")

4-3. 점수와 임계치(Threshold)

낮은 유사도의 결과를 걸러 노이즈 응답을 줄입니다.

docs_with_scores = faiss_store.similarity_search_with_score(query, k=8)
filtered = [d for d, score in docs_with_scores if score > 0.75]  # cosine 기준 예시

점수 스케일은 스토어/거리함수에 따라 달라집니다. 운영 전에 리콜/정확도 테스트로 임계값을 잡으세요.

---

5. 하이브리드 검색 (BM25 + 벡터)

정책/규정처럼 정확한 용어 매칭이 중요한 데이터는 하이브리드가 효과적입니다.

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

bm25 = BM25Retriever.from_documents(indexed_docs); bm25.k = 4
vec_ret = faiss_store.as_retriever(search_kwargs={"k": 4})
ensemble = EnsembleRetriever(retrievers=[bm25, vec_ret], weights=[0.4, 0.6])

docs = ensemble.get_relevant_documents("No-show 수수료와 체크인〮아웃 시간")

---

6. 리랭킹(Re-ranking)으로 최종 품질 개선 (선택)

리트리버 결과 상위 N개를 교차 인코더나 LLM 기반 평가로 재정렬합니다. (외부 모델 필요)

# (개념 예시) 상위 20개 → LLM에 "질문과의 관련성" 점수를 매기게 하고 상위 5개만 사용
from operator import itemgetter

candidates = faiss_store.similarity_search(query, k=20)
# pseudo: scores = llm_score(query, candidates)
# ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)[:5]

---

7. RAG 체인 연결 (LCEL)

검색 결과를 포맷팅해 LLM에 전달하여 최종 답변을 생성합니다.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

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

PROMPT = ChatPromptTemplate.from_template(
    """
    너는 고객지원 봇이야. 다음 컨텍스트를 참고해 정확하고 간결하게 답해.
    컨텍스트:\n{context}\n\n질문: {question}
    """
)

def format_docs(docs):
    return "\n\n".join(
        f"[source: {d.metadata.get('source','?')}]\n{d.page_content}" for d in docs
    )

retriever = mmr_retriever  # 위에서 만든 어떤 리트리버든 사용 가능

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

print(rag_chain.invoke("노쇼(No-show) 수수료는?"))

---

8. 운영 체크리스트

• 임베딩/DB 매치: 차원/거리함수 일치 확인 (예: 1536D + cosine)
• 메타데이터: source, id, version, locale, doc_type을 표준화
• 증분 업데이트: 해시/ID 기반으로 변경분만 재임베딩, 배치 업서트
• 성능: k, fetch_k, lambda_mult, 임계치 등 파라미터를 Dataset 기반으로 튜닝(LangSmith 권장)
• 관측: 트레이싱(토큰/지연/실패율), 캐시(Hit Ratio), 최종 답변 근거(출처) 로깅

---

9. FastAPI로 간단한 Semantic Search API 만들기

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()
retriever = chroma_store.as_retriever(search_kwargs={"k": 5, "filter": {"locale": "ko"}})

class Q(BaseModel):
    query: str

@app.post("/search")
def search(q: Q):
    hits = retriever.invoke(q.query)
    return [{
        "source": h.metadata.get("source"),
        "score": None,  # 필요시 with_score API 사용
        "snippet": h.page_content[:240]
    } for h in hits]

---

마무리

이제 인덱싱 → 검색 → 품질 향상(MMR/하이브리드/리랭킹) 까지 한 바퀴를 돌았습니다. 다음 글에서는 이 리트리버를 Generation 파트(프롬프트 엔지니어링, 대화 메모리) 와 결합해 진짜로 대답을 잘하는 RAG를 완성합니다.