크로스 인코더 리랭커

크로스 인코더 리랭커 (CrossEncoderReranker)

검색 결과 품질을 크게 끌어올릴 수 있는 방법 하나를 소개할게요. 바로 리랭킹(reranking) 이에요. 벡터 검색으로 상위 N개를 뽑은 뒤, 더 정확한 모델로 다시 순위를 매기는 방식이죠. 이 페이지에서 다루는 크로스 인코더는 각 (query, document) 쌍을 독립적으로 스코어링해서, 임베딩을 단독 비교하는 방식보다 훨씬 정확한 순서를 만들어요. 대신 문서 하나당 추론이 한 번 더 들어가는 비용이 있죠.

출처: 공식문서

왜 리랭킹이 중요한가

크로스 인코더 리랭킹은 RAG 파이프라인의 품질 개선 중에서도 효과가 가장 큰 방법 중 하나예요. 벡터 검색으로 상위 20개를 뽑았다면 리랭커로 상위 5개로 좁히는 식이죠. 여러 문서를 하나씩 처리하다 보니 비용이 들지만, 작은 크로스 인코더를 Hugging Face에서 가져와 로컬 CPU에서 무료로 돌릴 수 있어요.

이 가이드는 HuggingFaceCrossEncoder를 LangChain의 CrossEncoderReranker와 ContextualCompressionRetriever에 조합하는 방법을 보여줘요. 이 패턴은 Hugging Face의 모든 크로스 인코더 모델과 함께 쓸 수 있는데, BAAI/bge-reranker-*, mixedbread-ai/mxbai-rerank-*, Alibaba-NLP/gte-multilingual-reranker-*, Qwen/Qwen3-Reranker-*, 그리고 고전적인 cross-encoder/ms-marco-* 계열을 포함해요.

설정

pip install -qU langchain-huggingface langchain-community langchain-classic faiss-cpu

기본 리트리버 만들기

먼저 표준 벡터스토어 리트리버로 시작해요. 비교적 큰 k로 검색해 두고, 리랭커가 그 결과를 좁혀 줄 거예요.

from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import FAISS
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

documents = TextLoader("../../how_to/state_of_the_union.txt").load()
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=100)
texts = text_splitter.split_documents(documents)

embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-m3",
    encode_kwargs={"normalize_embeddings": True},
)
retriever = FAISS.from_documents(texts, embeddings).as_retriever(
    search_kwargs={"k": 20}
)

크로스 인코더로 리랭킹하기

CrossEncoderReranker는 어떤 크로스 인코더든 감싸서 ContextualCompressionRetriever에 연결해 줘요.

from langchain_classic.retrievers.contextual_compression import ContextualCompressionRetriever
from langchain_classic.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder

cross_encoder = HuggingFaceCrossEncoder(model_name="BAAI/bge-reranker-v2-m3")
reranker = CrossEncoderReranker(model=cross_encoder, top_n=3)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=reranker,
    base_retriever=retriever,
)

compressed_docs = compression_retriever.invoke("What is the plan for the economy?")
for i, doc in enumerate(compressed_docs, 1):
    print(f"Document {i}:\n{doc.page_content}\n")

크로스 인코더 고르기

Model	Size	Notes
`cross-encoder/ms-marco-MiniLM-L6-v2`	22M	Fastest; English only, 2022-era baseline
`BAAI/bge-reranker-v2-m3`	568M	Multilingual, strong default for most workloads
`mixedbread-ai/mxbai-rerank-large-v2`	1.5B	Top-tier English quality, GPU recommended
`Alibaba-NLP/gte-multilingual-reranker-base`	306M	Multilingual, 8192-token context
`Qwen/Qwen3-Reranker-0.6B`	595M	Instruction-aware, multilingual

HuggingFaceCrossEncoder는 사용 가능한 최상의 장치(CUDA > MPS > CPU)를 자동 선택해요. 특정 장치로 고정하려면 model_kwargs={"device": "cpu"} 같은 값을 전달하면 돼요.

SageMaker에 배포하기

크로스 인코더를 SageMaker 엔드포인트에 호스팅하고 SagemakerEndpointCrossEncoder를 쓸 수도 있어요. 아래 inference.py는 모델을 실행 시점에 로드하므로 model.tar.gz 산출물이 필요 없어요. 단계별 안내는 이 워크스루를 참고하세요.

import json
import logging
from typing import List

import torch
from sagemaker_inference import encoder
from transformers import AutoModelForSequenceClassification, AutoTokenizer

PAIRS = "pairs"
SCORES = "scores"


class CrossEncoder:
    def __init__(self) -> None:
        self.device = (
            torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu")
        )
        logging.info(f"Using device: {self.device}")
        model_name = "BAAI/bge-reranker-v2-m3"
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForSequenceClassification.from_pretrained(model_name)
        self.model = self.model.to(self.device)

    def __call__(self, pairs: List[List[str]]) -> List[float]:
        with torch.inference_mode():
            inputs = self.tokenizer(
                pairs,
                padding=True,
                truncation=True,
                return_tensors="pt",
                max_length=512,
            )
            inputs = inputs.to(self.device)
            scores = (
                self.model(**inputs, return_dict=True)
                .logits.view(
                    -1,
                )
                .float()
            )

        return scores.detach().cpu().tolist()


def model_fn(model_dir: str) -> CrossEncoder:
    try:
        return CrossEncoder()
    except Exception:
        logging.exception(f"Failed to load model from: {model_dir}")
        raise


def transform_fn(
    cross_encoder: CrossEncoder, input_data: bytes, content_type: str, accept: str
) -> bytes:
    payload = json.loads(input_data)
    model_output = cross_encoder(**payload)
    output = {SCORES: model_output}
    return encoder.encode(output, accept)

더 알아보기 (Learn more)

RAG 개요·아키텍처 — 검색 파이프라인과 리랭킹의 위치
Retriever 통합 — 다양한 리트리버
벡터스토어 통합 — 벡터 검색의 기반