mem0-python-server/mem0server.py

import os
import math
import json
import sqlite3
import httpx
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, FileResponse, HTMLResponse
from mem0 import Memory

# =============================================================================
# ENVIRONMENT
# =============================================================================

GROQ_API_KEY = os.environ.get("GROQ_API_KEY")
if not GROQ_API_KEY:
    raise RuntimeError("GROQ_API_KEY environment variable is not set.")

RERANKER_URL = os.environ.get("RERANKER_URL", "http://192.168.0.200:5200/rerank")
SQLITE_PATH  = os.path.expanduser("~/.mem0/history.db")

# =============================================================================
# SAFE JSON RESPONSE  (handles Infinity / NaN from Chroma / reranker scores)
# =============================================================================

def _sanitize(obj):
    if isinstance(obj, float):
        if math.isnan(obj) or math.isinf(obj):
            return None
    if isinstance(obj, dict):
        return {k: _sanitize(v) for k, v in obj.items()}
    if isinstance(obj, list):
        return [_sanitize(i) for i in obj]
    return obj


class SafeJSONResponse(JSONResponse):
    def render(self, content) -> bytes:
        return json.dumps(_sanitize(content), ensure_ascii=False).encode("utf-8")


# =============================================================================
# PROMPTS
# Edit these to change how each collection extracts and stores facts.
# =============================================================================

PROMPTS = {
    # Used by /memories — conversational, user-centric recall for OpenClaw.
    "conversational": {
        "fact_extraction": """
You are a personal memory assistant. Extract concise, standalone facts about
the user from the conversation below. Write each fact as a single sentence
starting with "User" — for example:
  - "User is interested in generative music."
  - "User is familiar with Python async patterns."
  - "User prefers dark mode interfaces."
Only extract facts that are clearly stated or strongly implied. Ignore filler,
greetings, and opinions the user is uncertain about.
""".strip(),
        "update_memory": """
You manage a long-term memory database for a personal AI assistant.
You receive existing memories and new information. Update, merge, or add
memories as needed. Keep each memory as a single concise sentence starting
with "User". Remove duplicates and outdated facts.
""".strip(),
    },

    # Used by /knowledge — objective, source-neutral facts for book/doc ingest.
    "knowledge": {
        "fact_extraction": """
You are a knowledge extraction system that reads source material and produces
a list of objective, encyclopedic facts. Write each fact as a precise,
self-contained sentence. Do NOT reframe facts as user preferences or interests.
Preserve names, terminology, and relationships exactly as they appear.
Examples:
  - "Silvio Gesell proposed demurrage as a mechanism to discourage hoarding of currency."
  - "The MIDI standard uses a 7-bit checksum for SysEx message validation."
Only extract verifiable facts. Ignore meta-commentary and transitional prose.
""".strip(),
        "update_memory": """
You manage a knowledge base that stores objective facts extracted from books,
documents, and reference material. You receive existing facts and new
information. Update, merge, or add facts as needed. Keep each fact as a
precise, self-contained sentence. Remove duplicates and outdated entries.
""".strip(),
    },
}

# =============================================================================
# MEM0 CONFIG FACTORY
# =============================================================================

def make_config(collection_name: str, prompt_key: str) -> dict:
    return {
        "llm": {
            "provider": "groq",
            "config": {
                "model": "meta-llama/llama-4-scout-17b-16e-instruct",
                "temperature": 0.025,
                "max_tokens": 1500,
            },
        },
        "vector_store": {
            "provider": "chroma",
            "config": {
                "host": "192.168.0.200",
                "port": 8001,
                "collection_name": collection_name,
            },
        },
        "embedder": {
            "provider": "ollama",
            "config": {
                "model": "nomic-embed-text",
                "ollama_base_url": "http://192.168.0.200:11434",
            },
        },
        "custom_prompts": PROMPTS[prompt_key],
    }


# =============================================================================
# MEMORY INSTANCES
# =============================================================================

memory_conv = Memory.from_config(make_config("openclaw_mem", "conversational"))
memory_know = Memory.from_config(make_config("knowledge_mem", "knowledge"))

# =============================================================================
# CHROMA EMPTY-FILTER PATCH  (applied to both instances)
# =============================================================================

NOOP_WHERE = {"$and": [
    {"user_id": {"$ne": ""}},
    {"user_id": {"$ne": ""}},
]}


def is_effectively_empty(filters) -> bool:
    if not filters:
        return True
    if filters in ({"AND": []}, {"OR": []}):
        return True
    return False


