Temporal Reasoning in RAG: Handling Document Freshness, Version Conflicts, and Time-Sensitive Retrieval in Production
Introduction
Your RAG system just confidently told a user that the company's parental leave policy offers eight weeks of paid leave. The problem: that policy was updated six months ago to fourteen weeks. Somewhere in your vector store, two documents coexist — the old policy and the new one — and the retrieval system surfaced the wrong one. The LLM had no idea, synthesized a confident answer, and now HR is getting angry emails.
This is the central failure mode of temporal reasoning in RAG, and it's far more insidious than the more commonly discussed hallucination problem. At least hallucinations are fabrications — you can sometimes catch them with a factuality check. Stale retrieval looks exactly like correct retrieval. The chunk scores well semantically because its content is genuinely relevant to the query; it just happens to describe reality as it was, not as it is. Production RAG systems operating on live knowledge bases — policy repositories, product documentation, regulatory filings, research literature — face this challenge constantly, and most teams don't build the infrastructure to handle it until they've already shipped bad answers to real users.
By the end of this lesson, you'll have the architecture, the code patterns, and the operational playbook to handle time-sensitive retrieval properly. We'll work through metadata schemas, temporal scoring functions, version conflict detection, and recency-aware prompt design.
What you'll learn:
- How to design a temporal metadata schema that enables freshness-aware retrieval at query time
- How to implement a scoring function that balances semantic similarity with document recency
- How to detect and resolve version conflicts when multiple document revisions coexist in your index
- How to build a query-time temporal classifier that adjusts retrieval strategy for time-sensitive questions
- How to monitor temporal drift in your knowledge base and build alerting before stale answers reach users
Prerequisites
You should be comfortable with the fundamentals of building and operating RAG pipelines. Specifically, you should understand how vector embeddings work and how similarity search operates, have hands-on experience building a RAG pipeline with documents, and understand how metadata filtering works in vector stores. If you've worked through the lessons on metadata filtering in RAG and indexing strategies for RAG, you'll find this lesson builds directly on those concepts. Experience with Python and either Pinecone, Weaviate, or pgvector is assumed.
Why Temporal Reasoning Is Hard in RAG
Before writing code, let's understand why this problem is genuinely difficult. RAG's retrieval mechanism is semantically indifferent to time. When you embed a document chunk and store it in a vector index, the embedding captures the semantic content — but temporal attributes like "this was written in 2021" or "this was superseded by version 3.2" are not encoded in the embedding itself. A chunk describing the old tax bracket structure and a chunk describing the new tax bracket structure will have very similar embeddings, because they discuss the same topic using similar vocabulary.
This creates three distinct failure modes:
Stale document retrieval: The most recent document exists in your index, but an older document scores higher for a given query because of slightly different phrasing that happens to match the query better. The retrieval system has no concept of "newer = more authoritative."
Version conflict synthesis: Both the old and new versions of a document exist in the index, and both are retrieved. The LLM receives contradictory information and either hallucinates a reconciliation, picks one arbitrarily, or (if well-prompted) expresses uncertainty it shouldn't need to express.
Temporal query mismatch: The user asks a time-anchored question ("What was the refund policy last year?") but the retrieval system has no mechanism to respect that temporal anchor. It just returns the most semantically similar chunks, which might be from this year's policy.
Key insight: The embedding model is not your temporal gatekeeper. Embeddings capture semantic similarity across time, not temporal ordering. Any freshness logic must be implemented at the metadata and scoring layer, not by hoping that newer documents embed differently than older ones.
The foundation of temporal reasoning in RAG is a well-designed metadata schema attached to every chunk at indexing time. Most teams attach a source_url and maybe a document_title — but that's not enough. Here's the schema you should be building toward:
from datetime import datetime, timezone
from typing import Optional
from dataclasses import dataclass, field
import uuid
@dataclass
class ChunkMetadata:
# Identity
chunk_id: str = field(default_factory=lambda: str(uuid.uuid4()))
document_id: str = "" # Stable ID for the document across versions
document_version: str = "" # Semantic version or revision string (e.g., "3.2.1")
version_sequence: int = 0 # Integer for ordering (higher = newer)
# Source information
source_url: str = ""
source_system: str = "" # "confluence", "sharepoint", "s3", etc.
# Temporal attributes
created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
content_date: Optional[datetime] = None # When the content was authored/effective
effective_date: Optional[datetime] = None # When the content became valid
expiry_date: Optional[datetime] = None # When the content ceases to be valid
last_modified: Optional[datetime] = None
indexed_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
# Versioning state
is_latest: bool = True # Is this the current version?
superseded_by: Optional[str] = None # document_id of the replacing document
supersedes: Optional[str] = None # document_id this document replaces
# Domain-specific temporal context
regulatory_period: Optional[str] = None # e.g., "FY2024", "Q3-2024"
jurisdiction_date: Optional[datetime] = None # For legal/regulatory docs
def to_vector_store_dict(self) -> dict:
"""Serialize for storage in vector metadata (timestamps as Unix epoch)."""
return {
"chunk_id": self.chunk_id,
"document_id": self.document_id,
"document_version": self.document_version,
"version_sequence": self.version_sequence,
"source_url": self.source_url,
"source_system": self.source_system,
"created_at": self.created_at.timestamp(),
"content_date": self.content_date.timestamp() if self.content_date else None,
"effective_date": self.effective_date.timestamp() if self.effective_date else None,
"expiry_date": self.expiry_date.timestamp() if self.expiry_date else None,
"last_modified": self.last_modified.timestamp() if self.last_modified else None,
"indexed_at": self.indexed_at.timestamp(),
"is_latest": self.is_latest,
"superseded_by": self.superseded_by,
"regulatory_period": self.regulatory_period,
}
Several design decisions here deserve explanation. Notice the distinction between content_date, effective_date, and expiry_date. These serve very different purposes. A policy document might be authored on January 15th, become effective on February 1st, and expire on December 31st. A regulatory filing might describe a fiscal period that ended years ago but was filed recently. Conflating these dates is a common source of bugs — particularly in financial and legal domains where "when was this written" and "what time period does this describe" are completely different questions.
The version_sequence field (an integer) is critical because string-based version comparisons are unreliable. "3.10" sorts before "3.9" alphabetically but after it numerically. Store an integer that strictly orders versions regardless of naming conventions.
The is_latest flag enables a crucial optimization: filtering out non-current documents at query time without requiring complex join logic. When you ingest a new version of a document, you update all existing chunks for that document_id to set is_latest = False before indexing the new chunks.
Warning: Don't rely solely on last_modified timestamps from your source systems. SharePoint, Confluence, and most CMS platforms update this field on every minor edit, including typo corrections. A document with "last_modified = yesterday" might have had only a comma added, while the substantive content dates from three years ago. Where possible, extract or maintain explicit content_date and effective_date fields.
Implementing Freshness-Weighted Scoring
Pure vector similarity search ranks documents by semantic closeness to the query. Adding temporal reasoning means implementing a composite scoring function that combines semantic similarity with a freshness signal.
The key insight is that freshness scoring should be adaptive, not a fixed offset. A highly relevant but slightly older document should still beat a marginally relevant but brand-new one. And the steepness of the freshness decay should be configurable per knowledge domain: regulatory documents decay slowly (last year's tax code is still mostly valid), news articles decay fast (yesterday's market data is nearly useless).
import math
from datetime import datetime, timezone
from typing import List, Tuple, Optional
from dataclasses import dataclass
@dataclass
class ScoredChunk:
chunk_id: str
document_id: str
text: str
semantic_score: float # From vector similarity (0-1)
freshness_score: float # Computed temporal score (0-1)
composite_score: float # Final ranking score
metadata: dict
class TemporalScorer:
"""
Computes composite scores blending semantic similarity with document freshness.
Uses an exponential decay model for freshness, parameterized by half-life.
Half-life: the age at which a document's freshness score drops to 0.5.
"""
# Domain-specific half-lives in days
HALF_LIVES = {
"news": 1,
"market_data": 0.5,
"product_documentation": 90,
"policy": 180,
"regulatory": 365,
"research": 730,
"legal_contract": 1825, # 5 years
"default": 180,
}
def __init__(
self,
semantic_weight: float = 0.7,
freshness_weight: float = 0.3,
domain: str = "default",
custom_half_life_days: Optional[float] = None,
):
assert abs(semantic_weight + freshness_weight - 1.0) < 0.001, \
"Weights must sum to 1.0"
self.semantic_weight = semantic_weight
self.freshness_weight = freshness_weight
self.half_life_days = (
custom_half_life_days or self.HALF_LIVES.get(domain, self.HALF_LIVES["default"])
)
def compute_freshness_score(
self,
metadata: dict,
reference_date: Optional[datetime] = None
) -> float:
"""
Exponential decay freshness score.
Score = 0.5 ^ (age_days / half_life_days)
Returns 1.0 for current documents, approaching 0 for very old ones.
Prefers effective_date > content_date > last_modified > indexed_at.
"""
now = reference_date or datetime.now(timezone.utc)
# Priority order for determining document age
timestamp_fields = [
"effective_date", "content_date", "last_modified", "indexed_at"
]
doc_timestamp = None
for field in timestamp_fields:
raw = metadata.get(field)
if raw is not None:
# Vector stores return Unix timestamps as floats
doc_timestamp = datetime.fromtimestamp(float(raw), tz=timezone.utc)
break
if doc_timestamp is None:
return 0.5 # Unknown age: neutral score
age_days = (now - doc_timestamp).total_seconds() / 86400
if age_days < 0:
# Future-dated document (e.g., pre-published policy)
return 1.0
# Exponential decay: score = 0.5^(age/half_life)
score = math.pow(0.5, age_days / self.half_life_days)
return max(0.0, min(1.0, score))
def apply_version_boost(self, metadata: dict) -> float:
"""
Apply a multiplicative boost to documents flagged as the latest version.
Non-latest documents receive a penalty to push them down in rankings.
"""
if metadata.get("is_latest", True):
return 1.0
else:
return 0.3 # Significant penalty for superseded documents
def score_chunks(
self,
retrieved_chunks: List[Tuple[str, float, dict]], # (text, semantic_score, metadata)
reference_date: Optional[datetime] = None,
) -> List[ScoredChunk]:
"""
Score and re-rank a list of retrieved chunks.
"""
scored = []
for text, semantic_score, metadata in retrieved_chunks:
freshness_score = self.compute_freshness_score(metadata, reference_date)
version_boost = self.apply_version_boost(metadata)
# Apply version boost to freshness component
adjusted_freshness = freshness_score * version_boost
composite_score = (
self.semantic_weight * semantic_score
+ self.freshness_weight * adjusted_freshness
)
scored.append(ScoredChunk(
chunk_id=metadata.get("chunk_id", ""),
document_id=metadata.get("document_id", ""),
text=text,
semantic_score=semantic_score,
freshness_score=adjusted_freshness,
composite_score=composite_score,
metadata=metadata,
))
return sorted(scored, key=lambda x: x.composite_score, reverse=True)
Let's trace through a concrete example to build intuition. Suppose you're building a policy RAG system. A user asks "What are the overtime rules for hourly employees?" You retrieve three chunks:
- Chunk A: Semantic score 0.91, from an HR policy last effective 2 years ago
- Chunk B: Semantic score 0.87, from the current policy effective 3 months ago
- Chunk C: Semantic score 0.82, from a Slack message summarizing the new policy, posted 2 weeks ago
With pure semantic scoring, Chunk A wins. With a policy domain half-life of 180 days and a 70/30 semantic/freshness split:
- Chunk A freshness:
0.5^(730/180) ≈ 0.059 → composite ≈ 0.7(0.91) + 0.3(0.059) ≈ 0.655
- Chunk B freshness:
0.5^(90/180) ≈ 0.707 → composite ≈ 0.7(0.87) + 0.3(0.707) ≈ 0.821
- Chunk C freshness:
0.5^(14/180) ≈ 0.947 → composite ≈ 0.7(0.82) + 0.3(0.947) ≈ 0.858
The Slack message now ranks first — which raises a different problem. Informal summaries shouldn't necessarily beat authoritative documents. This is where source_system weighting and the is_latest flag on your official policy document become important. You'd add a source authority score and tune accordingly.
Tip: Expose the semantic weight and freshness weight as runtime parameters rather than hardcoding them. Different query types need different balances. A user asking "what's the current policy" should have high freshness weight. A user asking "how has our refund policy evolved over time" needs historical documents and should use low freshness weight or none at all.
Detecting and Resolving Version Conflicts
Freshness weighting alone doesn't fully solve the version conflict problem. If you retrieve the top 5 chunks and three of them are from different versions of the same document, your LLM context will contain contradictory information regardless of how you've ranked them. You need a deduplication and conflict resolution layer between retrieval and generation.
from collections import defaultdict
from typing import Dict, List, Set
class VersionConflictResolver:
"""
Detects and resolves conflicts when multiple versions of the same
document are present in the retrieved set.
"""
def __init__(self, resolution_strategy: str = "latest_wins"):
"""
resolution_strategy:
- "latest_wins": Keep only the highest version_sequence for each document_id
- "flag_and_include": Keep all versions but annotate conflicts for the LLM
- "user_specified": Use the version_sequence closest to a target date
"""
self.resolution_strategy = resolution_strategy
def detect_conflicts(self, chunks: List[ScoredChunk]) -> Dict[str, List[ScoredChunk]]:
"""
Group chunks by document_id to find documents with multiple versions present.
Returns a dict of document_id -> list of chunks (only conflicts, i.e., len > 1).
"""
by_document = defaultdict(list)
for chunk in chunks:
by_document[chunk.document_id].append(chunk)
# Only return groups with actual conflicts (multiple chunks from same document_id
# AND different version_sequences)
conflicts = {}
for doc_id, doc_chunks in by_document.items():
version_sequences = set(
c.metadata.get("version_sequence", 0) for c in doc_chunks
)
if len(version_sequences) > 1:
conflicts[doc_id] = doc_chunks
return conflicts
def resolve(
self,
chunks: List[ScoredChunk],
target_date: Optional[datetime] = None,
) -> List[ScoredChunk]:
"""
Apply conflict resolution and return a clean, deduplicated chunk list.
"""
conflicts = self.detect_conflicts(chunks)
if not conflicts:
return chunks # No conflicts, pass through
conflict_doc_ids: Set[str] = set(conflicts.keys())
resolved_chunks = []
# Pass through all non-conflicted chunks unchanged
for chunk in chunks:
if chunk.document_id not in conflict_doc_ids:
resolved_chunks.append(chunk)
# Apply resolution strategy to conflicted document groups
for doc_id, conflicted in conflicts.items():
if self.resolution_strategy == "latest_wins":
winner = max(
conflicted,
key=lambda c: c.metadata.get("version_sequence", 0)
)
resolved_chunks.append(winner)
elif self.resolution_strategy == "flag_and_include":
# Keep all versions but annotate them
sorted_versions = sorted(
conflicted,
key=lambda c: c.metadata.get("version_sequence", 0),
reverse=True,
)
for i, chunk in enumerate(sorted_versions):
if i == 0:
chunk.text = f"[CURRENT VERSION - v{chunk.metadata.get('document_version', '?')}]\n{chunk.text}"
else:
chunk.text = (
f"[SUPERSEDED VERSION - v{chunk.metadata.get('document_version', '?')}] "
f"This version has been replaced by a newer document.\n{chunk.text}"
)
resolved_chunks.append(chunk)
elif self.resolution_strategy == "user_specified" and target_date:
# Find the version that was current as of target_date
valid_versions = [
c for c in conflicted
if c.metadata.get("effective_date") is not None
and datetime.fromtimestamp(
float(c.metadata["effective_date"]), tz=timezone.utc
) <= target_date
]
if valid_versions:
winner = max(
valid_versions,
key=lambda c: c.metadata.get("effective_date", 0)
)
resolved_chunks.append(winner)
else:
# No version was valid at that date; fall back to earliest
resolved_chunks.append(
min(conflicted, key=lambda c: c.metadata.get("version_sequence", 0))
)
return sorted(resolved_chunks, key=lambda x: x.composite_score, reverse=True)
The flag_and_include strategy deserves special attention. There are query types where you genuinely want multiple versions: "How has our return policy changed this year?" requires historical context. Automatically suppressing older versions would make this query unanswerable. The key is annotating the versions clearly so the LLM understands what it's looking at.
This connects to prompt engineering for RAG — you need to explicitly instruct the model on how to handle version annotations in the context. We'll revisit this in the prompt design section.
Building a Temporal Query Classifier
Not all queries are equally time-sensitive. "What is the capital of France?" doesn't need freshness scoring. "What are the current interest rates for our home equity line of credit?" absolutely does. Building a lightweight classifier that detects temporal intent in queries lets you adjust your retrieval strategy dynamically.
from enum import Enum
import re
from typing import Optional
from datetime import datetime, timezone
import dateparser # pip install dateparser
class TemporalIntent(Enum):
CURRENT = "current" # User wants the most recent information
HISTORICAL = "historical" # User wants information from a specific past time
COMPARATIVE = "comparative" # User wants to compare across time periods
ATEMPORAL = "atemporal" # Query has no meaningful temporal dimension
class TemporalQueryAnalysis:
def __init__(
self,
intent: TemporalIntent,
reference_date: Optional[datetime],
freshness_weight_override: Optional[float],
conflict_strategy: str,
reasoning: str,
):
self.intent = intent
self.reference_date = reference_date
self.freshness_weight_override = freshness_weight_override
self.conflict_strategy = conflict_strategy
self.reasoning = reasoning
class TemporalQueryClassifier:
"""
Analyzes query text to determine temporal intent and configure
retrieval parameters accordingly.
"""
CURRENT_PATTERNS = [
r'\bcurrent(ly)?\b', r'\blatest\b', r'\btoday\b', r'\bright now\b',
r'\bnowadays\b', r'\bpresent(ly)?\b', r'\bas of now\b', r'\bactive\b',
r'\bin effect\b', r'\beffective\b', r'\bup-to-date\b', r'\brecent(ly)?\b',
]
HISTORICAL_PATTERNS = [
r'\blast (year|month|quarter|week)\b',
r'\bin \d{4}\b',
r'\bprevious(ly)?\b', r'\bformerly\b', r'\bused to\b',
r'\boriginal(ly)?\b', r'\bbefore\b', r'\bhistorical(ly)?\b',
r'\bwas\b.*\bpolicy\b', r'\bold(er)?\b.*\bversion\b',
r'\bback in\b', r'\bprior to\b',
]
COMPARATIVE_PATTERNS = [
r'\bchanged?\b', r'\bupdated?\b', r'\bevolved?\b', r'\bdifference\b',
r'\bcompare\b', r'\bvs\b', r'\bversus\b', r'\bbefore and after\b',
r'\bover time\b', r'\bhistory of\b', r'\bprevious version\b',
]
def __init__(self):
self.current_re = [re.compile(p, re.IGNORECASE) for p in self.CURRENT_PATTERNS]
self.historical_re = [re.compile(p, re.IGNORECASE) for p in self.HISTORICAL_PATTERNS]
self.comparative_re = [re.compile(p, re.IGNORECASE) for p in self.COMPARATIVE_PATTERNS]
def _extract_reference_date(self, query: str) -> Optional[datetime]:
"""
Use dateparser to extract explicit date references from query text.
Falls back to None if no parseable date found.
"""
try:
parsed = dateparser.parse(
query,
settings={
"PREFER_DAY_OF_MONTH": "first",
"RETURN_AS_TIMEZONE_AWARE": True,
"PREFER_DATES_FROM": "past",
}
)
# Sanity check: don't accept dates far in the future or before 1970
if parsed and 0 < (datetime.now(timezone.utc) - parsed).days < 36500:
return parsed
except Exception:
pass
return None
def classify(self, query: str) -> TemporalQueryAnalysis:
current_matches = sum(1 for p in self.current_re if p.search(query))
historical_matches = sum(1 for p in self.historical_re if p.search(query))
comparative_matches = sum(1 for p in self.comparative_re if p.search(query))
reference_date = self._extract_reference_date(query)
# Determine dominant intent
scores = {
TemporalIntent.CURRENT: current_matches,
TemporalIntent.HISTORICAL: historical_matches + (2 if reference_date else 0),
TemporalIntent.COMPARATIVE: comparative_matches,
TemporalIntent.ATEMPORAL: 1, # Default baseline
}
intent = max(scores, key=scores.get)
# Configure retrieval based on intent
if intent == TemporalIntent.CURRENT:
return TemporalQueryAnalysis(
intent=intent,
reference_date=datetime.now(timezone.utc),
freshness_weight_override=0.4,
conflict_strategy="latest_wins",
reasoning="Query explicitly asks for current/recent information.",
)
elif intent == TemporalIntent.HISTORICAL:
return TemporalQueryAnalysis(
intent=intent,
reference_date=reference_date,
freshness_weight_override=0.0, # Don't penalize old docs
conflict_strategy="user_specified",
reasoning=f"Query references historical context. Reference date: {reference_date}",
)
elif intent == TemporalIntent.COMPARATIVE:
return TemporalQueryAnalysis(
intent=intent,
reference_date=None,
freshness_weight_override=0.1, # Light freshness, want multiple versions
conflict_strategy="flag_and_include",
reasoning="Query asks for comparison across time; multiple versions needed.",
)
else:
return TemporalQueryAnalysis(
intent=intent,
reference_date=None,
freshness_weight_override=None, # Use domain default
conflict_strategy="latest_wins",
reasoning="No strong temporal signal detected; using default strategy.",
)
This classifier connects naturally to query routing in RAG — temporal intent is one of the dimensions you should be routing on. A historical query might need to hit a separate archive index rather than your live knowledge base.
Note: The regex-based classifier above is fast and interpretable, but it will miss nuanced temporal intent. For production systems with significant traffic, consider adding an LLM-based classification step for queries that score below a confidence threshold — or route all queries through a small, fast model like a fine-tuned BERT classifier trained on examples from your domain. The regex layer acts as a cheap first pass.
Assembling the Full Temporal Retrieval Pipeline
Now let's wire all the pieces together into a cohesive retrieval function that your application can call:
from typing import Optional, List
from datetime import datetime, timezone
class TemporalRAGRetriever:
"""
End-to-end temporal-aware retrieval pipeline.
Orchestrates:
1. Query temporal classification
2. Metadata-filtered vector search
3. Freshness-weighted rescoring
4. Version conflict resolution
5. Context assembly with temporal annotations
"""
def __init__(
self,
vector_store, # Your Pinecone/Weaviate/pgvector client
embedding_model, # Your embedding function
domain: str = "default",
base_freshness_weight: float = 0.3,
top_k_retrieval: int = 20, # Retrieve more, then rerank and trim
top_k_final: int = 5,
):
self.vector_store = vector_store
self.embedding_model = embedding_model
self.classifier = TemporalQueryClassifier()
self.resolver = VersionConflictResolver()
self.top_k_retrieval = top_k_retrieval
self.top_k_final = top_k_final
self.domain = domain
self.base_freshness_weight = base_freshness_weight
def retrieve(
self,
query: str,
force_intent: Optional[TemporalIntent] = None,
metadata_filters: Optional[dict] = None,
) -> dict:
"""
Returns a dict with:
- chunks: List[ScoredChunk] ready for context assembly
- temporal_analysis: TemporalQueryAnalysis used
- conflicts_detected: bool
- retrieval_metadata: operational details for logging
"""
# Step 1: Classify temporal intent
analysis = self.classifier.classify(query)
if force_intent:
analysis.intent = force_intent
# Step 2: Build metadata pre-filters
# Always filter by expiry: don't retrieve expired documents
now_ts = datetime.now(timezone.utc).timestamp()
pre_filters = metadata_filters or {}
if analysis.intent == TemporalIntent.CURRENT:
# For current queries, also filter to is_latest=True
# This halves your search space and improves precision
pre_filters["is_latest"] = True
pre_filters["$or"] = [
{"expiry_date": {"$gt": now_ts}},
{"expiry_date": None},
]
# Step 3: Vector similarity search (over-fetch to allow reranking)
query_embedding = self.embedding_model.embed(query)
raw_results = self.vector_store.query(
vector=query_embedding,
top_k=self.top_k_retrieval,
filter=pre_filters if pre_filters else None,
include_metadata=True,
)
# Step 4: Build ScoredChunk objects from raw results
retrieved = [
(match["text"], match["score"], match["metadata"])
for match in raw_results["matches"]
]
# Step 5: Temporal rescoring
freshness_weight = (
analysis.freshness_weight_override
if analysis.freshness_weight_override is not None
else self.base_freshness_weight
)
scorer = TemporalScorer(
semantic_weight=1.0 - freshness_weight,
freshness_weight=freshness_weight,
domain=self.domain,
)
scored_chunks = scorer.score_chunks(retrieved, reference_date=analysis.reference_date)
# Step 6: Version conflict resolution
conflict_strategy = analysis.conflict_strategy
self.resolver.resolution_strategy = conflict_strategy
resolved_chunks = self.resolver.resolve(
scored_chunks,
target_date=analysis.reference_date
)
conflicts_detected = len(self.resolver.detect_conflicts(scored_chunks)) > 0
# Step 7: Trim to final top-k
final_chunks = resolved_chunks[:self.top_k_final]
return {
"chunks": final_chunks,
"temporal_analysis": analysis,
"conflicts_detected": conflicts_detected,
"retrieval_metadata": {
"raw_retrieved": len(retrieved),
"after_scoring": len(scored_chunks),
"after_resolution": len(resolved_chunks),
"final": len(final_chunks),
"freshness_weight_used": freshness_weight,
"strategy": conflict_strategy,
}
}
Notice the "over-fetch and rerank" pattern: retrieving top_k=20 from the vector store, then trimming to 5 after scoring. This is essential for temporal reranking to have any meaningful effect. If you retrieve only 5 documents, you're letting the embedding similarity entirely pre-select your pool, and temporal rescoring has almost nothing to work with. This connects to ideas explored in contextual compression in RAG — retrieve broadly, then compress and filter.
Temporal Context in Prompt Design
The retrieval layer alone doesn't complete the picture. Your prompt needs to communicate temporal context to the LLM so it can reason appropriately. There are three distinct cases:
def build_temporal_system_prompt(analysis: TemporalQueryAnalysis, conflicts_present: bool) -> str:
base = """You are a knowledgeable assistant with access to internal documents.
Answer questions based only on the provided context. Be precise about dates and versions.
"""
if analysis.intent == TemporalIntent.CURRENT:
temporal_instruction = """TEMPORAL CONTEXT: The retrieved documents are filtered to current,
active versions. Assume information reflects the present state unless a document explicitly
states an effective date in the future. If you see a document marked [CURRENT VERSION],
that is the authoritative source.
"""
elif analysis.intent == TemporalIntent.HISTORICAL:
ref_date_str = analysis.reference_date.strftime("%B %d, %Y") if analysis.reference_date else "a past date"
temporal_instruction = f"""TEMPORAL CONTEXT: The user is asking about the situation as of {ref_date_str}.
Documents marked with version information reflect what was in effect at that time.
Do not use your general knowledge to fill gaps — if the context doesn't cover the period in question,
say so explicitly.
"""
elif analysis.intent == TemporalIntent.COMPARATIVE:
temporal_instruction = """TEMPORAL CONTEXT: You have been provided with multiple versions
of documents, marked [CURRENT VERSION] and [SUPERSEDED VERSION]. When comparing, clearly
state which version you are drawing from. Note the effective dates of each version when relevant.
"""
else:
temporal_instruction = """TEMPORAL CONTEXT: Treat retrieved information as generally current
unless a document explicitly states an effective date or expiry.
"""
conflict_note = ""
if conflicts_present:
conflict_note = """IMPORTANT: Multiple versions of some documents are present in the context.
Always prefer information from the [CURRENT VERSION] unless the user explicitly asks about historical versions.
"""
return base + temporal_instruction + conflict_note
Key insight: The LLM cannot independently determine which of two contradictory passages is more authoritative. If you feed it both "overtime is calculated at 1.5x after 40 hours" from a 2022 policy and "overtime is calculated at 1.5x after 38 hours" from a 2024 policy, it may synthesize "approximately 38-40 hours" or pick one arbitrarily. Your prompt must make the temporal hierarchy explicit.
Monitoring Temporal Drift in Production
Building the retrieval pipeline is half the battle. The other half is catching problems before users do. Temporal drift — the gradual accumulation of stale content in your index — is silent and insidious. Here's a monitoring strategy:
from collections import Counter
from datetime import datetime, timezone, timedelta
import statistics
class TemporalDriftMonitor:
"""
Analyzes the temporal health of your vector index.
Run this on a schedule (daily or weekly) and alert on threshold breaches.
"""
def __init__(self, vector_store, domain: str = "default"):
self.vector_store = vector_store
self.domain = domain
def compute_index_health_report(self) -> dict:
"""
Samples document metadata across the index to compute freshness distribution.
"""
# Fetch metadata for a representative sample
# Implementation varies by vector store; this uses a generic interface
all_metadata = self.vector_store.fetch_all_metadata(sample_size=5000)
now = datetime.now(timezone.utc)
scorer = TemporalScorer(domain=self.domain)
freshness_scores = []
stale_doc_ids = []
expired_doc_ids = []
version_conflicts = Counter()
is_latest_true = 0
is_latest_false = 0
for meta in all_metadata:
# Check for expiry
expiry = meta.get("expiry_date")
if expiry and datetime.fromtimestamp(float(expiry), tz=timezone.utc) < now:
expired_doc_ids.append(meta.get("document_id"))
# Compute freshness
fs = scorer.compute_freshness_score(meta, reference_date=now)
freshness_scores.append(fs)
if fs < 0.2:
stale_doc_ids.append(meta.get("document_id"))
# Version tracking
if meta.get("is_latest"):
is_latest_true += 1
else:
is_latest_false += 1
return {
"sample_size": len(all_metadata),
"mean_freshness": statistics.mean(freshness_scores) if freshness_scores else 0,
"median_freshness": statistics.median(freshness_scores) if freshness_scores else 0,
"p10_freshness": sorted(freshness_scores)[int(len(freshness_scores) * 0.1)] if freshness_scores else 0,
"stale_chunk_count": len(stale_doc_ids),
"stale_pct": len(stale_doc_ids) / len(all_metadata) if all_metadata else 0,
"expired_chunk_count": len(expired_doc_ids),
"superseded_chunk_count": is_latest_false,
"superseded_pct": is_latest_false / (is_latest_true + is_latest_false) if (is_latest_true + is_latest_false) > 0 else 0,
"generated_at": now.isoformat(),
"alerts": self._generate_alerts(
stale_pct=len(stale_doc_ids) / len(all_metadata) if all_metadata else 0,
expired_count=len(expired_doc_ids),
superseded_pct=is_latest_false / max(1, is_latest_true + is_latest_false),
mean_freshness=statistics.mean(freshness_scores) if freshness_scores else 0,
)
}
def _generate_alerts(
self, stale_pct, expired_count, superseded_pct, mean_freshness
) -> list:
alerts = []
if stale_pct > 0.15:
alerts.append({
"severity": "high",
"message": f"{stale_pct:.1%} of index chunks have freshness score below 0.2. Consider re-ingestion."
})
if expired_count > 0:
alerts.append({
"severity": "critical",
"message": f"{expired_count} chunks have passed their expiry_date and should be removed immediately."
})
if superseded_pct > 0.3:
alerts.append({
"severity": "medium",
"message": f"{superseded_pct:.1%} of chunks are marked is_latest=False. High version conflict risk."
})
if mean_freshness < 0.4:
alerts.append({
"severity": "high",
"message": f"Mean freshness score is {mean_freshness:.2f}. Index may be significantly outdated."
})
return alerts
Connect this to your observability stack — emit these metrics to Datadog, Prometheus, or whatever you use. The P10 freshness score (the freshness of the bottom 10% of your index) is particularly valuable because it catches cases where most of your content is fresh but a significant minority is dangerously stale. This integrates with the broader monitoring approach covered in production RAG: caching, monitoring, and continuous improvement.
Hands-On Exercise
Work through this exercise to validate your understanding:
Scenario: You're building a RAG system for a financial services company. The knowledge base contains:
- Loan product terms (updated quarterly)
- Regulatory compliance documents (updated annually)
- FAQ articles (updated ad hoc, some are 3+ years old)
- Press releases (never updated after publication)
Task 1: Schema Design
Design a ChunkMetadata schema appropriate for this domain. What additional fields would you add beyond the baseline schema in this lesson? Consider: How do you handle documents that are temporarily withdrawn and then reinstated? How do you track jurisdictional validity (a product may be available in some states but not others)?
Task 2: Half-Life Configuration
Create a HALF_LIVES configuration for each document type in this corpus. Justify each value. Consider: Loan product terms might change quarterly, but a question about current rates implies maximum freshness sensitivity. How do you differentiate document half-life from query sensitivity?
Task 3: Conflict Scenario
The index contains the following chunks (all for the same mortgage product):
| Chunk |
Version Seq |
Effective Date |
is_latest |
Semantic Score |
| A |
1 |
2022-01-01 |
False |
0.89 |
| B |
2 |
2023-06-01 |
False |
0.85 |
| C |
3 |
2024-01-01 |
True |
0.78 |
A user asks: "What are the current origination fees for your 30-year fixed mortgage?" Trace the execution of your temporal pipeline: what does each stage output? What does the final context look like?
Task 4: Query Classification Edge Cases
Run your TemporalQueryClassifier on these queries and evaluate whether the classification is correct. If not, propose a fix:
- "Has the late payment fee always been $35?"
- "I want to understand the fee structure"
- "What's the best rate I could get today if I have a 780 credit score?"
- "Compare the terms from last year's product to this year's"
Common Mistakes & Troubleshooting
Mistake 1: Using last_modified as your primary freshness signal
Many teams default to whatever timestamp their CMS exports. In most content management systems, last_modified is updated by editors, automations, and sometimes by the CMS itself on migration or re-indexing events. The result is documents that appear "fresh" because someone corrected a typo last week, but whose substantive content is years old. Always seek to extract or maintain an explicit content_date or effective_date.
Mistake 2: Filtering out superseded documents entirely at query time
It's tempting to add is_latest = True as a permanent hard filter on all queries. This breaks historical and comparative queries completely. Reserve this filter for queries the classifier has determined are current-intent, and make it easy to override.
Mistake 3: Setting freshness weight too high
A freshness weight above 0.4-0.5 will start consistently beating semantic relevance. You'll retrieve fresh-but-irrelevant documents instead of relevant-but-older ones. Start at 0.2-0.3 and tune upward from evaluation data. Use evaluating RAG systems: precision, recall, and faithfulness methodologies to measure the impact of your freshness weight on answer quality.
Mistake 4: Not propagating is_latest = False on update
This is a race condition waiting to happen. When you ingest a new document version, you must atomically: (1) set is_latest = False on all existing chunks for that document_id, and (2) index the new chunks with is_latest = True. If these steps are not atomic, you'll have a window where both versions are marked is_latest = True. Implement this as a transactional update with a versioning lock if your vector store supports it, or use a coordinating metadata database (PostgreSQL is a natural fit) to manage version state separately from the vector index.
Warning: Most vector databases do not support true transactional updates. You cannot atomically modify metadata on existing vectors and insert new vectors in a single operation. Mitigate this by maintaining an authoritative version registry in a relational database, using it as the source of truth for is_latest state, and syncing to the vector store asynchronously. See the discussion of indexing strategies for RAG for patterns around index state management.
Mistake 5: Ignoring the distinction between document versions and document editions
A "version" (3.1 → 3.2) typically represents a minor update to the same document. An "edition" might represent a completely different document for a different year (2023 Annual Report vs 2024 Annual Report). These require different handling: version conflicts should be resolved by keeping only the latest. Edition conflicts are not conflicts at all — both documents are simultaneously valid and relevant to different query types. Your document_id scheme must distinguish these cases. Use a stable ID for documents that version, and different IDs for distinct editions.
Mistake 6: Not handling timezone-naive timestamps
The single most common bug in temporal metadata handling. Your source system exports "2024-03-15 14:30:00" — is that UTC? EST? The server's local timezone? Always normalize to UTC at ingestion time and store as timezone-aware Unix timestamps. Validate this with a unit test that ingests a document from a source system in a non-UTC timezone and verifies the stored timestamp is correct.
Tip: If you're ingesting documents from multiple source systems with different timezone conventions, build a SourceSystemTimezoneRegistry that maps each source to its expected timezone offset, and apply the normalization in your ingestion pipeline before the chunk ever reaches the vector store. This belongs in your document ingestion pipeline alongside your other cleaning steps.
Advanced Patterns
Temporal Expansion for Sparse Historical Coverage
When a user asks about a specific historical date and your index has sparse coverage around that date, pure semantic search may return nothing relevant. Consider a temporal expansion strategy: if no documents match within a narrow date window around the target date, progressively widen the window until you find candidates.
def retrieve_with_temporal_expansion(
query: str,
target_date: datetime,
initial_window_days: int = 30,
max_window_days: int = 365,
expansion_factor: float = 2.0,
) -> List[ScoredChunk]:
window = initial_window_days
while window <= max_window_days:
start = target_date - timedelta(days=window)
end = target_date + timedelta(days=window)
results = vector_store.query(
filter={
"effective_date": {"$gte": start.timestamp(), "$lte": end.timestamp()}
}
)
if results and len(results) >= 3:
return results
window = int(window * expansion_factor)
return [] # No coverage found; trigger fallback
Proactive Staleness Alerts in Responses
For high-stakes domains, consider having your pipeline automatically append a staleness warning when the context documents are older than a threshold:
def format_staleness_warning(chunks: List[ScoredChunk], threshold_days: int = 90) -> str:
now = datetime.now(timezone.utc)
old_chunks = [
c for c in chunks
if c.metadata.get("effective_date") and
(now - datetime.fromtimestamp(float(c.metadata["effective_date"]), tz=timezone.utc)).days > threshold_days
]
if not old_chunks:
return ""
oldest = min(old_chunks, key=lambda c: float(c.metadata.get("effective_date", 0)))
age_days = (now - datetime.fromtimestamp(float(oldest.metadata["effective_date"]), tz=timezone.utc)).days
return (
f"\n\n⚠️ **Note**: Some source documents referenced in this answer are "
f"up to {age_days} days old. Please verify with the primary source for time-sensitive decisions."
)
This pattern pairs well with guardrails for RAG pipelines — staleness warnings are one category of output annotation that your policy layer should be able to inject consistently.
Summary & Next Steps
Temporal reasoning in RAG is not a single feature — it's a discipline that touches your ingestion pipeline, your metadata schema, your retrieval scoring, your conflict resolution logic, your prompt design, and your production monitoring. The key architectural principles:
Embed temporal metadata richly at ingestion time. effective_date, expiry_date, version_sequence, and is_latest are non-negotiable for any knowledge base where content changes over time.
Never let the embedding model serve as your temporal filter. Temporal logic belongs in scoring functions and metadata filters, not in semantic similarity space.
Classify query temporal intent and adapt retrieval accordingly. Current, historical, and comparative queries require fundamentally different retrieval configurations.
Resolve version conflicts before context assembly. Multiple versions of the same document in the LLM's context is a recipe for synthesized misinformation.
Monitor your index's temporal health continuously. Stale content accumulates silently; alerts must be automatic.
The natural next topics to tackle from here:
- For multi-step temporal reasoning across complex queries, explore agentic RAG: building self-correcting retrieval pipelines — agents can iterate over temporal search strategies in ways that static pipelines cannot.
- For scenarios where temporal relationships between entities matter (e.g., "what was the regulatory status of this drug compound during the period when it was in clinical trials"), graph RAG gives you explicit time-anchored edges between entities.
- For production systems where temporal retrieval latency is a concern, the techniques in retrieval latency optimization discuss how to cache freshness scores and pre-compute temporal metadata filters without sacrificing accuracy.
