Qdrant 벡터 검색에서 Reranking까지, 실전 코드

왜 벡터 검색만으로는 부족한가?

"6개월 전에 정리한 Kubernetes 노트"와 "어제 작성한 Kubernetes 노트"가 있다고 합시다. 벡터 유사도(Vector Similarity)만으로 검색하면 둘 다 비슷한 점수를 받습니다. 의미적으로 비슷하니까요. 하지만 실제로는 어제 작성한 노트가 더 가치 있을 가능성이 높습니다.

이것이 바로 Multi-Stage Retrieval(다단계 검색)이 필요한 이유입니다. 벡터 검색으로 후보를 넓게 뽑고, 시간 감쇠(Time Decay)로 오래된 문서의 점수를 깎고, 마지막으로 Cross-Encoder로 정밀하게 재순위를 매기는 3단계 파이프라인을 만들면, 단순 벡터 검색보다 훨씬 정확한 결과를 얻을 수 있습니다.

실생활 활용 시나리오

시나리오 1: 개인 지식 관리 시스템
매일 업무 노트, 기술 메모, 회의록을 쌓아두는 개인 지식 베이스가 있습니다. "Docker 네트워크 설정 방법"을 검색하면 2년 전 노트와 지난주 노트가 함께 나옵니다. 시간 감쇠를 적용하면 최신 정보가 자연스럽게 상위에 올라옵니다.

시나리오 2: 사내 문서 검색
API 문서, 온보딩 가이드, 장애 보고서가 섞여 있는 사내 시스템에서 "결제 API 에러 처리"를 검색합니다. 벡터 검색은 후보를 100개 뽑지만, Cross-Encoder가 질문의 의도에 가장 정확히 맞는 문서 10개를 골라냅니다.

전체 파이프라인 한눈에 보기

핵심 아이디어는 "깔때기"입니다. 위에서 아래로 갈수록 후보가 줄어들지만 정확도는 올라갑니다. Stage 1은 빠르지만 대략적이고, Stage 3은 느리지만 정밀합니다. 각 단계가 다음 단계의 부하를 줄여주는 구조입니다.

Stage 1: Qdrant 벡터 검색

왜 첫 단계에서 벡터 검색을 쓰는가?

수십만 개의 문서에서 Cross-Encoder로 하나하나 비교하면 몇 분이 걸립니다. 벡터 검색은 ANN(Approximate Nearest Neighbor, 근사 최근접 이웃) 인덱스 덕분에 수백만 개 문서에서도 밀리초 단위로 후보를 뽑아냅니다.

임베딩 생성

먼저 사용자의 쿼리를 벡터(숫자 배열)로 변환해야 합니다. 여기서는 OpenAI의 text-embedding-3-small 모델을 사용합니다.

from openai import OpenAI

client = OpenAI()

def get_embedding(text: str) -> list[float]:
    """텍스트를 1536차원 벡터로 변환"""
    response = client.embeddings.create(
        input=text,
        model="text-embedding-3-small"
    )
    return response.data[0].embedding

query = "Kubernetes 네트워크 정책 설정 방법"
query_vector = get_embedding(query)
print(f"벡터 차원: {len(query_vector)}")  # 1536
print(f"첫 5개 값: {query_vector[:5]}")

실행 결과:

벡터 차원: 1536
첫 5개 값: [0.0123, -0.0456, 0.0789, -0.0234, 0.0567]

Qdrant 검색 요청

이제 이 벡터를 Qdrant에 보내서 최근 1년 이내 문서 중 유사한 것 100개를 뽑아봅시다. Qdrant의 강점인 메타데이터 필터링을 활용해서, 벡터 검색과 날짜 필터를 동시에 처리합니다.

from qdrant_client import QdrantClient
from qdrant_client.models import (
    Filter, FieldCondition, Range,
    SearchRequest, NamedVector
)
from datetime import datetime, timedelta

qdrant = QdrantClient(host="localhost", port=6333)

# 1년 전 날짜 계산 (Unix timestamp)
one_year_ago = datetime.now() - timedelta(days=365)
one_year_ago_ts = one_year_ago.timestamp()

# 검색 실행
results = qdrant.search(
    collection_name="knowledge_base",
    query_vector=query_vector,
    query_filter=Filter(
        must=[
            FieldCondition(
                key="created_at",
                range=Range(gte=one_year_ago_ts)
            )
        ]
    ),
    limit=100,
    with_payload=True,
    score_threshold=0.3  # 최소 유사도 컷오프
)

실제 HTTP 요청 JSON (REST API 기준)

