Treating RAG Chunks as Time-Aware Micro-Memory Stores
Traditional knowledge graphs (RAG) represent facts as unchanging entities (chunks), but in reality, information is dynamic and constantly evolving...
Intro
Traditional knowledge graphs, often leveraged in Retrieval-Augmented Generation (RAG) systems, represent facts as static, unchanging entities — typically in the form of fixed data chunks.
However, in real-world applications, information is inherently dynamic, constantly evolving with new updates, events, or contexts that can render prior data obsolete.
For instance, a company’s revenue figures from last quarter could become obsolete with new reports, leading to potential errors or flawed decisions if the graph fails to account for temporal changes.
Instead of re-running chunking process and replacing / updating chunks, why not create versions of a single chunks to act as micro-memory stores…
Temporal knowledge graphs (TKGs) address this by incorporating time dimensions — such as timestamps for validity periods — into the chunking structure, allowing facts to be associated with specific intervals or events.
This enables precise queries like “What was the CEO on a given date?” or analyses of how relationships and entities have evolved over time, ensuring AI systems or users base decisions on the most current and contextually relevant data.
To augment this, TKGs extend beyond basic static graphs by modelling temporal dynamics through elements like time-stamped triplets (subject-predicate-object with validity ranges), which capture evolving knowledge from sources such as documents or unstructured text.
Key benefits include improved contextual references, forecasting, planning and decision-making in dynamic environments, as they support advanced reasoning over historical patterns and future predictions.
Memory
One of the key building blocks of AI agents is memory.
This component provides agents with context from previous interactions, ensuring continuity and enhancing user experience by avoiding the need to repeat information that can be retrieved from conversation history.
While Retrieval-Augmented Generation (RAG) focuses on delivering up-to-date context during inference — leveraging the in-context learning (ICL) capabilities of language models — it is not primarily designed as a memory system.
Instead, RAG injects prompts with optimal data chunks: not too short to lack sufficient context, nor too long to inflate costs, overhead, or latency.
Temporal AI Agents represent a hybrid approach, blending elements of memory and RAG without fully replacing traditional memory systems.
They enhance RAG by adding contextual history and depth to responses, enabling agents to track how information evolves over time.
In this architecture, chunks — semantically segmented pieces of raw data — are treated as micro-memory stores.
Each chunk is processed into time-stamped triplets (subject-predicate-object structures) and events that capture specific facts or relationships, such as revenue figures or role changes, complete with temporal metadata like valid_at and expired_at.
Stored in a knowledge graph, these triplets serve as compact, contextual memory units that preserve both the content and its temporal relevance.
This design makes them effective micro-memory stores: discrete, structured representations tied to specific time frames, allowing the system to monitor evolving contexts for RAG queries or other applications.
Unlike comprehensive agent memory systems that include episodic or procedural elements, these focus on time-aware facts, creating a lightweight, history-preserving foundation for dynamic, adaptive data handling in real-world scenarios like financial analysis or customer preference tracking.
Practical Code Example
# Install required package
!pip install openai -q
import openai
from getpass import getpass
import json
from datetime import datetime
import uuid
# Prompt for OpenAI API key
api_key = getpass('Enter your OpenAI API key: ')
client = openai.OpenAI(api_key=api_key)
# In-memory knowledge graph: {id: {'subject': ..., 'predicate': ..., 'object': ..., 'valid_at': ..., 'expired_at': None, 'invalidated_by': None}}
knowledge_graph = {}
# Stub chunks
chunk1 = {
'id': 'chunk-1',
'text': 'TechNova Q1 2024: Revenue grew by 15% to $10 million. CEO is Matt Taylor.',
'metadata': {'quarter': 'Q1 2024', 'date': '2024-04-01'}
}
chunk2 = {
'id': 'chunk-2',
'text': 'TechNova Q2 2024: Revenue grew by 20% to $12 million. New CEO is Jane Lee starting July 1, 2024.',
'metadata': {'quarter': 'Q2 2024', 'date': '2024-07-01'}
}
def extraction_agent(chunk):
prompt = f"""Extract subject-predicate-object triplets from this text: {chunk['text']}
Also assign valid_at based on date: {chunk['metadata']['date']}.
Output as JSON list of dicts with keys: subject, predicate, object."""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "system", "content": "You are an extraction agent."},
{"role": "user", "content": prompt}],
response_format={"type": "json_object"}
)
extracted = json.loads(response.choices[0].message.content)['triplets']
# Add temporal metadata
valid_at = datetime.fromisoformat(chunk['metadata']['date']).isoformat()
for trip in extracted:
trip['valid_at'] = valid_at
trip['expired_at'] = None
trip['invalidated_by'] = None
return extracted
def invalidation_agent(new_triplets):
for new_trip in new_triplets:
new_id = str(uuid.uuid4())
knowledge_graph[new_id] = new_trip
# Check for conflicts (simple string match for demo)
for existing_id, existing_trip in list(knowledge_graph.items()):
if (existing_id != new_id and
existing_trip['subject'] == new_trip['subject'] and
existing_trip['predicate'] == new_trip['predicate'] and
existing_trip['object'] != new_trip['object']):
# Mark old as expired
knowledge_graph[existing_id]['expired_at'] = new_trip['valid_at']
knowledge_graph[existing_id]['invalidated_by'] = new_id
print(f"Invalidated {existing_id} by {new_id}")
def retrieval_agent(query_subject):
results = []
current_time = datetime.now().isoformat()
for id, trip in knowledge_graph.items():
if (trip['subject'] == query_subject and
(trip['expired_at'] is None or trip['expired_at'] > current_time)):
results.append(trip)
return results
# Demonstrate Agent Collaboration
# Step 1: Process first chunk
triplets1 = extraction_agent(chunk1)
invalidation_agent(triplets1)
print("After processing chunk1:", knowledge_graph)
# Step 2: Process second chunk (will invalidate old CEO if conflicting)
triplets2 = extraction_agent(chunk2)
invalidation_agent(triplets2)
print("After processing chunk2:", knowledge_graph)
# Step 3: Retrieve current data for TechNova
results = retrieval_agent("TechNova")
print("Current retrieval results:", results)
Chief Evangelist @ Kore.ai | I’m passionate about exploring the intersection of AI and language. From Language Models, AI Agents to Agentic Applications, Development Frameworks & Data-Centric Productivity Tools, I share insights and ideas on how these technologies are shaping the future.
Temporal Agents with Knowledge Graphs | OpenAI Cookbook
Temporal Agents with Knowledge Graphs Executive Summary 1.1. Purpose and Audience 1.2. Key Takeaways How to Use this…cookbook.openai.com
openai-cookbook/examples/partners/temporal_agents_with_knowledge_graphs/temporal_agents_with_knowled…
Examples and guides for using the OpenAI API. Contribute to openai/openai-cookbook development by creating an account…github.com
Temporal AI Agents
The Temporal AI Agent architecture treats chunks as micro-memory stores…cobusgreyling.medium.com
COBUS GREYLING
Where AI Meets Language | Language Models, AI Agents, Agentic Applications, Development Frameworks & Data-Centric…www.cobusgreyling.com