버전 관리 문서에서의 브랜치 인지 검색

시간: 25분 난이도: 중급

문서 말뭉치가 git 스타일의 브랜치로 버전 관리된다면, 일반적인 검색은 브랜치를 넘어 누출돼요. 다른 브랜치의 콘텐츠나, 현재 브랜치가 이미 교체한 버전을 반환하게 되죠.

이 튜토리얼은 그런 말뭉치를 Qdrant에 인덱싱하고, 각 쿼리를 단일 브랜치의 라이브 뷰(live view) 로 범위를 한정하는 방법을 보여줘요. 라이브 뷰란 자기 브랜치의 커밋에다가 조상으로부터 상속받은 것, 그리고 이후 커밋이 교체하지 않은 것만 포함하는 뷰예요.

이 패턴은 초안(draft)과 발행(published) 브랜치가 있는 문서 사이트, 지역별 포크가 있는 정책 저장소, 각 기능 브랜치마다 고유한 뷰가 필요한 코드베이스에 잘 맞아요.

출처: Qdrant 공식 문서 — branch-aware-search

설정 (Setup)

각 파일의 버전은 하나의 포인트이고, 의미적 검색을 위해 dense 모델로 임베딩돼요. Cloud Quickstart에서 클러스터 생성과 클라이언트 연결을 다뤄요.

이 튜토리얼은 임베딩을 서버 측에서 생성하는 Qdrant Cloud Inference를 사용해요. 무료 티어가 이 튜토리얼의 사용량을 충당해요. 셀프호스팅하려면 FastEmbed 같은 라이브러리로 클라이언트에서 dense 벡터를 생성하고 models.Document 대신 원시 벡터로 전달하세요.

from qdrant_client import QdrantClient, models
from typing import List, Tuple

# 384차원 dense 벡터, Cloud Inference에서 무료
MODEL = "sentence-transformers/all-MiniLM-L6-v2"

# url과 api_key를 https://cloud.qdrant.io 의 것으로 교체
client = QdrantClient(
    url="https://xyz-example.qdrant.io:6333",
    api_key="<your-api-key>",
    cloud_inference=True,
)

client.create_collection(
    collection_name="content",
    vectors_config=models.VectorParams(
        size=384, distance=models.Distance.COSINE),
)

모든 포인트는 파생 ID(derived ID) 를 갖는데, 같은 히스토리를 다시 재생하면 같은 ID가 만들어져서 리빌드 시 upsert가 포인트를 중복하지 않아요:

import uuid

NS = uuid.UUID("00000000-0000-0000-0000-000000000042")

def point_id(branch: str, seq: int, path: str) -> str:
    return str(uuid.uuid5(NS, f"{branch}|{seq}|{path}"))

각 단계는 그 포인트에서 처음 필요해지는 시점에 페이로드 인덱스를 만들어서, 각 인덱스가 왜 존재하는지 볼 수 있게 해요. 프로덕션에서는 데이터를 로드하기 전에 모든 인덱스를 만들어 추가 인덱싱 패스를 피하는 게 좋아요.

1단계: 파일 추가하기 (Step 1: Add a File)

단일 root 브랜치와 파일 하나를 추가하는 커밋 하나로 시작해요.

root  ●            seq 0: add pricing.md

커밋은 브랜치별 seq(0, 1, 2, ...)로 번호가 매겨져요. 각 커밋이 쓰는 파일은 파일을 쓴 branch, seq, 파일 path로 태그된 포인트가 돼요. 나중에 브랜치에서 파일을 찾으려면 path와 branch로 필터링하므로, 둘 다 페이로드 인덱스가 필요해요:

client.create_payload_index(
    collection_name="content", field_name="path",
    field_schema=models.PayloadSchemaType.KEYWORD,
)
client.create_payload_index(
    collection_name="content", field_name="branch",
    field_schema=models.PayloadSchemaType.KEYWORD,
)

쓰기는 콘텐츠를 임베딩하고 세 필드를 저장해요:

def update(file_name, branch, seq, content):
    # 콘텐츠를 MODEL로 임베딩하고, branch/seq/path 태그와 함께 upsert
    client.upsert(collection_name="content", points=[models.PointStruct(
        id=point_id(branch, seq, file_name),
        vector=models.Document(text=content, model=MODEL),
        payload={"branch": branch, "seq": seq, "path": file_name},
    )])

브랜치에서 파일을 다시 읽는 것은 path와 branch에 대한 필터예요. 필터에 맞는 포인트를 반환하는 scroll을 사용해요:

def lookup(file_name, branch):
    points, _ = client.scroll(
        collection_name="content",
        scroll_filter=models.Filter(must=[
            models.FieldCondition(
                key="path", match=models.MatchValue(value=file_name)),
            models.FieldCondition(
                key="branch", match=models.MatchValue(value=branch)),
        ]),
        limit=1,
    )
    return points[0] if points else None

