Temporal RAG: 왜 RAG는 '언제' 질문에 항상 틀릴까?

"2023년 당시 CEO가 누구였지?" "그럼 지금은?" — 이런 간단한 질문에 RAG가 엉뚱한 답을 하는 이유와 해결책.

들어가며: RAG의 시간 맹점

당신의 RAG 시스템에 이런 질문을 해보세요:

"OpenAI의 CEO는 누구야?"

답변: "Sam Altman입니다."

좋습니다. 이제 이 질문을 해보세요:

"2023년 11월 OpenAI CEO는 누구였어?"

답변: "Sam Altman입니다."

틀렸습니다. 2023년 11월 17일부터 22일까지 Sam Altman은 해고 상태였고, Mira Murati가 임시 CEO였습니다.

더 많은 실패 사례들

왜 이런 일이 발생하는가?

임베딩의 근본적 한계

벡터 임베딩은 의미적 유사도만 캡처합니다. 시간 정보는 포함되지 않습니다.

# 이 두 문장의 임베딩 유사도는 매우 높음
text1 = "Sam Altman is the CEO of OpenAI"  # 2024년 문서
text2 = "Sam Altman is the CEO of OpenAI"  # 2020년 문서

# 하지만 이 문장과도 높은 유사도
text3 = "Mira Murati is the CEO of OpenAI"  # 2023년 11월 문서

# 임베딩은 '언제'를 모름
similarity(embed(text1), embed(text2)) ≈ 1.0  # 같은 내용
similarity(embed(text1), embed(text3)) ≈ 0.85  # CEO 질문엔 둘 다 관련

시간 관련 질문의 유형

1. 특정 시점 질문 (Point-in-Time)

• "2023년 3분기 매출은?"
• "그 당시 정책은 뭐였지?"
• "작년 이맘때 상황은?"

2. 시간 범위 질문 (Time Range)

• "2020년부터 2023년까지 변화"
• "최근 3개월간 트렌드"
• "올해 들어 달라진 점"

3. 상대적 시간 질문 (Relative Time)

• "최근 뉴스" (언제 기준?)
• "예전에는 어땠어?" (얼마나 예전?)
• "그 후로 뭐가 바뀌었어?"

4. 시간 비교 질문 (Temporal Comparison)

• "전년 대비 성장률"
• "정책 변경 전후 차이"
• "CEO 교체 전후 실적"

5. 시계열 질문 (Time Series)

• "분기별 매출 추이"
• "연도별 사용자 증가"
• "월별 트래픽 변화"

해결책 1: Metadata Filtering (메타데이터 필터링)

가장 기본적인 접근법입니다. 문서에 시간 메타데이터를 추가하고 검색 시 필터링합니다.

구현

from datetime import datetime, timedelta
from typing import List, Optional
import chromadb

class TemporalVectorStore:
    """시간 인식 벡터 스토어"""

    def __init__(self):
        self.client = chromadb.Client()
        self.collection = self.client.create_collection("temporal_docs")

    def add_document(self, doc_id: str, text: str, timestamp: datetime,
                     source: str = None):
        """시간 메타데이터와 함께 문서 추가"""
        self.collection.add(
            ids=[doc_id],
            documents=[text],
            metadatas=[{
                "timestamp": timestamp.isoformat(),
                "year": timestamp.year,
                "month": timestamp.month,
                "quarter": (timestamp.month - 1) // 3 + 1,
                "source": source or "unknown"
            }]
        )

    def query_with_time_filter(
        self,
        query: str,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None,
        top_k: int = 5
    ) -> List[dict]:
        """시간 필터링된 검색"""

        where_filter = {}

        if start_date and end_date:
            where_filter = {
                "$and": [
                    {"timestamp": {"$gte": start_date.isoformat()}},
                    {"timestamp": {"$lte": end_date.isoformat()}}
                ]
            }
        elif start_date:
            where_filter = {"timestamp": {"$gte": start_date.isoformat()}}
        elif end_date:
            where_filter = {"timestamp": {"$lte": end_date.isoformat()}}

        results = self.collection.query(
            query_texts=[query],
            n_results=top_k,
            where=where_filter if where_filter else None
        )

        return results

시간 표현 파싱

import re
from dateutil import parser
from dateutil.relativedelta import relativedelta

