41 lines
1.5 KiB
Python
41 lines
1.5 KiB
Python
from .models import UserMemory
|
|
from .prompts import SIMBA_SYSTEM_PROMPT
|
|
from .context import estimate_tokens
|
|
|
|
MEMORY_BUDGET_TOKENS = 1000
|
|
|
|
|
|
async def get_memories_for_user(user_id: str) -> list[str]:
|
|
"""Return all memory content strings for a user, ordered by most recently updated."""
|
|
memories = await UserMemory.filter(user_id=user_id).order_by("-updated_at")
|
|
return [m.content for m in memories]
|
|
|
|
|
|
async def build_system_prompt_with_memories(user_id: str) -> str:
|
|
memories = await get_memories_for_user(user_id)
|
|
if not memories:
|
|
return SIMBA_SYSTEM_PROMPT
|
|
selected = []
|
|
used = 0
|
|
for memory in memories:
|
|
cost = estimate_tokens(f"- {memory}\n")
|
|
if used + cost <= MEMORY_BUDGET_TOKENS:
|
|
selected.append(memory)
|
|
used += cost
|
|
block = "\n".join(f"- {m}" for m in selected)
|
|
if len(selected) < len(memories):
|
|
block += "\n[Additional saved memories omitted from this prompt.]"
|
|
return f"{SIMBA_SYSTEM_PROMPT}\n\nUSER MEMORIES (facts the user has asked you to remember):\n{block}"
|
|
|
|
|
|
async def save_memory(user_id: str, content: str) -> str:
|
|
"""Save a new memory or touch an existing one (exact-match dedup)."""
|
|
existing = await UserMemory.filter(user_id=user_id, content=content).first()
|
|
if existing:
|
|
existing.updated_at = None # auto_now=True will refresh it on save
|
|
await existing.save(update_fields=["updated_at"])
|
|
return "Memory already exists (refreshed)."
|
|
|
|
await UserMemory.create(user_id=user_id, content=content)
|
|
return "Memory saved."
|