NexusOS/synapse/memory/extractor.py

"""Memory extraction — asks Mistral to evaluate a conversation exchange and
decide if it contains a new permanent personal fact worth saving."""

from __future__ import annotations

import json
import logging
import re
from typing import Optional

_log = logging.getLogger(__name__)

_PROMPT = """\
You are a memory curator for a personal AI assistant named Nexus.

A conversation just occurred. Decide if the USER revealed a NEW, PERMANENT \
personal fact that should be saved to long-term memory.

SAVE ONLY facts that are stable and biographical:
- Identity: full name, age, location, nationality
- Possessions: vehicle, home, devices
- Relationships: family members, partner, close friends
- Career: job title, employer, field, skills
- Long-running projects or goals (not today's to-dos)
- Durable preferences or habits explicitly stated

DO NOT SAVE — these are ephemeral and would clutter memory:
- Anything time-bound: "today I have...", "I'm working on X today", "I have a full day of work"
- Mood or energy: "I'm tired", "feeling good", "having a rough day"
- Greetings or small talk: "good morning", "how are you"
- Questions the user asked the assistant
- Near-duplicates of anything in the existing memory list below

Existing memory (do not re-save these or close variants):
{existing_texts}

---
USER: {user_message}
ASSISTANT: {assistant_response}
---

Respond with JSON only — no prose, no markdown fences:
{{"save": true, "section": "<section name, or one from the list above>", "text": "<concise fact about Jon, third person>"}}
OR
{{"save": false}}"""


async def extract_memory(
    user_message: str,
    assistant_response: str,
    existing_sections: list[str],
    existing_texts: list[str],
    ollama_manager,
) -> Optional[dict]:
    """Ask Mistral to extract a saveable memory fact from a conversation exchange.

    Returns {"section": ..., "text": ...} or None.
    """
    if existing_texts:
        texts_block = "\n".join(f"- {t}" for t in existing_texts[:40])
    else:
        texts_block = "(none yet)"
    prompt = _PROMPT.format(
        existing_texts=texts_block,
        user_message=user_message[:800],
        assistant_response=assistant_response[:800],
    )
    try:
        response = await ollama_manager.chat(
            messages=[{"role": "user", "content": prompt}],
            model="mistral:latest",
            stream=False,
            temperature=0.0,
        )
        if not response:
            return None

        text = response.strip()

        # Strip markdown code fences if the model added them
        if "```" in text:
            m = re.search(r"```(?:json)?\s*(.*?)\s*```", text, re.DOTALL)
            if m:
                text = m.group(1).strip()

        data = json.loads(text)
        if data.get("save") and data.get("section") and data.get("text"):
            return {
                "section": str(data["section"]).strip(),
                "text": str(data["text"]).strip(),
            }
    except Exception as e:
        _log.debug("memory extraction failed: %s", e)
    return None