def make_safe_search(mem_instance: Memory):
    orig = mem_instance.vector_store.search

    def safe_search(query, vectors, limit=10, filters=None):
        if is_effectively_empty(filters):
            return mem_instance.vector_store.collection.query(
                query_embeddings=vectors,
                n_results=limit,
                where=NOOP_WHERE,
            )
        try:
            return orig(query=query, vectors=vectors, limit=limit, filters=filters)
        except Exception as e:
            if "Expected where" in str(e):
                return mem_instance.vector_store.collection.query(
                    query_embeddings=vectors,
                    n_results=limit,
                    where=NOOP_WHERE,
                )
            raise

    return safe_search


memory_conv.vector_store.search = make_safe_search(memory_conv)
memory_know.vector_store.search = make_safe_search(memory_know)

# =============================================================================
# RERANKER
# =============================================================================

def rerank_results(query: str, items: list, top_k: int) -> list:
    """Re-order results via local reranker. Falls back gracefully."""
    if not items:
        return items

    documents = [r.get("memory", "") for r in items]
    try:
        resp = httpx.post(
            RERANKER_URL,
            json={"query": query, "documents": documents, "top_k": top_k},
            timeout=5.0,
        )
        resp.raise_for_status()
        reranked = resp.json()["results"]
    except Exception as exc:
        print(f"[reranker] unavailable, skipping rerank: {exc}")
        return items[:top_k]

    text_to_meta = {r.get("memory", ""): r for r in items}
    merged = []
    for r in reranked:
        meta = text_to_meta.get(r["text"])
        if meta:
            merged.append({**meta, "rerank_score": r["score"]})
    return merged

# =============================================================================
# SQLITE HELPER
# =============================================================================

def sqlite_delete_ids(memory_ids: list[str]) -> int:
    """Delete rows from mem0 SQLite by memory_id. Returns count deleted."""
    if not memory_ids:
        return 0
    try:
        conn = sqlite3.connect(SQLITE_PATH)
        cur  = conn.cursor()
        placeholders = ",".join("?" * len(memory_ids))
        cur.execute(
            f"DELETE FROM history WHERE memory_id IN ({placeholders})",
            memory_ids
        )
        deleted = cur.rowcount
        conn.commit()
        conn.close()
        return deleted
    except Exception as e:
        print(f"[sqlite] warning: {e}")
        return 0

# =============================================================================
# CHROMA PAGINATION HELPER
# =============================================================================

def chroma_get_all(collection, user_id: str, include: list = None) -> list[dict]:
    """
    Page through a Chroma collection in batches, filtering by user_id.
    Returns list of dicts with 'id' and any included fields.
    Bypasses mem0's 100-entry cap entirely.
    """
    if include is None:
        include = ["metadatas"]

    results = []
    batch   = 500
    offset  = 0

    while True:
        page = collection.get(
            where={"user_id": {"$eq": user_id}},
            limit=batch,
            offset=offset,
            include=include,
        )
        ids = page.get("ids", [])
        if not ids:
            break

        for i, id_ in enumerate(ids):
            row = {"id": id_}
            for field in include:
                values = page.get(field, [])
                row[field[:-1]] = values[i] if i < len(values) else None
            results.append(row)

        offset += len(ids)
        if len(ids) < batch:
            break

    return results

# =============================================================================
# SHARED HANDLERS
# =============================================================================

def extract_user_id(data: dict) -> str:
    return data.get("userId") or data.get("user_id") or "default"


async def handle_add(req: Request, mem: Memory, verbatim_allowed: bool = False):
    """
    Shared add handler for /memories and /knowledge.

    /knowledge  (verbatim_allowed=True)  — always stores verbatim (infer=False).
                                           The ingestor already summarised; skip
                                           the second LLM pass.
    /memories   (verbatim_allowed=False) — always uses LLM extraction for
                                           conversational recall.

    Supports:
      - text      — raw string (legacy)
      - messages  — list of {role, content} dicts (standard mem0)
      - metadata  — dict, passed through to mem0
      - user_id / userId
    """
    data     = await req.json()
    user_id  = extract_user_id(data)
    metadata = data.get("metadata") or {}
    messages = data.get("messages")
    text     = data.get("text")

    if not messages and not text:
        return SafeJSONResponse(
            content={"error": "Provide 'text' or 'messages'"}, status_code=400
        )

    if verbatim_allowed:
        # /knowledge — always verbatim, ingestor already summarised
        content = text or " ".join(
            m["content"] for m in messages if m.get("role") == "user"
        )
        result = mem.add(content, user_id=user_id, metadata=metadata, infer=False)
        print(f"[add verbatim] user={user_id} chars={len(content)} meta={metadata}")
        return SafeJSONResponse(content=result)

    # /memories — always LLM extraction
    if messages:
        result = mem.add(messages, user_id=user_id, metadata=metadata)
    else:
        result = mem.add(text, user_id=user_id, metadata=metadata)

    print(f"[add conversational] user={user_id} meta={metadata}")
    return SafeJSONResponse(content=result)


