forked from enderofwings/NexusOS
Initial commit: NexusOS - local AI assistant platform
This commit is contained in:
Jon Wingender
@@ -0,0 +1,169 @@
|
||||
"""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 ..chat import _synapse_trace # append curator reasoning to the same MindTrace log
|
||||
from ..nexus_config import DEFAULT_MEMORY_MODEL
|
||||
|
||||
_log = logging.getLogger(__name__)
|
||||
|
||||
_TR = "┅" * 55
|
||||
|
||||
_PROMPT = """\
|
||||
You are a memory curator for a personal AI assistant named Nexus.
|
||||
|
||||
Extract EVERY new, permanent personal fact the USER revealed in this exchange.
|
||||
There may be SEVERAL facts in one message — output one JSON object for each.
|
||||
|
||||
SAVE facts that are stable and biographical, such as: identity (name, age,
|
||||
location), relationships (family, partner, friends), pets, possessions (vehicles,
|
||||
home, devices), career (job, employer, skills), hobbies and interests,
|
||||
long-running projects or goals (not today's to-dos), and durable preferences.
|
||||
|
||||
DO NOT SAVE — these are ephemeral and would clutter memory:
|
||||
- Anything time-bound: "today I have...", "I'm working on X today"
|
||||
- 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
|
||||
|
||||
The existing memory is below FOR CONTEXT. If the user ADDS NEW DETAIL to
|
||||
something already known (e.g. a new detail about a known pet, car, or project),
|
||||
DO save that new detail as its own fact. Only skip a fact that is an EXACT
|
||||
restatement of one already listed.
|
||||
|
||||
Existing memory:
|
||||
{existing_texts}
|
||||
|
||||
For "section", REUSE one of these existing section names whenever it fits:
|
||||
{existing_sections}
|
||||
Only invent a new section if none fit, and make it a SHORT single word
|
||||
(e.g. Hobbies, Pets, Health). Never use a sentence or long phrase as a section.
|
||||
|
||||
---
|
||||
USER: {user_message}
|
||||
ASSISTANT: {assistant_response}
|
||||
---
|
||||
|
||||
Respond with JSON only — no prose, no markdown fences. Output one object PER
|
||||
new fact (several objects, one per line, if there are several):
|
||||
{{"save": true, "section": "<short section name>", "text": "<concise fact about the user, third person>"}}
|
||||
If there is nothing new to save, output exactly:
|
||||
{{"save": false}}"""
|
||||
|
||||
|
||||
async def extract_memory(
|
||||
user_message: str,
|
||||
assistant_response: str,
|
||||
existing_sections: list[str],
|
||||
existing_texts: list[str],
|
||||
ollama_manager,
|
||||
model: str = DEFAULT_MEMORY_MODEL,
|
||||
num_gpu: int | None = 0,
|
||||
) -> list[dict]:
|
||||
"""Ask Mistral to extract saveable memory facts from a conversation exchange.
|
||||
|
||||
Returns a list of {"section": ..., "text": ...} — possibly empty. A single
|
||||
exchange can hold several facts, and Mistral emits one JSON object per fact.
|
||||
"""
|
||||
if existing_texts:
|
||||
# ponytail: only the 12 most-recent facts go in the dedup context, not all
|
||||
# ~40. On a CPU-bound curator (num_gpu=0) prompt-eval dominates, and 40
|
||||
# facts made a ~1200-token prompt that took ~50s+ to process. If dedup
|
||||
# starts re-saving older facts, move dedup to a difflib check in the
|
||||
# service instead of stuffing every fact into the prompt.
|
||||
texts_block = "\n".join(f"- {t}" for t in existing_texts[-12:])
|
||||
else:
|
||||
texts_block = "(none yet)"
|
||||
# Give Mistral the real section names to reuse, so it stops inventing
|
||||
# sentence-long sections out of the category descriptions in the prompt.
|
||||
# Only offer SHORT, clean names — never feed a junk sentence-section (e.g. a
|
||||
# past bad "Long-running projects or goals") back as a valid choice.
|
||||
clean = sorted(s for s in existing_sections if s and len(s.split()) <= 2 and len(s) <= 24)
|
||||
sections_line = ", ".join(clean) if clean else (
|
||||
"Identity, Relationships, Pets, Possessions, Career, Hobbies, Projects, Preferences"
|
||||
)
|
||||
prompt = _PROMPT.format(
|
||||
existing_texts=texts_block,
|
||||
existing_sections=sections_line,
|
||||
user_message=user_message[:3000],
|
||||
assistant_response=assistant_response[:800],
|
||||
)
|
||||
# MindTrace: curator pre-flight (full prompt) so its reasoning is visible in
|
||||
# the same console as the frontline model, not just Python warnings on failure.
|
||||
_synapse_trace(
|
||||
f"\n{_TR}\n◆ CURATOR: {model} (num_gpu={num_gpu})\n"
|
||||
f" PROMPT ({len(prompt)} chars):\n{prompt}\n{_TR}\n"
|
||||
)
|
||||
try:
|
||||
# num_gpu=0 (the default) pins the curator fully in system RAM instead of
|
||||
# the GPU, so it coexists with the GPU-resident chat model instead of
|
||||
# evicting it. Without this, on a small GPU the two thrash: every
|
||||
# exchange cold-loads the curator (~45s) and extraction times out,
|
||||
# silently saving nothing. Boxes with spare VRAM override via settings.
|
||||
response = await ollama_manager.chat(
|
||||
messages=[{"role": "user", "content": prompt}],
|
||||
model=model,
|
||||
stream=False,
|
||||
temperature=0.0,
|
||||
num_gpu=num_gpu,
|
||||
)
|
||||
if not response:
|
||||
_synapse_trace("◆ CURATOR RAW: (empty response)\n")
|
||||
return []
|
||||
|
||||
text = response.strip()
|
||||
_synapse_trace(f"◆ CURATOR RAW:\n{text}\n")
|
||||
|
||||
# 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()
|
||||
|
||||
# For several facts Mistral is inconsistent: sometimes ONE JSON object
|
||||
# per fact newline-separated, sometimes a single JSON ARRAY of objects.
|
||||
# raw_decode pulls each top-level value (handles the newline case and
|
||||
# plain "Extra data"); we then flatten any array so both shapes save all
|
||||
# facts. Plain json.loads() would die on the newline case and skip the
|
||||
# array (a list isn't a dict), losing every fact either way.
|
||||
results: list[dict] = []
|
||||
|
||||
def _keep(o):
|
||||
if isinstance(o, dict) and o.get("save") and o.get("section") and o.get("text"):
|
||||
results.append({
|
||||
"section": str(o["section"]).strip(),
|
||||
"text": str(o["text"]).strip(),
|
||||
})
|
||||
|
||||
dec = json.JSONDecoder()
|
||||
idx = 0
|
||||
while idx < len(text):
|
||||
while idx < len(text) and text[idx] in " \t\r\n,":
|
||||
idx += 1
|
||||
if idx >= len(text):
|
||||
break
|
||||
try:
|
||||
obj, idx = dec.raw_decode(text, idx)
|
||||
except json.JSONDecodeError:
|
||||
break
|
||||
if isinstance(obj, list):
|
||||
for o in obj:
|
||||
_keep(o)
|
||||
else:
|
||||
_keep(obj)
|
||||
if not results:
|
||||
_log.warning("memory: nothing saved. mistral said: %.300r", text)
|
||||
_synapse_trace(f"◆ CURATOR VERDICT: nothing to save\n{_TR}\n\n")
|
||||
else:
|
||||
_facts = "; ".join(f"[{r['section']}] {r['text']}" for r in results)
|
||||
_synapse_trace(f"◆ CURATOR VERDICT: {len(results)} fact(s) — {_facts}\n{_TR}\n\n")
|
||||
return results
|
||||
except Exception as e:
|
||||
_log.warning("memory extraction failed: %s", e)
|
||||
_synapse_trace(f"◆ CURATOR ERROR: {e}\n{_TR}\n\n")
|
||||
return []
|
||||
@@ -0,0 +1,208 @@
|
||||
"""Dedicated memory curator service — run alongside Synapse on port 8001.
|
||||
|
||||
Endpoints:
|
||||
GET / health check
|
||||
GET /memories list all memory items (optional ?section= filter)
|
||||
POST /memories direct write — no LLM, saves immediately
|
||||
PATCH /memories/{id} update a memory item
|
||||
DELETE /memories/{id} delete a memory item
|
||||
POST /memories/extract LLM-curated: evaluate a conversation exchange and
|
||||
optionally save a new permanent fact
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import math
|
||||
import uuid
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from fastapi import FastAPI, HTTPException, Body
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from pydantic import BaseModel
|
||||
|
||||
from .store import store, MemoryItem
|
||||
from .extractor import extract_memory
|
||||
from ..ollama_manager import get_ollama_manager
|
||||
|
||||
app = FastAPI(title="Nexus Memory Service", version="1.0")
|
||||
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=["*"],
|
||||
allow_credentials=False,
|
||||
allow_methods=["*"],
|
||||
allow_headers=["*"],
|
||||
)
|
||||
|
||||
|
||||
@app.on_event("startup")
|
||||
async def _warm_curator():
|
||||
"""Preload the curator model (in RAM, num_gpu=0 by default) so the first
|
||||
extraction isn't a cold load that blows the timeout. Runs in the background
|
||||
so it never delays startup. keep_alive then holds it warm between messages."""
|
||||
async def _bg():
|
||||
try:
|
||||
mgr = get_ollama_manager()
|
||||
# Ollama is started by the Synapse backend (a separate process), so
|
||||
# at our startup it usually isn't reachable yet. Wait for it before
|
||||
# warming instead of failing with "All connection attempts failed" —
|
||||
# which leaves the curator cold and makes the first extraction slow.
|
||||
for _ in range(60): # up to ~2 min
|
||||
if await asyncio.to_thread(mgr.is_running):
|
||||
break
|
||||
await asyncio.sleep(2)
|
||||
else:
|
||||
return
|
||||
settings = store.get_settings()
|
||||
model = settings.get("memory_model") or await mgr.select_best_model()
|
||||
num_gpu = await mgr.resolve_num_gpu(settings.get("memory_gpu_offload", 0), model)
|
||||
await mgr.warm(model, num_gpu=num_gpu)
|
||||
except Exception:
|
||||
pass
|
||||
asyncio.create_task(_bg())
|
||||
|
||||
|
||||
@app.get("/")
|
||||
async def health():
|
||||
return {"status": "ok", "count": len(store.all())}
|
||||
|
||||
|
||||
@app.get("/memories")
|
||||
async def list_memories(section: Optional[str] = None):
|
||||
items = store.all()
|
||||
if section:
|
||||
items = [i for i in items if i.section.lower() == section.lower()]
|
||||
return {"items": [{"id": i.id, "section": i.section, "text": i.text, "tags": i.tags} for i in items]}
|
||||
|
||||
|
||||
@app.post("/memories")
|
||||
async def add_memory(payload: Dict[str, Any] = Body(...)):
|
||||
text = (payload.get("text") or "").strip()
|
||||
if not text:
|
||||
raise HTTPException(status_code=400, detail="Missing 'text'")
|
||||
item = MemoryItem(
|
||||
id=str(uuid.uuid4()),
|
||||
section=(payload.get("section") or "General").strip(),
|
||||
text=text,
|
||||
tags=payload.get("tags", []),
|
||||
)
|
||||
store.add(item)
|
||||
return {"id": item.id, "section": item.section, "text": item.text, "tags": item.tags}
|
||||
|
||||
|
||||
@app.patch("/memories/{item_id}")
|
||||
async def update_memory(item_id: str, payload: Dict[str, Any] = Body(...)):
|
||||
existing = store.get(item_id)
|
||||
if not existing:
|
||||
raise HTTPException(status_code=404, detail="Not found")
|
||||
updated = MemoryItem(
|
||||
id=item_id,
|
||||
section=(payload.get("section") or existing.section).strip(),
|
||||
text=(payload.get("text") or existing.text).strip(),
|
||||
tags=payload.get("tags", existing.tags),
|
||||
)
|
||||
store.update(updated)
|
||||
return {"id": updated.id, "section": updated.section, "text": updated.text, "tags": updated.tags}
|
||||
|
||||
|
||||
@app.delete("/memories/{item_id}")
|
||||
async def delete_memory(item_id: str):
|
||||
if not store.get(item_id):
|
||||
raise HTTPException(status_code=404, detail="Not found")
|
||||
store.delete(item_id)
|
||||
return {"status": "deleted"}
|
||||
|
||||
|
||||
def _cosine(a: list, b: list) -> float:
|
||||
dot = sum(x * y for x, y in zip(a, b))
|
||||
na = math.sqrt(sum(x * x for x in a))
|
||||
nb = math.sqrt(sum(y * y for y in b))
|
||||
return dot / (na * nb) if na and nb else 0.0
|
||||
|
||||
|
||||
class ExtractRequest(BaseModel):
|
||||
user_message: str
|
||||
assistant_response: str
|
||||
|
||||
|
||||
@app.post("/memories/extract")
|
||||
async def extract_and_save(req: ExtractRequest):
|
||||
"""LLM-curated extraction — asks Mistral to evaluate the exchange against all
|
||||
existing memory items and save only new, permanent personal facts."""
|
||||
existing = store.all()
|
||||
existing_sections = list({i.section for i in existing})
|
||||
existing_texts = [i.text for i in existing]
|
||||
|
||||
# Adaptable per-machine curator config (see store _SETTINGS_DEFAULTS):
|
||||
# which model does extraction, and whether it runs on CPU/RAM or the GPU.
|
||||
settings = store.get_settings()
|
||||
mgr = get_ollama_manager()
|
||||
model = settings.get("memory_model") or await mgr.select_best_model()
|
||||
num_gpu = await mgr.resolve_num_gpu(settings.get("memory_gpu_offload", 0), model)
|
||||
try:
|
||||
merge_threshold = float(settings.get("memory_merge_threshold", 0.88))
|
||||
except (TypeError, ValueError):
|
||||
merge_threshold = 0.88
|
||||
|
||||
try:
|
||||
results = await asyncio.wait_for(
|
||||
extract_memory(
|
||||
req.user_message,
|
||||
req.assistant_response,
|
||||
existing_sections,
|
||||
existing_texts,
|
||||
mgr,
|
||||
model=model,
|
||||
num_gpu=num_gpu,
|
||||
),
|
||||
timeout=300.0,
|
||||
)
|
||||
|
||||
# Embed existing facts once so each new fact can be matched against them.
|
||||
# A near-duplicate UPDATES the matched fact in place (edit with new info)
|
||||
# rather than appending a copy. Best effort: if embeddings are down we
|
||||
# fall back to plain append. Merge disabled unless 0 < threshold < 1.
|
||||
existing_embeds: dict = {}
|
||||
if results and 0 < merge_threshold < 1:
|
||||
vecs = await asyncio.gather(*(mgr.embed(it.text) for it in existing))
|
||||
existing_embeds = {it.id: v for it, v in zip(existing, vecs) if v}
|
||||
|
||||
saved = []
|
||||
for result in results:
|
||||
new_vec = await mgr.embed(result["text"]) if existing_embeds else None
|
||||
match_id, best = None, 0.0
|
||||
if new_vec:
|
||||
for eid, ev in existing_embeds.items():
|
||||
sim = _cosine(new_vec, ev)
|
||||
if sim > best:
|
||||
best, match_id = sim, eid
|
||||
if best < merge_threshold:
|
||||
match_id = None
|
||||
|
||||
target = store.get(match_id) if match_id else None
|
||||
if target:
|
||||
# Near-duplicate of an existing fact — overwrite with the newer
|
||||
# statement, keeping the original id/section/position.
|
||||
updated = MemoryItem(id=target.id, section=target.section,
|
||||
text=result["text"], tags=target.tags)
|
||||
store.update(updated)
|
||||
if new_vec:
|
||||
existing_embeds[updated.id] = new_vec # keep cache fresh for later facts in this batch
|
||||
saved.append({"id": updated.id, "section": updated.section,
|
||||
"text": updated.text, "updated": True})
|
||||
else:
|
||||
item = MemoryItem(id=str(uuid.uuid4()),
|
||||
section=result["section"], text=result["text"])
|
||||
store.add(item)
|
||||
if new_vec:
|
||||
existing_embeds[item.id] = new_vec
|
||||
saved.append({"id": item.id, "section": item.section, "text": item.text})
|
||||
if saved:
|
||||
first = {k: saved[0][k] for k in ("id", "section", "text")}
|
||||
return {"saved": True, "items": saved, **first}
|
||||
except asyncio.TimeoutError:
|
||||
pass
|
||||
except Exception:
|
||||
pass
|
||||
return {"saved": False}
|
||||
@@ -0,0 +1,676 @@
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
from pydantic import BaseModel
|
||||
|
||||
from ..nexus_config import DEFAULT_MEMORY_MODEL
|
||||
import sqlite3
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import time
|
||||
|
||||
|
||||
def _cosine(a: List[float], b: List[float]) -> float:
|
||||
"""Cosine similarity between two equal-length vectors. 0.0 on mismatch."""
|
||||
if not a or not b or len(a) != len(b):
|
||||
return 0.0
|
||||
dot = sum(x * y for x, y in zip(a, b))
|
||||
na = math.sqrt(sum(x * x for x in a))
|
||||
nb = math.sqrt(sum(y * y for y in b))
|
||||
if na == 0.0 or nb == 0.0:
|
||||
return 0.0
|
||||
return dot / (na * nb)
|
||||
|
||||
# -----------------------------
|
||||
# Models
|
||||
# -----------------------------
|
||||
class MemoryItem(BaseModel):
|
||||
id: str
|
||||
section: str = "General"
|
||||
text: str
|
||||
tags: List[str] = []
|
||||
position: int = 0
|
||||
|
||||
class MessageItem(BaseModel):
|
||||
role: str # "user" or "assistant"
|
||||
content: str
|
||||
timestamp: float
|
||||
model: Optional[str] = None
|
||||
tokens: Optional[int] = None
|
||||
|
||||
class ConversationItem(BaseModel):
|
||||
id: str
|
||||
messages: List[MessageItem] = []
|
||||
created_at: float
|
||||
updated_at: float
|
||||
title: Optional[str] = None
|
||||
|
||||
@property
|
||||
def preview(self) -> str:
|
||||
for msg in self.messages:
|
||||
if msg.role == "user":
|
||||
return msg.content[:80]
|
||||
return "Empty conversation"
|
||||
|
||||
@property
|
||||
def timestamp(self) -> float:
|
||||
return self.created_at
|
||||
|
||||
# -----------------------------
|
||||
# Persistent Store
|
||||
# -----------------------------
|
||||
class PersistentMemoryStore:
|
||||
def __init__(self, db_path: Path):
|
||||
self.db_path = db_path
|
||||
os.makedirs(self.db_path.parent, exist_ok=True)
|
||||
self._ensure_tables()
|
||||
self._cache: Dict[str, MemoryItem] = self._load_all_memory()
|
||||
|
||||
# -----------------------------
|
||||
# Internal helpers
|
||||
# -----------------------------
|
||||
def _connect(self):
|
||||
conn = sqlite3.connect(self.db_path)
|
||||
conn.row_factory = sqlite3.Row
|
||||
conn.execute("PRAGMA journal_mode=WAL;")
|
||||
return conn
|
||||
|
||||
def _ensure_tables(self):
|
||||
conn = self._connect()
|
||||
cur = conn.cursor()
|
||||
|
||||
cur.execute("""
|
||||
CREATE TABLE IF NOT EXISTS memory (
|
||||
id TEXT PRIMARY KEY,
|
||||
section TEXT NOT NULL DEFAULT 'General',
|
||||
text TEXT NOT NULL,
|
||||
tags TEXT
|
||||
)
|
||||
""")
|
||||
# Migrate: add section column if it doesn't exist yet
|
||||
try:
|
||||
cur.execute("ALTER TABLE memory ADD COLUMN section TEXT NOT NULL DEFAULT 'General'")
|
||||
except Exception:
|
||||
pass
|
||||
# Migrate: add position column for stable ordering
|
||||
try:
|
||||
cur.execute("ALTER TABLE memory ADD COLUMN position INTEGER NOT NULL DEFAULT 0")
|
||||
except Exception:
|
||||
pass
|
||||
# Backfill positions for rows added before this column existed
|
||||
cur.execute("SELECT COUNT(*) FROM memory WHERE position > 0")
|
||||
if cur.fetchone()[0] == 0:
|
||||
cur.execute("UPDATE memory SET position = rowid")
|
||||
|
||||
cur.execute("""
|
||||
CREATE TABLE IF NOT EXISTS conversations (
|
||||
id TEXT PRIMARY KEY,
|
||||
created_at REAL NOT NULL,
|
||||
updated_at REAL NOT NULL,
|
||||
title TEXT
|
||||
)
|
||||
""")
|
||||
# Migrate: add title column if it doesn't exist yet
|
||||
try:
|
||||
cur.execute("ALTER TABLE conversations ADD COLUMN title TEXT")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
cur.execute("""
|
||||
CREATE TABLE IF NOT EXISTS messages (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
conversation_id TEXT NOT NULL,
|
||||
role TEXT NOT NULL,
|
||||
content TEXT NOT NULL,
|
||||
timestamp REAL NOT NULL,
|
||||
model TEXT,
|
||||
tokens INTEGER,
|
||||
FOREIGN KEY (conversation_id) REFERENCES conversations(id)
|
||||
)
|
||||
""")
|
||||
for col, typedef in (("model", "TEXT"), ("tokens", "INTEGER")):
|
||||
try:
|
||||
cur.execute(f"ALTER TABLE messages ADD COLUMN {col} {typedef}")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
cur.execute("""
|
||||
CREATE INDEX IF NOT EXISTS idx_messages_conversation_id
|
||||
ON messages (conversation_id)
|
||||
""")
|
||||
|
||||
# Semantic recall: one embedding vector per message, stored as JSON.
|
||||
# Backfilled lazily by semantic_search_conversations so existing history
|
||||
# gets indexed on first search.
|
||||
cur.execute("""
|
||||
CREATE TABLE IF NOT EXISTS message_vectors (
|
||||
message_id INTEGER PRIMARY KEY,
|
||||
embedding TEXT NOT NULL,
|
||||
FOREIGN KEY (message_id) REFERENCES messages(id)
|
||||
)
|
||||
""")
|
||||
|
||||
cur.execute("""
|
||||
CREATE TABLE IF NOT EXISTS settings (
|
||||
key TEXT PRIMARY KEY,
|
||||
value TEXT NOT NULL
|
||||
)
|
||||
""")
|
||||
cur.execute(
|
||||
"DELETE FROM settings WHERE key IN ('anthropic_api_key', 'escalation_model')"
|
||||
)
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
# -----------------------------
|
||||
# Loaders
|
||||
# -----------------------------
|
||||
def _load_all_memory(self) -> Dict[str, MemoryItem]:
|
||||
conn = self._connect()
|
||||
cur = conn.cursor()
|
||||
cur.execute("SELECT id, section, text, tags, position FROM memory ORDER BY position ASC, rowid ASC")
|
||||
rows = cur.fetchall()
|
||||
conn.close()
|
||||
|
||||
cache = {}
|
||||
for row in rows:
|
||||
try:
|
||||
tags = json.loads(row["tags"]) if row["tags"] else []
|
||||
except Exception:
|
||||
tags = []
|
||||
cache[row["id"]] = MemoryItem(
|
||||
id=row["id"],
|
||||
section=row["section"] or "General",
|
||||
text=row["text"],
|
||||
tags=tags,
|
||||
position=row["position"] or 0,
|
||||
)
|
||||
|
||||
return cache
|
||||
|
||||
# -----------------------------
|
||||
# Memory API
|
||||
# -----------------------------
|
||||
def add(self, item: MemoryItem):
|
||||
if not item.position:
|
||||
conn = self._connect()
|
||||
try:
|
||||
row = conn.execute(
|
||||
"SELECT position FROM memory WHERE id = ?", (item.id,)
|
||||
).fetchone()
|
||||
if row and row["position"]:
|
||||
item.position = row["position"]
|
||||
else:
|
||||
row = conn.execute("SELECT MAX(position) AS max_position FROM memory").fetchone()
|
||||
item.position = (row["max_position"] or 0) + 1
|
||||
finally:
|
||||
conn.close()
|
||||
self._cache[item.id] = item
|
||||
conn = self._connect()
|
||||
try:
|
||||
cur = conn.cursor()
|
||||
cur.execute(
|
||||
"INSERT OR REPLACE INTO memory (id, section, text, tags, position) VALUES (?, ?, ?, ?, ?)",
|
||||
(item.id, item.section or "General", item.text, json.dumps(item.tags), item.position)
|
||||
)
|
||||
conn.commit()
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def update(self, item: MemoryItem):
|
||||
existing = self.get(item.id)
|
||||
if existing:
|
||||
item.position = existing.position
|
||||
self.add(item)
|
||||
|
||||
def delete(self, item_id: str):
|
||||
self._cache.pop(item_id, None)
|
||||
conn = self._connect()
|
||||
try:
|
||||
cur = conn.cursor()
|
||||
cur.execute("DELETE FROM memory WHERE id = ?", (item_id,))
|
||||
conn.commit()
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def get(self, item_id: str) -> Optional[MemoryItem]:
|
||||
conn = self._connect()
|
||||
try:
|
||||
row = conn.execute(
|
||||
"SELECT id, section, text, tags, position FROM memory WHERE id = ?",
|
||||
(item_id,),
|
||||
).fetchone()
|
||||
finally:
|
||||
conn.close()
|
||||
if not row:
|
||||
self._cache.pop(item_id, None)
|
||||
return None
|
||||
try:
|
||||
tags = json.loads(row["tags"]) if row["tags"] else []
|
||||
except Exception:
|
||||
tags = []
|
||||
item = MemoryItem(
|
||||
id=row["id"], section=row["section"] or "General", text=row["text"],
|
||||
tags=tags, position=row["position"] or 0,
|
||||
)
|
||||
self._cache[item_id] = item
|
||||
return item
|
||||
|
||||
def all(self) -> List[MemoryItem]:
|
||||
# Always read from DB — the memory service and backend run in separate processes
|
||||
# with separate caches, so the cache can be stale for facts extracted by the
|
||||
# memory service after this process started.
|
||||
conn = self._connect()
|
||||
cur = conn.cursor()
|
||||
cur.execute("SELECT id, section, text, tags, position FROM memory ORDER BY position ASC, rowid ASC")
|
||||
rows = cur.fetchall()
|
||||
conn.close()
|
||||
items = []
|
||||
for row in rows:
|
||||
try:
|
||||
tags = json.loads(row["tags"]) if row["tags"] else []
|
||||
except Exception:
|
||||
tags = []
|
||||
items.append(MemoryItem(
|
||||
id=row["id"],
|
||||
section=row["section"] or "General",
|
||||
text=row["text"],
|
||||
tags=tags,
|
||||
position=row["position"] or 0,
|
||||
))
|
||||
return items
|
||||
|
||||
def reorder_section(self, section: str, ordered_ids: List[str]) -> bool:
|
||||
"""Rewrite the order of items in a section using its existing position pool.
|
||||
ordered_ids must contain exactly the ids currently in the section."""
|
||||
section_norm = section or "General"
|
||||
in_section = [i for i in self.all() if (i.section or "General") == section_norm]
|
||||
if len(in_section) != len(ordered_ids):
|
||||
return False
|
||||
by_id = {i.id: i for i in in_section}
|
||||
siblings = []
|
||||
for id_ in ordered_ids:
|
||||
if id_ not in by_id:
|
||||
return False
|
||||
siblings.append(by_id[id_])
|
||||
positions = sorted([i.position for i in in_section])
|
||||
conn = self._connect()
|
||||
try:
|
||||
cur = conn.cursor()
|
||||
for s, new_pos in zip(siblings, positions):
|
||||
s.position = new_pos
|
||||
cur.execute("UPDATE memory SET position = ? WHERE id = ?", (new_pos, s.id))
|
||||
conn.commit()
|
||||
finally:
|
||||
conn.close()
|
||||
return True
|
||||
|
||||
# -----------------------------
|
||||
# Conversation API
|
||||
# -----------------------------
|
||||
def create_conversation(self, conversation_id: str) -> ConversationItem:
|
||||
now = time.time()
|
||||
conn = self._connect()
|
||||
try:
|
||||
cur = conn.cursor()
|
||||
cur.execute(
|
||||
"INSERT OR IGNORE INTO conversations (id, created_at, updated_at) VALUES (?, ?, ?)",
|
||||
(conversation_id, now, now)
|
||||
)
|
||||
conn.commit()
|
||||
finally:
|
||||
conn.close()
|
||||
return ConversationItem(id=conversation_id, created_at=now, updated_at=now)
|
||||
|
||||
def set_conversation_title(self, conversation_id: str, title: str):
|
||||
conn = self._connect()
|
||||
try:
|
||||
cur = conn.cursor()
|
||||
cur.execute(
|
||||
"UPDATE conversations SET title = ? WHERE id = ?",
|
||||
(title, conversation_id),
|
||||
)
|
||||
conn.commit()
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def add_message(self, conversation_id: str, role: str, content: str, model: Optional[str] = None, tokens: Optional[int] = None):
|
||||
now = time.time()
|
||||
conn = self._connect()
|
||||
try:
|
||||
cur = conn.cursor()
|
||||
cur.execute(
|
||||
"INSERT INTO messages (conversation_id, role, content, timestamp, model, tokens) VALUES (?, ?, ?, ?, ?, ?)",
|
||||
(conversation_id, role, content, now, model, tokens)
|
||||
)
|
||||
message_id = cur.lastrowid
|
||||
cur.execute(
|
||||
"UPDATE conversations SET updated_at = ? WHERE id = ?",
|
||||
(now, conversation_id)
|
||||
)
|
||||
conn.commit()
|
||||
return message_id
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def get_conversation(self, conversation_id: str) -> Optional[ConversationItem]:
|
||||
conn = self._connect()
|
||||
cur = conn.cursor()
|
||||
cur.execute("SELECT * FROM conversations WHERE id = ?", (conversation_id,))
|
||||
row = cur.fetchone()
|
||||
if not row:
|
||||
conn.close()
|
||||
return None
|
||||
cur.execute(
|
||||
"SELECT role, content, timestamp, model, tokens FROM messages WHERE conversation_id = ? ORDER BY timestamp ASC",
|
||||
(conversation_id,)
|
||||
)
|
||||
msg_rows = cur.fetchall()
|
||||
conn.close()
|
||||
|
||||
messages = [MessageItem(role=r["role"], content=r["content"], timestamp=r["timestamp"], model=r["model"], tokens=r["tokens"]) for r in msg_rows]
|
||||
return ConversationItem(
|
||||
id=row["id"],
|
||||
messages=messages,
|
||||
created_at=row["created_at"],
|
||||
updated_at=row["updated_at"],
|
||||
title=row["title"] if "title" in row.keys() else None,
|
||||
)
|
||||
|
||||
def all_conversations(self) -> List[ConversationItem]:
|
||||
conn = self._connect()
|
||||
cur = conn.cursor()
|
||||
cur.execute("""
|
||||
SELECT c.id, c.created_at, c.updated_at, c.title,
|
||||
m.role, m.content, m.timestamp, m.model, m.tokens
|
||||
FROM conversations c
|
||||
LEFT JOIN messages m ON m.conversation_id = c.id
|
||||
ORDER BY c.updated_at DESC, m.timestamp ASC
|
||||
""")
|
||||
rows = cur.fetchall()
|
||||
conn.close()
|
||||
|
||||
convs: Dict[str, ConversationItem] = {}
|
||||
order: list[str] = []
|
||||
for row in rows:
|
||||
cid = row["id"]
|
||||
if cid not in convs:
|
||||
convs[cid] = ConversationItem(
|
||||
id=cid,
|
||||
created_at=row["created_at"],
|
||||
updated_at=row["updated_at"],
|
||||
title=row["title"],
|
||||
)
|
||||
order.append(cid)
|
||||
if row["role"] is not None:
|
||||
convs[cid].messages.append(
|
||||
MessageItem(role=row["role"], content=row["content"], timestamp=row["timestamp"], model=row["model"], tokens=row["tokens"])
|
||||
)
|
||||
return [convs[cid] for cid in order]
|
||||
|
||||
def delete_conversation(self, conversation_id: str):
|
||||
conn = self._connect()
|
||||
try:
|
||||
cur = conn.cursor()
|
||||
cur.execute("DELETE FROM messages WHERE conversation_id = ?", (conversation_id,))
|
||||
cur.execute("DELETE FROM conversations WHERE id = ?", (conversation_id,))
|
||||
conn.commit()
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
# -----------------------------
|
||||
# Search API
|
||||
# -----------------------------
|
||||
def search_conversations(self, query: str, limit: int = 3) -> List[dict]:
|
||||
"""Return up to `limit` conversations that contain the query string,
|
||||
with full user+assistant exchange pairs around each match."""
|
||||
if not query or not query.strip():
|
||||
return []
|
||||
q = query.strip().lower()
|
||||
conn = self._connect()
|
||||
cur = conn.cursor()
|
||||
cur.execute("""
|
||||
SELECT DISTINCT c.id, c.created_at, c.updated_at
|
||||
FROM conversations c
|
||||
JOIN messages m ON m.conversation_id = c.id
|
||||
WHERE LOWER(m.content) LIKE ?
|
||||
ORDER BY c.updated_at DESC
|
||||
LIMIT ?
|
||||
""", (f"%{q}%", limit))
|
||||
rows = cur.fetchall()
|
||||
|
||||
results = []
|
||||
for row in rows:
|
||||
# Load all messages in order so we can find complete exchange pairs
|
||||
cur.execute("""
|
||||
SELECT role, content FROM messages
|
||||
WHERE conversation_id = ?
|
||||
ORDER BY timestamp ASC
|
||||
""", (row["id"],))
|
||||
all_msgs = [{"role": r["role"], "content": r["content"]} for r in cur.fetchall()]
|
||||
|
||||
# For each matching message, collect the full user+assistant pair around it
|
||||
seen_pairs: set = set()
|
||||
matches = []
|
||||
for i, msg in enumerate(all_msgs):
|
||||
if q not in msg["content"].lower():
|
||||
continue
|
||||
if msg["role"] == "user":
|
||||
start, end = i, i + 1 if i + 1 < len(all_msgs) else i
|
||||
else:
|
||||
start, end = (i - 1 if i > 0 else i), i
|
||||
if (start, end) in seen_pairs:
|
||||
continue
|
||||
seen_pairs.add((start, end))
|
||||
for m in all_msgs[start:end + 1]:
|
||||
matches.append({"role": m["role"], "content": m["content"][:500]})
|
||||
if len(seen_pairs) >= 2:
|
||||
break
|
||||
|
||||
if matches:
|
||||
results.append({
|
||||
"id": row["id"],
|
||||
"updated_at": row["updated_at"],
|
||||
"matches": matches,
|
||||
})
|
||||
|
||||
conn.close()
|
||||
return results
|
||||
|
||||
# nomic-embed-text is an asymmetric retrieval model: queries and stored
|
||||
# documents must be embedded with these task prefixes or similarity collapses
|
||||
# into noise. Cached vectors are document-embeddings (search_document:).
|
||||
_EMBED_QUERY_PREFIX = "search_query: "
|
||||
_EMBED_DOC_PREFIX = "search_document: "
|
||||
|
||||
async def semantic_search_conversations(
|
||||
self, query: str, embed_fn, limit: int = 3, min_score: float = 0.6
|
||||
) -> List[dict]:
|
||||
"""Recall past conversations relevant to `query` using hybrid retrieval.
|
||||
|
||||
Combines semantic similarity (embeddings — finds reworded matches with no
|
||||
shared keywords) with the existing lexical substring match (catches exact
|
||||
terms the embedding underweights), unioned and deduped by conversation.
|
||||
|
||||
`embed_fn` is an async callable returning an embedding vector for a string
|
||||
(typically OllamaManager.embed). Messages without a stored vector are
|
||||
embedded and cached on first use (lazy backfill). If embeddings are
|
||||
unavailable (no model / Ollama down) this degrades to pure lexical match,
|
||||
so recall never silently breaks.
|
||||
|
||||
Returns the same shape as `search_conversations`: a list of
|
||||
{id, updated_at, matches:[{role, content}]} with full user+assistant
|
||||
pairs around each match.
|
||||
"""
|
||||
if not query or not query.strip():
|
||||
return []
|
||||
|
||||
query_vec = await embed_fn(self._EMBED_QUERY_PREFIX + query.strip())
|
||||
if not query_vec:
|
||||
return self.search_conversations(query, limit=limit)
|
||||
|
||||
conn = self._connect()
|
||||
cur = conn.cursor()
|
||||
|
||||
# Lazy backfill: embed any messages that don't have a vector yet.
|
||||
cur.execute("""
|
||||
SELECT m.id, m.content
|
||||
FROM messages m
|
||||
LEFT JOIN message_vectors v ON v.message_id = m.id
|
||||
WHERE v.message_id IS NULL AND TRIM(m.content) != ''
|
||||
""")
|
||||
missing = cur.fetchall()
|
||||
for row in missing:
|
||||
vec = await embed_fn(self._EMBED_DOC_PREFIX + row["content"][:2000])
|
||||
if vec:
|
||||
cur.execute(
|
||||
"INSERT OR REPLACE INTO message_vectors (message_id, embedding) VALUES (?, ?)",
|
||||
(row["id"], json.dumps(vec)),
|
||||
)
|
||||
if missing:
|
||||
conn.commit()
|
||||
|
||||
# Score every stored message against the query vector.
|
||||
cur.execute("""
|
||||
SELECT v.message_id, v.embedding, m.conversation_id
|
||||
FROM message_vectors v
|
||||
JOIN messages m ON m.id = v.message_id
|
||||
""")
|
||||
scored = []
|
||||
for row in cur.fetchall():
|
||||
try:
|
||||
vec = json.loads(row["embedding"])
|
||||
except Exception:
|
||||
continue
|
||||
score = _cosine(query_vec, vec)
|
||||
if score >= min_score:
|
||||
scored.append((score, row["message_id"], row["conversation_id"]))
|
||||
|
||||
scored.sort(reverse=True)
|
||||
|
||||
results: List[dict] = []
|
||||
seen_convs: set = set()
|
||||
for score, message_id, conv_id in scored:
|
||||
if len(results) >= limit:
|
||||
break
|
||||
if conv_id in seen_convs:
|
||||
continue
|
||||
pair = self._exchange_pair(cur, conv_id, message_id)
|
||||
if pair:
|
||||
seen_convs.add(conv_id)
|
||||
results.append(pair)
|
||||
|
||||
conn.close()
|
||||
|
||||
# Hybrid union: fill any remaining slots with lexical matches the
|
||||
# embedding missed (e.g. exact proper nouns), skipping dupes.
|
||||
if len(results) < limit:
|
||||
for conv in self.search_conversations(query, limit=limit):
|
||||
if conv["id"] not in seen_convs:
|
||||
seen_convs.add(conv["id"])
|
||||
results.append(conv)
|
||||
if len(results) >= limit:
|
||||
break
|
||||
|
||||
return results
|
||||
|
||||
def _exchange_pair(self, cur, conv_id: str, message_id: int) -> Optional[dict]:
|
||||
"""Build a {id, updated_at, matches} record with the full user+assistant
|
||||
pair surrounding `message_id`, in the shape search callers expect."""
|
||||
cur.execute(
|
||||
"SELECT id, role, content FROM messages WHERE conversation_id = ? ORDER BY timestamp ASC",
|
||||
(conv_id,),
|
||||
)
|
||||
all_msgs = cur.fetchall()
|
||||
idx = next((i for i, m in enumerate(all_msgs) if m["id"] == message_id), None)
|
||||
if idx is None:
|
||||
return None
|
||||
if all_msgs[idx]["role"] == "user":
|
||||
start, end = idx, min(idx + 1, len(all_msgs) - 1)
|
||||
else:
|
||||
start, end = max(idx - 1, 0), idx
|
||||
matches = [
|
||||
{"role": all_msgs[i]["role"], "content": all_msgs[i]["content"][:500]}
|
||||
for i in range(start, end + 1)
|
||||
]
|
||||
cur.execute("SELECT updated_at FROM conversations WHERE id = ?", (conv_id,))
|
||||
crow = cur.fetchone()
|
||||
return {
|
||||
"id": conv_id,
|
||||
"updated_at": crow["updated_at"] if crow else 0,
|
||||
"matches": matches,
|
||||
}
|
||||
|
||||
# -----------------------------
|
||||
# Settings API
|
||||
# -----------------------------
|
||||
_SETTINGS_DEFAULTS: Dict[str, Any] = {
|
||||
"model": "",
|
||||
# Qwen3-style reasoning. Off by default: the hidden <think> block is pure
|
||||
# latency for chat/memory. Turn on for hard multi-step problems.
|
||||
"think": False,
|
||||
"temperature": 0.7,
|
||||
"system_prompt": "",
|
||||
"timeout": 120,
|
||||
# How long Ollama keeps the model resident in VRAM between messages.
|
||||
# "30m"/"-1" (never unload)/"0" (unload now). Avoids cold-reload latency
|
||||
# when you return to an idle chat. Empty → Ollama's 5-minute default.
|
||||
"keep_alive": "30m",
|
||||
# CPU/GPU offload: -1 = Auto (Ollama auto-fits layers to VRAM).
|
||||
# 0–100 = percent of model layers to force onto the GPU; the
|
||||
# remainder runs on CPU. See OllamaManager.get_model_layers.
|
||||
"gpu_offload": -1,
|
||||
# Memory curator (the model that extracts facts after each exchange).
|
||||
# Empty → same auto-selected model as chat. A dedicated model (e.g.
|
||||
# "mistral:latest") gives better extraction but must share VRAM.
|
||||
"memory_model": DEFAULT_MEMORY_MODEL,
|
||||
# Curator CPU/GPU offload — same scale as gpu_offload above. Default 0
|
||||
# (all CPU/RAM): OS-neutral and never evicts the chat model from a small
|
||||
# GPU. Boxes with spare VRAM can set -1 (Auto) or a percent to use the GPU.
|
||||
"memory_gpu_offload": 0,
|
||||
# Similar-fact merge: when a newly extracted fact's embedding is at least
|
||||
# this cosine-similar to an existing fact, UPDATE that fact in place
|
||||
# instead of appending a duplicate ("edit with new info"). 0 disables
|
||||
# (always append). Calibrated on nomic-embed-text: genuine updates
|
||||
# (mileage/title/location changes) score 0.81–0.99, while distinct facts
|
||||
# top out ~0.61 — so 0.80 catches updates and never merges unrelated
|
||||
# facts. Lower to catch looser rephrases; raise toward 1.0 to be stricter.
|
||||
"memory_merge_threshold": 0.80,
|
||||
}
|
||||
|
||||
def get_settings(self) -> Dict[str, Any]:
|
||||
conn = self._connect()
|
||||
cur = conn.cursor()
|
||||
cur.execute("SELECT key, value FROM settings")
|
||||
rows = cur.fetchall()
|
||||
conn.close()
|
||||
result = dict(self._SETTINGS_DEFAULTS)
|
||||
for row in rows:
|
||||
try:
|
||||
result[row["key"]] = json.loads(row["value"])
|
||||
except Exception:
|
||||
result[row["key"]] = row["value"]
|
||||
return result
|
||||
|
||||
def update_settings(self, data: Dict[str, Any]):
|
||||
conn = self._connect()
|
||||
try:
|
||||
cur = conn.cursor()
|
||||
for key, value in data.items():
|
||||
if key in self._SETTINGS_DEFAULTS:
|
||||
cur.execute(
|
||||
"INSERT OR REPLACE INTO settings (key, value) VALUES (?, ?)",
|
||||
(key, json.dumps(value))
|
||||
)
|
||||
conn.commit()
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
# -----------------------------
|
||||
# Store Instance
|
||||
# -----------------------------
|
||||
from ..nexus_config import MEMORY_DB
|
||||
|
||||
DB_PATH = MEMORY_DB
|
||||
store = PersistentMemoryStore(DB_PATH)
|
||||
Reference in New Issue
Block a user
GitNexus · part of the Nexus stack