Engram docs

docs/engram/_includes/check_run_status.py

@@ -0,0 +1,38 @@
+import os
+import time
+import uuid
+from engram import EngramClient
+
+client = EngramClient(
+    api_key=os.environ["ENGRAM_API_KEY"], base_url="https://api.engram.weaviate.io"
+)
+
+test_user_id = f"test-{uuid.uuid4().hex[:8]}"
+
+# Setup: store a memory to get a run_id
+run = client.memories.add(
+    "The user prefers dark mode",
+    user_id=test_user_id,
+    group="default",
+)
+
+# START PollRun
+status = client.runs.wait(run.run_id)
+
+print(status.run_id)
+print(status.status)
+print(status.committed_operations)
+# END PollRun
+
+assert status.status == "completed"
+assert status.committed_operations is not None
+assert len(status.memories_created) >= 1
+
+time.sleep(2)  # Allow tenant indexing to complete
+
+# Cleanup
+_all = client.memories.search(query="dark mode", user_id=test_user_id, group="default")
+for _m in _all:
+    client.memories.delete(_m.id, user_id=test_user_id, group="default")
+
+client.close()

docs/engram/_includes/cleanup_memories.py

@@ -0,0 +1,25 @@
+import os
+from engram import EngramClient
+
+client = EngramClient(
+    api_key=os.environ["ENGRAM_API_KEY"], base_url="https://api.engram.weaviate.io"
+)
+
+user_id = os.environ.get("ENGRAM_USER_ID", "user-uuid")
+group = os.environ.get("ENGRAM_GROUP", "default")
+
+deleted = 0
+for _ in range(20):
+    results = client.memories.search(query="user", user_id=user_id, group=group)
+    if len(results) == 0:
+        break
+    for m in results:
+        try:
+            client.memories.delete(m.id, topic=m.topic, user_id=user_id)
+            deleted += 1
+        except Exception:
+            pass
+
+print(f"Deleted {deleted} memories for user_id={user_id}, group={group}")
+
+client.close()

docs/engram/_includes/manage_memories.py

@@ -0,0 +1,58 @@
+import os
+import uuid
+from engram import EngramClient, APIError
+
+client = EngramClient(
+    api_key=os.environ["ENGRAM_API_KEY"], base_url="https://api.engram.weaviate.io"
+)
+
+test_user_id = f"test-{uuid.uuid4().hex[:8]}"
+
+# Setup: store a memory so we can get and delete it
+run = client.memories.add(
+    "The user prefers dark mode",
+    user_id=test_user_id,
+    group="default",
+)
+status = client.runs.wait(run.run_id)
+assert status.status == "completed"
+assert status.committed_operations is not None
+assert len(status.memories_created) >= 1
+
+memory_id = status.memories_created[0].memory_id
+
+# START GetMemory
+memory = client.memories.get(
+    memory_id,
+    user_id=test_user_id,
+    group="default",
+)
+
+print(memory.content)
+print(memory.topic)
+# END GetMemory
+
+assert memory.id == memory_id
+assert "dark mode" in memory.content
+
+# START DeleteMemory
+client.memories.delete(
+    memory_id,
+    user_id=test_user_id,
+    group="default",
+)
+# END DeleteMemory
+
+# Verify the memory was deleted
+try:
+    client.memories.get(memory_id, user_id=test_user_id, group="default")
+    assert False, "Expected memory to be deleted"
+except APIError:
+    pass  # Memory no longer exists
+
+# Cleanup
+_all = client.memories.search(query="dark mode", user_id=test_user_id, group="default")
+for _m in _all:
+    client.memories.delete(_m.id, user_id=test_user_id, group="default")
+
+client.close()

docs/engram/_includes/quickstart.py

@@ -0,0 +1,60 @@
+import os
+import time
+import uuid
+from engram import EngramClient
+
+# START Connect
+client = EngramClient(
+    api_key=os.environ["ENGRAM_API_KEY"], base_url="https://api.engram.weaviate.io"
+)
+# END Connect
+
+test_user_id = f"test-{uuid.uuid4().hex[:8]}"
+
+# START AddMemory
+run = client.memories.add(
+    "The user prefers dark mode and uses VS Code as their primary editor.",
+    user_id=test_user_id,
+    group="default",
+)
+
+print(run.run_id)
+print(run.status)
+# END AddMemory
+
+assert run.run_id is not None
+
+# START CheckRun
+status = client.runs.wait(run.run_id)
+
+print(status.status)
+# END CheckRun
+
+assert status.status == "completed"
+assert status.committed_operations is not None
+assert len(status.memories_created) >= 1
+
+time.sleep(3)
+
+# START SearchMemory
+results = client.memories.search(
+    query="What editor does the user prefer?",
+    user_id=test_user_id,
+    group="default",
+)
+
+for memory in results:
+    print(memory.content)
+# END SearchMemory
+
+assert len(results) >= 1
+assert any("VS Code" in m.content or "editor" in m.content or "dark mode" in m.content for m in results)
+
+# Cleanup
+for _m in results:
+    try:
+        client.memories.delete(_m.id, user_id=test_user_id, group="default")
+    except Exception:
+        pass
+
+client.close()

docs/engram/_includes/search_memories.py

