Personal Assistant
intermediateBuild an AI assistant that learns user preferences, routines, and habits over time — and gets smarter with each interaction.
What you'll build
An AI assistant that learns your preferences, remembers your routines, and adapts its behavior over time — without explicit configuration or re-training.
Prerequisites
| Requirement | Details |
|---|---|
| Account | Free tier or above |
| API key | From your dashboard |
| Language | Python (requests) or JavaScript (fetch) |
| Time | ~20 minutes |
The difference from a chatbot
A chatbot stores and recalls conversations. A personal assistant goes further:
| Feature | Chatbot | Personal Assistant |
|---|---|---|
| Remembers what was said | Yes | Yes |
| Learns preferences | Basic | Adaptive |
| Detects patterns | No | Yes (via consolidation) |
| Improves over time | Static | Self-organizes |
Steps
Store observations
Retrieve with context
Track feedback
Run consolidation
Review important memories
Complete code
import os
import requests
BASE_URL = "https://api.engramma-memory.com"
API_KEY = os.environ["ENGRAMMA_API_KEY"]
HEADERS = {
"X-API-Key": API_KEY,
"Content-Type": "application/json"
}
class PersonalAssistant:
def __init__(self, user_id: str):
self.user_id = user_id
def learn(self, observation: str, category: str = "preference"):
"""Store something learned about the user."""
resp = requests.post(
f"{BASE_URL}/v1/memory/text/store",
headers=HEADERS,
json={
"text": observation,
"metadata": {
"user_id": self.user_id,
"category": category,
"protected": category in ("preference", "fact", "feedback")
}
}
)
data = resp.json()
return data["pattern_id"]
def recall(self, context: str, top_k: int = 5):
"""Recall relevant knowledge about the user using Active Inference."""
resp = requests.post(
f"{BASE_URL}/v1/memory/text/recall",
headers=HEADERS,
json={"query": context, "top_k": top_k}
)
data = resp.json()
return [
{
"text": r["text"],
"similarity": r["similarity"],
"category": r["metadata"].get("category"),
"pattern_id": r["pattern_id"]
}
for r in data["results"]
if r["similarity"] > 0.5
]
def record_feedback(self, original: str, correction: str):
"""Store user feedback to update preferences."""
requests.post(
f"{BASE_URL}/v1/memory/text/store",
headers=HEADERS,
json={
"text": f"Correction: User said '{correction}' when assistant suggested '{original}'",
"metadata": {
"user_id": self.user_id,
"category": "feedback",
"protected": True
}
}
)
def get_important(self):
"""Retrieve the user's important protected memories."""
resp = requests.get(
f"{BASE_URL}/v1/memory/text/important",
headers=HEADERS
)
data = resp.json()
return [
{"text": m["text"], "category": m["category"], "protected": m["protected"]}
for m in data["important"]
]
def consolidate(self):
"""Run a consolidation sleep cycle to strengthen and prune memories."""
resp = requests.post(
f"{BASE_URL}/v1/memory/consolidation/sleep",
headers=HEADERS,
json={
"mode": "full",
"protect_categories": ["identity", "preference", "feedback"]
}
)
data = resp.json()
return {
"evicted": data["evicted"],
"strengthened": data["strengthened"],
"duration_ms": data["duration_ms"]
}
# Usage
assistant = PersonalAssistant("user_456")
# The assistant learns over multiple interactions
assistant.learn("User wakes up at 6:30am on weekdays", "routine")
assistant.learn("User prefers Python over JavaScript", "preference")
assistant.learn("User drinks oat milk lattes", "preference")
assistant.learn("User has standup at 9am every Monday", "routine")
# Later — assistant recalls relevant context
context = assistant.recall("morning schedule")
print(context)
# [{'text': 'User wakes up at 6:30am on weekdays', 'similarity': 0.89, 'category': 'routine', 'pattern_id': 'pat_...'},
# {'text': 'User has standup at 9am every Monday', 'similarity': 0.82, 'category': 'routine', 'pattern_id': 'pat_...'}]
# User corrects the assistant
assistant.record_feedback(
original="You usually have coffee at 7am",
correction="I actually switched to tea last month"
)
assistant.learn("User drinks tea in the morning (switched from coffee)", "preference")
# Run consolidation to strengthen important memories
result = assistant.consolidate()
print(result)
# {'evicted': 3, 'strengthened': 12, 'duration_ms': 2340}
# Check what the system considers important
important = assistant.get_important()
print(important)
# [{'text': 'User prefers Python over JavaScript', 'category': 'preference', 'protected': True}, ...]How preferences evolve
Engramma's consolidation cycle makes your assistant smarter over time:
| Time | What happens |
|---|---|
| Day 1 | Store raw observations: "User likes dark mode" |
| Day 7 | Consolidation strengthens repeated patterns: multiple "dark mode" mentions reinforce that memory |
| Day 14 | Weak or unprotected memories get evicted during sleep cycles |
| Day 30 | Protected feedback corrections persist; outdated unprotected observations fade naturally |
Organizing knowledge categories
Use metadata categories to structure what the assistant knows:
| Category | Examples | Protected? |
|---|---|---|
preference | "Prefers Python", "Likes dark mode" | Yes |
routine | "Standup at 9am Monday", "Gym on Wednesdays" | No |
fact | "Works at Acme Corp", "Team size is 5" | Yes |
feedback | "Correction: tea not coffee" | Yes |
observation | "Seemed stressed today", "Asked about vacations" | No |
Protect preferences, facts, and feedback — they represent core knowledge. Leave routines and observations unprotected so consolidation can evict outdated ones naturally. Use protect_categories in the sleep endpoint to safeguard entire categories during consolidation.
Using recall for smarter retrieval
The /v1/memory/text/recall endpoint uses Active Inference with semantic re-ranking, producing higher-quality results than basic /v1/memory/text/retrieve. Use recall when you need the assistant to surface contextually relevant memories:
# Basic retrieve — pure semantic similarity
resp = requests.post(
f"{BASE_URL}/v1/memory/text/retrieve",
headers=HEADERS,
json={"query": "What does the user need on Monday morning?", "top_k": 5}
)
retrieve_results = resp.json()
print(f"Retrieve latency: {retrieve_results['latency_ms']}ms")
# Active Inference recall — semantic + re-ranking
resp = requests.post(
f"{BASE_URL}/v1/memory/text/recall",
headers=HEADERS,
json={"query": "What does the user need on Monday morning?", "top_k": 5}
)
recall_results = resp.json()
print(f"Recall latency: {recall_results['latency_ms']}ms")
# Recall surfaces more contextually relevant results
for r in recall_results["results"]:
print(f"{r['text']} (similarity: {r['similarity']})")
# User has standup at 9am every Monday (similarity: 0.91)
# User wakes up at 6:30am on weekdays (similarity: 0.84)
# User drinks tea in the morning (similarity: 0.72)Retrieve vs Recall: Use /v1/memory/text/retrieve for fast, low-latency lookups (~2ms). Use /v1/memory/text/recall when you need Active Inference re-ranking for higher relevance (~8ms). Both return results with a similarity score.
Next steps
- Knowledge Base — Build a self-organizing knowledge system
- Consolidation — How memories strengthen and prune over time