Triggering Consolidation

Learn when and how to trigger memory consolidation cycles — scheduling patterns, monitoring, and best practices.

What you'll build

A consolidation strategy for your application — knowing when to trigger sleep cycles, how to merge duplicates, how to monitor effects, and how to schedule consolidation for optimal memory health.

Prerequisites

Requirement	Details
Account	Free tier or above
API key	From your dashboard
Libraries	`requests` (Python) or `fetch` (JavaScript)
Knowledge	Familiarity with Consolidation concepts
Time	~10 minutes

When to trigger consolidation

Scenario	Trigger type	Recommendation
Normal usage	Automatic	Let the engine handle it via scheduled sleep cycles
After bulk import	Manual	Trigger sleep immediately after import completes
High duplication detected	Manual	Preview merge candidates, merge if similarity > 0.85
Before a critical demo	Manual	Run full sleep cycle to optimize memory space
On a schedule	Cron/scheduled	Nightly or weekly for high-volume apps

Steps

Check current status

Query consolidation status and memory stats to see pattern count, last run time, and whether a cycle is already scheduled.

Preview the effects

Run a preview to see how many merge candidates exist and what the pattern count would look like after — without executing.

Trigger consolidation

Execute a sleep cycle (full consolidation) or merge duplicates only, depending on what's needed.

Verify the outcome

Check stats again to confirm improvement in pattern count and storage.

Set up monitoring

Track sleep-stats over time to understand your memory space's health.

Complete code

import os
import requests

BASE_URL = "https://api.engramma-memory.com/v1"
HEADERS = {"X-API-Key": os.environ["ENGRAMMA_API_KEY"]}

# Step 1: Check current status
status = requests.get(f"{BASE_URL}/memory/consolidation/status", headers=HEADERS).json()
print(f"Status: {status['status']}")
print(f"Last run: {status['last_run']}")
print(f"Next scheduled: {status['next_scheduled']}")

stats = requests.get(f"{BASE_URL}/memory/text/stats", headers=HEADERS).json()
print(f"Total patterns: {stats['total_patterns']}")
print(f"Storage: {stats['storage_bytes']} bytes")

# Step 2: Preview what would happen
preview = requests.get(f"{BASE_URL}/memory/consolidation/preview", headers=HEADERS).json()
print(f"\nPreview:")
print(f"  Merge candidates: {preview['merge_candidates']}")
print(f"  Estimated patterns after: {preview['estimated_patterns_after']}")

# Step 3: Decide and execute
if preview['merge_candidates'] > 5:
    print("\nTriggering full sleep consolidation...")
    result = requests.post(
        f"{BASE_URL}/memory/consolidation/sleep",
        headers=HEADERS,
        json={"mode": "full", "protect_categories": ["identity", "health"]}
    ).json()
    print(f"  Status: {result['status']}")
    print(f"  Evicted: {result['evicted']}")
    print(f"  Strengthened: {result['strengthened']}")
    print(f"  Duration: {result['duration_ms']}ms")
else:
    print("\nNo consolidation needed — memory space is healthy.")

# Step 4: Verify
stats_after = requests.get(f"{BASE_URL}/memory/text/stats", headers=HEADERS).json()
print(f"\nAfter consolidation:")
print(f"  Total patterns: {stats_after['total_patterns']}")
print(f"  Storage: {stats_after['storage_bytes']} bytes")

# Wake the system back up
wake = requests.post(f"{BASE_URL}/memory/consolidation/wake", headers=HEADERS).json()
print(f"\nWake status: {wake['status']}")

Merging duplicates

If you only need to deduplicate without running a full sleep cycle, use the merge-duplicates endpoint directly.

import os
import requests

BASE_URL = "https://api.engramma-memory.com/v1"
HEADERS = {"X-API-Key": os.environ["ENGRAMMA_API_KEY"]}

# Inspect merge candidates first
candidates = requests.get(
    f"{BASE_URL}/memory/consolidation/merge-candidates",
    headers=HEADERS
).json()

for c in candidates['candidates']:
    print(f"{c['pattern_a']} <-> {c['pattern_b']} (similarity: {c['similarity']})")

# Merge patterns above a similarity threshold
result = requests.post(
    f"{BASE_URL}/memory/consolidation/merge-duplicates",
    headers=HEADERS,
    json={"threshold": 0.85}
).json()

print(f"Merged: {result['merged']}")
print(f"Patterns: {result['patterns_before']} -> {result['patterns_after']}")

Scheduling patterns

After bulk imports

import os
import requests

BASE_URL = "https://api.engramma-memory.com/v1"
HEADERS = {"X-API-Key": os.environ["ENGRAMMA_API_KEY"]}

