Hierarchical Memory
Memory OS supports hierarchical relationships between memories through parent-child linking. This enables you to organize memories into structured groups, track conversation threads, build project hierarchies, and maintain contextual relationships.
Overview
Hierarchical memory allows you to:
- Link related memories: Create parent-child relationships between memories
- Build conversation threads: Track multi-turn conversations with context
- Organize projects: Group task-related memories under project memories
- Navigate context: Retrieve memories with their full hierarchical context
Tier Legend:
- 🟢 Long-term (Root): Project-level memories
- 🔵 Medium-term (Children): Feature-level memories
- 🟡 Short-term (Grandchildren): Task-level memories
Use Cases
1. Conversation Threads
Track multi-turn conversations where each message relates to the conversation context.
Conversation (parent)
└── Turn 1: User asks about React hooks
└── Turn 2: Assistant explains useState
└── Turn 3: User asks follow-up about useEffect
└── Turn 4: Assistant provides useEffect examples2. Project Hierarchies
Organize knowledge about projects and their components.
Project: E-commerce Platform (parent)
└── Feature: Authentication
└── Decision: Use JWT tokens
└── Issue: Token refresh bug
└── Feature: Shopping Cart
└── Requirement: Real-time updates
└── Implementation: WebSocket approach3. Task Chains
Track multi-step tasks with their individual steps.
Task: Deploy new version (parent)
└── Step 1: Run tests (completed)
└── Step 2: Build artifacts (completed)
└── Step 3: Deploy to staging (in progress)
└── Step 4: Run smoke tests (pending)4. Knowledge Graphs
Build structured knowledge with relationships.
Topic: Machine Learning (parent)
└── Concept: Supervised Learning
└── Algorithm: Linear Regression
└── Algorithm: Decision Trees
└── Concept: Unsupervised Learning
└── Algorithm: K-Means
└── Algorithm: PCACreating Hierarchical Memories
Basic Parent-Child Relationship
import { MemoryOS } from '@memory-os/sdk';
const memory = new MemoryOS({ apiKey: process.env.MEMORY_OS_API_KEY });
// Create a parent memory
const projectMemory = await memory.memories.create({
content: "E-commerce platform project for retail client. Launch target: Q2 2024.",
tier: "long",
content_type: "document",
memory_nature: "semantic",
metadata: {
type: "project",
project_name: "e-commerce-platform",
status: "active"
}
});
console.log(`Created project: ${projectMemory.id}`);
// Create child memories linked to the parent
const featureMemory = await memory.memories.create({
content: "Authentication feature using OAuth 2.0 with Google and GitHub providers.",
tier: "medium",
content_type: "document",
memory_nature: "semantic",
parent_memory_id: projectMemory.id, // Link to parent
metadata: {
type: "feature",
feature_name: "authentication",
status: "in_progress"
}
});
const cartMemory = await memory.memories.create({
content: "Shopping cart with real-time sync across devices using WebSocket.",
tier: "medium",
content_type: "document",
memory_nature: "semantic",
parent_memory_id: projectMemory.id, // Same parent
metadata: {
type: "feature",
feature_name: "shopping-cart",
status: "planned"
}
});
console.log(`Created features under project ${projectMemory.id}`);import os
from memoryos import MemoryOS
memory = MemoryOS(api_key=os.environ["MEMORY_OS_API_KEY"])
# Create a parent memory
project_memory = memory.memories.create(
content="E-commerce platform project for retail client. Launch target: Q2 2024.",
tier="long",
content_type="document",
memory_nature="semantic",
metadata={
"type": "project",
"project_name": "e-commerce-platform",
"status": "active"
}
)
print(f"Created project: {project_memory['id']}")
# Create child memories linked to the parent
feature_memory = memory.memories.create(
content="Authentication feature using OAuth 2.0 with Google and GitHub providers.",
tier="medium",
content_type="document",
memory_nature="semantic",
parent_memory_id=project_memory["id"], # Link to parent
metadata={
"type": "feature",
"feature_name": "authentication",
"status": "in_progress"
}
)
cart_memory = memory.memories.create(
content="Shopping cart with real-time sync across devices using WebSocket.",
tier="medium",
content_type="document",
memory_nature="semantic",
parent_memory_id=project_memory["id"], # Same parent
metadata={
"type": "feature",
"feature_name": "shopping-cart",
"status": "planned"
}
)
print(f"Created features under project {project_memory['id']}")# Create parent memory
PARENT_ID=$(curl -X POST "https://api.mymemoryos.com/api/v1/memories" \
-H "Authorization: Bearer mos_live_<your_key>" \
-H "Content-Type: application/json" \
-d '{
"content": "E-commerce platform project for retail client.",
"tier": "long",
"content_type": "document",
"memory_nature": "semantic",
"metadata": {
"type": "project",
"project_name": "e-commerce-platform"
}
}' | jq -r '.data.id')
echo "Created project: $PARENT_ID"
# Create child memory linked to parent
curl -X POST "https://api.mymemoryos.com/api/v1/memories" \
-H "Authorization: Bearer mos_live_<your_key>" \
-H "Content-Type: application/json" \
-d "{
\"content\": \"Authentication feature using OAuth 2.0.\",
\"tier\": \"medium\",
\"content_type\": \"document\",
\"parent_memory_id\": \"$PARENT_ID\",
\"metadata\": {
\"type\": \"feature\",
\"feature_name\": \"authentication\"
}
}"Building Deep Hierarchies
Create multi-level hierarchies for complex structures.
import { MemoryOS } from '@memory-os/sdk';
const memory = new MemoryOS({ apiKey: process.env.MEMORY_OS_API_KEY });
async function buildProjectHierarchy() {
// Level 1: Project
const project = await memory.memories.create({
content: "Machine Learning Pipeline Project",
tier: "long",
content_type: "document",
memory_nature: "semantic",
metadata: { level: 1, type: "project" }
});
// Level 2: Phases
const phases = await Promise.all([
memory.memories.create({
content: "Data Preparation Phase",
tier: "medium",
content_type: "document",
parent_memory_id: project.id,
metadata: { level: 2, type: "phase", phase: "data-prep" }
}),
memory.memories.create({
content: "Model Training Phase",
tier: "medium",
content_type: "document",
parent_memory_id: project.id,
metadata: { level: 2, type: "phase", phase: "training" }
}),
memory.memories.create({
content: "Deployment Phase",
tier: "medium",
content_type: "document",
parent_memory_id: project.id,
metadata: { level: 2, type: "phase", phase: "deployment" }
})
]);
// Level 3: Tasks under Data Preparation
const dataTasks = await Promise.all([
memory.memories.create({
content: "Clean and normalize dataset",
tier: "short",
content_type: "event",
parent_memory_id: phases[0].id,
metadata: { level: 3, type: "task", status: "completed" }
}),
memory.memories.create({
content: "Feature engineering",
tier: "short",
content_type: "event",
parent_memory_id: phases[0].id,
metadata: { level: 3, type: "task", status: "in_progress" }
})
]);
return {
project,
phases,
dataTasks
};
}
const hierarchy = await buildProjectHierarchy();
console.log('Hierarchy created:', hierarchy);import os
from typing import Dict, List, Any
from memoryos import MemoryOS
memory = MemoryOS(api_key=os.environ["MEMORY_OS_API_KEY"])
def build_project_hierarchy() -> Dict[str, Any]:
# Level 1: Project
project = memory.memories.create(
content="Machine Learning Pipeline Project",
tier="long",
content_type="document",
memory_nature="semantic",
metadata={"level": 1, "type": "project"}
)
# Level 2: Phases
phases = [
memory.memories.create(
content="Data Preparation Phase",
tier="medium",
content_type="document",
parent_memory_id=project["id"],
metadata={"level": 2, "type": "phase", "phase": "data-prep"}
),
memory.memories.create(
content="Model Training Phase",
tier="medium",
content_type="document",
parent_memory_id=project["id"],
metadata={"level": 2, "type": "phase", "phase": "training"}
),
memory.memories.create(
content="Deployment Phase",
tier="medium",
content_type="document",
parent_memory_id=project["id"],
metadata={"level": 2, "type": "phase", "phase": "deployment"}
)
]
# Level 3: Tasks under Data Preparation
data_tasks = [
memory.memories.create(
content="Clean and normalize dataset",
tier="short",
content_type="event",
parent_memory_id=phases[0]["id"],
metadata={"level": 3, "type": "task", "status": "completed"}
),
memory.memories.create(
content="Feature engineering",
tier="short",
content_type="event",
parent_memory_id=phases[0]["id"],
metadata={"level": 3, "type": "task", "status": "in_progress"}
)
]
return {
"project": project,
"phases": phases,
"data_tasks": data_tasks
}
hierarchy = build_project_hierarchy()
print(f"Hierarchy created: {hierarchy['project']['id']}")Querying Hierarchical Memories
Finding Children of a Memory
import { MemoryOS } from '@memory-os/sdk';
const memory = new MemoryOS({ apiKey: process.env.MEMORY_OS_API_KEY });
async function getChildren(parentId) {
// Search for memories with this parent
const results = await memory.search({
query: `parent:${parentId}`,
limit: 50,
threshold: 0.0 // Get all, regardless of similarity
});
// Filter by parent_memory_id in metadata
const children = results.results.filter(r =>
r.metadata?.parent_memory_id === parentId ||
r.parent_memory_id === parentId
);
return children;
}
async function getDescendants(parentId, depth = 0, maxDepth = 3) {
if (depth >= maxDepth) return [];
const children = await getChildren(parentId);
const descendants = [...children];
// Recursively get grandchildren
for (const child of children) {
const grandchildren = await getDescendants(child.id, depth + 1, maxDepth);
descendants.push(...grandchildren);
}
return descendants;
}
// Get all descendants of a project
const projectId = "550e8400-e29b-41d4-a716-446655440000";
const allDescendants = await getDescendants(projectId);
console.log(`Found ${allDescendants.length} descendants`);from typing import List, Dict, Any
from memoryos import MemoryOS
memory = MemoryOS(api_key=os.environ["MEMORY_OS_API_KEY"])
def get_children(parent_id: str) -> List[Dict[str, Any]]:
"""Get direct children of a memory."""
results = memory.search(
query=f"parent:{parent_id}",
limit=50,
threshold=0.0
)
children = [
r for r in results["results"]
if r.get("parent_memory_id") == parent_id
or r.get("metadata", {}).get("parent_memory_id") == parent_id
]
return children
def get_descendants(
parent_id: str,
depth: int = 0,
max_depth: int = 3
) -> List[Dict[str, Any]]:
"""Recursively get all descendants of a memory."""
if depth >= max_depth:
return []
children = get_children(parent_id)
descendants = list(children)
for child in children:
grandchildren = get_descendants(child["id"], depth + 1, max_depth)
descendants.extend(grandchildren)
return descendants
# Get all descendants of a project
project_id = "550e8400-e29b-41d4-a716-446655440000"
all_descendants = get_descendants(project_id)
print(f"Found {len(all_descendants)} descendants")Building Full Context from Hierarchy
import { MemoryOS } from '@memory-os/sdk';
const memory = new MemoryOS({ apiKey: process.env.MEMORY_OS_API_KEY });
async function getAncestors(memoryId) {
const ancestors = [];
let currentId = memoryId;
while (currentId) {
try {
const mem = await memory.memories.get(currentId);
ancestors.unshift(mem); // Add to beginning
currentId = mem.parent_memory_id;
} catch (error) {
break; // Memory not found or no parent
}
}
return ancestors;
}
async function getFullContext(memoryId) {
// Get ancestors (path from root to this memory)
const ancestors = await getAncestors(memoryId);
// Get siblings (other children of the parent)
const parentId = ancestors[ancestors.length - 2]?.id;
const siblings = parentId ? await getChildren(parentId) : [];
// Get children of this memory
const children = await getChildren(memoryId);
return {
path: ancestors.map(a => ({ id: a.id, content: a.content })),
siblings: siblings.filter(s => s.id !== memoryId),
children
};
}
// Example: Get full context for a task
const taskId = "task-memory-id";
const context = await getFullContext(taskId);
console.log("Path from root:", context.path.map(p => p.content));
console.log("Sibling tasks:", context.siblings.length);
console.log("Subtasks:", context.children.length);from typing import List, Dict, Any, Optional
from memoryos import MemoryOS
from memoryos.exceptions import MemoryOSError
memory = MemoryOS(api_key=os.environ["MEMORY_OS_API_KEY"])
def get_ancestors(memory_id: str) -> List[Dict[str, Any]]:
"""Get all ancestors from root to this memory."""
ancestors = []
current_id = memory_id
while current_id:
try:
mem = memory.memories.get(current_id)
ancestors.insert(0, mem) # Add to beginning
current_id = mem.get("parent_memory_id")
except MemoryOSError:
break
return ancestors
def get_full_context(memory_id: str) -> Dict[str, Any]:
"""Get complete hierarchical context for a memory."""
# Get ancestors
ancestors = get_ancestors(memory_id)
# Get siblings
parent_id = ancestors[-2]["id"] if len(ancestors) >= 2 else None
siblings = get_children(parent_id) if parent_id else []
# Get children
children = get_children(memory_id)
return {
"path": [{"id": a["id"], "content": a["content"]} for a in ancestors],
"siblings": [s for s in siblings if s["id"] != memory_id],
"children": children
}
# Example usage
task_id = "task-memory-id"
context = get_full_context(task_id)
print(f"Path from root: {[p['content'] for p in context['path']]}")
print(f"Sibling tasks: {len(context['siblings'])}")
print(f"Subtasks: {len(context['children'])}")Conversation Thread Example
Build and query conversation threads with hierarchical memory.
import { MemoryOS } from '@memory-os/sdk';
const memory = new MemoryOS({ apiKey: process.env.MEMORY_OS_API_KEY });
class ConversationThread {
constructor(userId) {
this.userId = userId;
this.threadId = null;
}
async startThread(topic) {
// Create the root thread memory
const thread = await memory.memories.create({
content: `Conversation thread: ${topic}`,
tier: 'medium',
content_type: 'conversation',
memory_nature: 'episodic',
metadata: {
user_id: this.userId,
type: 'thread',
topic,
started_at: new Date().toISOString()
}
});
this.threadId = thread.id;
return thread;
}
async addTurn(role, content) {
if (!this.threadId) {
throw new Error('No active thread. Call startThread() first.');
}
const turn = await memory.memories.create({
content: `${role}: ${content}`,
tier: 'short',
content_type: 'conversation',
memory_nature: 'episodic',
parent_memory_id: this.threadId,
metadata: {
user_id: this.userId,
type: 'turn',
role,
turn_number: await this.getTurnCount() + 1,
timestamp: new Date().toISOString()
}
});
return turn;
}
async getTurnCount() {
const children = await this.getThreadHistory();
return children.length;
}
async getThreadHistory() {
if (!this.threadId) return [];
const results = await memory.memories.list({
limit: 100
});
return results.data
.filter(m => m.parent_memory_id === this.threadId)
.sort((a, b) =>
(a.metadata?.turn_number || 0) - (b.metadata?.turn_number || 0)
);
}
async getThreadContext() {
const history = await this.getThreadHistory();
return history
.map(turn => turn.content)
.join('\n\n');
}
async summarizeThread() {
const history = await this.getThreadHistory();
if (history.length > 10) {
// Store summary as a long-term memory
const summaryContent = `Summary of conversation about ${this.threadId}: ${history.length} turns`;
await memory.memories.create({
content: summaryContent,
tier: 'long',
content_type: 'document',
memory_nature: 'semantic',
parent_memory_id: this.threadId,
metadata: {
user_id: this.userId,
type: 'summary',
turns_summarized: history.length
}
});
}
}
}
// Usage
const thread = new ConversationThread('user_123');
await thread.startThread('React Performance Optimization');
await thread.addTurn('user', 'How can I optimize React renders?');
await thread.addTurn('assistant', 'There are several strategies: memo, useMemo, useCallback...');
await thread.addTurn('user', 'Can you explain useMemo in detail?');
await thread.addTurn('assistant', 'useMemo memoizes expensive computations...');
const context = await thread.getThreadContext();
console.log('Thread context:\n', context);import os
from datetime import datetime
from typing import Optional, List, Dict, Any
from memoryos import MemoryOS
memory = MemoryOS(api_key=os.environ["MEMORY_OS_API_KEY"])
class ConversationThread:
def __init__(self, user_id: str):
self.user_id = user_id
self.thread_id: Optional[str] = None
def start_thread(self, topic: str) -> Dict[str, Any]:
"""Create a new conversation thread."""
thread = memory.memories.create(
content=f"Conversation thread: {topic}",
tier="medium",
content_type="conversation",
memory_nature="episodic",
metadata={
"user_id": self.user_id,
"type": "thread",
"topic": topic,
"started_at": datetime.utcnow().isoformat()
}
)
self.thread_id = thread["id"]
return thread
def add_turn(self, role: str, content: str) -> Dict[str, Any]:
"""Add a turn to the conversation."""
if not self.thread_id:
raise ValueError("No active thread. Call start_thread() first.")
turn = memory.memories.create(
content=f"{role}: {content}",
tier="short",
content_type="conversation",
memory_nature="episodic",
parent_memory_id=self.thread_id,
metadata={
"user_id": self.user_id,
"type": "turn",
"role": role,
"turn_number": self.get_turn_count() + 1,
"timestamp": datetime.utcnow().isoformat()
}
)
return turn
def get_turn_count(self) -> int:
"""Get the number of turns in the thread."""
return len(self.get_thread_history())
def get_thread_history(self) -> List[Dict[str, Any]]:
"""Get all turns in the thread."""
if not self.thread_id:
return []
results = memory.memories.list(limit=100)
turns = [
m for m in results["data"]
if m.get("parent_memory_id") == self.thread_id
]
return sorted(
turns,
key=lambda x: x.get("metadata", {}).get("turn_number", 0)
)
def get_thread_context(self) -> str:
"""Get formatted thread context for LLM."""
history = self.get_thread_history()
return "\n\n".join(turn["content"] for turn in history)
def summarize_thread(self):
"""Create a summary for long threads."""
history = self.get_thread_history()
if len(history) > 10:
summary_content = (
f"Summary of conversation about {self.thread_id}: "
f"{len(history)} turns"
)
memory.memories.create(
content=summary_content,
tier="long",
content_type="document",
memory_nature="semantic",
parent_memory_id=self.thread_id,
metadata={
"user_id": self.user_id,
"type": "summary",
"turns_summarized": len(history)
}
)
# Usage
thread = ConversationThread("user_123")
thread.start_thread("React Performance Optimization")
thread.add_turn("user", "How can I optimize React renders?")
thread.add_turn("assistant", "There are several strategies: memo, useMemo, useCallback...")
thread.add_turn("user", "Can you explain useMemo in detail?")
thread.add_turn("assistant", "useMemo memoizes expensive computations...")
context = thread.get_thread_context()
print(f"Thread context:\n{context}")Best Practices
1. Tier Selection for Hierarchies
Choose appropriate tiers based on the hierarchy level:
| Level | Recommended Tier | Rationale |
|---|---|---|
| Root (projects, topics) | Long | Persistent anchors |
| Mid-level (features, phases) | Medium | Semi-persistent groupings |
| Leaf (tasks, messages) | Short | Ephemeral details |
2. Metadata Conventions
Use consistent metadata to make hierarchies queryable:
// Recommended metadata structure
{
type: "project" | "feature" | "task" | "turn",
level: 1 | 2 | 3, // Hierarchy depth
parent_type: "project" | "feature",
status: "active" | "completed" | "archived"
}3. Pruning and Archival
Regularly clean up hierarchies:
async function archiveCompletedProjects(cutoffDays = 30) {
const cutoff = new Date();
cutoff.setDate(cutoff.getDate() - cutoffDays);
const projects = await memory.search({
query: "project completed",
tier: "long",
limit: 100
});
for (const project of projects.results) {
if (project.metadata?.status === "completed" &&
new Date(project.updated_at) < cutoff) {
// Get all descendants
const descendants = await getDescendants(project.id);
// Archive or delete based on your policy
for (const desc of descendants) {
if (desc.tier === "short") {
await memory.memories.delete(desc.id);
}
}
// Update project status
await memory.memories.update(project.id, {
metadata: { ...project.metadata, status: "archived" }
});
}
}
}4. Context Window Optimization
When building LLM context from hierarchies, prioritize appropriately:
async function buildHierarchicalContext(targetId, maxTokens = 2000) {
const ancestors = await getAncestors(targetId);
const siblings = await getSiblings(targetId);
const children = await getChildren(targetId);
// Allocate tokens: 30% ancestors, 20% siblings, 50% target + children
const ancestorBudget = Math.floor(maxTokens * 0.3);
const siblingBudget = Math.floor(maxTokens * 0.2);
const childBudget = Math.floor(maxTokens * 0.5);
return {
path: truncateToTokens(ancestors, ancestorBudget),
context: truncateToTokens(siblings, siblingBudget),
details: truncateToTokens(children, childBudget)
};
}