2단계: 파일 업데이트하기 (Step 2: Update a File)

파일을 업데이트하면 이전 버전은 그 브랜치에서 더 이상 보이지 않아야 해요. 새 버전을 쓰기 전에 이전 버전을 "대체됨(superseded)"으로 표시해요:

def update(file_name, branch, seq, content):
    prev = lookup(file_name, branch)
    if prev:
        supersede(prev, by=branch, seq=seq)
    client.upsert(collection_name="content", points=[models.PointStruct(
        id=point_id(branch, seq, file_name),
        vector=models.Document(text=content, model=MODEL),
        payload={"branch": branch, "seq": seq, "path": file_name},
    )])

overwritten_in 배열 페이로드에 대체 정보를 기록하고, 그 필드에 인덱스를 만들어요:

client.create_payload_index(
    collection_name="content", field_name="overwritten_in[].by",
    field_schema=models.PayloadSchemaType.KEYWORD,
)

lookup은 이제 nested filter가 포함된 must_not을 추가해서, 이 브랜치의 마크를 가진 모든 버전을 제외해요:

def visibility_filter(branch, path=None):
    """
    브랜치의 파일 하나(또는 path가 None이면 모든 파일)의 뷰에
    맞는 필터를 반환한다.

    결과의 단순화된 예:

    filter: {
        must = [
            branch == "root",
            path == "pricing.md"
        ],
        must_not = [
            overwritten_in[].by == "root"
        ]
    }
    """

3단계: 파일 삭제하기 (Step 3: Delete a File)

삭제도 같은 원리예요. 파일을 실제로 지우는 대신, "삭제됨(deleted)" 마크를 넣어 브랜치의 라이브 뷰에서 가려지게 해요. tombstone 같은 페이로드 필드로 표시하고, 필터에서 제외하면 돼요.

4단계: 루트에서 브랜치 만들기 (Step 4: Branch Off the Root)

이제 읽기가 두 개 이상의 브랜치에 걸쳐 있으므로, 브랜치 로직을 자체 branch_filter로 분리해요. 현재 브랜치 먼저, 그다음 모든 조상 브랜치에 대해 후보 절 하나와 제외 절 하나를 만드는 구조예요.

후보들(should라서 어느 하나라도 매치 가능)은 각 브랜치의 버전을 모으고, 제외들(must_not)은 그 브랜치가 교체한 것을 떨어뜨려요:

def branch_filter(
        branch: str,
        ancestry: List[Tuple[str, int]]
    ) -> models.Filter:
    """
    현재 브랜치의 뷰에서 보이는 파일만 매치하는 필터를 반환한다.

    Ancestry는 (branch, fork_seq) 쌍의 목록으로, 각 브랜치가
    자식 브랜치에 상속하는 내용을 결정한다.
    """
    # 후보 should 절: 각 (브랜치, fork_seq) 쌍에서 해당 브랜치의 버전을 고름
    # 제외 must_not 절: 각 브랜치가 돌연변이로 대체/삭제한 것을 제외
    ...

6단계: 포크 이후 루트에 커밋하기 (Step 6: Commit on the Root After the Fork)

seq와 대체 시퀀스도 정수 필터로 쓰이므로 정수 인덱스를 만들어요:

client.create_payload_index(
    collection_name="content", field_name="seq",
    field_schema=models.PayloadSchemaType.INTEGER,
)
client.create_payload_index(
    collection_name="content", field_name="overwritten_in[].seq",
    field_schema=models.PayloadSchemaType.INTEGER,
)

브랜치 검색하기 (Searching a Branch)

lookup은 알려진 경로 하나를 해석하는 반면, 검색은 전체 말뭉치를 관련도로 순위화해요. 동일한 가시성 필터를 경로 스크롤 대신 의미적 쿼리에 전달하면 돼요:

def search(query, branch, ancestry, limit=5):
    return client.query_points(
        collection_name="content",
        query=models.Document(text=query, model=MODEL),
        query_filter=visibility_filter(branch, ancestry),
        limit=limit, with_payload=True,
    ).points

같은 쿼리를 각 브랜치로 범위를 한정하면 그 브랜치의 라이브 버전 페이지를 반환해요.

프로덕션 규모로 확장하기 (Scaling to Production)

  • Qdrant는 파생 인덱스예요. 브랜치 조상과 히스토리는 버전 관리 시스템에 살아 있어요. 그 히스토리를 재생해서 컬렉션을 만들고, 인덱스가 어긋나면 리빌드하세요. 파생 point ID 덕분에 전체 재생이 멱등적이에요. 같은 커밋이 같은 포인트를 쓰므로 upsert가 깔끔하게 덮어써요.

더 알아보기 (Learn more)