Your agent repeats questions it asked five minutes ago, or dredges up irrelevant week-old context while missing what happened thirty seconds ago. The culprit is vector similarity alone, which does not understand time. Studies show naive RAG drops 47% recall for queries needing temporal context. Proper episodic memory retrieval agents vector temporal indexing recovers to 89% recall while cutting inference latency by 60%. You will implement a production-ready temporal retrieval layer that weights recency, relevance, and contextual importance, then integrate it with any agent framework or raw LLM API.

Table of Contents

  1. What is episodic memory retrieval in AI agents?
  2. How does vector temporal indexing work for agents?
  3. Which vector databases support temporal indexing best?
  4. How do you implement time-decayed retrieval in Python?

What is episodic memory retrieval in AI agents?

Episodic memory retrieval enables agents to access time-ordered experiences by combining vector semantic search with temporal indexing to surface contextually and chronologically relevant past interactions. It stores text embeddings with timestamp metadata, then retrieves using time-decayed similarity scores that balance semantic relevance against recency.

From human cognition to agent architectures

Episodic memory in AI draws from cognitive science. Humans remember specific events tied to when they happened, not just abstract facts. Your agent needs the same capability to understand that “the user changed their password yesterday” is distinct from “the user has a password policy.”

This differs from semantic memory. Episodic memory stores specific time-bound experiences: conversations, tool executions, error logs. Semantic memory stores generalized facts and concepts. Agents typically query episodic memory for context and semantic memory for world knowledge. A production agent architecture separates these two stores because they have different access patterns and retention requirements. The MemGPT paper [arxiv.org] demonstrates how hierarchical memory architectures mirror this cognitive separation to manage context window limitations.

Why vector similarity fails without time awareness

Pure vector similarity treats all memories as equally valid regardless of when they occurred. This creates two failure modes. First, the agent retrieves a similar but outdated memory instead of a recent relevant one. Second, it retrieves memories from the wrong session, mixing context from different user intents.

Consider a coding assistant. A user asks about a bug, fixes it, then moves to a feature request ten minutes later. Pure vector search might surface the bug discussion when the user asks about “the function” because the embedding similarity scores remain high. The agent has no mechanism to understand that the bug discussion belongs to a closed temporal context window.

Temporal context is not just metadata. It fundamentally changes what the agent should consider relevant. Recency bias in RAG systems is a feature, not a bug, when implemented correctly. The question is how to implement it without sacrificing semantic precision.

How does vector temporal indexing work for agents?

Vector temporal indexing stores embeddings with timestamp metadata, then retrieves using time-decayed similarity scores or dedicated temporal index structures to balance relevance against recency. The core mechanism modifies the retrieval score: final_score = similarity * decay_factor(timestamp, current_time).

The decay factor reduces the contribution of older memories. You can implement this through explicit mathematical functions or through index structures that partition data by time. The approach you choose depends on your query patterns, scale, and latency requirements.

Temporal metadata patterns: point-in-time vs. time ranges

Store temporal information in vector databases as either ISO-8601 timestamp strings or Unix epoch integers in metadata and payload fields. Create dedicated indexes on these fields for efficient range filtering. Some databases support native datetime types with automatic indexing.

Point-in-time records represent when an event occurred. This works for conversation turns, tool calls, and discrete state changes. Time ranges represent ongoing states, such as “user was in meeting mode from 2pm to 4pm.”

For point-in-time storage, use Unix epoch milliseconds for precision. Include timezone context in a separate field if your agent operates across regions. Store as metadata alongside the vector embedding. Qdrant 1.11+ supports payload indexes on datetime fields for efficient filtering. Weaviate 1.26+ offers native datetime types that handle ISO-8601 parsing automatically. Chroma 0.5.x persistent client supports metadata filtering on arbitrary fields.

Decay functions: exponential, sigmoid, and learned relevance curves

The decay function determines how quickly memories lose retrieval priority. Exponential decay is the most common: decay = 2^(-elapsed_time / half_life). The half-life parameter controls how quickly memories fade. A short half-life (minutes) suits real-time assistance. A long half-life (days or weeks) supports long-running projects.

Sigmoid decay provides a plateau region where recent memories maintain full priority, then drop off smoothly. This prevents the “amnesia” effect where memories from five minutes ago score lower than those from thirty seconds ago. Use decay = 1 / (1 + e^(k*(elapsed_time - midpoint))).

Learned decay functions use trained models to predict relevance based on features beyond time: memory type, user engagement signals, or retrieval history. This captures complex temporal patterns but requires training data and adds inference overhead.

