Addresses @kartik-mem0's review on mem0#7340, verified against the live staging profiles API on a neuron: - generate_profile / sample_profiles accept a caller-supplied idempotency_key, so retrying a lost request reuses the job instead of creating a second billable one (Python sync+async and TS) - TS uses the uuid dependency instead of the global crypto.randomUUID(), which throws on the supported Node 18 target - update_profile_settings distinguishes an omitted argument from an explicit None, so schema / custom_instructions can be cleared (Python sentinel) - export ProfileJobResponse and ProfileJobStatus; drop the deprecated ProfileTriggerResponse / ProfileSamplesResponse aliases and the unpopulated results field; correct usageUnits -> entityCountReserved; add error to ProfileResponse - openapi: nest schema / custom_instructions under entities in the settings request and response, add capabilities, and mark entity_type required on the job body - docs sample example polls to a terminal job status with a timeout, then reads the create response's entity_ids (status.results raised KeyError) - notebook: include PARTIALLY_SUCCEEDED in terminal states, raise on timeout, and snapshot/restore project settings so a shared env is left as found - tests: real job_id create shape, idempotency-key reuse, and clear-with-None Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
32 KiB
32 KiB
In [ ]:
# This notebook drives the SDK from this worktree, not the published mem0ai:
# the profile fixes below are not released yet.
%pip install -q -e ../..
In [ ]:
import json
import os
import time
import uuid
import mem0
from mem0 import MemoryClient
API_KEY = os.environ.get("MEM0_API_KEY")
if not API_KEY:
import getpass
API_KEY = getpass.getpass("API key: ")
client = MemoryClient(api_key=API_KEY, host=os.environ.get("MEM0_API_HOST") or None)
# Fresh id each run, so nothing below is stale from a previous pass.
USER_ID = f"demo_{uuid.uuid4().hex[:8]}"
# Snapshot the project's profile settings up front. This notebook overwrites the
# shared project schema/instructions/enabled below; the cleanup cell restores this.
ORIGINAL_SETTINGS = client.get_profile_settings()
print("sdk :", mem0.__file__) # must be this worktree
print("host :", client.host)
print("demo user:", USER_ID)In [ ]:
SCHEMA = {
"type": "object",
"properties": {
"occupation": {
"type": "string",
"description": "The person's current job title, in one short phrase.",
},
"location": {
"type": "string",
"description": "The city or region the person currently lives in.",
},
"interests": {
"type": "array",
"items": {"type": "string"},
"description": "Hobbies and topics they return to, as short lowercase tags.",
},
"dietary_restrictions": {
"type": "array",
"items": {"type": "string"},
"description": "Foods the person avoids, and why, if they said.",
},
"communication_style": {
"type": "string",
"enum": ["concise", "detailed", "casual", "formal"],
"description": "How this person prefers to be answered.",
},
"expertise_level": {
"type": "string",
"enum": ["beginner", "intermediate", "advanced"],
"description": "Their technical depth, judged from how they discuss their work.",
},
},
}
settings = client.update_profile_settings(
enabled=True,
schema=SCHEMA,
custom_instructions=(
"Prefer facts the person stated outright over anything inferred. "
"Leave a field empty rather than guessing."
),
)
# Sorted, because JSONB storage does not preserve the key order you sent.
# Compare a stored schema by SET, never by string or by key order.
stored = settings["entities"]["user"]["schema"]
print("schema fields:", sorted(stored["properties"]))
print("enabled :", settings["enabled"])
print("capabilities :", settings["capabilities"])
In [ ]:
TURNS = [
("user", "Hey — I just moved to Berlin for a new job."),
("assistant", "Congratulations! What's the new role?"),
("user", "Senior data engineer at a logistics company. Mostly Spark and Airflow."),
("assistant", "Nice stack. How are you finding the pipelines there?"),
("user", "Honestly the DAGs are a mess. I've been rewriting the partitioning to cut shuffle."),
("assistant", "That usually pays off fast. Anything blocking you?"),
("user", "Just time. Keep it short when you answer me, I skim everything."),
("assistant", "Understood — short answers from here."),
("user", "Outside work I climb most weekends, and I'm learning German."),
("assistant", "Bouldering or ropes?"),
("user", "Bouldering. Also — I'm vegetarian, so skip meat in any recipe suggestions."),
("assistant", "Noted, vegetarian only."),
]
response = client.add(
[{"role": r, "content": c} for r, c in TURNS],
user_id=USER_ID,
)
print(json.dumps(response, indent=2)[:300])
In [ ]:
event_id = response["event_id"]
deadline = time.time() + 300
# 1. The add itself. Terminal status, not a sleep.
event_status = None
while time.time() < deadline:
event_status = client.client.get(f"/v1/event/{event_id}/").json().get("status")
if event_status in ("SUCCEEDED", "FAILED"):
break
print(f" add {event_status}")
time.sleep(5)
print(f"add finished: {event_status}")
assert event_status != "FAILED", "the add itself failed — nothing downstream can work"
# 2. Extraction lands in batches, so the FIRST non-empty page is not the whole set.
# Wait for the count to stop growing instead of breaking on the first result.
memories, stable = [], 0
while time.time() < deadline:
page = client.get_all(filters={"user_id": USER_ID}, page_size=50)
found = page.get("results", []) if isinstance(page, dict) else page
stable = stable + 1 if found and len(found) == len(memories) else 0
memories = found
if stable >= 2: # two identical polls in a row
break
print(f" ... {len(memories)} so far")
time.sleep(5)
if memories:
print(f"\n{len(memories)} memories extracted:\n")
for m in memories:
print(" \u2022", m.get("memory"))
else:
# Say so. Reporting '0 memories' as a result hides a busy or broken environment
# and makes the profile below look like it came from nothing.
print("\nNO memories yet — extraction is still catching up, or the ingestion")
print("worker is down. Everything below will report insufficient_data.")
In [ ]:
envelope = client.get_profile(USER_ID)
print(json.dumps(envelope, indent=2))
print("\nstatus vocabulary:")
print(" succeeded terminal — a generation ran AND the profile has content")
print(" pending queued or running")
print(" failed terminal — the last generation did not complete")
print(" not_enabled feature off, or plan below Pro")
print(" insufficient_data no content to show: no row yet, queued, or a")
print(" generation that legitimately found nothing")
print()
print("`succeeded` is decided by the profile BODY, not by generation_count: an")
print("empty extraction still increments the counter, so counting generations")
print("reports 'done' for a profile with nothing in it.")
In [ ]:
TERMINAL = {"succeeded", "failed", "not_enabled"}
def wait_for_profile(entity_id, timeout=300, interval=5, since=None):
"""Poll until terminal.
`since` waits for a generation_count ABOVE that value, which is how you wait
for an UPDATE rather than accepting the profile you already had.
`insufficient_data` is NOT terminal by itself — it also covers 'queued', so
poll through it and give up on the timeout instead.
"""
deadline = time.time() + timeout
body = None
while time.time() < deadline:
body = client.get_profile(entity_id)
status = (body.get("status") or "").lower()
count = body.get("generation_count") or 0
fresh = count > since if since is not None else True
if status == "succeeded" and fresh:
return body
if status in ("failed", "not_enabled"):
raise RuntimeError(f"generation stopped: {status}")
print(f" ... {status} (generation_count={count})")
time.sleep(interval)
raise TimeoutError(f"not ready in {timeout}s: {body}")
print(json.dumps(client.generate_profile(USER_ID), indent=2))
print("\npolling...")
try:
body = wait_for_profile(USER_ID)
print("\n=== PROFILE ===")
print(json.dumps(body["profile"], indent=2))
print(f"\nstatus={body['status']} generations={body['generation_count']} updated={body['updated_at']}")
except TimeoutError as e:
# Say so plainly and let the rest of the notebook skip, rather than raising
# a NameError in every cell below and burying the real cause.
body = None
print(f"\nNO PROFILE: {e}")
print("Generation never finished. Usually the ingestion worker is down, or")
print("this project has no memories for the user yet.")
In [ ]:
# The model must not invent fields outside your schema — the forced tool call is
# what makes that structural rather than a request.
if body is None:
print("skipped — no profile was generated above")
else:
extra = set(body["profile"]) - set(SCHEMA["properties"])
print("fields outside the schema:", extra or "none")
# A forced JSON-Schema response makes the model emit SOMETHING for every property,
# so 'I found nothing' arrives as a type default: 0, "", [].
filled = {k: v for k, v in body["profile"].items() if v not in (None, "", [], {}, 0)}
print(f"genuinely populated: {len(filled)}/{len(SCHEMA['properties'])} -> {list(filled)}")
In [ ]:
if body is None:
print("skipped — no profile was generated above")
else:
before = body["generation_count"]
MORE_TURNS = [
("user", "Update — I got promoted to staff engineer last week."),
("assistant", "Congratulations. Same team?"),
("user", "Same company, but I'm relocating to Munich for it."),
("assistant", "Big move. How do you feel about it?"),
("user", "Good. I've stopped climbing though — knee injury. Picked up cycling instead."),
("assistant", "Sorry about the knee. Cycling's kinder on it."),
]
followup = client.add(
[{"role": r, "content": c} for r, c in MORE_TURNS],
user_id=USER_ID,
)
# Wait for the add to land before triggering: a generation queued before the new
# memories exist rewrites the profile from the OLD ones and looks like a no-op.
deadline = time.time() + 300
while time.time() < deadline:
status = client.client.get(f"/v1/event/{followup['event_id']}/").json().get("status")
if status in ("SUCCEEDED", "FAILED"):
break
time.sleep(5)
print("follow-up add:", status)
time.sleep(15) # let extraction settle
print(json.dumps(client.generate_profile(USER_ID), indent=2))
print(f"\npolling for a NEW generation (count must exceed {before})...")
updated = wait_for_profile(USER_ID, since=before)
In [ ]:
if body is None:
print("skipped — no profile was generated above")
else:
print(f"{'field':<22} {'before':<34} after")
print("-" * 92)
for field in SCHEMA["properties"]:
b = json.dumps(body["profile"].get(field))
a = json.dumps(updated["profile"].get(field))
mark = " " if a == b else "->"
print(f"{mark} {field:<20} {b[:32]:<34} {a[:32]}")
In [ ]:
def build_system_prompt(entity_id):
result = client.get_profile(entity_id)
if result["status"] != "succeeded":
# Branch on status, never on an empty profile: a user whose profile is
# still building is not a user you know nothing about.
return "You are a helpful assistant."
p = result["profile"]
return f"""You are helping {entity_id}.
Occupation: {p.get("occupation", "unknown")}
Location: {p.get("location", "unknown")}
Interests: {", ".join(p.get("interests", [])) or "unknown"}
Dietary restrictions: {", ".join(p.get("dietary_restrictions", [])) or "none stated"}
Preferred style: {p.get("communication_style", "unknown")}
Match their style and do not explain what they already know."""
print(build_system_prompt(USER_ID))
In [ ]:
# 202, not 200: the sample generations are queued, not finished.
#
# A 409 `already_running` means a sample from an earlier run is still going.
# That is the cooldown working, not an error — reuse that job rather than
# failing the notebook.
try:
job = client.sample_profiles(limit=3)
print(json.dumps(job, indent=2)[:400])
print("\nsampled", job.get("sampled"), "entities:", job.get("entity_ids"))
except Exception as e:
detail = str(e)
print("sample refused:", detail[:200])
running = json.loads(detail).get("error", {}).get("job_id") if detail.startswith("{") else None
job = {"job_id": running, "status_url": f"/v2/profiles/jobs/{running}/"} if running else None
print("reusing the running job:", running)
In [ ]:
JOB_TERMINAL = {"SUCCEEDED", "PARTIALLY_SUCCEEDED", "FAILED", "CANCELLED"}
def wait_for_job(job_response, timeout=300, interval=5):
"""Poll a generation job. Prefer status_url over a bare job id, so a route
change needs no client update. Raise on timeout so an unfinished job is never
mistaken for a finished one."""
handle = job_response.get("status_url") or job_response["job_id"]
deadline = time.time() + timeout
status = None
while time.time() < deadline:
status = client.get_profile_job(handle)["job"]
total = status.get("total")
print(
f" {status['status']} "
f"completed={status.get('completed', 0)}/{total if total is not None else '?'} "
f"succeeded={status.get('succeeded', 0)} "
f"failed={status.get('failed', 0)} "
f"skipped={status.get('skipped', 0)}"
)
if str(status.get("status", "")).upper() in JOB_TERMINAL:
return status
time.sleep(interval)
raise TimeoutError(
f"job not terminal in {timeout}s (last status: {status.get('status') if status else 'none'})"
)
if job is None:
print("no sample job to poll")
else:
final = wait_for_job(job)
print("\n--- what the sample produced ---")
for entity_id in job.get("entity_ids", []):
got = client.get_profile(entity_id)
print(f"\n{entity_id} [{got['status']}]")
print(" ", json.dumps(got["profile"])[:220])In [ ]:
BAD_SCHEMAS = [
({"type": "object", "properties": {}}, "empty — bills you to extract nothing"),
({"type": "array", "items": {"type": "string"}}, "root must be an object"),
(
{
"type": "object",
"properties": {"tone": {"type": "string", "description": "Preferred tone."}},
# At the schema ROOT, which is where the envelope's own keys live.
"_profile_config_version": 1,
"entities": {"user": {}},
},
"reserved settings keys at the schema root",
),
]
for bad, why in BAD_SCHEMAS:
try:
client.update_profile_settings(schema=bad)
print(f"ACCEPTED (unexpected): {why}")
except Exception as e:
print(f"rejected [{why}]:\n {str(e)[:160]}\n")
# NOT rejected: a property with no description. The validator allows it and the
# model then has nothing to go on, so the field comes back empty. Descriptions are
# your job, not the validator's.
try:
client.update_profile_settings(schema={"type": "object", "properties": {"x": {"type": "string"}}})
print("accepted [no description on 'x'] <- the trap: valid to store, useless to generate")
finally:
client.update_profile_settings(schema=SCHEMA) # put the good one back
restored = client.get_profile_settings()["entities"]["user"]["schema"]
assert set(restored["properties"]) == set(SCHEMA["properties"])
print("\nschema restored:", sorted(restored["properties"]))
In [ ]:
from mem0.exceptions import MemoryNotFoundError
try:
client.get_profile("user_who_never_existed")
print("ACCEPTED (unexpected)")
except MemoryNotFoundError as e:
print("404 as intended:", str(e)[:120])
# ...versus a real user who simply has no profile row yet.
fresh = f"demo_never_profiled_{uuid.uuid4().hex[:6]}"
client.add([{"role": "user", "content": "One passing remark."}], user_id=fresh)
time.sleep(5)
print("known but unprofiled:", client.get_profile(fresh)["status"])
In [ ]:
client.update_profile_settings(enabled=False)
print("read (demo user) :", client.get_profile(USER_ID)["status"])
print("read (never profiled) :", client.get_profile(fresh)["status"])
try:
client.generate_profile(USER_ID)
print("trigger: ACCEPTED (unexpected)")
except Exception as e:
print("trigger:", str(e)[:160])
back = client.update_profile_settings(enabled=True) # put it back
print("\nrestored:", back["enabled"])
print("schema survived:", bool(back["entities"]["user"]["schema"]))
In [ ]:
for entity_id in (USER_ID, fresh):
r = client.client.delete(f"/v2/entities/user/{entity_id}/")
print(entity_id, "->", r.status_code)
# Restore the project's profile settings to the start-of-run snapshot, so a shared
# env is left exactly as we found it. Passing the original values (including None)
# clears anything this notebook set — the SDK distinguishes an explicit None from an
# omitted argument.
_user = ORIGINAL_SETTINGS.get("entities", {}).get("user", {})
client.update_profile_settings(
enabled=ORIGINAL_SETTINGS.get("enabled", False),
schema=_user.get("schema"),
custom_instructions=_user.get("custom_instructions"),
)
print("profile settings restored to the pre-notebook snapshot")