Memory OS

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.

TEXT
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 examples

2. Project Hierarchies

Organize knowledge about projects and their components.

TEXT
Project: E-commerce Platform (parent)
  └── Feature: Authentication
      └── Decision: Use JWT tokens
      └── Issue: Token refresh bug
  └── Feature: Shopping Cart
      └── Requirement: Real-time updates
      └── Implementation: WebSocket approach

3. Task Chains

Track multi-step tasks with their individual steps.

TEXT
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.

TEXT
Topic: Machine Learning (parent)
  └── Concept: Supervised Learning
      └── Algorithm: Linear Regression
      └── Algorithm: Decision Trees
  └── Concept: Unsupervised Learning
      └── Algorithm: K-Means
      └── Algorithm: PCA

Creating Hierarchical Memories

Basic Parent-Child Relationship

JavaScript
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}`);
Python
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']}")
Bash
# 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.

JavaScript
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);
Python
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

JavaScript
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`);
Python
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

JavaScript
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);
Python
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.

JavaScript
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);
Python
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:

LevelRecommended TierRationale
Root (projects, topics)LongPersistent anchors
Mid-level (features, phases)MediumSemi-persistent groupings
Leaf (tasks, messages)ShortEphemeral details

2. Metadata Conventions

Use consistent metadata to make hierarchies queryable:

JavaScript
// 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:

JavaScript
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:

JavaScript
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)
  };
}

Ctrl+Shift+C to copy