quickstart.py

"""
MemoryMesh Quick-Start Demo
===========================
Run this after starting the API server:
  uvicorn memorymesh.api.main:app --reload

This script demonstrates the full lifecycle:
  1. Store memories of every type
  2. Hybrid semantic search
  3. Type-filtered search
  4. Memory graph
  5. Stats
  6. Update & delete
"""

import asyncio
import json

import httpx

API = "http://localhost:8000/v1"

MEMORIES = [
    {
        "content": "Devdutt S is a software engineer from Kochi, Kerala. GitHub: 0DevDutt0. LinkedIn: devdutts.",
        "agent_id": "demo-agent",
        "memory_type": "semantic",
        "importance": 0.95,
        "metadata": {"category": "identity"},
    },
    {
        "content": "On 2026-06-15 the user built MemoryMesh, a production-grade persistent memory server with MCP support.",
        "agent_id": "demo-agent",
        "memory_type": "episodic",
        "importance": 0.9,
        "metadata": {"project": "MemoryMesh", "date": "2026-06-15"},
    },
    {
        "content": "To launch MemoryMesh: pip install -e '.[dev]', then uvicorn memorymesh.api.main:app --reload",
        "agent_id": "demo-agent",
        "memory_type": "procedural",
        "importance": 0.85,
        "metadata": {"category": "setup"},
    },
    {
        "content": "The user prefers concise answers, dark mode UI, Python over JavaScript, asyncio over threading, and pytest for testing.",
        "agent_id": "demo-agent",
        "memory_type": "preference",
        "importance": 0.8,
        "metadata": {"category": "style"},
    },
    {
        "content": "MemoryMesh uses BAAI/bge-large-en-v1.5 for 1024-dimensional embeddings with FAISS for vector search.",
        "agent_id": "demo-agent",
        "memory_type": "semantic",
        "importance": 0.75,
        "metadata": {"category": "architecture"},
    },
    {
        "content": "The user asked how hierarchical compression works in MemoryMesh: Groq handles tier-1, Mistral handles tier-2 cluster merges.",
        "agent_id": "demo-agent",
        "memory_type": "episodic",
        "importance": 0.7,
        "metadata": {"topic": "compression"},
    },
    {
        "content": "Decay formula: R = exp(-t/S) where S depends on memory type multiplier, importance, and access count.",
        "agent_id": "demo-agent",
        "memory_type": "procedural",
        "importance": 0.8,
        "metadata": {"category": "algorithm"},
    },
]


def _box(title: str) -> str:
    line = "─" * (len(title) + 4)
    return f"\n┌{line}┐\n│  {title}  │\n└{line}┘"


async def run_demo() -> None:
    async with httpx.AsyncClient(timeout=30.0) as client:

        # ── Health check ─────────────────────────────────────────────
        print(_box("1. Health Check"))
        r = await client.get("http://localhost:8000/health")
        print(json.dumps(r.json(), indent=2))

        # ── Store all memories ───────────────────────────────────────
        print(_box("2. Storing 7 Memories"))
        stored_ids: list[str] = []
        for mem in MEMORIES:
            r = await client.post(f"{API}/memories/", json=mem)
            r.raise_for_status()
            m = r.json()
            stored_ids.append(m["id"])
            print(f"  ✓ [{m['memory_type']:12s}] id={m['id'][:8]}…  importance={m['importance']}")

        # ── Stats ────────────────────────────────────────────────────
        print(_box("3. Memory Stats"))
        r = await client.get(f"{API}/stats")
        stats = r.json()
        print(json.dumps(stats, indent=2))

        # ── Semantic search ──────────────────────────────────────────
        print(_box("4. Semantic Search — 'user communication style'"))
        r = await client.post(f"{API}/memories/search", json={
            "query": "how does the user like to communicate and what are their preferences?",
            "agent_id": "demo-agent",
            "k": 3,
            "semantic_weight": 0.5,
            "recency_weight": 0.3,
            "importance_weight": 0.2,
        })
        r.raise_for_status()
        for result in r.json():
            mem = result["memory"]
            print(f"  #{result['rank']} score={result['score']:.4f}  [{mem['memory_type']}]")
            print(f"     {mem['content'][:90]}…")

        # ── Type-filtered search ─────────────────────────────────────
        print(_box("5. Procedural-only Search — 'how to start server'"))
        r = await client.post(f"{API}/memories/search", json={
            "query": "how do I run and start the server?",
            "agent_id": "demo-agent",
            "memory_type": "procedural",
            "k": 3,
        })
        r.raise_for_status()
        for result in r.json():
            mem = result["memory"]
            print(f"  #{result['rank']} score={result['score']:.4f}  {mem['content'][:90]}…")

        # ── Memory graph ─────────────────────────────────────────────
        print(_box("6. Memory Graph (threshold=0.7)"))
        r = await client.post(
            f"{API}/memories/agent/demo-agent/graph",
            params={"threshold": 0.7},
        )
        r.raise_for_status()
        graph = r.json()
        print(f"  Nodes: {len(graph['nodes'])}  |  Edges: {len(graph['edges'])}")
        for edge in graph["edges"][:5]:
            print(f"  Edge  weight={edge['weight']:.4f}  {edge['source'][:8]}… ↔ {edge['target'][:8]}…")

        # ── Update a memory ──────────────────────────────────────────
        print(_box("7. Update Memory Importance"))
        target_id = stored_ids[0]
        r = await client.patch(f"{API}/memories/{target_id}", json={"importance": 0.99})
        r.raise_for_status()
        updated = r.json()
        print(f"  id={updated['id'][:8]}…  importance: old=0.95 → new={updated['importance']}")

        # ── List agent memories ──────────────────────────────────────
        print(_box("8. List Agent Memories"))
        r = await client.get(f"{API}/memories/agent/demo-agent", params={"limit": 10})
        r.raise_for_status()
        mems = r.json()
        print(f"  Total returned: {len(mems)}")
        for m in mems:
            print(f"  [{m['memory_type']:12s}] imp={m['importance']:.2f}  acc={m['access_count']}  decay={m['decay_score']:.4f}  {m['content'][:50]}…")

        # ── Final stats ──────────────────────────────────────────────
        print(_box("9. Final Stats"))
        r = await client.get(f"{API}/stats")
        stats = r.json()
        print(f"  Total memories : {stats['total_memories']}")
        print(f"  Compressed     : {stats['compressed_memories']}")
        print(f"  Avg decay score: {stats['avg_decay_score']:.4f}")
        print(f"  By type        : {stats['by_type']}")

        print("\n✅  Demo complete — all operations succeeded.\n")


if __name__ == "__main__":
    asyncio.run(run_demo())