class TemporalQueryParser:
    """쿼리에서 시간 정보 추출"""

    def __init__(self):
        self.patterns = {
            # 절대 시간
            r'(\d{4})년': 'year',
            r'(\d{4})년\s*(\d{1,2})월': 'year_month',
            r'(\d{1,2})분기': 'quarter',

            # 상대 시간
            r'최근\s*(\d+)일': 'recent_days',
            r'최근\s*(\d+)개월': 'recent_months',
            r'지난\s*(\d+)년': 'past_years',
            r'작년': 'last_year',
            r'올해': 'this_year',
            r'이번\s*달': 'this_month',
            r'지난\s*달': 'last_month',

            # 특수 표현
            r'당시|그때|그\s*당시': 'contextual',  # 컨텍스트 필요
            r'현재|지금|오늘': 'now',
            r'예전|과거': 'past_general',
        }

    def parse(self, query: str, reference_date: datetime = None) -> dict:
        """쿼리에서 시간 범위 추출"""
        if reference_date is None:
            reference_date = datetime.now()

        result = {
            "original_query": query,
            "start_date": None,
            "end_date": None,
            "temporal_type": "none"
        }

        # 절대 연도
        year_match = re.search(r'(\d{4})년', query)
        if year_match:
            year = int(year_match.group(1))
            result["start_date"] = datetime(year, 1, 1)
            result["end_date"] = datetime(year, 12, 31)
            result["temporal_type"] = "absolute_year"
            return result

        # 최근 N일/개월
        recent_days = re.search(r'최근\s*(\d+)일', query)
        if recent_days:
            days = int(recent_days.group(1))
            result["start_date"] = reference_date - timedelta(days=days)
            result["end_date"] = reference_date
            result["temporal_type"] = "relative_recent"
            return result

        recent_months = re.search(r'최근\s*(\d+)개월', query)
        if recent_months:
            months = int(recent_months.group(1))
            result["start_date"] = reference_date - relativedelta(months=months)
            result["end_date"] = reference_date
            result["temporal_type"] = "relative_recent"
            return result

        # 작년/올해
        if '작년' in query:
            last_year = reference_date.year - 1
            result["start_date"] = datetime(last_year, 1, 1)
            result["end_date"] = datetime(last_year, 12, 31)
            result["temporal_type"] = "relative_year"
            return result

        if '올해' in query:
            result["start_date"] = datetime(reference_date.year, 1, 1)
            result["end_date"] = reference_date
            result["temporal_type"] = "relative_year"
            return result

        # 현재/지금
        if any(kw in query for kw in ['현재', '지금', '오늘']):
            result["start_date"] = reference_date - timedelta(days=7)  # 최근 1주
            result["end_date"] = reference_date
            result["temporal_type"] = "current"
            return result

        return result

한계점

메타데이터 필터링은 단순하지만 한계가 있습니다:

1. 하드 필터링: 경계 바로 밖 문서는 완전히 제외
2. 희소성 문제: 특정 기간에 문서가 없으면 결과 없음
3. 복합 시간 표현 처리 어려움: "2020년대 초반" 같은 표현

해결책 2: Temporal Decay (시간 감쇠)

최신 문서에 높은 가중치를 부여하는 방식입니다.

구현

import numpy as np
from datetime import datetime

