Build persistent knowledge graphs from LLM conversations

mnemo

Local-first AI memory layer for any LLM. Persistent knowledge graph, entity extraction, semantic retrieval - no cloud required.

What is mnemo?

LLM apps built on custom pipelines have no persistent memory between sessions. mnemo is a local sidecar that extracts entities, builds a knowledge graph, and injects scored context back into your prompts - no cloud, no Python runtime, no vendor lock-in.

mnemo is a sidecar service that watches every conversation you feed it, extracts named entities and relationships using an LLM, builds a persistent knowledge graph in SQLite, and injects relevant context back into future prompts - automatically, in under 50ms. It works with Ollama (fully local, free), OpenAI, Anthropic, or any OpenAI-compatible API. It ships as a single static binary with zero cloud dependency.

How it works

  your app
     │
     ▼
  POST /ingest ──► entity extraction (LLM) ──► knowledge graph (SQLite + petgraph)
                                                        │
  POST /retrieve ◄── scoring + ranking ◄── graph traversal + full-text search
     │
     ▼
  context_prompt  ──► inject into your LLM prompt

1. You POST raw text to /ingest (a conversation turn, a document, a note).
2. mnemo sends it to your configured LLM and extracts entities (people, tools, places, concepts) and the relationships between them.
3. Entities are deduplicated by name+type, aliases are merged, and everything is written to SQLite. The in-memory petgraph is updated atomically.
4. On POST /retrieve, mnemo runs a 6-stage pipeline: full-text chunk search → entity name search → graph expansion (BFS over the knowledge graph) → relation filter → score+rank → assemble a context_prompt string.
5. You inject context_prompt into your LLM's system prompt. Done.

Why mnemo

There are a lot of AI memory tools. Here's what makes mnemo different:

mnemo is not for everyone. If you're using a managed agent harness that handles memory for you, you don't need it. mnemo is for developers building custom LLM pipelines who need persistent, structured, local memory they fully control.

The graph layer is the real differentiator - entities are deduplicated across sessions, relationships are weighted and traversed at query time, and graph-expanded results score at 0.5x so direct matches always rank higher than inferred ones.

Quickstart

Path A - Docker + Ollama (fully free, recommended)

git clone https://github.com/zaydmulani09/mnemo
cd mnemo
docker compose up -d

# Pull the llama3 model the first time (~4 GB)
docker exec mnemo-ollama ollama pull llama3

# Verify everything is healthy
curl http://localhost:8080/health

Path B - Binary (Ollama or OpenAI running separately)

cargo install --path crates/mnemo-api

# With Ollama
export MNEMO_LLM_BASE_URL=http://localhost:11434/v1
mnemo-api

# With OpenAI
export MNEMO_LLM_BASE_URL=https://api.openai.com/v1
export MNEMO_LLM_API_KEY=sk-...
export MNEMO_LLM_MODEL=gpt-4o-mini
export MNEMO_LLM_PROVIDER=openai
mnemo-api

Path C - Python SDK

pip install mnemo-sdk

from mnemo import MnemoClient

client = MnemoClient()  # server at http://localhost:8080

# Store a memory
client.ingest("I'm building a Rust vector database called vecdb")

# Get context for injection into your next LLM prompt
print(client.get_context("what am I working on?"))

API Reference

All endpoints accept and return application/json. Base URL: http://localhost:8080.

Key request/response types:

// IngestRequest
{
  "content": "string",         // required — text to store
  "source":  "string",         // required — e.g. "chat", "email", "cli"
  "session_id": "string|null", // optional — group related chunks
  "metadata": {}               // optional — arbitrary JSON
}

// RetrievalQuery
{
  "text": "string",            // required — query text
  "session_id": "string|null", // optional — filter by session
  "max_chunks": 10,            // default 10
  "max_entities": 20,          // default 20
  "min_confidence": 0.5,       // default 0.5
  "include_graph": true,       // default true — expand via knowledge graph
  "graph_depth": 2             // default 2 — BFS depth for graph expansion
}

