Spaces:
Running on Zero
Running on Zero
File size: 5,350 Bytes
f7e32a5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 | """Memory manager — orchestrates short-term, long-term, and vector memory."""
import json
from datetime import datetime
from typing import Optional
from config import config
from database.repositories import MemoryRepo, UserFactRepo
class MemoryManager:
"""Central memory system for Synapse V8."""
def __init__(self, user_id: str = "guest"):
self.user_id = user_id
self.short_term: list[dict] = []
self._short_term_limit = config.memory.short_term_limit
async def add_short_term(self, role: str, content: str,
metadata: dict = None) -> dict:
entry = {
"role": role,
"content": content,
"timestamp": datetime.utcnow().isoformat(),
"metadata": metadata or {},
}
self.short_term.append(entry)
if len(self.short_term) > self._short_term_limit:
await self._consolidate_oldest()
return entry
async def _consolidate_oldest(self):
if not self.short_term:
return
oldest = self.short_term.pop(0)
importance = self._calculate_importance(oldest["content"])
if importance >= config.memory.long_term_threshold:
await MemoryRepo.create(
user_id=self.user_id,
content=oldest["content"],
memory_type="consolidated",
importance=importance,
metadata=oldest.get("metadata"),
)
def _calculate_importance(self, content: str) -> float:
score = 0.3
important_markers = [
"remember", "important", "never forget", "always",
"my name", "i prefer", "i like", "i don't like",
"my birthday", "my email", "password", "secret",
"goal", "project", "deadline", "meeting",
]
lower = content.lower()
for marker in important_markers:
if marker in lower:
score += 0.15
if len(content) > 200:
score += 0.1
if "?" in content:
score += 0.05
return min(score, 1.0)
def get_short_term_context(self, max_messages: int = 20) -> list[dict]:
return self.short_term[-max_messages:]
async def search_long_term(self, query: str, limit: int = 5) -> list[dict]:
results = await MemoryRepo.search(self.user_id, query)
return results[:limit]
async def get_pinned_memories(self) -> list[dict]:
return await MemoryRepo.get_pinned(self.user_id)
async def add_memory(self, content: str, memory_type: str = "long_term",
importance: float = 0.5, tags: list = None,
conversation_id: str = None) -> dict:
return await MemoryRepo.create(
user_id=self.user_id,
content=content,
memory_type=memory_type,
importance=importance,
tags=tags,
conversation_id=conversation_id,
)
async def add_user_fact(self, fact: str, category: str = "general",
source: str = None) -> dict:
return await UserFactRepo.create(
user_id=self.user_id,
fact=fact,
category=category,
source=source,
)
async def get_user_facts(self, category: str = None) -> list[dict]:
return await UserFactRepo.get_user_facts(self.user_id, category)
async def get_all_memories(self, memory_type: str = None,
limit: int = 50) -> list[dict]:
return await MemoryRepo.get_user_memories(
self.user_id, memory_type=memory_type, limit=limit
)
async def toggle_pin_memory(self, memory_id: str) -> None:
await MemoryRepo.toggle_pin(memory_id)
async def delete_memory(self, memory_id: str) -> bool:
return await MemoryRepo.delete(memory_id)
async def build_memory_context(self) -> str:
parts = []
pinned = await self.get_pinned_memories()
if pinned:
parts.append("Pinned memories:")
for m in pinned[:10]:
parts.append(f"- {m['content']}")
facts = await self.get_user_facts()
if facts:
parts.append("\nKnown facts about user:")
for f in facts[:10]:
parts.append(f"- {f['fact']} ({f['category']})")
recent_memories = await self.get_all_memories(limit=5)
if recent_memories:
parts.append("\nRecent memories:")
for m in recent_memories:
parts.append(f"- {m['content']}")
return "\n".join(parts) if parts else ""
async def consolidate(self, conversation_messages: list[dict]) -> None:
if len(conversation_messages) < config.memory.consolidation_interval:
return
summary_parts = []
for msg in conversation_messages[-config.memory.consolidation_interval:]:
summary_parts.append(f"{msg['role']}: {msg['content'][:200]}")
summary = "\n".join(summary_parts)
importance = self._calculate_importance(summary)
if importance >= 0.4:
await self.add_memory(
content=f"Conversation summary: {summary[:500]}",
memory_type="summary",
importance=importance,
)
|