Initial commit: TrueRecall v2.2 with 30b curator and timer-based curation
This commit is contained in:
root
350
tr-continuous/curator_timer.py
Executable file
350
tr-continuous/curator_timer.py
Executable file
@@ -0,0 +1,350 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
TrueRecall Timer Curator: Runs every 30 minutes via cron.
|
||||
|
||||
- Queries all uncurated memories from memories_tr
|
||||
- Sends batch to qwen3 for gem extraction
|
||||
- Stores gems to gems_tr
|
||||
- Marks processed memories as curated=true
|
||||
|
||||
Usage:
|
||||
python3 curator_timer.py --config curator_config.json
|
||||
python3 curator_timer.py --config curator_config.json --dry-run
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import argparse
|
||||
import requests
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import List, Dict, Any, Optional
|
||||
import hashlib
|
||||
|
||||
# Load config
|
||||
def load_config(config_path: str) -> Dict[str, Any]:
|
||||
with open(config_path, 'r') as f:
|
||||
return json.load(f)
|
||||
|
||||
# Default paths
|
||||
SCRIPT_DIR = Path(__file__).parent
|
||||
DEFAULT_CONFIG = SCRIPT_DIR / "curator_config.json"
|
||||
|
||||
# Curator prompt path
|
||||
CURATOR_PROMPT_PATH = Path("/root/.openclaw/workspace/.projects/true-recall-v2/curator-prompt.md")
|
||||
|
||||
|
||||
def load_curator_prompt() -> str:
|
||||
"""Load the curator system prompt."""
|
||||
try:
|
||||
with open(CURATOR_PROMPT_PATH, 'r') as f:
|
||||
return f.read()
|
||||
except FileNotFoundError:
|
||||
print(f"⚠️ Curator prompt not found at {CURATOR_PROMPT_PATH}")
|
||||
return """You are The Curator. Extract meaningful gems from conversation history.
|
||||
Extract facts, insights, decisions, preferences, and context that would be valuable to remember.
|
||||
Output a JSON array of gems with fields: gem, context, snippet, categories, importance (1-5), confidence (0-0.99)."""
|
||||
|
||||
|
||||
def get_uncurated_memories(qdrant_url: str, collection: str, user_id: str, max_batch: int) -> List[Dict[str, Any]]:
|
||||
"""Query Qdrant for uncurated memories."""
|
||||
filter_data = {
|
||||
"must": [
|
||||
{"key": "user_id", "match": {"value": user_id}},
|
||||
{"key": "curated", "match": {"value": False}}
|
||||
]
|
||||
}
|
||||
|
||||
all_points = []
|
||||
offset = None
|
||||
iterations = 0
|
||||
max_iterations = 10
|
||||
|
||||
while len(all_points) < max_batch and iterations < max_iterations:
|
||||
iterations += 1
|
||||
scroll_data = {
|
||||
"limit": min(100, max_batch - len(all_points)),
|
||||
"with_payload": True,
|
||||
"filter": filter_data
|
||||
}
|
||||
|
||||
if offset:
|
||||
scroll_data["offset"] = offset
|
||||
|
||||
try:
|
||||
response = requests.post(
|
||||
f"{qdrant_url}/collections/{collection}/points/scroll",
|
||||
json=scroll_data,
|
||||
headers={"Content-Type": "application/json"},
|
||||
timeout=30
|
||||
)
|
||||
response.raise_for_status()
|
||||
result = response.json()
|
||||
points = result.get("result", {}).get("points", [])
|
||||
|
||||
if not points:
|
||||
break
|
||||
|
||||
all_points.extend(points)
|
||||
offset = result.get("result", {}).get("next_page_offset")
|
||||
if not offset:
|
||||
break
|
||||
except Exception as e:
|
||||
print(f"Error querying Qdrant: {e}", file=sys.stderr)
|
||||
break
|
||||
|
||||
# Convert to simple dicts
|
||||
memories = []
|
||||
for point in all_points:
|
||||
payload = point.get("payload", {})
|
||||
memories.append({
|
||||
"id": point.get("id"),
|
||||
"content": payload.get("content", ""),
|
||||
"role": payload.get("role", ""),
|
||||
"timestamp": payload.get("timestamp", ""),
|
||||
"turn": payload.get("turn", 0),
|
||||
**payload
|
||||
})
|
||||
|
||||
return memories[:max_batch]
|
||||
|
||||
|
||||
def extract_gems(memories: List[Dict[str, Any]], ollama_url: str) -> List[Dict[str, Any]]:
|
||||
"""Send memories to qwen3 for gem extraction."""
|
||||
if not memories:
|
||||
return []
|
||||
|
||||
prompt = load_curator_prompt()
|
||||
|
||||
# Build conversation from memories
|
||||
conversation_lines = []
|
||||
for mem in memories:
|
||||
role = mem.get("role", "unknown")
|
||||
content = mem.get("content", "")
|
||||
if content:
|
||||
conversation_lines.append(f"{role}: {content}")
|
||||
|
||||
conversation_text = "\n".join(conversation_lines)
|
||||
|
||||
try:
|
||||
response = requests.post(
|
||||
f"{ollama_url}/api/generate",
|
||||
json={
|
||||
"model": "qwen3:30b-a3b-instruct-2507-q8_0",
|
||||
"system": prompt,
|
||||
"prompt": f"## Input Conversation\n\n{conversation_text}\n\n## Output\n",
|
||||
"stream": False,
|
||||
"options": {
|
||||
"temperature": 0.1,
|
||||
"num_predict": 4000
|
||||
}
|
||||
},
|
||||
timeout=120
|
||||
)
|
||||
response.raise_for_status()
|
||||
except Exception as e:
|
||||
print(f"Error calling Ollama: {e}", file=sys.stderr)
|
||||
return []
|
||||
|
||||
result = response.json()
|
||||
output = result.get('response', '').strip()
|
||||
|
||||
# Extract JSON from output
|
||||
if '```json' in output:
|
||||
output = output.split('```json')[1].split('```')[0].strip()
|
||||
elif '```' in output:
|
||||
output = output.split('```')[1].split('```')[0].strip()
|
||||
|
||||
try:
|
||||
start_idx = output.find('[')
|
||||
end_idx = output.rfind(']')
|
||||
if start_idx != -1 and end_idx != -1 and end_idx > start_idx:
|
||||
output = output[start_idx:end_idx+1]
|
||||
|
||||
# Fix common JSON issues from LLM output
|
||||
# Replace problematic escape sequences
|
||||
output = output.replace('\\n', '\n').replace('\\t', '\t')
|
||||
# Fix single quotes in content that break JSON
|
||||
output = output.replace("\\'", "'")
|
||||
|
||||
gems = json.loads(output)
|
||||
if not isinstance(gems, list):
|
||||
gems = [gems] if gems else []
|
||||
return gems
|
||||
except json.JSONDecodeError as e:
|
||||
# Try to extract gems with regex fallback
|
||||
import re
|
||||
gem_matches = re.findall(r'"gem"\s*:\s*"([^"]+)"', output)
|
||||
if gem_matches:
|
||||
gems = []
|
||||
for gem_text in gem_matches:
|
||||
gems.append({
|
||||
"gem": gem_text,
|
||||
"context": "Extracted via fallback",
|
||||
"categories": ["extracted"],
|
||||
"importance": 3,
|
||||
"confidence": 0.7
|
||||
})
|
||||
print(f"⚠️ Fallback extraction: {len(gems)} gems", file=sys.stderr)
|
||||
return gems
|
||||
print(f"Error parsing curator output: {e}", file=sys.stderr)
|
||||
print(f"Raw output: {repr(output[:500])}...", file=sys.stderr)
|
||||
return []
|
||||
|
||||
|
||||
def get_embedding(text: str, ollama_url: str) -> Optional[List[float]]:
|
||||
"""Get embedding from Ollama."""
|
||||
try:
|
||||
response = requests.post(
|
||||
f"{ollama_url}/api/embeddings",
|
||||
json={"model": "mxbai-embed-large", "prompt": text},
|
||||
timeout=30
|
||||
)
|
||||
response.raise_for_status()
|
||||
return response.json()['embedding']
|
||||
except Exception as e:
|
||||
print(f"Error getting embedding: {e}", file=sys.stderr)
|
||||
return None
|
||||
|
||||
|
||||
def store_gem(gem: Dict[str, Any], user_id: str, qdrant_url: str, target_collection: str, ollama_url: str) -> bool:
|
||||
"""Store a single gem to Qdrant."""
|
||||
embedding_text = f"{gem.get('gem', '')} {gem.get('context', '')} {gem.get('snippet', '')}"
|
||||
vector = get_embedding(embedding_text, ollama_url)
|
||||
|
||||
if vector is None:
|
||||
return False
|
||||
|
||||
# Generate ID
|
||||
hash_content = f"{user_id}:{gem.get('conversation_id', '')}:{gem.get('turn_range', '')}:{gem.get('gem', '')[:50]}"
|
||||
hash_bytes = hashlib.sha256(hash_content.encode()).digest()[:8]
|
||||
gem_id = int.from_bytes(hash_bytes, byteorder='big') % (2**63)
|
||||
|
||||
payload = {
|
||||
"user_id": user_id,
|
||||
**gem,
|
||||
"curated_at": datetime.now(timezone.utc).isoformat()
|
||||
}
|
||||
|
||||
try:
|
||||
response = requests.put(
|
||||
f"{qdrant_url}/collections/{target_collection}/points",
|
||||
json={
|
||||
"points": [{
|
||||
"id": abs(gem_id),
|
||||
"vector": vector,
|
||||
"payload": payload
|
||||
}]
|
||||
},
|
||||
timeout=30
|
||||
)
|
||||
response.raise_for_status()
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"Error storing gem: {e}", file=sys.stderr)
|
||||
return False
|
||||
|
||||
|
||||
def mark_curated(memory_ids: List, qdrant_url: str, collection: str) -> bool:
|
||||
"""Mark memories as curated in Qdrant using POST /points/payload format."""
|
||||
if not memory_ids:
|
||||
return True
|
||||
|
||||
try:
|
||||
response = requests.post(
|
||||
f"{qdrant_url}/collections/{collection}/points/payload",
|
||||
json={
|
||||
"points": memory_ids,
|
||||
"payload": {
|
||||
"curated": True,
|
||||
"curated_at": datetime.now(timezone.utc).isoformat()
|
||||
}
|
||||
},
|
||||
timeout=30
|
||||
)
|
||||
response.raise_for_status()
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"Error marking curated: {e}", file=sys.stderr)
|
||||
return False
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="TrueRecall Timer Curator")
|
||||
parser.add_argument("--config", "-c", default=str(DEFAULT_CONFIG), help="Config file path")
|
||||
parser.add_argument("--dry-run", "-n", action="store_true", help="Don't write, just preview")
|
||||
args = parser.parse_args()
|
||||
|
||||
config = load_config(args.config)
|
||||
|
||||
qdrant_url = os.getenv("QDRANT_URL", "http://10.0.0.40:6333")
|
||||
ollama_url = os.getenv("OLLAMA_URL", "http://10.0.0.10:11434")
|
||||
|
||||
user_id = config.get("user_id", "rob")
|
||||
source_collection = config.get("source_collection", "memories_tr")
|
||||
target_collection = config.get("target_collection", "gems_tr")
|
||||
max_batch = config.get("max_batch_size", 100)
|
||||
|
||||
print(f"🔍 TrueRecall Timer Curator")
|
||||
print(f"👤 User: {user_id}")
|
||||
print(f"📥 Source: {source_collection}")
|
||||
print(f"💎 Target: {target_collection}")
|
||||
print(f"📦 Max batch: {max_batch}")
|
||||
if args.dry_run:
|
||||
print("🏃 DRY RUN MODE")
|
||||
print()
|
||||
|
||||
# Get uncurated memories
|
||||
print("📥 Fetching uncurated memories...")
|
||||
memories = get_uncurated_memories(qdrant_url, source_collection, user_id, max_batch)
|
||||
print(f"✅ Found {len(memories)} uncurated memories")
|
||||
|
||||
if not memories:
|
||||
print("🤷 Nothing to curate. Exiting.")
|
||||
return
|
||||
|
||||
# Extract gems
|
||||
print(f"\n🧠 Sending {len(memories)} memories to curator...")
|
||||
gems = extract_gems(memories, ollama_url)
|
||||
print(f"✅ Extracted {len(gems)} gems")
|
||||
|
||||
if not gems:
|
||||
print("⚠️ No gems extracted. Nothing to store.")
|
||||
# Still mark as curated so we don't reprocess
|
||||
memory_ids = [m["id"] for m in memories] # Keep as integers
|
||||
mark_curated(memory_ids, qdrant_url, source_collection)
|
||||
return
|
||||
|
||||
# Preview
|
||||
print("\n💎 Gems preview:")
|
||||
for i, gem in enumerate(gems[:3], 1):
|
||||
print(f" {i}. {gem.get('gem', 'N/A')[:80]}...")
|
||||
if len(gems) > 3:
|
||||
print(f" ... and {len(gems) - 3} more")
|
||||
|
||||
if args.dry_run:
|
||||
print("\n🏃 DRY RUN: Not storing gems or marking curated.")
|
||||
return
|
||||
|
||||
# Store gems
|
||||
print(f"\n💾 Storing {len(gems)} gems...")
|
||||
stored = 0
|
||||
for gem in gems:
|
||||
if store_gem(gem, user_id, qdrant_url, target_collection, ollama_url):
|
||||
stored += 1
|
||||
print(f"✅ Stored: {stored}/{len(gems)}")
|
||||
|
||||
# Mark memories as curated
|
||||
print("\n📝 Marking memories as curated...")
|
||||
memory_ids = [m["id"] for m in memories] # Keep as integers
|
||||
if mark_curated(memory_ids, qdrant_url, source_collection):
|
||||
print(f"✅ Marked {len(memory_ids)} memories as curated")
|
||||
else:
|
||||
print(f"⚠️ Failed to mark some memories as curated")
|
||||
|
||||
print("\n🎉 Curation complete!")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user