7장: 리랭킹으로 검색 정밀도 높이기

프로덕션 RAG 검색 패턴:
  질문 --> 1차 검색 (top-20) --> 리랭킹 (top-5) --> LLM 생성
 
1차 검색: 빠르지만 정밀도 낮음 (Bi-Encoder)
리랭킹:   느리지만 정밀도 높음 (Cross-Encoder)

Bi-Encoder:
  쿼리: "재택근무 신청"  --> [0.2, 0.8, ...] --+
                                                |--> 코사인 유사도
  문서: "원격 근무 규정"  --> [0.3, 0.7, ...] --+
  
  쿼리와 문서를 독립적으로 인코딩 --> 빠르지만 상호작용 없음

Cross-Encoder:
  입력: "[CLS] 재택근무 신청 [SEP] 원격 근무 규정 [SEP]"
  
  --> Transformer 전체 어텐션 --> 관련성 점수: 0.92
  
  쿼리와 문서를 함께 인코딩 --> 느리지만 정밀

import cohere
 
co = cohere.Client(api_key="your-api-key")
 
query = "재택근무는 주 몇 회까지 가능한가요?"
documents = [
    "재택근무는 팀장 승인 후 주 2회까지 신청할 수 있습니다.",
    "출장 시 교통비는 실비 정산이 원칙입니다.",
    "원격 근무 시 VPN 접속이 필수이며, 보안 규정을 준수해야 합니다.",
    "연차는 입사일 기준으로 매년 15일이 부여됩니다.",
    "재택근무 신청은 전주 금요일까지 팀장에게 제출해야 합니다.",
]
 
response = co.rerank(
    model="rerank-v3.5",
    query=query,
    documents=documents,
    top_n=3,
    return_documents=True
)
 
for result in response.results:
    print(f"순위: {result.index + 1}")
    print(f"점수: {result.relevance_score:.4f}")
    print(f"내용: {result.document.text[:100]}")
    print("---")
 
# 출력:
# 순위: 1, 점수: 0.9821
# 내용: 재택근무는 팀장 승인 후 주 2회까지 신청할 수 있습니다.
#
# 순위: 2, 점수: 0.7234
# 내용: 재택근무 신청은 전주 금요일까지 팀장에게 제출해야 합니다.
#
# 순위: 3, 점수: 0.4102
# 내용: 원격 근무 시 VPN 접속이 필수이며, 보안 규정을 준수해야 합니다.

from sentence_transformers import CrossEncoder
 
# 모델 로딩 (첫 실행 시 다운로드)
reranker = CrossEncoder("BAAI/bge-reranker-v2-m3", max_length=512)
 
query = "재택근무는 주 몇 회까지 가능한가요?"
documents = [
    "재택근무는 팀장 승인 후 주 2회까지 신청할 수 있습니다.",
    "출장 시 교통비는 실비 정산이 원칙입니다.",
    "원격 근무 시 VPN 접속이 필수입니다.",
    "연차는 입사일 기준으로 매년 15일이 부여됩니다.",
]
 
# 쿼리-문서 쌍으로 점수 계산
pairs = [(query, doc) for doc in documents]
scores = reranker.predict(pairs)
 
# 점수 기준 정렬
ranked = sorted(
    zip(documents, scores),
    key=lambda x: x[1],
    reverse=True
)
 
for doc, score in ranked:
    print(f"점수: {score:.4f} | {doc[:60]}")

from flashrank import Ranker, RerankRequest
 
ranker = Ranker(model_name="ms-marco-MiniLM-L-12-v2")
 
request = RerankRequest(
    query="재택근무 신청 방법",
    passages=[
        {"text": "재택근무는 팀장 승인 후 주 2회까지 신청할 수 있습니다."},
        {"text": "출장 시 교통비는 실비 정산이 원칙입니다."},
        {"text": "재택근무 신청은 전주 금요일까지 팀장에게 제출해야 합니다."},
    ]
)
 
results = ranker.rerank(request)
for result in results:
    print(f"점수: {result['score']:.4f} | {result['text'][:60]}")

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
import cohere
 
class RAGWithReranking:
    """리랭킹이 포함된 RAG 파이프라인"""
 
    def __init__(self, vectorstore, cohere_api_key):
        self.vectorstore = vectorstore
        self.co = cohere.Client(api_key=cohere_api_key)
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
 
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", """주어진 컨텍스트만을 사용하여 질문에 답변하세요.
컨텍스트에 없는 내용은 답변하지 마세요."""),
            ("human", """컨텍스트:
{context}
 
