시간 가중 벡터저장소 검색기(TimeWeightedVectorStoreRetriever)

1 TimeWeightedVectorStoreRetriever 개요

1.1 핵심 개념

TimeWeightedVectorStoreRetriever는 의미론적 유사성(semantic similarity)과 시간에 따른 감쇠(time decay)를 결합하여 사용하는 검색 도구이다. 문서의 관련성(relevance)과 신선도(freshness)를 동시에 평가하여 가장 적합한 결과를 제공한다.

1.2 적용 사례

• 뉴스 검색: 유사한 주제의 기사 중 최신 기사를 우선 검색
  
• 규정집/법률 문서: 개정된 최신 조항을 우선 반환
  
• 기술 문서: 업데이트된 최신 버전의 문서를 우선 검색
  
• 고객 문의 응대: 최근 자주 참조되는 FAQ를 우선 제공

1.3 스코어링 알고리즘

검색 점수는 다음 공식으로 계산된다:

\[\text{score} = \text{semantic\_similarity} + (1.0 - \text{decay\_rate})^{\text{hours\_passed}}\]

수식 구성 요소

• semantic_similarity: 쿼리와 문서 간의 의미적 유사도 (벡터 임베딩 기반 코사인 유사도)
  
• decay_rate: 시간 감쇠율 (0 ~ 1 사이의 값, 높을수록 과거 문서에 큰 페널티)
  
• hours_passed: 마지막 접근 이후 경과한 시간 (시간 단위)

decay_rate 설정 효과

• 높은 값 (0.999): 과거 문서에 큰 페널티 → 최신 문서 우선 검색 (뉴스, 실시간 정보)
  
• 낮은 값 (0.0001): 시간 페널티 최소화 → 의미적 유사도 우선 (지식 베이스, 영구적 문서)
  
• 중간 값 (0.5): 최신성과 관련성의 균형 (일반적인 문서 검색)

1.4 주요 특징

1. 마지막 접근 시간 기준

문서 생성 시점이 아닌 마지막으로 접근된 시점을 기준으로 신선도를 평가한다. 자주 참조되는 문서는 계속 “최신” 상태를 유지하므로, 실제로 중요하고 유용한 정보가 상위에 노출된다.

2. 동적 가중치 조정

시간이 지남에 따라 자동으로 가중치가 감소하여, 별도의 재인덱싱 없이도 검색 결과가 최신 트렌드를 반영한다.

3. 의미적 유사도와의 조합

단순히 최신 문서만 반환하는 것이 아니라, 쿼리와의 관련성도 함께 고려하여 균형잡힌 검색 결과를 제공한다.

2 환경 설정

2.1 API 키 로드

# API 키를 환경변수로 관리하기 위한 설정 파일  
from dotenv import load_dotenv  

# API 키 정보 로드  
load_dotenv()  

2.2 LangSmith 추적 설정

LangSmith를 활용하면 시간 가중치 계산 과정과 검색 점수 변화를 시각적으로 추적할 수 있다.

# LangSmith 추적 설정 (https://smith.langchain.com)  
# !pip install langchain-teddynote  
from langchain_teddynote import logging  

# 프로젝트 이름 입력  
logging.langsmith("CH10-Retriever")  

3 낮은 감쇠율 (의미적 유사도 우선)

3.1 개념 설명

낮은 decay_rate (0에 가까운 값)는 시간이 지나도 문서의 점수가 거의 감소하지 않음을 의미한다. 즉, 시간보다 의미적 유사도가 검색 결과에 더 큰 영향을 미친다.

특징

• decay_rate = 0: 시간 감쇠 없음 → 일반적인 벡터 유사도 검색과 동일
  
• decay_rate ≈ 0.0000001: 거의 감쇠 없음 → 과거 문서도 유사도가 높으면 상위 노출

적용 시나리오

• 시간에 구애받지 않는 지식 베이스 (수학 공식, 역사적 사실)
  
• 영구적인 참조 문서 (API 문서, 기술 사양서)
  
• 오래된 문서도 여전히 가치 있는 경우

3.2 Retriever 초기화

벡터 저장소와 매우 낮은 감쇠율을 설정하여 retriever를 생성한다.

from datetime import datetime, timedelta  

import faiss  
from langchain.docstore import InMemoryDocstore  
from langchain.retrievers import TimeWeightedVectorStoreRetriever  
from langchain_community.vectorstores import FAISS  
from langchain_core.documents import Document  
from langchain_openai import OpenAIEmbeddings  