@@ -0,0 +1,92 @@
+import os
+import uuid
+from engram import EngramClient, RetrievalConfig
+
+client = EngramClient(
+    api_key=os.environ["ENGRAM_API_KEY"], base_url="https://api.engram.weaviate.io"
+)
+
+test_user_id = f"test-{uuid.uuid4().hex[:8]}"
+
+# Setup: store a memory so we have data to search
+run = client.memories.add(
+    "The user prefers dark mode and works primarily in Python. They are building a RAG application.",
+    user_id=test_user_id,
+    group="default",
+)
+status = client.runs.wait(run.run_id)
+assert status.status == "completed"
+assert len(status.memories_created) >= 1
+
+# START BasicSearch
+results = client.memories.search(
+    query="What programming language does the user prefer?",
+    user_id=test_user_id,
+    group="default",
+    retrieval_config=RetrievalConfig(retrieval_type="hybrid", limit=5),
+)
+
+for memory in results:
+    print(memory.content)
+# END BasicSearch
+
+assert len(results) >= 1
+
+# START VectorSearch
+results = client.memories.search(
+    query="What programming language does the user prefer?",
+    user_id=test_user_id,
+    group="default",
+    retrieval_config=RetrievalConfig(retrieval_type="vector", limit=10),
+)
+# END VectorSearch
+
+assert len(results) >= 1
+
+# START BM25Search
+results = client.memories.search(
+    query="What programming language does the user prefer?",
+    user_id=test_user_id,
+    group="default",
+    retrieval_config=RetrievalConfig(retrieval_type="bm25", limit=10),
+)
+# END BM25Search
+
+assert len(results) >= 1
+
+# START HybridSearch
+results = client.memories.search(
+    query="What programming language does the user prefer?",
+    user_id=test_user_id,
+    group="default",
+    retrieval_config=RetrievalConfig(retrieval_type="hybrid", limit=10),
+)
+# END HybridSearch
+
+assert len(results) >= 1
+
+# Discover actual topic name from existing results
+topic = results[0].topic
+
+# START TopicFilter
+results = client.memories.search(
+    query="user preferences",
+    topics=[topic],
+    user_id=test_user_id,
+    group="default",
+    retrieval_config=RetrievalConfig(retrieval_type="hybrid", limit=10),
+)
+
+for memory in results:
+    print(memory.content)
+# END TopicFilter
+
+assert len(results) >= 1
+assert all(m.topic == topic for m in results)
+
+# Cleanup
+_all = client.memories.search(query="user", user_id=test_user_id, group="default")
+for _m in _all:
+    client.memories.delete(_m.id, user_id=test_user_id, group="default")
+
+client.close()

docs/engram/_includes/store_memories.py

@@ -0,0 +1,84 @@
+import os
+import time
+import uuid
+from engram import EngramClient, PreExtractedContent
+
+client = EngramClient(
+    api_key=os.environ["ENGRAM_API_KEY"], base_url="https://api.engram.weaviate.io"
+)
+
+test_user_id = f"test-{uuid.uuid4().hex[:8]}"
+
+# START StoreString
+run = client.memories.add(
+    "The user prefers dark mode and works primarily in Python. They are building a RAG application.",
+    user_id=test_user_id,
+    group="default",
+)
+
+print(run.run_id)
+print(run.status)
+# END StoreString
+
+assert run.run_id is not None
+status = client.runs.wait(run.run_id)
+assert status.status == "completed"
+assert len(status.memories_created) >= 1
+
+time.sleep(3)
+
+results = client.memories.search(query="Python RAG dark mode", user_id=test_user_id, group="default")
+assert len(results) >= 1
+assert any("Python" in m.content or "dark mode" in m.content or "RAG" in m.content for m in results)
+
+# START StorePreExtracted
+run = client.memories.add(
+    PreExtractedContent(
+        content="User prefers dark mode",
+    ),
+    user_id=test_user_id,
+    group="default",
+)
+
+print(run.run_id)
+print(run.status)
+# END StorePreExtracted
+
+assert run.run_id is not None
+status = client.runs.wait(run.run_id)
+assert status.status == "completed"
+
+results = client.memories.search(query="dark mode preference", user_id=test_user_id, group="default")
+assert len(results) >= 1
+assert any("dark mode" in m.content for m in results)
+
+# START StoreConversation
+run = client.memories.add(
+    [
+        {"role": "user", "content": "I just moved to Berlin and I am looking for a good coffee shop."},
+        {"role": "assistant", "content": "Welcome to Berlin! Here are some popular coffee shops in the city..."},
+        {"role": "user", "content": "I prefer specialty coffee, not chains."},
+    ],
+    user_id=test_user_id,
+    conversation_id=str(uuid.uuid4()),
+    group="default",
+)
+
+print(run.run_id)
+print(run.status)
+# END StoreConversation
+
+assert run.run_id is not None
+status = client.runs.wait(run.run_id)
+assert status.status == "completed"
+
+results = client.memories.search(query="Berlin coffee specialty", user_id=test_user_id, group="default")
+assert len(results) >= 1
+assert any("Berlin" in m.content or "coffee" in m.content for m in results)
+
+# Cleanup
+_all = client.memories.search(query="user", user_id=test_user_id, group="default")
+for _m in _all:
+    client.memories.delete(_m.id, user_id=test_user_id, group="default")
+
+client.close()