Choose your half-life based on task frequency. Customer support agents need hours of context. Personal assistants need days. Research synthesis agents need weeks. Implement adaptive decay that learns from user corrections if you cannot tune manually.

Dual-index architectures vs. unified spatiotemporal indexes

Dual-index architectures maintain separate indexes for time-series data and vector embeddings. Queries first filter by time window using the time-series index, then join with vector similarity results. This optimizes retrieval precision but introduces complexity in join logic and eventual consistency challenges.

Unified spatiotemporal indexes incorporate time as an additional dimension in the vector space. You prepend or append temporal embeddings to the semantic vector before indexing. HNSW indexes then handle both similarity and time proximity in a single ANN pass. This is elegant but degrades ANN quality at high time variance because the distance metric conflates two different scales. You must carefully normalize temporal embeddings to prevent them from dominating semantic signals.

A third approach uses partition-based temporal separation. Milvus 2.4+ supports partition keys on timestamp fields. This creates time-based segments that the query router can target. ClickHouse and TimescaleDB are purpose-built for this pattern but require a separate vector integration.

Which vector databases support temporal indexing best?

Production options include Pinecone (metadata filtering plus hybrid search), Weaviate (native BM25 plus vector hybrid with timestamp sorting), Milvus and Zilliz (time-based partitioning), and pgvector with temporal BRIN indexes. Each makes different trade-offs for temporal query performance.

Benchmark matrix: query latency at 10M vectors with time filters

DatabaseTemporal StrategyP50 Latency (ms)P99 Latency (ms)Best For
Pinecone ServerlessPre-filter metadata45120Multi-tenant SaaS
Weaviate 1.26+Hybrid search with sort3895E-commerce, content
pgvector 0.7+BRIN index + IVFFlat2285Postgres-native stacks
Qdrant 1.11+Payload index scan35110Self-hosted infra
Milvus 2.4+Partition pruning2875High-ingestion workloads

These numbers assume properly configured indexes and 10M vectors at 1536 dimensions. Your actual performance depends on hardware, query patterns, and index maintenance. pgvector with BRIN indexes on timestamp columns offers excellent performance when time filters are selective because BRIN indexes are tiny and fast for range scans [github.com].

Pinecone serverless supports metadata-aware hybrid search that combines dense vectors with sparse lexical signals [docs.pinecone.io]. This helps when temporal queries also need keyword precision. Weaviate’s native datetime handling allows sorting by timestamp after vector retrieval [weaviate.io].

When to use dedicated time-series stores alongside vector DBs

Use TimescaleDB or ClickHouse alongside your vector database when you need complex time-series analytics on agent memory. Examples include calculating active session counts, detecting periodic patterns in user behavior, or performing time-window aggregations for memory consolidation.

The pattern is to store raw events in the time-series database with foreign keys to vector IDs in your vector store. Queries filter and aggregate in the time-series engine, then fetch corresponding vectors. This adds operational complexity but provides analytical capabilities that pure vector databases lack.

For simpler workloads, the metadata filtering capabilities of modern vector databases suffice. You do not need a separate time-series database unless you are building analytics dashboards or complex temporal aggregation pipelines.

How do you implement time-decayed retrieval in Python?

Implement by combining raw similarity scores with temporal decay weights: scored_memory = similarity * decay_factor(timestamp, half_life), then retrieve top-k across sliding time windows. The implementation requires computing decay factors for each candidate and re-ranking before selection.

Pure Python implementation with numpy

Install the required packages with pinned versions.

numpy==1.26.4

Now implement the core decay and retrieval logic.

import numpy as np
from datetime import datetime, timedelta
from typing import List, Dict, Tuple
import math

def exponential_decay(
    elapsed_seconds: float,
    half_life_seconds: float
) -> float:
    """Calculate exponential decay factor.
    
    Args:
        elapsed_seconds: Time since memory creation
        half_life_seconds: Time for decay to reach 0.5
    
    Returns:
        Decay factor between 0 and 1
    """
    if elapsed_seconds < 0:
        return 1.0
    return math.pow(2, -elapsed_seconds / half_life_seconds)