# 임베딩 모델을 정의합니다.  
embeddings_model = OpenAIEmbeddings(model="text-embedding-3-small")  

# 벡터 저장소를 빈 상태로 초기화합니다.  
embedding_size = 1536  
index = faiss.IndexFlatL2(embedding_size)  
vectorstore = FAISS(embeddings_model, index, InMemoryDocstore({}), {})  

# 시간 가중치가 적용된 벡터 저장소 검색기를 초기화합니다. (여기서는, 낮은 감쇠율을 적용합니다)  
retriever = TimeWeightedVectorStoreRetriever(  
    vectorstore=vectorstore, decay_rate=0.0000000000000000000000001, k=1  
)  

3.3 샘플 데이터 추가

두 개의 문서를 추가한다. 첫 번째 문서는 어제 작성된 것으로, 두 번째 문서는 방금 작성된 것으로 설정한다.

# 어제 날짜를 계산  
yesterday = datetime.now() - timedelta(days=1)  

retriever.add_documents(  
    # 문서를 추가하고, metadata에 어제 날짜를 설정  
    [  
        Document(  
            page_content="테디노트 구독해 주세요.",  
            metadata={"last_accessed_at": yesterday},  
        )  
    ]  
)  

# 다른 문서를 추가 (metadata는 별도로 설정하지 않음 = 현재 시간으로 자동 설정)  
retriever.add_documents([Document(page_content="테디노트 구독 해주실꺼죠? Please!")])  

['a6c732c4-adb2-45d1-bcbb-a5108a9778f7']  

3.4 검색 결과 확인

• decay_rate가 거의 0에 가까우므로, 시간 페널티가 거의 없어 의미적으로 더 유사한 문서가 먼저 반환된다.
  • 과거 문서가 최신 문서로 계산되는 효과가 있다

# "테디노트 구독해 주세요."가 먼저 반환됨  
# 이유: 감쇠율이 0에 가까워 시간 페널티가 거의 없고,   
#       의미적 유사도가 더 높기 때문  
retriever.invoke("테디노트")  

[Document(metadata={'last_accessed_at': datetime.datetime(2024, 8, 30, 22, 1, 49, 841379), 'created_at': datetime.datetime(2024, 8, 30, 22, 1, 44, 410635), 'buffer_idx': 0}, page_content='테디노트 구독해 주세요.')]  

결과 분석

• 어제 작성된 문서가 반환되는 이유: decay_rate ≈ 0이므로 24시간이 지나도 점수 감소가 미미함
  
• 시간보다 콘텐츠의 관련성이 우선시됨

4 높은 감쇠율 (최신성 우선)

4.1 개념 설명

높은 decay_rate (1에 가까운 값)는 시간이 지남에 따라 문서의 점수가 급격히 감소함을 의미한다. 즉, 최신 문서가 검색 결과에 더 큰 영향을 미친다.

특징

• decay_rate = 1: 모든 과거 문서의 시간 점수가 0 → 최신 문서만 반환
  
• decay_rate ≈ 0.999: 시간이 조금만 지나도 점수가 급격히 감소 → 최신성이 매우 중요

적용 시나리오

• 뉴스 기사 검색 (최신 뉴스 우선)
  
• 실시간 정보 (주식 시세, 날씨)
  
• 자주 업데이트되는 문서 (소프트웨어 릴리즈 노트)
  
• 시간에 민감한 정보 (이벤트 공지, 할인 정보)

4.2 Retriever 초기화

벡터 저장소와 높은 감쇠율을 설정하여 retriever를 생성한다.

# 임베딩 모델을 정의  
embeddings_model = OpenAIEmbeddings(model="text-embedding-3-small")  

# 벡터 저장소를 빈 상태로 초기화  
embedding_size = 1536  
index = faiss.IndexFlatL2(embedding_size)  
vectorstore = FAISS(embeddings_model, index, InMemoryDocstore({}), {})  

# 시간 가중치가 적용된 벡터 저장소 검색기를 초기화 (높은 감쇠율 적용)  
retriever = TimeWeightedVectorStoreRetriever(  
    vectorstore=vectorstore, decay_rate=0.999, k=1  
)  

4.3 샘플 데이터 추가

동일한 두 개의 문서를 추가한다. 첫 번째는 어제 작성, 두 번째는 방금 작성으로 설정한다.