Python 클라이언트가 내부적으로 보내는 HTTP 요청을 직접 살펴보면 이렇습니다. API를 직접 호출해야 하는 상황(다른 언어, curl 테스트 등)에서 참고할 수 있습니다.

POST /collections/knowledge_base/points/search
{
    "vector": [0.0123, -0.0456, 0.0789, "... (1536차원)"],
    "filter": {
        "must": [
            {
                "key": "created_at",
                "range": {
                    "gte": 1711036800.0
                }
            }
        ]
    },
    "limit": 100,
    "with_payload": true,
    "score_threshold": 0.3,
    "params": {
        "hnsw_ef": 128,
        "exact": false
    }
}

params.hnsw_ef는 검색 정확도를 조절하는 파라미터입니다. 값이 클수록 정확하지만 느려집니다. 기본값(128)이면 대부분의 경우 충분합니다. 100개 정도를 뽑을 때는 이 값을 올릴 필요가 거의 없습니다.

응답 JSON

{
    "result": [
        {
            "id": "note-20260315-k8s-netpol",
            "version": 42,
            "score": 0.89,
            "payload": {
                "title": "Kubernetes NetworkPolicy 실전 가이드",
                "content": "K8s 네트워크 정책은 Pod 간 트래픽을 제어하는...",
                "created_at": 1710460800.0,
                "source": "work/ai-llm/k8s-network.md",
                "tags": ["kubernetes", "networking", "security"]
            }
        },
        {
            "id": "note-20250901-k8s-cni",
            "version": 38,
            "score": 0.82,
            "payload": {
                "title": "CNI 플러그인 비교: Calico vs Cilium",
                "content": "Kubernetes CNI(Container Network Interface)는...",
                "created_at": 1693526400.0,
                "source": "work/ai-llm/k8s-cni.md",
                "tags": ["kubernetes", "cni", "calico", "cilium"]
            }
        },
        {
            "id": "note-20250620-docker-network",
            "version": 25,
            "score": 0.78,
            "payload": {
                "title": "Docker 네트워크 모드 정리",
                "content": "Docker의 bridge, host, overlay 네트워크...",
                "created_at": 1687219200.0,
                "source": "work/infra/docker-network.md",
                "tags": ["docker", "networking"]
            }
        }
    ],
    "status": "ok",
    "time": 0.0034
}

응답에서 주목할 점:

• score: cosine similarity 값 (0~1, 높을수록 유사). 우리가 지정한 0.3 이상만 반환됩니다
• time: 0.0034초. 벡터 검색이 얼마나 빠른지 보여줍니다
• payload: 저장해둔 메타데이터. 다음 단계에서 created_at을 시간 감쇠에 사용합니다

Stage 2: 시간 감쇠 (Time Decay)

왜 시간 감쇠가 필요한가?

Stage 1에서 score 0.89인 3개월 전 문서와 score 0.85인 어제 작성한 문서가 있다면, 어떤 것이 더 유용할까요? 많은 경우 최신 문서입니다. 시간 감쇠는 오래된 문서의 점수를 자연스럽게 낮춰서, 동일한 유사도라면 최신 문서가 상위에 오도록 만듭니다.

Exponential Decay 공식

위 그래프에서 보듯이, 반감기(half-life)를 365일로 설정하면:

• 오늘 작성된 문서: decay = 1.0 (감쇠 없음)
• 6개월 전 문서: decay ≈ 0.71 (점수 29% 감소)
• 1년 전 문서: decay ≈ 0.50 (점수 절반)

구현 코드

import math
from datetime import datetime
from dataclasses import dataclass

@dataclass
class ScoredDocument:
    id: str
    title: str
    content: str
    created_at: float      # Unix timestamp
    similarity: float      # Stage 1에서 받은 cosine similarity
    decay_factor: float = 1.0
    decayed_score: float = 0.0
    rerank_score: float = 0.0
    final_score: float = 0.0

def apply_time_decay(
    documents: list[ScoredDocument],
    half_life_days: float = 365.0,
    decay_weight: float = 0.3
) -> list[ScoredDocument]:
    """
    시간 감쇠를 적용한 점수 재계산.

    Args:
        documents: Stage 1에서 받은 문서 리스트
        half_life_days: 반감기 (이 일수가 지나면 decay가 0.5)
        decay_weight: 시간 감쇠가 최종 점수에 미치는 비중 (0~1)
                      0이면 시간 무시, 1이면 시간이 전부
    """
    # λ 계산: half_life에서 decay = 0.5가 되도록
    decay_lambda = math.log(2) / half_life_days
    now = datetime.now().timestamp()

    for doc in documents:
        days_old = (now - doc.created_at) / 86400  # 초 → 일
        doc.decay_factor = math.exp(-decay_lambda * days_old)

        # 가중 합산: similarity와 decay의 밸런스
        doc.decayed_score = (
            (1 - decay_weight) * doc.similarity
            + decay_weight * doc.decay_factor
        )

    # decayed_score 기준 내림차순 정렬
    documents.sort(key=lambda d: d.decayed_score, reverse=True)
    return documents