class TemporalDecayScorer:
    """시간 기반 점수 감쇠"""

    def __init__(self, half_life_days: int = 30):
        """
        half_life_days: 점수가 절반이 되는 기간
        예: 30일이면, 30일 전 문서는 현재 문서의 50% 점수
        """
        self.half_life_days = half_life_days
        self.decay_rate = np.log(2) / half_life_days

    def exponential_decay(self, doc_date: datetime,
                          reference_date: datetime = None) -> float:
        """지수 감쇠 함수"""
        if reference_date is None:
            reference_date = datetime.now()

        age_days = (reference_date - doc_date).days
        return np.exp(-self.decay_rate * age_days)

    def gaussian_decay(self, doc_date: datetime,
                       target_date: datetime,
                       sigma_days: int = 30) -> float:
        """
        가우시안 감쇠 - 특정 시점 근처에 피크
        특정 시점 질문에 적합
        """
        diff_days = abs((target_date - doc_date).days)
        return np.exp(-(diff_days ** 2) / (2 * sigma_days ** 2))

    def apply_temporal_score(
        self,
        results: List[dict],
        query_type: str = "recent",
        target_date: datetime = None
    ) -> List[dict]:
        """검색 결과에 시간 점수 적용"""

        scored_results = []

        for result in results:
            doc_date = datetime.fromisoformat(result['metadata']['timestamp'])
            semantic_score = result.get('score', 1.0)

            if query_type == "recent":
                # 최신 문서 선호
                temporal_score = self.exponential_decay(doc_date)
            elif query_type == "point_in_time" and target_date:
                # 특정 시점 근처 선호
                temporal_score = self.gaussian_decay(doc_date, target_date)
            else:
                temporal_score = 1.0

            # 최종 점수 = 의미 점수 * 시간 점수
            final_score = semantic_score * temporal_score

            scored_results.append({
                **result,
                'semantic_score': semantic_score,
                'temporal_score': temporal_score,
                'final_score': final_score
            })

        # 최종 점수로 재정렬
        scored_results.sort(key=lambda x: x['final_score'], reverse=True)

        return scored_results

감쇠 함수 비교

def visualize_decay_functions():
    """감쇠 함수 시각화"""
    import matplotlib.pyplot as plt

    days = np.arange(0, 365)
    scorer = TemporalDecayScorer(half_life_days=30)

    # 지수 감쇠 (최신성 질문용)
    exp_decay = [np.exp(-np.log(2)/30 * d) for d in days]

    # 가우시안 (특정 시점 질문용, 100일 전 기준)
    target_day = 100
    gaussian = [np.exp(-((d - target_day)**2) / (2 * 30**2)) for d in days]

    # 선형 감쇠
    linear = [max(0, 1 - d/365) for d in days]

    plt.figure(figsize=(12, 6))
    plt.plot(days, exp_decay, label='Exponential (최신성 질문)', linewidth=2)
    plt.plot(days, gaussian, label=f'Gaussian (Day {target_day} 기준)', linewidth=2)
    plt.plot(days, linear, label='Linear', linewidth=2, linestyle='--')

    plt.xlabel('Days Ago')
    plt.ylabel('Score Weight')
    plt.title('Temporal Decay Functions')
    plt.legend()
    plt.grid(True, alpha=0.3)
    plt.show()

해결책 3: Time-Aware Embedding (시간 인식 임베딩)

시간 정보를 임베딩 자체에 인코딩하는 방식입니다.

방법 1: 시간 토큰 추가

class TimeAwareEmbedder:
    """시간 정보를 텍스트에 추가하여 임베딩"""

    def __init__(self, embedding_model):
        self.model = embedding_model

    def add_temporal_context(self, text: str, timestamp: datetime) -> str:
        """텍스트에 시간 컨텍스트 추가"""
        time_prefix = f"[DATE: {timestamp.strftime('%Y-%m-%d')}] "
        return time_prefix + text

    def embed_with_time(self, text: str, timestamp: datetime) -> np.ndarray:
        """시간 컨텍스트가 포함된 임베딩 생성"""
        temporal_text = self.add_temporal_context(text, timestamp)
        return self.model.encode(temporal_text)

방법 2: 시간 임베딩 결합

class TemporalEmbedding:
    """텍스트 임베딩 + 시간 임베딩 결합"""

    def __init__(self, text_dim: int = 768, time_dim: int = 32):
        self.text_dim = text_dim
        self.time_dim = time_dim

        # 시간 인코딩 가중치 (학습 가능)
        self.time_encoder = self._build_time_encoder()

    def _build_time_encoder(self):
        """Positional Encoding 스타일 시간 인코더"""
        import torch.nn as nn

        return nn.Sequential(
            nn.Linear(6, 64),  # [year, month, day, hour, day_of_week, day_of_year]
            nn.ReLU(),
            nn.Linear(64, self.time_dim)
        )

    def encode_time(self, timestamp: datetime) -> np.ndarray:
        """시간을 벡터로 인코딩"""
        features = np.array([
            timestamp.year / 3000,  # 정규화
            timestamp.month / 12,
            timestamp.day / 31,
            timestamp.hour / 24,
            timestamp.weekday() / 7,
            timestamp.timetuple().tm_yday / 366
        ])
        return features

    def combine_embeddings(self, text_emb: np.ndarray,
                           time_emb: np.ndarray,
                           alpha: float = 0.1) -> np.ndarray:
        """텍스트와 시간 임베딩 결합"""
        # 간단한 concatenation
        # 또는 weighted combination
        combined = np.concatenate([
            text_emb * (1 - alpha),
            time_emb * alpha
        ])
        return combined / np.linalg.norm(combined)

