How to give a personal AI assistant memory across sessions

Building a personal AI assistant long-term memory system is not about appending every chat log to a vector database and hoping retrieval magic fixes context. It demands a typed memory schema, disciplined write-back after each session, and a retrieval path that respects recency and contradiction. This guide gives you an end-to-end implementation you can run on a laptop today.

Step 1: Choose a durable storage backend

The foundation of any personal AI assistant long-term memory is a store that survives process restarts and OS updates. SQLite is enough for a single-user assistant. It needs no server, handles thousands of records with sub-millisecond indexed lookups, and ships with every Python install.

Enable WAL mode so writes don’t block reads during a session:

import sqlite3

conn = sqlite3.connect("assistant_memory.db")
conn.execute("PRAGMA journal_mode=WAL;")
conn.execute("""
CREATE TABLE IF NOT EXISTS memories (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    type TEXT NOT NULL,
    content TEXT NOT NULL,
    embedding BLOB,
    created_at REAL NOT NULL,
    last_accessed REAL NOT NULL,
    confidence REAL DEFAULT 1.0
);
""")
conn.execute("CREATE INDEX IF NOT EXISTS idx_type ON memories(type);")
conn.commit()

For multi-device sync, swap SQLite for Postgres or Redis, but keep the column shape identical so your retrieval code never changes.

Step 2: Model memory as typed records

Raw text blobs become unmanageable after a week of conversations. Define three types up front: fact (user is vegetarian), preference (wants bullet-point answers), event (attended conference on June 1). Typing lets you apply different decay rates and retrieval weights later.

MEMORY_TYPES = {"fact", "preference", "event"}

def add_memory(conn, type: str, content: str, embedding: bytes, now: float):
    assert type in MEMORY_TYPES, f"unknown type {type}"
    conn.execute(
        "INSERT INTO memories (type, content, embedding, created_at, last_accessed) VALUES (?,?,?,?,?)",
        (type, content, embedding, now, now),
    )
    conn.commit()

Confidence starts at 1.0. You will discount it when newer contradictory evidence arrives. Provenance is implicit: the created_at timestamp and type tell you whether the assistant inferred something or the user stated it directly.

Step 3: Embed memories for semantic retrieval

Use an OpenAI-compatible embedding endpoint. The snippet below targets any such gateway; if you route through n4n.ai, you get automatic fallback when a provider is rate-limited or degraded, so memory writes never block on a single vendor outage.

import openai, struct

client = openai.OpenAI(base_url="https://api.n4n.ai/v1", api_key="YOUR_KEY")

def embed(text: str) -> bytes:
    resp = client.embeddings.create(model="text-embedding-3-small", input=text)
    vec = resp.data[0].embedding
    return struct.pack(f"{len(vec)}f", *vec)

Store the packed float buffer directly in the embedding column. For retrieval, unpack and compute cosine similarity in Python. At personal-assistant scale (<10k rows) this runs in microseconds—no need for a dedicated vector index yet.

If you prefer zero network calls, sentence-transformers with all-MiniLM-L6-v2 produces a 384-dim vector locally. Swap the embed function; the rest of the pipeline is unchanged.

Step 4: Retrieve relevant context at session start

Load candidate memories, score by cosine similarity, and boost by recency. This is where personal AI assistant long-term memory becomes useful: you surface the right fact without flooding the prompt with irrelevant history.

import math, time, struct

def cosine(a, b):
    dot = sum(x*y for x,y in zip(a,b))
    na = math.sqrt(sum(x*x for x in a))
    nb = math.sqrt(sum(y*y for y in b))
    return dot/(na*nb + 1e-9)

def retrieve(conn, query_vec, top_k=5, now=None):
    now = now or time.time()
    rows = conn.execute(
        "SELECT id, type, content, embedding, last_accessed FROM memories"
    ).fetchall()
    scored = []
    for id_, type_, content, emb, accessed in rows:
        vec = struct.unpack(f"{len(emb)//4}f", emb)
        sim = cosine(query_vec, vec)
        recency = 1.0 / (1.0 + (now - accessed)/86400.0)
        score = 0.8*sim + 0.2*recency
        scored.append((score, id_, type_, content))
    scored.sort(reverse=True)
    best = scored[:top_k]
    for score, id_, _, _ in best:
        conn.execute("UPDATE memories SET last_accessed=? WHERE id=?", (now, id_))
    conn.commit()
    return best