적용 예시

# Stage 1 결과를 ScoredDocument로 변환
docs = []
for hit in results:
    docs.append(ScoredDocument(
        id=hit.id,
        title=hit.payload["title"],
        content=hit.payload["content"],
        created_at=hit.payload["created_at"],
        similarity=hit.score
    ))

# 시간 감쇠 적용
docs = apply_time_decay(docs, half_life_days=365, decay_weight=0.3)

# 상위 20개만 다음 단계로
top_20 = docs[:20]

for doc in top_20[:5]:
    print(f"[{doc.decayed_score:.3f}] (sim={doc.similarity:.2f}, "
          f"decay={doc.decay_factor:.2f}) {doc.title}")

실행 결과:

[0.836] (sim=0.85, decay=0.97) Kubernetes NetworkPolicy 최신 변경사항
[0.833] (sim=0.89, decay=0.72) Kubernetes NetworkPolicy 실전 가이드
[0.790] (sim=0.82, decay=0.65) CNI 플러그인 비교: Calico vs Cilium
[0.762] (sim=0.78, decay=0.60) Docker 네트워크 모드 정리
[0.745] (sim=0.76, decay=0.72) Pod 간 통신 디버깅 체크리스트

주목: 원래 similarity가 0.89로 1위였던 "실전 가이드"가 0.85인 "최신 변경사항"에게 역전당했습니다. 3개월 전 문서(decay=0.72)보다 1주일 전 문서(decay=0.97)가 더 높은 가중치를 받았기 때문입니다.

decay_weight 튜닝 가이드

decay_weight	의미	적합한 상황
0.0	시간 무시, 순수 유사도	법률 문서, 학술 자료 (오래돼도 가치 유지)
0.1~0.2	시간을 약하게 반영	기술 문서 (내용이 더 중요하지만 최신성도 고려)
0.3	균형잡힌 기본값	개인 지식 베이스, 사내 위키
0.5 이상	최신성 강조	뉴스, 장애 보고서, 릴리스 노트

Stage 3: Cross-Encoder Reranking

왜 Reranking이 필요한가?

Stage 1의 벡터 검색은 Bi-Encoder 방식입니다. 쿼리와 문서를 각각 따로 벡터로 만들고 거리를 비교합니다. 빠르지만 "쿼리와 문서 사이의 미묘한 관계"를 놓칠 수 있습니다.

Cross-Encoder는 다릅니다. 쿼리와 문서를 하나의 입력으로 합쳐서 모델에 넣고, 관련도를 직접 계산합니다. 정확도가 Bi-Encoder보다 높지만, 문서 하나당 모델 추론이 한 번 필요하므로 훨씬 느립니다. 그래서 Stage 2에서 20개로 줄인 후보에만 적용합니다.

구현 코드

from sentence_transformers import CrossEncoder

# Cross-Encoder 모델 로드
# ms-marco-MiniLM은 가볍고 빠른 reranking 전용 모델
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-12-v2")

def rerank_documents(
    query: str,
    documents: list[ScoredDocument],
    rerank_weight: float = 0.5
) -> list[ScoredDocument]:
    """
    Cross-Encoder로 문서를 재순위 매김.

    Args:
        query: 원본 사용자 쿼리
        documents: Stage 2에서 받은 상위 문서들
        rerank_weight: rerank 점수의 비중 (0~1)
    """
    # (query, document) 쌍 생성
    pairs = [(query, doc.content) for doc in documents]

    # Cross-Encoder 점수 계산 (배치 처리)
    scores = reranker.predict(pairs)

    # 점수를 0~1 범위로 정규화 (sigmoid)
    import torch
    normalized = torch.sigmoid(torch.tensor(scores)).tolist()

    for doc, score in zip(documents, normalized):
        doc.rerank_score = score
        # 최종 점수: decay 적용된 점수와 rerank 점수의 가중 합산
        doc.final_score = (
            (1 - rerank_weight) * doc.decayed_score
            + rerank_weight * doc.rerank_score
        )

    documents.sort(key=lambda d: d.final_score, reverse=True)
    return documents