Full endpoint documentation with curl examples: docs/api.md

Configuration

Environment variables

TOML config file

Pass --config path/to/config.toml to mnemo-api. See mnemo.example.toml:

db_path = "mnemo.db"
port = 8080

[llm]
provider = "ollama"
base_url = "http://localhost:11434/v1"
model = "llama3"
api_key = "ollama"
timeout_secs = 30
max_retries = 3
max_tokens = 2048
temperature = 0.1

Environment variables take precedence over TOML values. The active config source is reported in GET /health → config_source.

CLI

Install:

cargo install --path crates/mnemo-cli

Usage:

# Store a memory
mnemo ingest "I use Neovim and prefer dark mode"

# Retrieve relevant context
mnemo search "what editor do I use?"

# List all extracted entities
mnemo entities

# Show entity detail + graph neighbors
mnemo entity <uuid> --neighbors

# List memory chunks
mnemo chunks

# Server health
mnemo health

# Memory statistics
mnemo stats

# Delete everything (prompts for confirmation)
mnemo wipe

# Skip confirmation prompt
mnemo wipe --yes

# Point at a non-default server
mnemo --server http://192.168.1.10:8080 stats

Python SDK

Install:

pip install mnemo-sdk

See sdk/python/README.md for the full API reference.

Async example:

import asyncio
from mnemo import AsyncMnemoClient

async def main():
    async with AsyncMnemoClient() as client:
        await client.ingest(
            "Alice is a principal engineer at Stripe working on payment infrastructure.",
            session_id="session-001",
        )
        context = await client.get_context(
            "what does Alice work on?",
            session_id="session-001",
        )
        print(context)

asyncio.run(main())

A working standalone example: examples/basic_usage.py

Architecture

Four Rust crates wired together:

Full architecture documentation: docs/architecture.md

Performance

Benchmarked on Apple M2, SQLite WAL mode, in-memory petgraph. Debug build numbers - release build (--release) is 3-5× faster.

Run cargo run -p mnemo-bench to benchmark on your hardware.

Testing

Rust

cargo test --workspace          # run all 122 tests
make coverage                  # HTML coverage report (requires cargo-llvm-cov)
make coverage-summary          # summary to stdout

Python SDK

cd sdk/python && pytest tests/ -v

Benchmarks

cargo run -p mnemo-bench                    # all 12 benchmarks
cargo run -p mnemo-bench -- --filter graph  # graph benchmarks only
cargo run -p mnemo-bench -- --json out.json # save results to JSON

Current test counts: 122 Rust tests · 21 Python tests · 12 benchmarks

examples/basic_usage.py

"""
mnemo basic usage example
=========================

Demonstrates the full mnemo Python SDK workflow:
  - health check
  - ingesting memories across multiple sessions
  - retrieving context for various queries
  - listing and inspecting entities
  - graph neighbor traversal
  - stats
  - optional wipe

Prerequisites:
  1. Start the mnemo server:
       cargo run -p mnemo-api
     or:
       docker compose up -d

  2. Install the SDK:
       pip install mnemo-sdk
     or (from repo root):
       pip install -e sdk/python

Run:
  python examples/basic_usage.py
"""

import sys

try:
    from mnemo import MnemoClient
    from mnemo.exceptions import MnemoConnectionError, MnemoNotFoundError
except ImportError:
    print("ERROR: mnemo SDK not installed.")
    print("       Run: pip install mnemo-sdk")
    print("       Or:  pip install -e sdk/python  (from repo root)")
    sys.exit(1)