# 어제 날짜를 계산  
yesterday = datetime.now() - timedelta(days=1)  

retriever.add_documents(  
    # 문서를 추가하고, metadata에 어제 날짜를 설정  
    [  
        Document(  
            page_content="테디노트 구독해 주세요.",  
            metadata={"last_accessed_at": yesterday},  
        )  
    ]  
)  

# 다른 문서를 추가 (metadata는 별도로 설정하지 않음)  
retriever.add_documents([Document(page_content="테디노트 구독 해주실꺼죠? Please!")])  

['c3349ba9-75c7-49ec-be7a-017bc0917fa2']  

4.4 검색 결과 확인

decay_rate가 0.999로 높으므로, 24시간이 지난 문서는 시간 페널티를 크게 받아 최신 문서가 먼저 반환된다.

# "테디노트 구독 해주실꺼죠? Please!"가 먼저 반환됨  
# 이유: 감쇠율이 0.999로 높아서 어제 작성된 문서는   
#       시간 페널티를 크게 받아 점수가 급격히 하락함  
retriever.invoke("테디노트")  

[Document(metadata={'last_accessed_at': datetime.datetime(2024, 8, 30, 22, 3, 18, 331780), 'created_at': datetime.datetime(2024, 8, 30, 22, 2, 44, 618745), 'buffer_idx': 1}, page_content='테디노트 구독 해주실꺼죠? Please!')]  

결과 분석

• 방금 작성된 문서가 반환되는 이유: 높은 decay_rate로 인해 24시간 지난 문서의 시간 점수가 거의 0에 수렴
  
• 의미적 유사도가 낮더라도 최신성이 우선시됨
  
• 실제 계산: \((1.0 - 0.999)^{24} ≈ 0\) (24시간 경과 시 시간 점수가 거의 0)

5 decay_rate 설정 가이드

5.1 값에 따른 동작 비교

decay_rate	시간 감쇠 속도	우선순위	적합한 사용 사례
0.0 ~ 0.1	매우 느림	의미적 유사도 우선	지식 베이스, 영구 문서, 역사적 자료
0.1 ~ 0.5	느림	유사도와 최신성 균형	일반 문서 검색, FAQ
0.5 ~ 0.9	빠름	최신성 우선	업데이트되는 가이드, 제품 정보
0.9 ~ 0.999	매우 빠름	최신성 강력 우선	뉴스, 실시간 정보
1.0	즉시 감쇠	최신 문서만	실시간 데이터 스트림

5.2 설정 원칙

낮은 decay_rate (0 ~ 0.1)

• 정보를 거의 “잊지 않음”
  
• 시간이 지나도 점수 변화가 미미함
  
• 의미적 유사도가 검색의 주요 기준
  
• 예: 수학 공식, API 문서, 기술 사양

중간 decay_rate (0.1 ~ 0.5)

• 적절한 시간 감쇠 적용
  
• 최신성과 관련성의 균형
  
• 주기적으로 업데이트되지만 과거 정보도 유효한 경우
  
• 예: 블로그 포스트, 제품 리뷰, 사용자 가이드

높은 decay_rate (0.9 ~ 0.999)

• 과거 정보를 빠르게 “잊음”
  
• 최신 정보에 압도적으로 높은 점수
  
• 시간에 민감한 정보 검색
  
• 예: 뉴스 기사, 실시간 알림, 이벤트 공지

5.3 실무 권장값

# 지식 베이스 (변하지 않는 정보)  
decay_rate = 0.01  

# 일반 문서 검색 (균형)  
decay_rate = 0.3  

# 뉴스/실시간 정보 (최신성 중시)  
decay_rate = 0.95  

6 고급: 가상 시간을 이용한 테스트

6.1 개념 설명

LangChain의 mock_now 유틸리티를 사용하면 현재 시간을 임의로 설정하여 시간 감쇠 효과를 시뮬레이션할 수 있다. 이를 통해 실제로 시간이 경과하지 않아도 다양한 시간대에서의 검색 동작을 테스트할 수 있다.

활용 목적

1. 최적의 decay_rate 값 탐색
   
2. 시간 경과에 따른 검색 결과 변화 예측
   
3. 프로덕션 배포 전 시간 기반 로직 검증
   
4. 과거 시점의 검색 결과 재현

6.2 mock_now 사용법

import datetime  

from langchain.utils import mock_now  

