Temporal RAG: 왜 RAG는 '언제' 질문에 항상 틀릴까?
"2023년 당시 CEO가 누구였지?" "그럼 지금은?" — 이런 간단한 질문에 RAG가 엉뚱한 답을 하는 이유와 해결책.
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였습니다.
더 많은 실패 사례들
| 질문 | 기대 답변 | RAG 답변 | 문제 |
|---|---|---|---|
| "테슬라 2022년 주가는?" | $100-400 범위 | "현재 $248입니다" | 시점 무시 |
| "작년 애플 실적 vs 올해?" | 비교 분석 | 뒤섞인 데이터 | 시점 혼재 |
| "트위터 CEO가 누구야?" | "Linda Yaccarino" (2024) | "Elon Musk" (2022) | 오래된 정보 |
| "코로나 확진자 현황" | 최신 데이터 | 2021년 피크 데이터 | 최신성 실패 |
| "그때 회사 정책은?" | 과거 시점 정책 | 현재 정책 | 과거 추적 불가 |
왜 이런 일이 발생하는가?
임베딩의 근본적 한계
벡터 임베딩은 의미적 유사도만 캡처합니다. 시간 정보는 포함되지 않습니다.
python
# 이 두 문장의 임베딩 유사도는 매우 높음
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 (메타데이터 필터링)
가장 기본적인 접근법입니다. 문서에 시간 메타데이터를 추가하고 검색 시 필터링합니다.
구현
python
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시간 표현 파싱
python
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한계점
메타데이터 필터링은 단순하지만 한계가 있습니다:
- 하드 필터링: 경계 바로 밖 문서는 완전히 제외
- 희소성 문제: 특정 기간에 문서가 없으면 결과 없음
- 복합 시간 표현 처리 어려움: "2020년대 초반" 같은 표현
해결책 2: Temporal Decay (시간 감쇠)
최신 문서에 높은 가중치를 부여하는 방식입니다.
구현
python
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감쇠 함수 비교
python
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: 시간 토큰 추가
python
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: 시간 임베딩 결합
python
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을 사용하여 시간 관련성을 재평가합니다.
구현
python
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 예시로 정확도 향상
python
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를 구축합니다.
개념
text
기존 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)구현
python
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 시스템
모든 기법을 통합한 완전한 시스템입니다.
python
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 변경 이력
python
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: 재무 데이터 시계열
python
# 질문: 비교 분석
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: 정책 변경 추적
python
# 질문: 정책 변경 전
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. 시간 인덱스 분할
python
# 연도별 별도 컬렉션
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. 시간 기반 캐싱
python
# 시간 범위별 캐시
cache_key = f"{query_hash}_{start_date}_{end_date}"
cached_result = cache.get(cache_key)
if cached_result:
return cached_result3. 점진적 인덱싱
python
# 새 문서만 추가 (전체 재인덱싱 방지)
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
핵심 문제
- 벡터 임베딩은 시간 정보를 인코딩하지 않음
- "최근", "당시", "현재" 같은 시간 표현 이해 불가
- 시점별 팩트 변경 추적 불가
해결책 비교
| 방법 | 장점 | 단점 | 적합한 경우 |
|---|---|---|---|
| Metadata Filtering | 구현 간단, 빠름 | 하드 필터링, 경계 문제 | 명확한 시간 범위 질문 |
| Temporal Decay | 최신성 자연스럽게 반영 | 과거 시점 질문에 부적합 | "최근 뉴스" 유형 |
| Time-Aware Embedding | 근본적 해결 | 학습 필요, 복잡 | 대규모 시스템 |
| Temporal Reranking | 정확도 높음 | LLM 비용, 느림 | 높은 정확도 필요시 |
| Temporal KG | 팩트 변경 완벽 추적 | 구축 비용 높음 | 정형 지식 중심 |
권장 조합
- 빠른 시작: Metadata Filtering + Temporal Decay
- 균형: 위 + Temporal Reranking
- 완전한 솔루션: 모든 기법 + Temporal KG
다음 단계
- Multi-hop Temporal Reasoning
- Event-based Temporal Indexing
- Temporal Question Decomposition
References
이메일로 받아보기
관련 포스트
AI Research
VibeTensor: AI가 딥러닝 프레임워크를 직접 만들 수 있을까?
NVIDIA 연구진이 공개한 VibeTensor는 LLM 기반 AI 에이전트가 생성한 완전한 딥러닝 런타임입니다. 6만 줄 이상의 C++/CUDA 코드가 AI에 의해 작성되었으며, 이 프로젝트가 보여주는 가능성과 한계를 분석합니다.
Models & Algorithms
Qwen3-Max-Thinking 스냅샷 공개: 추론형 AI의 새로운 기준
최근 LLM 시장의 트렌드는 단순히 '더 많은 데이터'를 학습하는 것을 넘어, 모델이 '어떻게 생각하느냐'에 집중하고 있습니다. 알리바바 클라우드(Alibaba Cloud)가 자사의 가장 강력한 모델 Qwen3-Max-Thinking의 API 스냅샷(qwen3-max-2026-01-23)을 공개했습니다.
Engineering
배포 전 30분 행동 QA: 바이브코딩 앱에서 '진짜' 터지는 버그 12가지
정적 스캔으로 안 잡히는 세션·권한·중복요청·LLM 내성 테스트