async def handle_search(req: Request, mem: Memory):
    data    = await req.json()
    query   = (data.get("query") or "").strip()
    user_id = extract_user_id(data)
    limit   = int(data.get("limit", 5))

    if not query:
        return SafeJSONResponse(content={"results": []})

    fetch_k = max(limit * 3, 15)
    try:
        result = mem.search(query, user_id=user_id, limit=fetch_k)
    except Exception:
        all_res = mem.get_all(user_id=user_id)
        items   = (
            all_res.get("results", [])
            if isinstance(all_res, dict)
            else (all_res if isinstance(all_res, list) else [])
        )
        q     = query.lower()
        items = [r for r in items if q in r.get("memory", "").lower()]
        result = {"results": items}

    items = result.get("results", [])
    items = rerank_results(query, items, top_k=limit)
    print(f"[search] user={user_id} query={query!r} hits={len(items)}")
    return SafeJSONResponse(content={"results": items})


async def handle_recent(req: Request, mem: Memory):
    data    = await req.json()
    user_id = extract_user_id(data)
    if not user_id:
        return SafeJSONResponse(content={"error": "Missing userId"}, status_code=400)
    limit = int(data.get("limit", 5))

    try:
        results = mem.get_all(user_id=user_id)
    except Exception:
        results = mem.search(query="recent", user_id=user_id)

    items = results.get("results", [])
    items = sorted(items, key=lambda r: r.get("created_at", ""), reverse=True)
    return SafeJSONResponse(content={"results": items[:limit]})

# =============================================================================
# APP
# =============================================================================

app = FastAPI(title="mem0 server")


# ---------------------------------------------------------------------------
# DASHBOARD
# ---------------------------------------------------------------------------

DASHBOARD_HTML = open("dashboard.html").read()

@app.get("/dashboard")
async def dashboard():
    return HTMLResponse(content=DASHBOARD_HTML)


# ---------------------------------------------------------------------------
# HEALTH
# ---------------------------------------------------------------------------

@app.get("/health")
async def health():
    return SafeJSONResponse(content={
        "status": "ok",
        "reranker_url": RERANKER_URL,
        "collections": {
            "conversational": "openclaw_mem",
            "knowledge":      "knowledge_mem",
        },
        "prompts": {
            k: {pk: pv[:80] + "…" for pk, pv in pv_dict.items()}
            for k, pv_dict in PROMPTS.items()
        },
    })


# ---------------------------------------------------------------------------
# /memories  — conversational, OpenClaw
# ---------------------------------------------------------------------------

@app.post("/memories")
async def add_memory(req: Request):
    return await handle_add(req, memory_conv, verbatim_allowed=False)


@app.post("/memories/search")
async def search_memories(req: Request):
    return await handle_search(req, memory_conv)


@app.post("/memories/recent")
async def recent_memories(req: Request):
    return await handle_recent(req, memory_conv)


@app.delete("/memories")
async def delete_memory(req: Request):
    data = await req.json()
    return SafeJSONResponse(content=memory_conv.delete(data.get("filter", {})))


# ---------------------------------------------------------------------------
# /knowledge  — objective facts, book-ingestor
# ---------------------------------------------------------------------------

@app.post("/knowledge")
async def add_knowledge(req: Request):
    return await handle_add(req, memory_know, verbatim_allowed=True)


@app.post("/knowledge/search")
async def search_knowledge(req: Request):
    return await handle_search(req, memory_know)


@app.post("/knowledge/recent")
async def recent_knowledge(req: Request):
    return await handle_recent(req, memory_know)


@app.delete("/knowledge")
async def delete_knowledge(req: Request):
    data = await req.json()
    return SafeJSONResponse(content=memory_know.delete(data.get("filter", {})))