# 현재 시간을 특정 시점으로 설정  
mock_now(datetime.datetime(2024, 8, 30, 00, 00))  

# 현재 시간 출력  
print(datetime.datetime.now())  

2024-08-30 22:05:01.844175  

6.3 시간 변경 테스트

다양한 시간대에서 검색 결과가 어떻게 달라지는지 확인하여 적절한 decay_rate를 찾을 수 있다.

# 과거 시점 (2024년 8월 29일)으로 설정하여 검색  
with mock_now(datetime.datetime(2024, 8, 29, 00, 00)):  
    # 해당 시점에서의 검색 결과 확인  
    print(retriever.invoke("테디노트"))  

주의사항

• 너무 오래된 시간으로 설정하면 decay_rate 계산 시 오류가 발생할 수 있음 (지수 연산 오버플로우)
  
• 일반적으로 수년 이내의 시간 범위에서 테스트 권장

6.4 decay_rate 튜닝 예제

다양한 시간 간격에서 검색 결과를 테스트하여 최적의 decay_rate를 찾는다.

# 다양한 시간대 테스트  
test_times = [  
    ("1시간 후", datetime.timedelta(hours=1)),  
    ("1일 후", datetime.timedelta(days=1)),  
    ("1주일 후", datetime.timedelta(weeks=1)),  
    ("1개월 후", datetime.timedelta(days=30)),  
]  

for label, delta in test_times:  
    future_time = datetime.datetime.now() + delta  
    with mock_now(future_time):  
        result = retriever.invoke("테디노트")  
        print(f"{label}: {result[0].page_content}")  

[Document(metadata={'last_accessed_at': MockDateTime(2024, 8, 29, 0, 0), 'created_at': datetime.datetime(2024, 8, 30, 22, 2, 44, 618745), 'buffer_idx': 1}, page_content='테디노트 구독 해주실꺼죠? Please!')]  

7 실전 활용 전략

7.1 전략 1: 하이브리드 검색 시스템

서로 다른 decay_rate를 가진 여러 retriever를 조합하여 다양한 사용자 요구를 충족한다.

# 최신성 우선 retriever  
recent_retriever = TimeWeightedVectorStoreRetriever(  
    vectorstore=vectorstore,  
    decay_rate=0.95,  # 높은 감쇠율  
    k=3  
)  

# 관련성 우선 retriever  
relevant_retriever = TimeWeightedVectorStoreRetriever(  
    vectorstore=vectorstore,  
    decay_rate=0.1,  # 낮은 감쇠율  
    k=3  
)  

# 사용자 쿼리 의도에 따라 선택  
if "최신" in user_query or "최근" in user_query:  
    results = recent_retriever.invoke(user_query)  
else:  
    results = relevant_retriever.invoke(user_query)  

적용 시나리오: 뉴스 플랫폼에서 “최신 뉴스”는 높은 decay_rate, “관련 뉴스”는 낮은 decay_rate 적용

7.2 전략 2: 동적 decay_rate 조정

문서 유형이나 카테고리에 따라 자동으로 decay_rate를 조정한다.

# 문서 카테고리별 decay_rate 매핑  
category_decay_rates = {  
    "news": 0.95,        # 뉴스는 최신성 중요  
    "tutorial": 0.2,     # 튜토리얼은 시간 무관  
    "api_doc": 0.05,     # API 문서는 영구적  
    "blog": 0.4,         # 블로그는 중간  
    "regulation": 0.7,   # 규정은 업데이트 빈번  
}  

def get_retriever_for_category(category):  
    decay_rate = category_decay_rates.get(category, 0.3)  
    return TimeWeightedVectorStoreRetriever(  
        vectorstore=vectorstore,  
        decay_rate=decay_rate,  
        k=5  
    )  

적용 시나리오: 통합 문서 관리 시스템에서 문서 카테고리를 자동 감지하여 최적의 검색 설정 적용

7.3 전략 3: 주기적 접근 시간 업데이트

중요한 문서의 last_accessed_at를 주기적으로 업데이트하여 항상 상위에 유지한다.

import datetime  

# 중요 문서의 접근 시간 갱신  
def refresh_important_docs(important_doc_ids):  
    for doc_id in important_doc_ids:  
        # 문서의 메타데이터 업데이트  
        vectorstore.update_metadata(  
            doc_id,  
            {"last_accessed_at": datetime.datetime.now()}  
        )  