MEMORIES = [
    # session-work
    {
        "content": "I am a software engineer at Acme Corp working on backend infrastructure.",
        "source": "chat",
        "session_id": "session-work",
    },
    {
        "content": "I use Rust and Python daily — Rust for systems code, Python for ML pipelines.",
        "source": "chat",
        "session_id": "session-work",
    },
    {
        "content": (
            "My current project is a vector database called vecdb built in Rust "
            "using HNSW indexing for approximate nearest-neighbour search."
        ),
        "source": "chat",
        "session_id": "session-work",
    },
    {
        "content": (
            "My team at Acme uses GitHub for source control, Linear for project management, "
            "and Slack for communication."
        ),
        "source": "chat",
        "session_id": "session-work",
    },
    # session-prefs
    {
        "content": "I prefer dark mode everywhere. My terminal is Wezterm with fish shell.",
        "source": "settings",
        "session_id": "session-prefs",
    },
    {
        "content": "I edit code in Neovim with LazyVim. I never use VS Code.",
        "source": "settings",
        "session_id": "session-prefs",
    },
    {
        "content": "My keyboard is a ZSA Moonlander with Gateron Yellow switches.",
        "source": "settings",
        "session_id": "session-prefs",
    },
    # session-personal
    {
        "content": "I live in San Francisco and commute by bicycle.",
        "source": "profile",
        "session_id": "session-personal",
    },
    {
        "content": "I am learning Japanese. I have been studying for about six months.",
        "source": "profile",
        "session_id": "session-personal",
    },
    {
        "content": (
            "I maintain an open-source CLI tool called mnemo that gives LLMs persistent memory. "
            "It is written in Rust and has a Python SDK."
        ),
        "source": "profile",
        "session_id": "session-personal",
    },
]

QUERIES = [
    ("what programming languages do I use?", None),
    ("what editor do I use?", "session-prefs"),
    ("what am I building at work?", "session-work"),
    ("tell me about my open source projects", None),
    ("what city do I live in?", "session-personal"),
]


def separator(title: str) -> None:
    width = 60
    print(f"\n{'─' * width}")
    print(f"  {title}")
    print(f"{'─' * width}")


def step_1_health(client: MnemoClient) -> bool:
    separator("1. Health check")
    health = client.health()
    print(f"  status:        {health.status}")
    print(f"  version:       {health.version}")
    print(f"  db_connected:  {health.db_connected}")
    print(f"  llm_reachable: {health.llm_reachable}")
    print(f"  provider:      {health.provider_type} / {health.provider_model}")
    print(f"  uptime:        {health.uptime_seconds}s")

    if health.status != "ok" or not health.db_connected:
        print("\n  ERROR: server is not healthy. Check server logs.")
        return False
    if not health.llm_reachable:
        print("\n  WARNING: LLM is unreachable — entity extraction will be skipped.")
        print("           Memories will still be stored as text chunks.")
    return True


def step_2_ingest(client: MnemoClient) -> None:
    separator("2. Ingesting 10 memories across 3 sessions")
    for mem in MEMORIES:
        result = client.ingest(
            mem["content"],
            source=mem["source"],
            session_id=mem["session_id"],
        )
        session_label = mem["session_id"] or "—"
        print(
            f"  ✓ [{session_label:<18}] "
            f"entities={result.entities_extracted:2d}  "
            f"relations={result.relations_extracted:2d}  "
            f"{result.processing_time_ms}ms"
        )


def step_3_retrieve(client: MnemoClient) -> None:
    separator("3. Retrieving context for 5 queries")
    for query_text, session_id in QUERIES:
        result = client.retrieve(
            query_text,
            session_id=session_id,
            max_chunks=3,
            max_entities=5,
            min_confidence=0.0,
            include_graph=True,
            graph_depth=2,
        )
        print(f"\n  Query: \"{query_text}\"")
        if session_id:
            print(f"  Session filter: {session_id}")
        print(
            f"  → {len(result.chunks)} chunks, "
            f"{len(result.entities)} entities, "
            f"{len(result.relations)} relations"
        )
        if result.context_prompt:
            # Show first 300 chars of context_prompt
            preview = result.context_prompt[:300].replace("\n", "\n    ")
            print(f"  Context preview:\n    {preview}{'...' if len(result.context_prompt) > 300 else ''}")
        else:
            print("  (no context returned)")


