RAG 기초 (4) - Vector Store: 저장하고 찾아오기

RAG 기초 시리즈의 4편입니다. 전체 목차는 0편에 있습니다.

numpy로도 되는데

3편의 검색은 다섯 줄이었습니다. 그런데도 별도 저장소를 쓰는 이유는 그 다섯 줄이 감당하지 못하는 것들 때문입니다.

한계	내용
영속성	프로세스를 끄면 벡터가 사라집니다. 서버를 재시작할 때마다 전체 문서를 다시 임베딩해야 합니다
증분 갱신	문서 하나가 바뀌었을 때 그 문서의 chunk만 갈아끼우는 코드를 직접 짜야 합니다
메타데이터 필터	“배송 문서 안에서만 검색”을 하려면 인덱스 관리 코드가 필요합니다
규모	chunk가 수십만 개가 되면 전체 내적이 매 질의마다 부담이 됩니다

vector store는 이 네 가지를 대신해 줍니다. 이 시리즈에서는 로컬 파일로 동작하고 설치가 간단한 Chroma를 씁니다.

저장하기

import chromadb
from sentence_transformers import SentenceTransformer
from chunking import build_chunks          # 2편

EMBED_MODEL = "jhgan/ko-sroberta-multitask"
model = SentenceTransformer(EMBED_MODEL)

client = chromadb.PersistentClient(path="./chroma")      # ① 디스크에 남는다
collection = client.get_or_create_collection(
    name="policy",
    metadata={"hnsw:space": "cosine"},                   # ② 거리 기준을 코사인으로
)

chunks = build_chunks()
collection.upsert(                                       # ③ 같은 id면 덮어쓴다
    ids=[c["id"] for c in chunks],
    documents=[c["text"] for c in chunks],
    embeddings=model.encode(
        [c["text"] for c in chunks], normalize_embeddings=True
    ).tolist(),
    metadatas=[{"source": c["source"], "title": c["title"]} for c in chunks],
)

print(collection.count())   # 7

• ① PersistentClient는 지정한 경로에 데이터를 남깁니다. 다음 실행에서 그대로 읽힙니다
• ② 기본 거리 기준은 코사인이 아니므로 명시합니다. 이 설정은 collection을 만들 때만 적용되고 나중에 바꿀 수 없습니다
• ③ add 대신 upsert를 쓰면 같은 id를 다시 넣을 때 덮어씁니다. 재색인 스크립트를 반복 실행해도 중복이 쌓이지 않습니다

id를 membership-0처럼 문서와 순번으로 만들어 둔 2편의 결정이 여기서 쓰입니다. 문서가 바뀌면 그 문서의 id만 다시 넣으면 됩니다.

찾아오기

def search(query: str, k: int = 3, source: str | None = None):
    qvec = model.encode(query, normalize_embeddings=True).tolist()
    result = collection.query(
        query_embeddings=[qvec],
        n_results=k,
        where={"source": source} if source else None,
    )
    return [
        {
            "text": doc,
            "source": meta["source"],
            "score": 1 - dist,          # 코사인 거리를 유사도로 되돌린다
        }
        for doc, meta, dist in zip(
            result["documents"][0], result["metadatas"][0], result["distances"][0]
        )
    ]

for hit in search("골드 등급이면 배송비가 무료인가요?"):
    print(f"{hit['score']:.3f}  {hit['source']}  {hit['text'][:40]}...")
# 0.734  membership.md  ## 등급별 혜택 적립률은 실버 1%, 골드 2%...
# 0.612  shipping.md    # 배송 정책 ## 배송비 기본 배송비는 3,000원...
# 0.401  refund.md      ## 단순 변심 상품 수령일로부터 7일 이내에...

query는 결과를 리스트의 리스트로 돌려줍니다. 질문을 여러 개 한 번에 넣을 수 있기 때문이고, 하나만 넣었으므로 [0]으로 꺼냅니다.