# 주기적으로 실행 (예: 매일 자정)  
refresh_important_docs(["faq_main", "policy_core", "guide_essential"])  

적용 시나리오: FAQ 시스템에서 핵심 질문들을 항상 상위에 노출

7.4 전략 4: 시간대별 가중치 조정

업무 시간대와 비업무 시간대에 다른 검색 전략 적용

import datetime  

def get_time_aware_retriever():  
    current_hour = datetime.datetime.now().hour  
    
    # 업무 시간 (9-18시): 최신 문서 우선  
    if 9 <= current_hour <= 18:  
        decay_rate = 0.8  
    # 비업무 시간: 관련성 우선  
    else:  
        decay_rate = 0.2  
    
    return TimeWeightedVectorStoreRetriever(  
        vectorstore=vectorstore,  
        decay_rate=decay_rate,  
        k=5  
    )  

적용 시나리오: 고객 지원 챗봇에서 업무 시간에는 최신 공지사항 우선 제공

7.5 전략 5: A/B 테스팅을 통한 최적화

다양한 decay_rate 값으로 A/B 테스팅을 수행하여 사용자 만족도가 높은 값을 찾는다.

import random  

# A/B 테스트 그룹 설정  
def get_ab_test_retriever(user_id):  
    # 사용자 ID 기반 그룹 할당  
    group = hash(user_id) % 3  
    
    decay_rates = {  
        0: 0.3,   # 그룹 A: 낮은 감쇠  
        1: 0.6,   # 그룹 B: 중간 감쇠  
        2: 0.9,   # 그룹 C: 높은 감쇠  
    }  
    
    return TimeWeightedVectorStoreRetriever(  
        vectorstore=vectorstore,  
        decay_rate=decay_rates[group],  
        k=5  
    )  

# 사용자 피드백 수집 후 최적값 선택  

적용 시나리오: 검색 서비스에서 클릭률(CTR)이 가장 높은 decay_rate 값 발견

7.6 전략 6: 계절성/주기성 고려

특정 시즌이나 주기에 따라 검색 전략을 조정한다.

import datetime  

def get_seasonal_retriever():  
    current_month = datetime.datetime.now().month  
    
    # 연말(11-12월): 연간 리포트 등 최신 정보 중요  
    if current_month in [11, 12]:  
        decay_rate = 0.85  
    # 연초(1-2월): 과거 데이터 참조 중요  
    elif current_month in [1, 2]:  
        decay_rate = 0.3  
    # 일반 기간  
    else:  
        decay_rate = 0.5  
    
    return TimeWeightedVectorStoreRetriever(  
        vectorstore=vectorstore,  
        decay_rate=decay_rate,  
        k=5  
    )  

적용 시나리오: 재무 문서 검색 시스템에서 분기별/연도별 보고서 우선순위 자동 조정

8 요약

TimeWeightedVectorStoreRetriever는 의미적 유사도와 시간 기반 신선도를 결합하여 상황에 맞는 최적의 검색 결과를 제공한다. decay_rate 값을 조정하여 최신성과 관련성의 균형을 세밀하게 제어할 수 있으며, 뉴스, 규정집, 기술 문서 등 다양한 도메인에서 활용 가능하다. 실전에서는 문서 특성, 사용자 의도, 시간대 등을 고려한 동적 전략을 수립하여 검색 품질을 극대화할 수 있다.

with mock_now(datetime.datetime(2024, 8, 29, 00, 00)):  
    # 해당 시점에서의 검색 결과 확인  
    print(retriever.invoke("테디노트"))  

주의사항

• 너무 오래된 시간으로 설정하면 decay_rate 계산 시 오류가 발생할 수 있음 (지수 연산 오버플로우)
  
• 일반적으로 수년 이내의 시간 범위에서 테스트 권장

8.1 decay_rate 튜닝 예제

다양한 시간 간격에서 검색 결과를 테스트하여 최적의 decay_rate를 찾는다.

# 다양한 시간대 테스트  
test_times = [  
    ("1시간 후", datetime.timedelta(hours=1)),  
    ("1일 후", datetime.timedelta(days=1)),  
    ("1주일 후", datetime.timedelta(weeks=1)),  
    ("1개월 후", datetime.timedelta(days=30)),  
]  

for label, delta in test_times:  
    future_time = datetime.datetime.now() + delta  
    with mock_now(future_time):  
        result = retriever.invoke("테디노트")  
        print(f"{label}: {result[0].page_content}")