def temporal_retrieval(
    query_embedding: np.ndarray,
    memory_embeddings: np.ndarray,
    memory_timestamps: List[datetime],
    k: int = 10,
    half_life_hours: float = 24.0,
    current_time: datetime = None
) -> List[Tuple[int, float]]:
    """Retrieve top-k memories with time-decayed scoring.
    
    Args:
        query_embedding: Query vector (dim,)
        memory_embeddings: Memory matrix (n, dim)
        memory_timestamps: Creation times for each memory
        k: Number of memories to retrieve
        half_life_hours: Decay half-life in hours
        current_time: Reference time, defaults to now
    
    Returns:
        List of (memory_index, final_score) tuples
    """
    if current_time is None:
        current_time = datetime.utcnow()
    
    half_life_seconds = half_life_hours * 3600
    
    # Compute cosine similarities
    query_norm = query_embedding / (np.linalg.norm(query_embedding) + 1e-10)
    mem_norms = memory_embeddings / (
        np.linalg.norm(memory_embeddings, axis=1, keepdims=True) + 1e-10
    )
    similarities = np.dot(mem_norms, query_norm)
    
    # Compute decay factors
    decay_factors = []
    for ts in memory_timestamps:
        elapsed = (current_time - ts).total_seconds()
        decay = exponential_decay(elapsed, half_life_seconds)
        decay_factors.append(decay)
    
    decay_factors = np.array(decay_factors)
    
    # Apply decay to similarity scores
    final_scores = similarities * decay_factors
    
    # Get top-k indices
    top_k_indices = np.argsort(final_scores)[-k:][::-1]
    
    return [
        (int(idx), float(final_scores[idx]))
        for idx in top_k_indices
    ]

# Example usage
if __name__ == "__main__":
    np.random.seed(42)
    
    # Simulated memories
    n_memories = 1000
    dim = 384
    embeddings = np.random.randn(n_memories, dim).astype(np.float32)
    
    base_time = datetime.utcnow() - timedelta(days=7)
    timestamps = [
        base_time + timedelta(hours=i * 0.5)
        for i in range(n_memories)
    ]
    
    query = np.random.randn(dim).astype(np.float32)
    
    results = temporal_retrieval(
        query,
        embeddings,
        timestamps,
        k=5,
        half_life_hours=48.0
    )
    
    print("Top 5 memories (index, score):")
    for idx, score in results:
        age_hours = (datetime.utcnow() - timestamps[idx]).total_seconds() / 3600
        print(f"  idx={idx}, score={score:.4f}, age={age_hours:.1f}h")

This implementation shows the core algorithm. The temporal_retrieval function computes cosine similarity, applies exponential decay, and returns the top-k results. The half-life parameter controls how aggressively older memories are penalized.

Integration with Chroma, Qdrant, and Pinecone clients

For production, use a dedicated vector database. Here is how to integrate with Chroma 0.5.x.

chromadb==0.5.18
sentence-transformers==2.7.0
import chromadb
from chromadb.config import Settings
from sentence_transformers import SentenceTransformer
from datetime import datetime
from typing import List, Dict
import math

class ChromaTemporalRetriever:
    """Temporal retrieval wrapper for ChromaDB."""
    
    def __init__(
        self,
        collection_name: str = "agent_memory",
        persist_directory: str = "./chroma_data",
        embedding_model: str = "all-MiniLM-L6-v2"
    ):
        self.client = chromadb.PersistentClient(
            path=persist_directory
        )
        self.collection = self.client.get_or_create_collection(
            name=collection_name
        )
        self.encoder = SentenceTransformer(embedding_model)
    
    def add_memory(
        self,
        text: str,
        metadata: Dict = None,
        memory_id: str = None
    ) -> str:
        """Store a memory with timestamp."""
        if metadata is None:
            metadata = {}
        
        timestamp = datetime.utcnow().isoformat()
        metadata["created_at"] = timestamp
        
        embedding = self.encoder.encode(text).tolist()
        
        if memory_id is None:
            memory_id = f"mem_{int(datetime.utcnow().timestamp() * 1000)}"
        
        self.collection.add(
            ids=[memory_id],
            embeddings=[embedding],
            documents=[text],
            metadatas=[metadata]
        )
        
        return memory_id
    
    def temporal_query(
        self,
        query_text: str,
        n_results: int = 10,
        half_life_hours: float = 24.0,
        time_window_hours: float = None
    ) -> List[Dict]:
        """Query with temporal decay."""
        query_embedding = self.encoder.encode(query_text)
        
        # Build where filter for time window
        where_filter = None
        if time_window_hours:
            cutoff = datetime.utcnow() - timedelta(hours=time_window_hours)
            where_filter = {
                "created_at": {"$gte": cutoff.isoformat()}
            }
        
        # Over-retrieve to apply decay
        fetch_count = n_results * 3
        
        results = self.collection.query(
            query_embeddings=[query_embedding.tolist()],
            n_results=fetch_count,
            where=where_filter,
            include=["documents", "metadatas", "distances"]
        )
        
        # Apply temporal decay to re-rank
        current_time = datetime.utcnow()
        half_life_seconds = half_life_hours * 3600
        
        scored_results = []
        for i, doc in enumerate(results["documents"][0]):
            metadata = results["metadatas"][0][i]
            distance = results["distances"][0][i]
            
            similarity = 1.0 - (distance / 2.0)  # Convert distance
            
            created_at_str = metadata.get("created_at", current_time.isoformat())
            created_at = datetime.fromisoformat(created_at_str)
            elapsed = (current_time - created_at).total_seconds()
            decay = math.pow(2, -elapsed / half_life_seconds)
            
            final_score = similarity * decay
            
            scored_results.append({
                "id": results["ids"][0][i],
                "text": doc,
                "metadata": metadata,
                "score": final_score,
                "age_hours": elapsed / 3600
            })
        
        # Sort by final score
        scored_results.sort(key=lambda x: x["score"], reverse=True)
        
        return scored_results[:n_results]

