Build an AI Memory Store: User Isolation, Updates and Expiry

Before connecting an AI assistant to persistent memory, test who can retrieve each record, how updates replace old facts, and what happens when information expires or is deleted. This lab implements those storage operations in Qdrant using synthetic data and a small Python program.

Tested October 14, 2026: Qdrant 1.19.2 in a local Linux Docker container and Python 3.12.14 on Windows. Tests verified user separation, updates, scoped deletion, expiry exclusion and persistence after a container restart. This is not a complete chatbot, a VPS benchmark or an authentication system.

Start with a private test database

Start the loopback-only Compose lab in the vector database guide. The code below expects its HTTP API at http://127.0.0.1:16333. Use only disposable records; do not point this exercise at a production database.

The three-number vectors below are deliberately synthetic. They let us create two users with identical similarity scores and verify that the ownership filter, not ranking luck, separates them. They are not semantic embeddings of the text. For real document embeddings, use the separate retrieval lab.

Implement owned records and scoped operations

Put the example in memory.py and run python3 memory.py. Each new demonstration gets its own collection. The same collection name can be passed to MemoryStore later to reopen it without deleting its data.

import json
import re
import time
import uuid
from urllib.error import HTTPError
from urllib.request import Request, urlopen

BASE = "http://127.0.0.1:16333"


def api(method, path, data=None):
    body = None if data is None else json.dumps(data).encode("utf-8")
    request = Request(BASE + path, data=body, method=method,
                      headers={"Content-Type": "application/json"})
    with urlopen(request, timeout=30) as response:
        return json.load(response)


class MemoryStore:
    def __init__(self, collection):
        if not re.fullmatch(r"tips_memory_[a-f0-9]{32}", collection):
            raise ValueError("Use a dedicated demo collection name.")
        self.path = "/collections/" + collection
        try:
            config = api("GET", self.path)["result"]["config"]["params"]["vectors"]
        except HTTPError as error:
            if error.code != 404:
                raise
            api("PUT", self.path, {"vectors": {"size": 3, "distance": "Cosine"}})
        else:
            if config != {"size": 3, "distance": "Cosine"}:
                raise ValueError("Existing collection has an incompatible configuration.")

    def remember(self, owner, key, text, vector, expires_at):
        if not owner or not key or not text or expires_at <= time.time():
            raise ValueError("Owner, key, text and a future expiry are required.")
        point_id = str(uuid.uuid5(uuid.NAMESPACE_URL, json.dumps([owner, key])))
        api("PUT", self.path + "/points?wait=true", {"points": [{
            "id": point_id, "vector": vector,
            "payload": {"owner": owner, "key": key, "text": text,
                        "expires_at": expires_at}
        }]})

    def recall(self, owner, vector):
        if not owner:
            raise ValueError("An authenticated owner is required.")
        result = api("POST", self.path + "/points/query", {
            "query": vector, "limit": 3, "with_payload": True,
            "filter": {"must": [
                {"key": "owner", "match": {"value": owner}},
                {"key": "expires_at", "range": {"gt": time.time()}}
            ]}
        })
        return result["result"]["points"]

    def forget(self, owner, key):
        if not owner or not key:
            raise ValueError("Owner and key are required.")
        api("POST", self.path + "/points/delete?wait=true", {"filter": {"must": [
            {"key": "owner", "match": {"value": owner}},
            {"key": "key", "match": {"value": key}}
        ]}})


if __name__ == "__main__":
    name = "tips_memory_" + uuid.uuid4().hex
    store = MemoryStore(name)
    expiry = time.time() + 3600
    store.remember("alice", "style", "Use brief replies.", [1, 0, 0], expiry)
    store.remember("bob", "style", "Use detailed replies.", [1, 0, 0], expiry)
    found = store.recall("alice", [1, 0, 0])
    assert len(found) == 1 and found[0]["payload"]["owner"] == "alice"
    print("Collection:", name)
    print("Alice's text:", found[0]["payload"]["text"])

The program prints Alice’s text, Use brief replies., even though Bob has an identical vector. A UUID based on owner and key means that updating Alice’s style record replaces that record rather than adding another point. Bob’s record uses a different identifier.

Understand the security boundary

The owner argument must come from your application’s authenticated session. Never let an LLM or a client-supplied form choose it freely. This example is a local storage layer, not an exposed HTTP service; adding an API requires authentication, authorization, input validation and abuse controls.

Both retrieval and deletion include an owner filter. A payload field alone is not access control if another code path can query the database without that filter. Keep database credentials and direct network access away from end users. Consult Qdrant’s multitenancy documentation when designing a larger deployment.

What the tests established

• A query for Alice excluded Bob’s equally similar record, and an unknown owner received no records.
• Writing a new value for Alice’s existing key returned only the new text.
• Deleting Bob’s key left Alice’s record intact.
• A record with an expired timestamp was excluded from retrieval.
• Empty owners and non-demo collection names were rejected.
• Reopening the store after a container restart preserved Alice’s updated record.

These are small functional tests, not proof of production isolation under every failure mode. The expiry filter prevents retrieval but does not remove expired points from disk. Add a deliberate cleanup process and account for caches and backup retention. We did not test disaster recovery in this lab.

Connect a model only after the storage checks pass

Replace the toy vectors with a consistent embedding model and create a separate collection with its correct dimensions and distance metric. Do not mix old and new embeddings merely because their dimensions match. Keep authoritative fields such as billing status in the system of record rather than in a speculative memory summary.

Retrieve only authorized, current records, then build a bounded context with source labels. Preserve real newline separators, not the literal letters n or nn. Treat the retrieved material as untrusted data and do not allow it to grant tool permissions. Similarity scores are not confidence probabilities; evaluate relevance and no-answer cases with your own test questions.

No LLM generation was executed for this memory lab. Adding a local or hosted model requires a separate test of prompt limits, data handling, answer accuracy and tool permissions. The earlier article’s claimed VPS timings, memory usage and generated responses are not supported by this test and have been removed.

Operations checklist

Keep the database on a private network or loopback, protect remote access, restrict credentials and test backups in a separate instance. Monitor collection growth and failures. Do not expose the unauthenticated lab port publicly or schedule automatic writes until ownership, update and deletion behavior are established.

The broader agent memory design guide explains when a structured profile or ordinary database is a better fit. The InterServer MCP Server guide covers a different concern: connecting agents to hosting-account tools.