Jarvis-like Memory for OpenClaw

#!/usr/bin/env python3 """ Store conversation turns to Qdrant (Mem0-style). Each turn creates 3 embeddings: user message, AI response, summary. """ import argparse, hashlib, json, os, sys, urllib.request, uuid from datetime import datetime from typing import List, Optional, Dict, Any QDRANT_URL = "http://localhost:6333" COLLECTION_NAME = "openclaw_memories" OLLAMA_URL = "http://localhost:11434/v1" def get_embedding(text: str) -> Optional[List[float]]: """Generate embedding using snowflake-arctic-embed2""" data = json.dumps({ "model": "snowflake-arctic-embed2", "input": text[:8192] }).encode() req = urllib.request.Request( f"{OLLAMA_URL}/embeddings", data=data, headers={"Content-Type": "application/json"} ) try: with urllib.request.urlopen(req, timeout=30) as response: result = json.loads(response.read().decode()) return result["data"][0]["embedding"] # Ollama format except Exception as e: print(f"Embedding error: {e}", file=sys.stderr) return None

def store_memory_point(user_id, text, speaker, date_str, conversation_id, turn_number, tags): """Store a single memory point to Qdrant""" embedding = get_embedding(text) if not embedding: return None point_id = str(uuid.uuid4()) payload = { "user_id": user_id, # Mem0-style: persistent user ID "text": text, "date": date_str, "tags": tags, "source_type": speaker, # "user" or "assistant" "conversation_id": conversation_id, "turn_number": turn_number, "created_at": datetime.now().isoformat() } # Qdrant upsert format upsert_data = { "points": [{ "id": point_id, "vector": embedding, "payload": payload }] } req = urllib.request.Request( f"{QDRANT_URL}/collections/{COLLECTION_NAME}/points?wait=true", data=json.dumps(upsert_data).encode(), headers={"Content-Type": "application/json"}, method="PUT" ) try: with urllib.request.urlopen(req, timeout=10) as response: result = json.loads(response.read().decode()) return point_id if result.get("status") == "ok" else None except Exception as e: print(f"Storage error: {e}", file=sys.stderr) return None

#!/usr/bin/env python3 """Mem0-style retrieval: Search by user_id across ALL conversations.""" import argparse, json, sys, urllib.request from typing import List, Optional QDRANT_URL = "http://localhost:6333" COLLECTION_NAME = "openclaw_memories" OLLAMA_URL = "http://localhost:11434/v1" def search_user_memories(user_id: str, query: str, limit: int = 10): """Search memories for a specific user across all conversations.""" # Get embedding for query data = json.dumps({"model": "snowflake-arctic-embed2", "input": query[:8192]}) req = urllib.request.Request(f"{OLLAMA_URL}/embeddings", data=data.encode(), headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=30) as resp: embedding = json.loads(resp.read())["data"][0]["embedding"] # Search WITH user_id filter (Mem0-style) search_data = json.dumps({ "vector": embedding, "limit": limit, "with_payload": True, "filter": { "must": [ {"key": "user_id", "match": {"value": user_id}} ] } }).encode() req = urllib.request.Request( f"{QDRANT_URL}/collections/{COLLECTION_NAME}/points/search", data=search_data, headers={"Content-Type": "application/json"}, method="POST" ) with urllib.request.urlopen(req, timeout=10) as resp: return json.loads(resp.read()).get("result", [])

#!/usr/bin/env python3 """Heartbeat: Append new turns to Redis short-term buffer.""" import os, sys, json, redis from datetime import datetime, timezone from pathlib import Path REDIS_HOST = os.getenv("REDIS_HOST", "localhost") USER_ID = os.getenv("USER_ID", "user") STATE_FILE = Path("/root/.openclaw/workspace/.mem_last_turn") def get_session_transcript(): """Find current OpenClaw session JSONL file.""" sessions_dir = Path("/root/.openclaw/agents/main/sessions") files = list(sessions_dir.glob("*.jsonl")) return max(files, key=lambda p: p.stat().st_mtime) if files else None def parse_turns_since(last_turn_num): """Extract conversation turns since last processed.""" # ... parses OpenClaw session JSONL ... # Returns list of turn dicts with role, content, timestamp pass def main(): r = redis.Redis(host=REDIS_HOST, port=6379, decode_responses=True) last_turn = int(STATE_FILE.read_text().strip()) if STATE_FILE.exists() else 0 new_turns = parse_turns_since(last_turn) if not new_turns: print(f"No new turns since turn {last_turn}") sys.exit(0) key = f"mem:{USER_ID}" for turn in new_turns: r.lpush(key, json.dumps(turn)) STATE_FILE.write_text(str(max(t['turn'] for t in new_turns))) print(f"✅ Appended {len(new_turns)} turns to Redis")

#!/usr/bin/env python3 """Daily: Flush Redis → Qdrant, then clear Redis.""" import os, sys, json, redis, urllib.request from datetime import datetime QDRANT_URL = "http://localhost:6333" OLLAMA_URL = "http://localhost:11434/v1" COLLECTION = "openclaw_memories" REDIS_HOST = os.getenv("REDIS_HOST", "localhost") def get_embedding(text): data = json.dumps({"model": "snowflake-arctic-embed2", "input": text[:8192]}) req = urllib.request.Request(f"{OLLAMA_URL}/embeddings", data=data.encode(), headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=30) as resp: return json.loads(resp.read())["data"][0]["embedding"] def main(): r = redis.Redis(host=REDIS_HOST, port=6379, decode_responses=True) items = r.lrange("mem:user", 0, -1) print(f"Backing up {len(items)} items...") for item_json in items: item = json.loads(item_json) text = f"{item.get('role')}: {item.get('content')}" vector = get_embedding(text) # Store to Qdrant... r.delete("mem:user") print("✅ Backup complete, Redis cleared")

--- name: qdrant-memory description: Full conversation memory storage to Qdrant (Mem0-style). --- # Qdrant Memory - Mem0-Style Usage ## Quick Commands - `save q` → Store conversation to Qdrant - `search q topic` → Search memories ## Configuration - Qdrant: http://localhost:6333 - Ollama: http://localhost:11434 - Model: snowflake-arctic-embed2 (1024 dims)