def step_4_entities(client: MnemoClient) -> str | None:
    """List entities, print summary table, return ID of most-referenced entity."""
    separator("4. Entities extracted from memory")
    entities = client.list_entities(limit=50)
    if not entities:
        print("  (no entities yet — LLM may be offline)")
        return None

    # Sort by source_count descending
    entities.sort(key=lambda e: e.source_count, reverse=True)

    col_name = 30
    col_type = 14
    print(
        f"  {'Name':<{col_name}} {'Type':<{col_type}} {'Conf':>5} {'Refs':>4}"
    )
    print(f"  {'─' * col_name} {'─' * col_type} {'─' * 5} {'─' * 4}")
    for e in entities[:20]:  # show top 20
        name = e.name[:col_name - 1] if len(e.name) >= col_name else e.name
        etype = e.entity_type[:col_type - 1] if len(e.entity_type) >= col_type else e.entity_type
        print(
            f"  {name:<{col_name}} {etype:<{col_type}} "
            f"{e.confidence:>5.2f} {e.source_count:>4}"
        )
    if len(entities) > 20:
        print(f"  ... and {len(entities) - 20} more")

    top_entity = entities[0]
    print(f"\n  Most-referenced entity: \"{top_entity.name}\" (refs={top_entity.source_count})")
    return str(top_entity.id)


def step_5_neighbors(client: MnemoClient, entity_id: str) -> None:
    separator("5. Knowledge graph neighbors")
    try:
        entity = client.get_entity(entity_id)
        print(f"  Entity: \"{entity.name}\" ({entity.entity_type})")
        neighbors = client.get_neighbors(entity_id, depth=2)
        if neighbors:
            print(f"  Neighbors (depth=2):")
            for n in neighbors:
                print(f"    → {n['name']} ({n['entity_type']})")
        else:
            print("  No neighbors found (graph may be sparse if LLM was offline).")
    except MnemoNotFoundError:
        print(f"  Entity {entity_id} not found.")


def step_6_stats(client: MnemoClient) -> None:
    separator("6. Memory statistics")
    stats = client.stats()
    print(f"  entities:    {stats.entity_count}")
    print(f"  chunks:      {stats.chunk_count}")
    print(f"  graph nodes: {stats.node_count}")
    print(f"  graph edges: {stats.edge_count}")
    print(f"  uptime:      {stats.uptime_seconds}s")


def step_7_wipe(client: MnemoClient) -> None:
    separator("7. Optional wipe")
    print("  This will delete ALL memories from the server.")
    try:
        answer = input("  Wipe all memories? [y/N] ").strip().lower()
    except (EOFError, KeyboardInterrupt):
        answer = "n"

    if answer == "y":
        client.wipe()
        print("  ✓ All memories wiped.")
        stats = client.stats()
        print(f"  entity_count={stats.entity_count} chunk_count={stats.chunk_count}")
    else:
        print("  Skipped.")


def main() -> None:
    print("\nmnemo basic_usage.py")
    print("====================")
    print("Connecting to http://localhost:8080 ...")

    client = MnemoClient(base_url="http://localhost:8080", timeout=60)

    # 1. Health
    try:
        if not step_1_health(client):
            sys.exit(1)
    except MnemoConnectionError:
        print("\nERROR: Cannot connect to mnemo server at http://localhost:8080")
        print("\nStart the server with one of:")
        print("  cargo run -p mnemo-api")
        print("  docker compose up -d")
        sys.exit(1)

    # 2. Ingest
    step_2_ingest(client)

    # 3. Retrieve
    step_3_retrieve(client)

    # 4. Entities
    top_entity_id = step_4_entities(client)

    # 5. Neighbors
    if top_entity_id:
        step_5_neighbors(client, top_entity_id)

    # 6. Stats
    step_6_stats(client)

    # 7. Optional wipe
    step_7_wipe(client)

    print("\nDone.\n")


if __name__ == "__main__":
    main()