해결책 4: Temporal Reranking (시간 기반 재순위화)

검색 후 LLM을 사용하여 시간 관련성을 재평가합니다.

구현

class TemporalReranker:
    """LLM 기반 시간 인식 재순위화"""

    def __init__(self, llm_client):
        self.llm = llm_client

    def rerank(self, query: str, documents: List[dict],
               temporal_context: dict) -> List[dict]:
        """시간 컨텍스트를 고려한 재순위화"""

        prompt = f"""Given the query and temporal context, rank these documents by relevance.

Query: {query}
Temporal Context: {temporal_context}

Documents:
"""
        for i, doc in enumerate(documents):
            prompt += f"""
[{i+1}] Date: {doc['metadata']['timestamp']}
Content: {doc['text'][:500]}...
"""

        prompt += """
For each document, provide:
1. Temporal relevance score (0-1): How well does the document's date match the query's temporal intent?
2. Content relevance score (0-1): How relevant is the content?
3. Final ranking

Output as JSON array."""

        response = self.llm.generate(prompt)
        rankings = self._parse_rankings(response)

        return self._apply_rankings(documents, rankings)

Few-shot 예시로 정확도 향상

TEMPORAL_RERANK_EXAMPLES = """
Example 1:
Query: "2023년 OpenAI CEO는 누구?"
Document A (2024-01): "Sam Altman returned as CEO of OpenAI"
Document B (2023-11): "Mira Murati appointed as interim CEO"
Document C (2023-03): "Sam Altman leads OpenAI's GPT-4 launch"

Analysis:
- Query asks about 2023
- Doc A is from 2024 → Low temporal relevance
- Doc B is from Nov 2023, discusses CEO → High temporal relevance
- Doc C is from Mar 2023, CEO context → Medium temporal relevance

Ranking: B > C > A

Example 2:
Query: "최근 테슬라 실적"
Document A (2024-01): "Tesla Q4 2023 earnings report"
Document B (2023-06): "Tesla Q1 2023 results"
Document C (2022-12): "Tesla annual report 2022"

Analysis:
- Query asks for "recent" → Prefer latest
- Doc A is most recent with earnings info → High relevance
- Doc B is older but relevant content → Medium relevance
- Doc C is too old → Low relevance

Ranking: A > B > C
"""

해결책 5: Temporal Knowledge Graph

시간 축을 가진 Knowledge Graph를 구축합니다.

개념

기존 KG: (Sam Altman) --[CEO_OF]--> (OpenAI)

Temporal KG: (Sam Altman) --[CEO_OF {start: 2019, end: 2023-11-17}]--> (OpenAI)
             (Mira Murati) --[CEO_OF {start: 2023-11-17, end: 2023-11-20}]--> (OpenAI)
             (Emmett Shear) --[CEO_OF {start: 2023-11-20, end: 2023-11-22}]--> (OpenAI)
             (Sam Altman) --[CEO_OF {start: 2023-11-22, end: null}]--> (OpenAI)

구현

from dataclasses import dataclass
from datetime import datetime
from typing import Optional, List

@dataclass
class TemporalTriple:
    """시간 정보를 가진 트리플"""
    subject: str
    predicate: str
    object: str
    valid_from: datetime
    valid_to: Optional[datetime] = None  # None = 현재까지 유효
    confidence: float = 1.0
    source: str = ""