질문: {question}""")
        ])
 
    def query(self, question, initial_k=20, final_k=5):
        """리랭킹 포함 RAG 질의"""
 
        # 1단계: 넓은 범위로 1차 검색 (top-20)
        initial_results = self.vectorstore.similarity_search(
            question, k=initial_k
        )
 
        if not initial_results:
            return "관련 문서를 찾을 수 없습니다."
 
        # 2단계: 리랭킹으로 정밀 선별 (top-5)
        documents = [doc.page_content for doc in initial_results]
        rerank_response = self.co.rerank(
            model="rerank-v3.5",
            query=question,
            documents=documents,
            top_n=final_k,
            return_documents=True
        )
 
        # 3단계: 리랭킹 결과로 컨텍스트 구성
        reranked_docs = []
        for result in rerank_response.results:
            original_doc = initial_results[result.index]
            source = original_doc.metadata.get("source", "알 수 없음")
            reranked_docs.append(
                f"[출처: {source}] (관련도: {result.relevance_score:.2f})\n"
                + result.document.text
            )
 
        context = "\n\n---\n\n".join(reranked_docs)
 
        # 4단계: LLM 생성
        chain = self.prompt | self.llm | StrOutputParser()
        answer = chain.invoke({
            "context": context,
            "question": question
        })
 
        return answer
 
# 사용
rag = RAGWithReranking(vectorstore, cohere_api_key="your-key")
answer = rag.query("재택근무는 주 몇 회까지 가능한가요?")
print(answer)

# initial_k에 따른 성능 변화 (예시)
# initial_k=10: Recall@5=0.72, 리랭킹 시간=0.3초
# initial_k=20: Recall@5=0.85, 리랭킹 시간=0.6초
# initial_k=50: Recall@5=0.88, 리랭킹 시간=1.5초
# initial_k=100: Recall@5=0.89, 리랭킹 시간=3.0초
 
# 일반적으로 initial_k=20~30이 비용 대비 효과가 좋음

def rerank_with_threshold(co, query, documents, threshold=0.3, top_n=5):
    """관련도 임계값을 적용한 리랭킹"""
    response = co.rerank(
        model="rerank-v3.5",
        query=query,
        documents=documents,
        top_n=top_n,
        return_documents=True
    )
 
    # 임계값 이상인 결과만 반환
    filtered = [
        result for result in response.results
        if result.relevance_score >= threshold
    ]
 
    if not filtered:
        # 임계값을 충족하는 결과가 없으면 최소 1개는 반환
        return [response.results[0]] if response.results else []
 
    return filtered

from concurrent.futures import ThreadPoolExecutor
 
def batch_rerank(co, queries_and_docs, max_workers=5):
    """여러 쿼리에 대한 병렬 리랭킹"""
 
    def rerank_single(item):
        query, docs = item
        response = co.rerank(
            model="rerank-v3.5",
            query=query,
            documents=docs,
            top_n=5
        )
        return response.results
 
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        results = list(executor.map(rerank_single, queries_and_docs))
 
    return results

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
 
def llm_rerank(query, documents, top_n=5):
    """LLM 기반 리랭킹"""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
 
    prompt = ChatPromptTemplate.from_messages([
        ("system", """다음 질문에 대한 각 문서의 관련성을 0~10 점수로 평가하세요.
JSON 배열 형식으로만 응답하세요. 예시: [8, 3, 7, 1, 5]"""),
        ("human", """질문: {query}
 
문서 목록:
{documents}""")
    ])
 
    doc_text = "\n\n".join([
        f"[문서 {i+1}]\n{doc}" for i, doc in enumerate(documents)
    ])
 
    response = (prompt | llm).invoke({
        "query": query,
        "documents": doc_text
    })
 
    import json
    scores = json.loads(response.content)
 
    ranked = sorted(
        zip(documents, scores),
        key=lambda x: x[1],
        reverse=True
    )
 
    return [doc for doc, score in ranked[:top_n]]

리랭킹 없음:
  검색 (top-5) --> LLM 생성
  총 시간: ~1.5초
  Precision@5: 0.60
 
리랭킹 포함:
  검색 (top-20) --> 리랭킹 (top-5) --> LLM 생성
  총 시간: ~2.5초
  Precision@5: 0.82
 
--> 정밀도 37% 향상, 지연 시간 1초 증가