examples/13_migration.py

Rama main · réplica de lectura del repositorio original.

"""Example 13 — one-command migration off the incumbents.

The Trojan-Horse tooling from the strategic plan: read a competitor's
local store and convert it into neuro-symbolic memory, μ=0 re-extraction
on raw text where possible, timestamps preserved.
"""

import datetime as dt
import json
import sqlite3
import tempfile
from pathlib import Path

from cortexm import Memory

TS = dt.datetime(2026, 1, 10, tzinfo=dt.timezone.utc)
td = Path(tempfile.mkdtemp())

# --- build a realistic mem0 store (history table, JSON payloads) -----------
mem0_db = td / "mem0.db"
conn = sqlite3.connect(mem0_db)
conn.execute("CREATE TABLE history (id INTEGER PRIMARY KEY, data TEXT, "
             "created_at TEXT, updated_at TEXT)")
conn.execute("INSERT INTO history(data, created_at, updated_at) VALUES(?,?,?)",
             (json.dumps({
                 "messages": [
                     {"role": "user", "content": "My name is Marco Silva."},
                     {"role": "user",
                      "content": "I work at Shopify as a data engineer."}],
                 "memories": ["Name is Marco Silva"],
             }), "2026-02-01T10:00:00", "2026-02-01T10:00:01"))
conn.execute("INSERT INTO history(data, created_at, updated_at) VALUES(?,?,?)",
             (json.dumps({
                 "messages": [{"role": "user",
                              "content": "I moved to Porto in March."}],
             }), "2026-03-05T09:00:00", "2026-03-05T09:00:01"))
conn.commit()
conn.close()

# --- migrate ----------------------------------------------------------------
from cortexm.migrate.importers import import_mem0

m = Memory()
report = import_mem0(m, str(mem0_db), user_id="marco")
print("migration report:", report)

for q in ("What is Marco's name?", "Where does Marco work?",
          "Where does Marco live?"):
    out = m.search(q, user_id="marco")
    line = out["context_block"].splitlines()
    print(f"{q:28s} ->", line[1][:65] if len(line) > 1 else "(nothing)")

# --- the CLI equivalent ------------------------------------------------------
#   cortexm migrate --from mem0 --path mem0.db --user-id marco
#   cortexm migrate --from zep   --path zep_export.jsonl
#   cortexm migrate --from chroma --path chroma.sqlite3
print("\nCLI: cortexm migrate --from mem0 --path <store>")