class TemporalKnowledgeGraph:
    """시간 인식 Knowledge Graph"""

    def __init__(self):
        self.triples: List[TemporalTriple] = []
        self.entity_index = {}  # entity -> triples
        self.time_index = {}    # (year, month) -> triples

    def add_triple(self, triple: TemporalTriple):
        """트리플 추가 및 인덱싱"""
        self.triples.append(triple)

        # 엔티티 인덱스
        for entity in [triple.subject, triple.object]:
            if entity not in self.entity_index:
                self.entity_index[entity] = []
            self.entity_index[entity].append(triple)

        # 시간 인덱스
        time_key = (triple.valid_from.year, triple.valid_from.month)
        if time_key not in self.time_index:
            self.time_index[time_key] = []
        self.time_index[time_key].append(triple)

    def query_at_time(self, subject: str, predicate: str,
                      at_time: datetime) -> List[TemporalTriple]:
        """특정 시점에 유효한 트리플 조회"""
        results = []

        if subject in self.entity_index:
            for triple in self.entity_index[subject]:
                if triple.predicate != predicate:
                    continue

                # 시간 유효성 검사
                if triple.valid_from <= at_time:
                    if triple.valid_to is None or triple.valid_to >= at_time:
                        results.append(triple)

        return results

    def query_history(self, subject: str, predicate: str) -> List[TemporalTriple]:
        """엔티티의 특정 관계 히스토리 조회"""
        results = []

        if subject in self.entity_index:
            for triple in self.entity_index[subject]:
                if triple.predicate == predicate:
                    results.append(triple)

        # 시간순 정렬
        results.sort(key=lambda x: x.valid_from)
        return results


# 사용 예시
tkg = TemporalKnowledgeGraph()

# OpenAI CEO 히스토리 추가
tkg.add_triple(TemporalTriple(
    subject="Sam Altman",
    predicate="CEO_OF",
    object="OpenAI",
    valid_from=datetime(2019, 3, 1),
    valid_to=datetime(2023, 11, 17),
    source="news_001"
))

tkg.add_triple(TemporalTriple(
    subject="Mira Murati",
    predicate="CEO_OF",
    object="OpenAI",
    valid_from=datetime(2023, 11, 17),
    valid_to=datetime(2023, 11, 20),
    source="news_002"
))

tkg.add_triple(TemporalTriple(
    subject="Emmett Shear",
    predicate="CEO_OF",
    object="OpenAI",
    valid_from=datetime(2023, 11, 20),
    valid_to=datetime(2023, 11, 22),
    source="news_003"
))

tkg.add_triple(TemporalTriple(
    subject="Sam Altman",
    predicate="CEO_OF",
    object="OpenAI",
    valid_from=datetime(2023, 11, 22),
    valid_to=None,  # 현재까지
    source="news_004"
))

# 쿼리
print("2023년 11월 18일 OpenAI CEO:")
result = tkg.query_at_time("OpenAI", "CEO_OF", datetime(2023, 11, 18))
# 역방향 쿼리 필요 -> 수정된 버전에서는 object도 검색

print("\nOpenAI CEO 히스토리:")
for triple in tkg.query_history("Sam Altman", "CEO_OF"):
    end = triple.valid_to.strftime('%Y-%m-%d') if triple.valid_to else "현재"
    print(f"  {triple.valid_from.strftime('%Y-%m-%d')} ~ {end}")

통합 Temporal RAG 시스템

모든 기법을 통합한 완전한 시스템입니다.