적용 및 최종 결과

query = "Kubernetes 네트워크 정책 설정 방법"

# Stage 3: Reranking
final_results = rerank_documents(query, top_20, rerank_weight=0.5)

# 최종 Top 10
print("=" * 70)
print(f"{'순위':>4} {'최종점수':>8} {'유사도':>6} {'감쇠':>5} "
      f"{'리랭크':>6}  제목")
print("=" * 70)

for i, doc in enumerate(final_results[:10], 1):
    print(f"{i:4d} {doc.final_score:8.3f} {doc.similarity:6.2f} "
          f"{doc.decay_factor:5.2f} {doc.rerank_score:6.3f}  {doc.title}")

실행 결과:

======================================================================
순위  최종점수  유사도  감쇠  리랭크  제목
======================================================================
   1    0.852   0.89  0.72  0.871  Kubernetes NetworkPolicy 실전 가이드
   2    0.847   0.85  0.97  0.858  Kubernetes NetworkPolicy 최신 변경사항
   3    0.738   0.82  0.65  0.685  CNI 플러그인 비교: Calico vs Cilium
   4    0.701   0.76  0.72  0.657  Pod 간 통신 디버깅 체크리스트
   5    0.689   0.78  0.60  0.615  Docker 네트워크 모드 정리
   6    0.652   0.71  0.55  0.594  Calico 설치 및 설정 가이드
   7    0.623   0.68  0.80  0.496  Service Mesh 개요: Istio vs Linkerd
   8    0.601   0.65  0.45  0.591  네트워크 트러블슈팅 명령어 모음
   9    0.578   0.72  0.38  0.542  iptables와 K8s kube-proxy 관계
  10    0.545   0.60  0.82  0.439  최근 인프라 변경 이력 정리
======================================================================

주목할 점:

• 1위 역전: Stage 2에서는 "최신 변경사항"이 1위였지만, Cross-Encoder가 쿼리("네트워크 정책 설정 방법")와 "실전 가이드"의 의미적 연관성이 더 높다고 판단하여 다시 역전
• 7위 Service Mesh: 벡터 유사도(0.68)는 낮지만 최신 문서(decay=0.80)라 상위에 남음. 하지만 Cross-Encoder(0.496)가 "네트워크 정책"과 직접적 관련이 낮다고 판단
• 이처럼 3단계를 거치면 유사도, 최신성, 의미적 정확도가 균형 잡힌 결과를 얻습니다

전체 파이프라인 통합 코드

지금까지의 3단계를 하나로 엮은 완전한 코드입니다. 복사해서 바로 사용할 수 있습니다.

"""
Multi-Stage Retrieval Pipeline
Stage 1: Qdrant 벡터 검색 (top-100)
Stage 2: Time Decay (top-20)
Stage 3: Cross-Encoder Reranking (top-10)
"""
import math
from datetime import datetime, timedelta
from dataclasses import dataclass

from openai import OpenAI
from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition, Range
from sentence_transformers import CrossEncoder
import torch


# ── 데이터 모델 ──────────────────────────────────────────────

@dataclass
class ScoredDocument:
    id: str
    title: str
    content: str
    created_at: float
    similarity: float
    decay_factor: float = 1.0
    decayed_score: float = 0.0
    rerank_score: float = 0.0
    final_score: float = 0.0


# ── 클라이언트 초기화 ────────────────────────────────────────

openai_client = OpenAI()
qdrant = QdrantClient(host="localhost", port=6333)
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-12-v2")


# ── Stage 1: 벡터 검색 ──────────────────────────────────────

def vector_search(
    query: str,
    collection: str = "knowledge_base",
    limit: int = 100,
    lookback_days: int = 365,
    score_threshold: float = 0.3
) -> list[ScoredDocument]:
    """Qdrant에서 벡터 유사도 + 날짜 필터로 후보 추출."""
    # 쿼리 임베딩
    resp = openai_client.embeddings.create(
        input=query, model="text-embedding-3-small"
    )
    query_vector = resp.data[0].embedding

    # 날짜 필터
    cutoff = (datetime.now() - timedelta(days=lookback_days)).timestamp()

    results = qdrant.search(
        collection_name=collection,
        query_vector=query_vector,
        query_filter=Filter(
            must=[FieldCondition(key="created_at", range=Range(gte=cutoff))]
        ),
        limit=limit,
        with_payload=True,
        score_threshold=score_threshold,
    )

    return [
        ScoredDocument(
            id=hit.id,
            title=hit.payload["title"],
            content=hit.payload["content"],
            created_at=hit.payload["created_at"],
            similarity=hit.score,
        )
        for hit in results
    ]