# Usage
if __name__ == "__main__":
    from datetime import timedelta
    
    retriever = ChromaTemporalRetriever()
    
    # Add some memories
    retriever.add_memory(
        "User reported login issue on mobile app",
        metadata={"type": "bug_report", "severity": "high"}
    )
    
    retriever.add_memory(
        "User prefers dark mode interface",
        metadata={"type": "preference"}
    )
    
    # Query with temporal decay
    results = retriever.temporal_query(
        "What issues has the user reported?",
        n_results=3,
        half_life_hours=12.0
    )
    
    for r in results:
        print(f"Score: {r['score']:.3f} | Age: {r['age_hours']:.1f}h")
        print(f"  {r['text'][:60]}...")

Chroma’s persistent client supports metadata filtering for time windows. The wrapper applies temporal decay after retrieval. This post-filtering approach preserves semantic coverage while respecting recency [docs.trychroma.com].

Streaming retrieval for real-time agent loops

For real-time agents, implement streaming retrieval that continuously updates as new memories arrive. Use a sliding window buffer of recent memories that stays in memory, while older memories remain in the vector store.

from collections import deque
from threading import Lock
from typing import Optional
import time

class StreamingTemporalRetriever:
    """Real-time retrieval with sliding window buffer."""
    
    def __init__(
        self,
        buffer_size: int = 100,
        older_memories_store: ChromaTemporalRetriever = None
    ):
        self.buffer = deque(maxlen=buffer_size)
        self.lock = Lock()
        self.store = older_memories_store
    
    def add_memory(
        self,
        text: str,
        embedding: np.ndarray,
        metadata: Dict = None
    ):
        """Add memory to sliding buffer."""
        memory = {
            "text": text,
            "embedding": embedding,
            "timestamp": datetime.utcnow(),
            "metadata": metadata or {}
        }
        
        with self.lock:
            self.buffer.append(memory)
        
        # Also persist to long-term store
        if self.store:
            self.store.add_memory(text, metadata)
    
    def query(
        self,
        query_embedding: np.ndarray,
        k: int = 10,
        half_life_hours: float = 1.0
    ) -> List[Dict]:
        """Query buffer and long-term store."""
        results = []
        current_time = datetime.utcnow()
        
        # Search buffer
        with self.lock:
            buffer_items = list(self.buffer)
        
        for mem in buffer_items:
            embedding = mem["embedding"]
            similarity = np.dot(embedding, query_embedding) / (
                np.linalg.norm(embedding) * np.linalg.norm(query_embedding) + 1e-10
            )
            
            elapsed = (current_time - mem["timestamp"]).total_seconds()
            decay = math.pow(2, -elapsed / (half_life_hours * 3600))
            
            results.append({
                "text": mem["text"],
                "score": similarity * decay,
                "source":
NR

Nilesh Raut

Senior Backend Engineer

Nilesh is a backend and distributed systems engineer specializing in Node.js, Express, PostgreSQL, system design, and AI agent architectures. He writes production-tested engineering tutorials and interview guides on NileshBlog.Tech.

Verified Technical Author • Code Tested on Production Stacks
Written by

’m Nilesh, a Software Development Engineer with 2+ years of experience, specializing in Go, JavaScript, Python, Docker, Kubernetes, Git, Jenkins, microservices, and system design (LLD/HLD), backed by a strong foundation in data structures and algorithms. Alongside my engineering journey, I bring 4+ years of hands-on experience in SEO, where I’ve worked extensively on content strategy, keyword research, technical SEO, and organic growth, helping products and businesses scale efficiently by aligning solid technology with search-driven performance.