distances는 거리라서 작을수록 가깝습니다. 코사인 공간에서는 1 - 거리가 3편에서 계산한 코사인 유사도와 같습니다. 유사도로 뒤집어 두면 사람이 읽기 편하고 앞 편의 숫자와도 비교됩니다.

메타데이터 필터

where로 검색 범위를 좁힙니다.

for hit in search("며칠까지 가능한가요?", source="refund.md"):
    print(hit["source"], hit["text"][:30])

“며칠까지”라는 질문은 환불 7일, 하자 30일, 배송 2~3일 어디에나 걸립니다. 사용자가 환불 화면에서 물었다는 맥락을 알고 있다면 문서를 좁히는 것만으로 정확도가 크게 오릅니다.

실무에서 자주 쓰는 필터는 문서 종류, 작성일, 접근 권한입니다. 특히 권한은 검색 단계에서 걸러야 합니다. 볼 수 없는 문서의 내용을 프롬프트에 넣으면 모델이 그대로 답해 버립니다.

k를 몇으로 둘까

가져올 chunk 개수입니다.

k	결과
1~2	정확한 chunk 하나만 필요한 단순 질문에 충분합니다. 놓치면 답이 아예 나오지 않습니다
3~5	대부분의 경우에 무난합니다. 이 시리즈는 3을 씁니다
10 이상	관련 없는 내용이 섞여 답이 흐려지고 입력 토큰이 늡니다

k를 늘리면 정답 chunk가 포함될 확률은 올라가지만 답변 품질은 어느 지점부터 떨어집니다. 두 지표를 나눠 재는 방법이 7편에 있습니다. k를 늘린 뒤 순위를 다시 매겨 상위 몇 개만 남기는 방법은 6편의 reranking입니다.

색인 스크립트로 정리하기

여기까지를 다시 돌릴 수 있는 스크립트 하나로 묶어 둡니다.

# index.py
import chromadb
from sentence_transformers import SentenceTransformer
from chunking import build_chunks

EMBED_MODEL = "jhgan/ko-sroberta-multitask"
COLLECTION = "policy"

def reindex(path: str = "./chroma") -> int:
    model = SentenceTransformer(EMBED_MODEL)
    client = chromadb.PersistentClient(path=path)
    collection = client.get_or_create_collection(
        name=COLLECTION, metadata={"hnsw:space": "cosine"}
    )

    chunks = build_chunks()
    texts = [c["text"] for c in chunks]
    collection.upsert(
        ids=[c["id"] for c in chunks],
        documents=texts,
        embeddings=model.encode(texts, normalize_embeddings=True).tolist(),
        metadatas=[{"source": c["source"], "title": c["title"]} for c in chunks],
    )

    stale = set(collection.get()["ids"]) - {c["id"] for c in chunks}   # 사라진 chunk 정리
    if stale:
        collection.delete(ids=list(stale))
    return collection.count()

if __name__ == "__main__":
    print(f"{reindex()}개 chunk 색인 완료")

마지막 정리 부분이 중요합니다. 문서가 짧아져서 chunk 수가 줄면 예전 id가 저장소에 남아 있습니다. 지우지 않으면 이미 삭제된 정책이 계속 검색되어 답에 섞입니다.

다른 선택지

저장소	특징
Chroma	로컬 파일 기반, 설치가 간단합니다. 프로토타입과 소규모에 적합합니다
FAISS	라이브러리 형태로 매우 빠릅니다. 메타데이터 관리는 직접 해야 합니다
pgvector	PostgreSQL 확장입니다. 이미 쓰는 DB에 벡터 컬럼을 더하는 방식이라 운영 부담이 적습니다
Pinecone, Qdrant 등	관리형 서비스입니다. 규모가 커지고 가용성이 필요할 때 검토합니다

인터페이스는 다르지만 하는 일은 같습니다. 벡터와 메타데이터를 넣고, 질의 벡터에 가까운 것을 k개 돌려주는 것입니다. 3편에서 numpy로 짜본 그 동작입니다.

다음 글: RAG 기초 (5) - Generation: 근거를 넣고 출처와 함께 답하기