twin memory

Moving from Mem0.

The same shape: add what your users say, search it later. Your user_id becomes a header, each memory becomes a dated record with its source, and every call has a route.

Mem0
from mem0 import MemoryClient

client = MemoryClient()
client.search("Where does Alice live?", user_id="alice")
twin memory
curl https://api.twinmemory.dev/v1/memory/search \
  -H "Authorization: Bearer $TWIN_MEMORY_API_KEY" \
  -H "X-Twin-Account: alice" \
  -H "Content-Type: application/json" \
  -d '{ "query": "Where does Alice live?" }'

What maps to what

Two differences matter. The end user is named by a header, not a body field, so no request can reach another user's memory. And memory keeps what you added as you sent it, so every record can be traced back to the words it was read from.

Concepts
Mem0twin memoryWhat changes
user_idX-Twin-Account headerSent on every request, never in a body. Each end user is a database file of their own.
agent_idabout on an item, as agent:<id>The tag the console's Import and the docs' script both write. It is stored; nothing filters on it.
app_ida projectA key names your organisation and one project; use a project per app.
run_id, session_idsession on an itemMessages that share a session are read as one thread.
metadata, categoriesmetadata, aboutStored and returned; nothing filters on them. Categories become about tags, as category:name.
expiration_dateexpires on an itemForgotten on time, every copy and version.
a memorya recordRead from what you added, dated, with the address of the passage it came from.
infer=Falseno equivalentItems are always kept as sent, and read once into records.
Calls
Mem0twin memoryWhat changes
add(messages, user_id)POST /v1/memory/add202, searchable a moment later; wait: true waits.
search(query, user_id)POST /v1/memory/searchRecords first, then whole passages.
get_all(user_id)POST /v1/memory/itemsThe items you added, newest first, with id, text, about, metadata and expiry. Paged, up to 200 at a time. Free.
get(memory_id)POST /v1/memory/itemOne item, whole, by the id you gave it; found: false for an id memory does not hold. Free.
update(memory_id, text)POST /v1/memory/add, same idOnly to fix the same statement. When something changed, add a new item: memory keeps both, dated.
delete(memory_id)POST /v1/memory/forgetEvery copy, with a receipt per table.
history(memory_id)POST /v1/memory/history, versionsBy subject, or by item id.

Copy every user's memories

The script reads one user's memories from Mem0 and adds each as a note under the same user id. Ids are prefixed, and an id sent twice is one item, so a run that stops halfway can be run again.

If you kept the raw conversations, send those instead. Memory reads its own records from them, with the date and source of each.

Add reference →
import_mem0.py
# Copy each user's Mem0 memories into twin memory, one note per memory.
import hashlib, os, re
import requests
from mem0 import MemoryClient

mem0 = MemoryClient(api_key=os.environ["MEM0_API_KEY"])
ADD = "https://api.twinmemory.dev/v1/memory/add"
KEY = {"Authorization": f"Bearer {os.environ['TWIN_WRITE_KEY']}"}
FITS = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,34}")  # 35 characters always fits


def account_for(user_id: str) -> str:  # an email does not fit: send a stable digest
    if FITS.fullmatch(user_id):
        return user_id
    return "u_" + hashlib.sha256(user_id.encode()).hexdigest()[:32]


def item_for(m: dict) -> dict:
    about = [f"category:{c}" for c in m.get("categories") or []]
    if m.get("agent_id"):
        about.append(f"agent:{m['agent_id']}")
    item = {
        "id": f"mem0-{m['id']}",  # same id, same item: a rerun is safe
        "kind": "note",
        "text": m["memory"],
        "at": m["created_at"],
        "metadata": {**(m.get("metadata") or {}), "mem0_id": m["id"]},
    }
    if about:
        item["about"] = about[:20]
    if m.get("expiration_date"):
        item["expires"] = m["expiration_date"]
    if m.get("session_id") or m.get("run_id"):
        item["session"] = m.get("session_id") or m.get("run_id")
    return item


def move(user_id: str) -> int:
    page = mem0.get_all(filters={"user_id": user_id})
    memories = page["results"] if isinstance(page, dict) else page
    items = [item_for(m) for m in memories]
    for start in range(0, len(items), 500):  # an add carries up to 500 items
        requests.post(
            ADD,
            headers=KEY | {"X-Twin-Account": account_for(user_id)},
            json={"items": items[start : start + 500]},
            timeout=30,
        ).raise_for_status()
    return len(items)

Import from the console

Or skip the script: in the console, open Import and choose a Mem0 export, a chat JSONL or a CSV. It reads the file in your browser, shows each user and the end user id they will be sent as, then adds them with a result per request. Each memory is mapped as the migration guide maps it.

Migration guide →

Open Import in the console's sidebar, under the project you are moving into.

Start with 10,000 events free.

No card needed. Run the script on a handful of users and compare answers side by side.