@app.post("/knowledge/sources")
async def knowledge_sources(req: Request):
    """
    Return distinct source_file values with entry counts.
    Pages through Chroma directly — no mem0 100-entry cap.
    """
    data    = await req.json()
    user_id = extract_user_id(data) or "knowledge_base"

    rows   = chroma_get_all(memory_know.vector_store.collection, user_id)
    counts = {}
    for row in rows:
        src = (row.get("metadata") or {}).get("source_file", "(no source)")
        counts[src] = counts.get(src, 0) + 1

    sources = [
        {"source_file": k, "count": v}
        for k, v in sorted(counts.items(), key=lambda x: -x[1])
    ]
    print(f"[sources] user={user_id} total={len(rows)} books={len(sources)}")
    return SafeJSONResponse(content={"sources": sources, "total": len(rows)})


@app.delete("/knowledge/by-source")
async def delete_knowledge_by_source(req: Request):
    """
    Delete all knowledge entries for a given source_file.
    Pages through Chroma directly, then cleans SQLite.
    """
    data        = await req.json()
    source_file = data.get("source_file")
    user_id     = extract_user_id(data) or "knowledge_base"

    if not source_file:
        return SafeJSONResponse(
            content={"error": "Missing source_file"}, status_code=400
        )

    rows = chroma_get_all(memory_know.vector_store.collection, user_id)
    to_delete = [
        row["id"] for row in rows
        if (row.get("metadata") or {}).get("source_file") == source_file
    ]

    if not to_delete:
        return SafeJSONResponse(
            content={"deleted": 0, "message": "no entries found for that source"}
        )

    # 1. Chroma bulk delete
    try:
        memory_know.vector_store.collection.delete(ids=to_delete)
    except Exception as e:
        return SafeJSONResponse(
            content={"error": f"chroma delete failed: {e}"}, status_code=500
        )

    # 2. SQLite cleanup
    sqlite_deleted = sqlite_delete_ids(to_delete)

    print(f"[delete by-source] source={source_file} "
          f"chroma={len(to_delete)} sqlite={sqlite_deleted}")

    return SafeJSONResponse(content={
        "deleted":        len(to_delete),
        "sqlite_deleted": sqlite_deleted,
        "source_file":    source_file,
    })


# ---------------------------------------------------------------------------
# /memory/{id}  — single entry delete (knowledge or conversational)
# ---------------------------------------------------------------------------

@app.delete("/memory/{memory_id}")
async def delete_single_memory(memory_id: str, req: Request):
    """
    Delete a single memory by ID from either collection.
    Body: { "collection": "knowledge" | "conversational" }
    Cleans both Chroma and SQLite.
    """
    data       = await req.json()
    collection = data.get("collection", "knowledge")
    mem        = memory_know if collection == "knowledge" else memory_conv

    # 1. Chroma delete
    try:
        mem.vector_store.collection.delete(ids=[memory_id])
    except Exception as e:
        return SafeJSONResponse(
            content={"error": f"chroma delete failed: {e}"}, status_code=500
        )

    # 2. SQLite cleanup
    sqlite_delete_ids([memory_id])

    print(f"[delete single] id={memory_id} collection={collection}")
    return SafeJSONResponse(content={"deleted": memory_id})


# ---------------------------------------------------------------------------
# /search  — merged results from both collections (OpenClaw autorecall)
# ---------------------------------------------------------------------------

@app.post("/search")
async def search_all(req: Request):
    """
    Query both collections and merge results.
    Results tagged with _source: conversational | knowledge.
    Accepts same payload as /memories/search.
    """
    data    = await req.json()
    query   = (data.get("query") or "").strip()
    user_id = extract_user_id(data)
    limit   = int(data.get("limit", 5))

    if not query:
        return SafeJSONResponse(content={"results": []})

    fetch_k = max(limit * 3, 15)

    def fetch(mem: Memory, tag: str):
        try:
            r     = mem.search(query, user_id=user_id, limit=fetch_k)
            items = r.get("results", [])
        except Exception:
            items = []
        for item in items:
            item["_source"] = tag
        return items

    conv_items = fetch(memory_conv, "conversational")
    know_items = fetch(memory_know, "knowledge")
    merged     = rerank_results(query, conv_items + know_items, top_k=limit)

    print(
        f"[search/all] user={user_id} query={query!r} "
        f"conv={len(conv_items)} know={len(know_items)} merged={len(merged)}"
    )
    return SafeJSONResponse(content={"results": merged})