For hybrid retrieval, add a LIKE filter on content for proper nouns before the vector scan. This catches “Berlin” even if the embedding drifted.

Step 5: Inject memories into the system prompt

Do not concatenate raw rows into the user message. Build a separate system block with explicit tags so the model knows what is a user-stated fact versus an inferred preference.

def build_system_prompt(memories):
    lines = ["You are a personal assistant. Known context:"]
    for _, _, type_, content in memories:
        lines.append(f"- [{type_}] {content}")
    lines.append("Use this context only when relevant. Never invent details beyond it.")
    return "\n".join(lines)

Token budget matters. With a 5-row cap and ~20 tokens per row, you spend under 150 tokens on memory—cheap insurance against repeating questions the user already answered.

Step 6: Extract and write new memories after each session

After the session ends, ask the model to extract durable items from the transcript. Constrain output to JSON matching your schema. This closes the loop for personal AI assistant long-term memory.

def extract_memories(client, transcript: str):
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "Extract durable facts, preferences, or events from the conversation. Return JSON: {\"memories\": [{\"type\": \"fact|preference|event\", \"content\": \"...\"}]}"},
            {"role": "user", "content": transcript}
        ],
        response_format={"type": "json_object"}
    )
    import json
    data = json.loads(resp.choices[0].message.content)
    return data.get("memories", [])

Persist each item with its embedding in a single batched transaction. Avoid extracting ephemeral chatter (“nice weather”) by prompting for “durable” explicitly and reviewing a sample of extractions weekly.

Step 7: Resolve contradictions and apply decay

When a new fact conflicts with an existing one (same type, embedding distance < 0.1), lower the old row’s confidence instead of deleting it. A daily job decays confidence by 1% per day; rows below 0.3 get archived.

def decay(conn, now):
    conn.execute(
        "UPDATE memories SET confidence = confidence * 0.99 WHERE created_at < ?",
        (now - 86400,)
    )
    conn.execute("DELETE FROM memories WHERE confidence < 0.3")
    conn.commit()

def merge_conflicts(conn, new_embedding, new_type, threshold=0.1):
    rows = conn.execute("SELECT id, embedding FROM memories WHERE type=?", (new_type,)).fetchall()
    for id_, emb in rows:
        vec = struct.unpack(f"{len(emb)//4}f", emb)
        if cosine(vec, new_embedding) > 1 - threshold:
            conn.execute("UPDATE memories SET confidence = confidence * 0.5 WHERE id=?", (id_,))
    conn.commit()

This keeps the personal AI assistant long-term memory honest when the user changes their mind (“I eat fish now”) without silently overwriting history.

Step 8: Verify the pipeline end-to-end

Write a small script that exercises the full path:

1. Delete assistant_memory.db to start clean.
2. Embed and store: “User is allergic to peanuts” as type fact.
3. Close the connection and process.
4. Reopen, embed query “what food restrictions should I respect”, call retrieve.
5. Assert the peanut row is in the top 3.
6. Send a chat request with the built system prompt asking “can I eat pad thai?” — the response should mention peanut risk.

If retrieve returns the memory and the model’s answer reflects it, the system works. Automate this with pytest and run it in CI so schema changes break the build.

Operational notes

• Keep embeddings and content in the same row to avoid join overhead on every retrieval.
• Batch embedding calls: collect session extractions and send them as a single input list to the embeddings endpoint.
• For a personal AI assistant long-term memory deployment on mobile, compile the SQLite file into app storage and run the same logic via a thin native layer.
• Log every write with a session ID. When the assistant gives a wrong answer, you can replay which memories were injected.

The architecture above is deliberately boring. Boring stores survive; clever vector hacks don’t. Build the write path first, then tune retrieval weights once you have real usage data.