# ── Stage 2: 시간 감쇠 ──────────────────────────────────────

def apply_time_decay(
    docs: list[ScoredDocument],
    half_life_days: float = 365.0,
    decay_weight: float = 0.3
) -> list[ScoredDocument]:
    """Exponential decay로 오래된 문서 점수 하향 조정."""
    decay_lambda = math.log(2) / half_life_days
    now = datetime.now().timestamp()

    for doc in docs:
        days_old = (now - doc.created_at) / 86400
        doc.decay_factor = math.exp(-decay_lambda * days_old)
        doc.decayed_score = (
            (1 - decay_weight) * doc.similarity
            + decay_weight * doc.decay_factor
        )

    docs.sort(key=lambda d: d.decayed_score, reverse=True)
    return docs


# ── Stage 3: Cross-Encoder Reranking ────────────────────────

def rerank(
    query: str,
    docs: list[ScoredDocument],
    rerank_weight: float = 0.5
) -> list[ScoredDocument]:
    """Cross-Encoder로 query-document 쌍의 관련도를 정밀 평가."""
    pairs = [(query, doc.content) for doc in docs]
    scores = reranker.predict(pairs)
    normalized = torch.sigmoid(torch.tensor(scores)).tolist()

    for doc, score in zip(docs, normalized):
        doc.rerank_score = score
        doc.final_score = (
            (1 - rerank_weight) * doc.decayed_score
            + rerank_weight * doc.rerank_score
        )

    docs.sort(key=lambda d: d.final_score, reverse=True)
    return docs


# ── 파이프라인 실행 ──────────────────────────────────────────

def retrieve(
    query: str,
    top_k: int = 10,
    stage1_limit: int = 100,
    stage2_limit: int = 20,
    half_life_days: float = 365.0,
    decay_weight: float = 0.3,
    rerank_weight: float = 0.5,
) -> list[ScoredDocument]:
    """3단계 Multi-Stage Retrieval 파이프라인."""
    # Stage 1: 벡터 검색
    candidates = vector_search(query, limit=stage1_limit)
    print(f"Stage 1: {len(candidates)}개 후보 추출")

    # Stage 2: 시간 감쇠
    decayed = apply_time_decay(
        candidates, half_life_days=half_life_days,
        decay_weight=decay_weight
    )[:stage2_limit]
    print(f"Stage 2: 시간 감쇠 적용 → 상위 {len(decayed)}개 선별")

    # Stage 3: Reranking
    final = rerank(query, decayed, rerank_weight=rerank_weight)[:top_k]
    print(f"Stage 3: Reranking 완료 → 최종 {len(final)}개")

    return final


# ── 사용 예시 ────────────────────────────────────────────────

if __name__ == "__main__":
    results = retrieve("Kubernetes 네트워크 정책 설정 방법")

    for i, doc in enumerate(results, 1):
        print(f"\n--- #{i} (score: {doc.final_score:.3f}) ---")
        print(f"제목: {doc.title}")
        print(f"유사도: {doc.similarity:.2f} | "
              f"감쇠: {doc.decay_factor:.2f} | "
              f"리랭크: {doc.rerank_score:.3f}")

정리: 각 단계별 역할과 비용

단계	입력	출력	속도	역할
Stage 1 벡터 검색	전체 컬렉션	100개 후보	~3ms	빠르게 관련 후보 추출 (recall 극대화)
Stage 2 시간 감쇠	100개	20개	<1ms	최신성 반영, 후보 축소
Stage 3 Reranking	20개	10개	~200ms	의미적 정확도 극대화 (precision 극대화)

3단계를 합쳐도 총 소요 시간은 약 200ms입니다. 사용자가 체감하기엔 거의 즉시 응답하는 수준이죠. 핵심은 각 단계가 다음 단계의 부하를 줄여주는 깔때기 구조라는 것입니다. 100만 개 문서에서 Cross-Encoder를 돌리면 몇 분이 걸리지만, 20개에만 돌리면 200ms면 충분합니다.

이 파이프라인은 Reranking까지만 다뤘습니다. 실제 RAG 시스템에서는 이 결과를 LLM에 전달해서 답변을 생성하는 단계가 뒤따르지만, 그건 또 다른 이야기입니다. 좋은 검색 결과가 좋은 답변의 시작이라는 점만 기억해두세요.

---

이 글은 생성형 AI의 도움을 받아 작성되었습니다. 원본 자료를 기반으로 AI가 초안을 생성하고, 작성자가 검토·편집하였습니다.