memory_client.py
Rama main · réplica de lectura del repositorio original.
"""
Memory Client for OpenVoiceUI
Direct access to clawdbot's memory system:
- SQLite FTS5 full-text search
- Discord session search
- Ambient transcript retrieval
- Combined context compilation
This gives the voice agent the same memory access as the Discord agent.
"""
import os
import sqlite3
import json
import logging
from pathlib import Path
from datetime import datetime, timedelta
from typing import List, Dict, Optional
logger = logging.getLogger(__name__)
# Paths to clawdbot data (configure via env vars or leave empty if not using memory features)
_home = Path.home()
MEMORY_DB = Path(os.getenv('CLAWDBOT_MEMORY_DB', str(_home / '.clawdbot/memory/main.sqlite')))
SESSIONS_DIR = Path(os.getenv('CLAWDBOT_SESSIONS_DIR', str(_home / '.clawdbot/agents/main/sessions/')))
VOICE_EVENTS = Path('/tmp/openvoiceui-events.jsonl')
# Ambient transcripts directory
AMBIENT_DIR = Path(__file__).parent / "ambient_transcripts"
class MemoryClient:
"""Direct access to clawdbot memory and sessions."""
def __init__(self):
self.memory_db = str(MEMORY_DB)
self.sessions_dir = SESSIONS_DIR
def search_memory(self, query: str, limit: int = 5) -> List[Dict]:
"""
Search memory chunks using FTS5.
Args:
query: Search query (FTS5 syntax supported)
limit: Max results to return
Returns:
List of {text, source, score} dicts
"""
if not MEMORY_DB.exists():
logger.warning(f"Memory DB not found: {MEMORY_DB}")
return []
try:
conn = sqlite3.connect(self.memory_db)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
# Clean query for FTS5 (remove special chars that break it)
clean_query = ''.join(c for c in query if c.isalnum() or c.isspace())
if not clean_query:
return []
# FTS5 search - the table has all columns, no join needed
cursor.execute("""
SELECT text, path, bm25(chunks_fts) as score
FROM chunks_fts
WHERE chunks_fts MATCH ?
ORDER BY bm25(chunks_fts)
LIMIT ?
""", (clean_query, limit))
results = []
for row in cursor.fetchall():
results.append({
'text': row['text'][:500], # Truncate long chunks
'source': row['path'].split('/')[-1] if row['path'] else 'memory',
'score': row['score']
})
conn.close()
logger.info(f"Memory search for '{query}': {len(results)} results")
return results
except Exception as e:
logger.error(f"Memory search error: {e}")
return []
def search_sessions(self, query: str, limit: int = 3) -> List[Dict]:
"""
Search Discord session files for relevant conversations.
Args:
query: Search terms
limit: Max results
Returns:
List of {content, role, session} dicts
"""
if not self.sessions_dir.exists():
logger.warning(f"Sessions dir not found: {self.sessions_dir}")
return []
results = []
query_lower = query.lower()
# Get most recent sessions first
session_files = sorted(
self.sessions_dir.glob('*.jsonl'),
key=lambda p: p.stat().st_mtime,
reverse=True
)[:20] # Only search last 20 sessions
for session_file in session_files:
try:
with open(session_file, 'r') as f:
for line in f:
try:
entry = json.loads(line)
if entry.get('type') != 'message':
continue
msg = entry.get('message', {})
content = msg.get('content', '')
role = msg.get('role', 'unknown')
if query_lower in content.lower():
results.append({
'content': content[:400],
'role': role,
'session': session_file.stem[:20]
})
if len(results) >= limit:
return results
except json.JSONDecodeError:
continue
except Exception as e:
logger.debug(f"Error reading session {session_file}: {e}")
continue
logger.info(f"Session search for '{query}': {len(results)} results")
return results
def get_recent_ambient(self, user_id: str, minutes: int = 30, max_chars: int = 1500) -> List[Dict]:
"""
Get recent ambient transcripts for a user.
These are background audio recordings that were transcribed while the
user was not actively conversing with the agent.
Args:
user_id: Clerk user ID
minutes: How many minutes back to look (default: 30)
max_chars: Maximum total characters to return
Returns:
List of {transcript, timestamp, has_wake_word} dicts
"""
if not user_id:
return []
user_dir = AMBIENT_DIR / user_id
if not user_dir.exists():
return []
threshold = datetime.now() - timedelta(minutes=minutes)
entries = []
total_chars = 0
# Check today's file and yesterday's
for days_ago in [0, 1]:
date = datetime.now() - timedelta(days=days_ago)
date_str = date.strftime('%Y-%m-%d')
transcript_file = user_dir / f"{date_str}.jsonl"
if not transcript_file.exists():
continue
try:
with open(transcript_file, 'r') as f:
for line in reversed(list(f)): # Start from newest
try:
entry = json.loads(line)
# Parse timestamp
ts_str = entry.get('timestamp', '')
try:
# Handle various ISO formats
entry_time = datetime.fromisoformat(
ts_str.replace('Z', '+00:00').replace('+00:00', '')
)
if entry_time < threshold:
continue
except ValueError:
pass # Include if we can't parse timestamp
text = entry.get('transcript', '')
if total_chars + len(text) > max_chars:
break
entries.append({
'transcript': text,
'timestamp': entry.get('timestamp', ''),
'has_wake_word': entry.get('has_wake_word', False),
'duration_seconds': entry.get('duration_seconds', 0)
})
total_chars += len(text)
except json.JSONDecodeError:
continue
except Exception as e:
logger.debug(f"Error reading ambient transcripts: {e}")
# Reverse to chronological order
entries.reverse()
logger.info(f"Retrieved {len(entries)} ambient transcripts for user {user_id}")
return entries
def search_voice_transcripts(self, query: str, limit: int = 3) -> List[Dict]:
"""
Search past voice conversations from events file.
Args:
query: Search terms
limit: Max results
Returns:
List of {content, role, time} dicts
"""
if not VOICE_EVENTS.exists():
return []
results = []
query_lower = query.lower()
try:
with open(VOICE_EVENTS, 'r') as f:
for line in f:
try:
event = json.loads(line)
if event.get('type') != 'conversation':
continue
message = event.get('message', '')
if query_lower in message.lower():
results.append({
'content': message[:300],
'role': event.get('role', 'unknown'),
'time': event.get('timestamp', 'unknown')
})
if len(results) >= limit:
break
except:
continue
except Exception as e:
logger.error(f"Voice transcript search error: {e}")
return results
def get_full_context(self, user_message: str, user_id: str = None) -> Dict:
"""
Get combined context from all sources.
Args:
user_message: What user just said
user_id: Optional Clerk user ID for ambient transcripts
Returns:
Dict with memory, sessions, transcripts, and ambient context
"""
# Extract key terms from message
key_terms = self._extract_keywords(user_message)
search_query = ' '.join(key_terms[:5])
context = {
'query': search_query,
'memory': self.search_memory(search_query, limit=5),
'sessions': self.search_sessions(search_query, limit=3),
'voice_transcripts': self.search_voice_transcripts(search_query, limit=2),
'generated_at': datetime.now().isoformat()
}
# Add ambient transcripts if user_id provided
if user_id:
context['ambient'] = self.get_recent_ambient(user_id, minutes=30, max_chars=1000)
return context
def format_context_for_prompt(self, context: Dict, max_tokens: int = 1500) -> str:
"""
Format context for injection into system prompt.
Args:
context: Context dict from get_full_context()
max_tokens: Approximate max tokens (chars / 4)
Returns:
Formatted string for prompt injection
"""
parts = ["\n--- RELEVANT CONTEXT ---"]
char_limit = max_tokens * 4 # Rough chars estimate
current_chars = 0
# Add memory results
if context.get('memory'):
parts.append("\nFrom memory:")
for item in context['memory'][:3]:
text = f"\n- [{item['source']}] {item['text'][:200]}"
if current_chars + len(text) > char_limit:
break
parts.append(text)
current_chars += len(text)
# Add session results
if context.get('sessions'):
parts.append("\nFrom Discord conversations:")
for item in context['sessions'][:2]:
text = f"\n- {item['role']}: {item['content'][:150]}"
if current_chars + len(text) > char_limit:
break
parts.append(text)
current_chars += len(text)
# Add voice transcripts
if context.get('voice_transcripts'):
parts.append("\nFrom past voice calls:")
for item in context['voice_transcripts'][:2]:
text = f"\n- {item['role']}: {item['content'][:100]}"
if current_chars + len(text) > char_limit:
break
parts.append(text)
current_chars += len(text)
# Add ambient transcripts (background audio context)
# This is special context about what was heard around the user
if context.get('ambient'):
ambient_intro = """
--- SURROUNDING AUDIO CONTEXT ---
You have ears that can hear the sounds and audio around you when the user has this feature enabled.
This is what you recently heard in the background. Determine if this requires a response or is just
background noise (TV, music, other people talking, etc). Use this to understand the user's context
and make relevant comments when they wake you up. Be playful and observant!
"""
parts.append(ambient_intro)
for item in context['ambient'][:5]:
# Format timestamp for readability
ts = item.get('timestamp', '')
if 'T' in ts:
ts = ts.split('T')[1][:8] # Just the time HH:MM:SS
text = f"\n[{ts}] \"{item.get('transcript', '')[:200]}\""
if item.get('has_wake_word'):
text += " [WAKE WORD DETECTED]"
if current_chars + len(text) > char_limit:
break
parts.append(text)
current_chars += len(text)
parts.append("\n--- END SURROUNDING AUDIO ---")
parts.append("\n--- END CONTEXT ---")
return '\n'.join(parts) if current_chars > 0 else ""
def _extract_keywords(self, text: str) -> List[str]:
"""Extract meaningful keywords from text."""
# Stop words to ignore
stop_words = {
'the', 'a', 'an', 'is', 'are', 'was', 'were', 'be', 'been',
'being', 'have', 'has', 'had', 'do', 'does', 'did', 'will',
'would', 'could', 'should', 'may', 'might', 'must', 'can',
'what', 'which', 'who', 'whom', 'when', 'where', 'why', 'how',
'this', 'that', 'these', 'those', 'i', 'me', 'my', 'we', 'our',
'you', 'your', 'he', 'him', 'his', 'she', 'her', 'it', 'its',
'they', 'them', 'their', 'and', 'or', 'but', 'if', 'then',
'else', 'so', 'for', 'with', 'about', 'into', 'to', 'from',
'up', 'down', 'in', 'out', 'on', 'off', 'over', 'under',
'again', 'further', 'once', 'here', 'there', 'all', 'each',
'few', 'more', 'most', 'other', 'some', 'such', 'only', 'just'
}
# Extract words
words = []
for word in text.lower().split():
# Clean word
clean = ''.join(c for c in word if c.isalnum())
# Keep if long enough and not stop word
if len(clean) > 3 and clean not in stop_words:
words.append(clean)
# Return unique words, preserving order
seen = set()
keywords = []
for w in words:
if w not in seen:
seen.add(w)
keywords.append(w)
return keywords[:10]
# Singleton instance
_client = None
def get_memory_client() -> MemoryClient:
"""Get or create memory client instance."""
global _client
if _client is None:
_client = MemoryClient()
return _client
# Convenience functions for direct import
def search_memory(query: str, limit: int = 5) -> List[Dict]:
return get_memory_client().search_memory(query, limit)
def search_sessions(query: str, limit: int = 3) -> List[Dict]:
return get_memory_client().search_sessions(query, limit)
def get_full_context(user_message: str, user_id: str = None) -> Dict:
return get_memory_client().get_full_context(user_message, user_id=user_id)
def format_context_for_prompt(context: Dict, max_tokens: int = 1500) -> str:
return get_memory_client().format_context_for_prompt(context, max_tokens)
if __name__ == '__main__':
# Test the client
logging.basicConfig(level=logging.INFO)
client = get_memory_client()
# Test memory search
print("Testing memory search...")
results = client.search_memory("steve call project")
for r in results:
print(f" [{r['source']}] {r['text'][:100]}...")
# Test session search
print("\nTesting session search...")
results = client.search_sessions("discord bot")
for r in results:
print(f" {r['role']}: {r['content'][:100]}...")
# Test full context
print("\nTesting full context...")
context = client.get_full_context("what did we talk about yesterday")
print(client.format_context_for_prompt(context))