class TemporalRAG:
    """통합 Temporal RAG 시스템"""

    def __init__(self, vector_store, knowledge_graph, llm, embedding_model):
        self.vector_store = vector_store
        self.kg = knowledge_graph
        self.llm = llm
        self.embedder = TimeAwareEmbedder(embedding_model)
        self.query_parser = TemporalQueryParser()
        self.decay_scorer = TemporalDecayScorer(half_life_days=30)
        self.reranker = TemporalReranker(llm)

    def query(self, query: str, reference_date: datetime = None) -> dict:
        """통합 시간 인식 검색"""

        if reference_date is None:
            reference_date = datetime.now()

        # 1. 쿼리에서 시간 정보 추출
        temporal_info = self.query_parser.parse(query, reference_date)

        # 2. 쿼리 유형 결정
        query_type = self._determine_query_type(temporal_info)

        # 3. 검색 전략 선택 및 실행
        if query_type == "point_in_time":
            results = self._point_in_time_search(query, temporal_info)
        elif query_type == "time_range":
            results = self._time_range_search(query, temporal_info)
        elif query_type == "recent":
            results = self._recency_search(query, reference_date)
        elif query_type == "historical":
            results = self._historical_search(query, temporal_info)
        else:
            results = self._default_search(query)

        # 4. Knowledge Graph 보강
        kg_facts = self._query_knowledge_graph(query, temporal_info)

        # 5. 재순위화
        reranked = self.reranker.rerank(query, results, temporal_info)

        # 6. 답변 생성
        answer = self._generate_answer(query, reranked, kg_facts, temporal_info)

        return {
            "query": query,
            "temporal_info": temporal_info,
            "query_type": query_type,
            "documents": reranked[:5],
            "kg_facts": kg_facts,
            "answer": answer
        }

    def _determine_query_type(self, temporal_info: dict) -> str:
        """시간 쿼리 유형 결정"""
        if temporal_info["temporal_type"] == "none":
            return "default"
        elif temporal_info["temporal_type"] in ["absolute_year", "absolute_date"]:
            return "point_in_time"
        elif temporal_info["temporal_type"] == "relative_recent":
            return "recent"
        elif temporal_info["temporal_type"] in ["relative_year", "range"]:
            return "time_range"
        else:
            return "default"

    def _point_in_time_search(self, query: str, temporal_info: dict) -> List[dict]:
        """특정 시점 검색"""
        target_date = temporal_info.get("start_date")

        # 1. 시간 필터링된 벡터 검색
        results = self.vector_store.query_with_time_filter(
            query,
            start_date=target_date - timedelta(days=30),
            end_date=target_date + timedelta(days=30),
            top_k=20
        )

        # 2. Gaussian decay로 점수 조정 (target_date 근처 선호)
        scored = self.decay_scorer.apply_temporal_score(
            results,
            query_type="point_in_time",
            target_date=target_date
        )

        return scored

    def _recency_search(self, query: str, reference_date: datetime) -> List[dict]:
        """최신성 검색"""
        # 1. 최근 문서만 검색
        results = self.vector_store.query_with_time_filter(
            query,
            start_date=reference_date - timedelta(days=90),
            end_date=reference_date,
            top_k=20
        )

        # 2. Exponential decay로 최신 문서 선호
        scored = self.decay_scorer.apply_temporal_score(
            results,
            query_type="recent"
        )

        return scored

    def _query_knowledge_graph(self, query: str, temporal_info: dict) -> List[dict]:
        """Knowledge Graph에서 시간 인식 팩트 조회"""
        facts = []

        # 쿼리에서 엔티티 추출 (간단한 구현)
        # 실제로는 NER 또는 LLM 사용
        entities = self._extract_entities(query)

        target_time = temporal_info.get("start_date", datetime.now())

        for entity in entities:
            # 해당 시점의 팩트 조회
            entity_facts = self.kg.query_at_time(entity, None, target_time)
            facts.extend(entity_facts)

        return facts

    def _generate_answer(self, query: str, documents: List[dict],
                        kg_facts: List, temporal_info: dict) -> str:
        """시간 컨텍스트를 포함한 답변 생성"""

        prompt = f"""Answer the following question using the provided context.
Pay special attention to the temporal aspect of the question.

Question: {query}
Temporal Context: {temporal_info}

Knowledge Graph Facts (time-aware):
"""
        for fact in kg_facts[:5]:
            end_date = fact.valid_to.strftime('%Y-%m-%d') if fact.valid_to else "present"
            prompt += f"- {fact.subject} {fact.predicate} {fact.object} (from {fact.valid_from.strftime('%Y-%m-%d')} to {end_date})\n"

        prompt += "\nRelevant Documents:\n"
        for doc in documents[:3]:
            prompt += f"[{doc['metadata']['timestamp']}] {doc['text'][:300]}...\n\n"

        prompt += """
Instructions:
1. Consider the time period specified in the question
2. Prioritize information from the relevant time period
3. If the question asks about a specific point in time, answer for that exact time
4. If comparing time periods, clearly distinguish between them
5. Acknowledge if information from the requested time period is not available

Answer:"""

        return self.llm.generate(prompt)

실제 사용 예시

예시 1: CEO 변경 이력

rag = TemporalRAG(vector_store, kg, llm, embedder)

# 질문 1: 과거 특정 시점
result = rag.query("2023년 11월 18일 OpenAI CEO는 누구였어?")
print(result["answer"])
# 출력: "2023년 11월 18일 당시 OpenAI의 CEO는 Mira Murati였습니다.
#        Sam Altman이 11월 17일 해임된 후 임시 CEO로 임명되었으며,
#        이후 11월 20일 Emmett Shear로 교체되었습니다."