# ... after importing documents ...

# Always consolidate after bulk operations
result = requests.post(
    f"{BASE_URL}/memory/consolidation/sleep",
    headers=HEADERS,
    json={"mode": "full", "protect_categories": ["identity"]}
).json()
print(f"Post-import: evicted {result['evicted']}, strengthened {result['strengthened']}")

# Wake when done
requests.post(f"{BASE_URL}/memory/consolidation/wake", headers=HEADERS)

Nightly scheduled consolidation

# Run nightly via cron job or scheduled task
import os
import requests

BASE_URL = "https://api.engramma-memory.com/v1"
HEADERS = {"X-API-Key": os.environ["ENGRAMMA_API_KEY"]}

# Check if consolidation is worthwhile
preview = requests.get(f"{BASE_URL}/memory/consolidation/preview", headers=HEADERS).json()

if preview['merge_candidates'] > 3:
    result = requests.post(
        f"{BASE_URL}/memory/consolidation/sleep",
        headers=HEADERS,
        json={"mode": "full", "protect_categories": ["identity", "health"]}
    ).json()
    print(f"Nightly consolidation complete:")
    print(f"  Evicted: {result['evicted']}")
    print(f"  Strengthened: {result['strengthened']}")
    print(f"  Duration: {result['duration_ms']}ms")
    
    # Wake after consolidation
    requests.post(f"{BASE_URL}/memory/consolidation/wake", headers=HEADERS)
else:
    print(f"Skipped — only {preview['merge_candidates']} merge candidates found")

Threshold-based (in your application)

import os
import requests

BASE_URL = "https://api.engramma-memory.com/v1"
HEADERS = {"X-API-Key": os.environ["ENGRAMMA_API_KEY"]}

# Track stores and consolidate every N operations
store_counter = 0

def store_with_auto_consolidation(text, metadata=None, threshold=100):
    global store_counter
    # ... store your pattern ...
    store_counter += 1

    if store_counter >= threshold:
        requests.post(
            f"{BASE_URL}/memory/consolidation/sleep",
            headers=HEADERS,
            json={"mode": "full", "protect_categories": ["identity"]}
        )
        requests.post(f"{BASE_URL}/memory/consolidation/wake", headers=HEADERS)
        store_counter = 0

Monitoring consolidation health

Use the sleep-stats endpoint to track consolidation history over time:

import os
import requests

BASE_URL = "https://api.engramma-memory.com/v1"
HEADERS = {"X-API-Key": os.environ["ENGRAMMA_API_KEY"]}

# Get sleep cycle history
sleep_stats = requests.get(f"{BASE_URL}/memory/consolidation/sleep-stats", headers=HEADERS).json()
print(f"Total cycles: {sleep_stats['total_cycles']}")
print(f"Last cycle started: {sleep_stats['last_cycle']['started_at']}")
print(f"Last cycle duration: {sleep_stats['last_cycle']['duration_ms']}ms")
print(f"Last cycle evicted: {sleep_stats['last_cycle']['evicted']}")
print(f"Last cycle strengthened: {sleep_stats['last_cycle']['strengthened']}")

# Check importance rankings
importance = requests.get(f"{BASE_URL}/memory/consolidation/importance", headers=HEADERS).json()
print(f"\nImportance rankings: {importance}")

# List protected patterns
protected = requests.get(f"{BASE_URL}/memory/consolidation/protected", headers=HEADERS).json()
print(f"Protected patterns: {protected}")

Key metrics to track over time:

Metric	Source	What to watch
`total_patterns`	`GET /memory/text/stats`	Growing too large — time to consolidate
`merge_candidates`	`GET /memory/consolidation/preview`	High count means duplicates accumulating
`total_cycles`	`GET /memory/consolidation/sleep-stats`	Cycles running regularly
`evicted` / `strengthened`	Sleep cycle result	Healthy ratio means effective consolidation

Warning

Don't trigger sleep cycles after every single store operation. Each cycle takes time (typically 1-5 seconds) and reorganizes your memory space. Let 50-100 new memories accumulate between cycles for best results. Use the preview endpoint to decide if consolidation is worthwhile.

Tip

Use GET /v1/memory/consolidation/preview before triggering a sleep cycle. It's free and tells you how many merge candidates exist and what your pattern count would look like afterward.

Info

Remember to call POST /v1/memory/consolidation/wake after a sleep cycle completes. This re-activates the memory system for normal read/write operations.

Next steps

Webhooks Setup — Get notified when consolidation completes
Consolidation — Deep dive into what happens during a cycle
Regimes — How regime state affects consolidation behavior