# 질문 2: 현재
result = rag.query("지금 OpenAI CEO는?")
print(result["answer"])
# 출력: "현재(2024년 1월 기준) OpenAI의 CEO는 Sam Altman입니다.
#        2023년 11월 22일 복귀하여 현재까지 재임 중입니다."

# 질문 3: 히스토리
result = rag.query("OpenAI CEO가 바뀐 적 있어?")
print(result["answer"])
# 출력: "네, OpenAI CEO는 여러 번 바뀌었습니다.
#        - Sam Altman (2019.3 ~ 2023.11.17)
#        - Mira Murati 임시 CEO (2023.11.17 ~ 2023.11.20)
#        - Emmett Shear 임시 CEO (2023.11.20 ~ 2023.11.22)
#        - Sam Altman 복귀 (2023.11.22 ~ 현재)"

예시 2: 재무 데이터 시계열

# 질문: 비교 분석
result = rag.query("테슬라 2022년 vs 2023년 매출 비교해줘")
print(result["answer"])
# 출력: "테슬라 연간 매출 비교:
#        - 2022년: $81.5B (전년 대비 51% 증가)
#        - 2023년: $96.8B (전년 대비 19% 증가)
#        2023년에도 성장했으나 성장률은 둔화되었습니다."

# 질문: 특정 분기
result = rag.query("테슬라 2023년 3분기 실적은?")
print(result["answer"])
# 출력: "테슬라 2023년 Q3 실적:
#        - 매출: $23.4B
#        - 순이익: $1.9B
#        - 차량 인도: 435,059대"

예시 3: 정책 변경 추적

# 질문: 정책 변경 전
result = rag.query("트위터가 X로 바뀌기 전 인증 정책은?")
print(result["answer"])
# 출력: "X(구 트위터) 리브랜딩 전(2023년 7월 이전) 인증 정책:
#        - 무료 파란 체크마크 (유명인, 기관, 언론인)
#        - 신원 확인 프로세스 필요
#        - 2022년 Elon Musk 인수 후 유료화 전환 시작"

# 질문: 정책 변경 후
result = rag.query("지금 X 인증 정책은?")
print(result["answer"])
# 출력: "현재 X의 인증 정책 (2024년 기준):
#        - X Premium 구독으로 파란 체크마크 구매 가능 ($8/월)
#        - 기업용 골드 체크마크 (X Verified Organizations)
#        - 정부/기관용 회색 체크마크"

성능 최적화 팁

1. 시간 인덱스 분할

# 연도별 별도 컬렉션
collections = {
    2022: chroma.create_collection("docs_2022"),
    2023: chroma.create_collection("docs_2023"),
    2024: chroma.create_collection("docs_2024"),
}

# 쿼리 시 관련 연도만 검색
def query_by_year(query, year):
    if year in collections:
        return collections[year].query(query)

2. 시간 기반 캐싱

# 시간 범위별 캐시
cache_key = f"{query_hash}_{start_date}_{end_date}"
cached_result = cache.get(cache_key)

if cached_result:
    return cached_result

3. 점진적 인덱싱

# 새 문서만 추가 (전체 재인덱싱 방지)
def incremental_index(new_docs):
    for doc in new_docs:
        if doc.timestamp > last_indexed_time:
            vector_store.add(doc)

    # Knowledge Graph도 업데이트
    kg.update_from_docs(new_docs)

Summary

핵심 문제

• 벡터 임베딩은 시간 정보를 인코딩하지 않음
• "최근", "당시", "현재" 같은 시간 표현 이해 불가
• 시점별 팩트 변경 추적 불가

해결책 비교

권장 조합

1. 빠른 시작: Metadata Filtering + Temporal Decay
2. 균형: 위 + Temporal Reranking
3. 완전한 솔루션: 모든 기법 + Temporal KG

다음 단계

• Multi-hop Temporal Reasoning
• Event-based Temporal Indexing
• Temporal Question Decomposition

References

• TimeQA: A Question Answering Dataset with Temporal Information
• Temporal Knowledge Graph Reasoning
• TRAM: Time-aware Retrieval Augmentation for Open-domain Question Answering