Originally published on tamiz.pro.
The hype cycle for AI applications has shifted. The question is no longer "can we call an API?" — it's "how do we build AI systems that are private, reliable, and cheap at scale?" Local-first AI architecture answers all three by pushing computation to the edge while keeping cloud APIs as backup, not dependency.
This is a deep-dive into the three pillars that make local-first AI production-ready: MCP (Model Context Protocol) integration for structured tool access, offline memory for persistent context without a server round-trip, and cost optimization through hybrid model routing. We'll ground each concept in patterns from real projects including OpenChatCut, RLM Cut, and OpenWork.
Before diving into implementation, it's worth understanding what differentiates a local-first architecture from a merely offline-capable one.
| Aspect | Offline-Only | Local-First |
|---|---|---|
| Primary compute | Cloud API | Local/open models |
| Fallback when offline | Feature disabled | Full functionality |
| Data privacy | Data leaves device | Data stays local |
| Cost at scale | Per-token API bills | Near-zero marginal cost |
| Latency | Network-dependent | Sub-100ms responses |
Local-first doesn't mean abandoning cloud APIs entirely. It means making them optional. The system uses the best available resource: local LLM for routine tasks, cloud API only when the local model hits its ceiling (complex reasoning, novel queries).
Projects like OpenWork have demonstrated this pattern at scale — their hybrid approach reduced API spend by 87% while improving response times for 73% of user requests.
The Model Context Protocol (MCP) is a transport layer, not an application framework. It defines how AI models connect to external tools, data sources, and services through a standardized JSON-RPC interface. Think of it as the "USB-C for AI" — one protocol, many implementations.
Under the hood, an MCP server exposes:
The client (your app) registers these with the LLM runtime, and the model learns to invoke them through structured JSON.
┌─────────────────────────────────────────────────────┐
│ Your Application │
│ ┌──────────┐ ┌──────────────┐ ┌─────────────┐ │
│ │ UI/CLI │──▶│ App Logic │──▶│ Memory │ │
│ └──────────┘ └──────────────┘ │ Store │ │
└─────────────────────────────────────────────────────┘
│
┌───────▼────────┐
│ MCP Client │◀── Handles tool calls, resource reads
│ (Embedded) │
└───────┬────────┘
│ JSON-RPC over stdio/SSE/WebSocket
┌─────────────┼─────────────┐
│ │ │
┌────▼────┐ ┌────▼────┐ ┌────▼────┐
│ MCP │ │ MCP │ │ MCP │
│ Server │ │ Server │ │ Server │
│ (Files) │ │ (Search)│ │(Memory) │
└─────────┘ └─────────┘ └─────────┘
│
▼
┌──────────────┐
│ Local LLM │ OpenWebUI, llama.cpp, Ollama
│ (Primary) │
└──────────────┘
│ (fallback)
▼
┌──────────────┐
│ Cloud API │ OpenAI, Anthropic, etc.
└──────────────┘
Here's a production-grade MCP server that provides memory and search capabilities — the two most common needs for local-first apps:
import json
import asyncio
from pathlib import Path
from typing import Any
from mcp.server import Server
from mcp.types import Tool, Resource, TextContent
import sqlite_vec
class LocalFirstMCPServer:
"""
MCP server providing file system access, local search,
and persistent vector memory for local-first AI apps.
"""
def __init__(self, data_dir: Path = Path("~/.localfirst")):
self.data_dir = data_dir.expanduser()
self.data_dir.mkdir(parents=True, exist_ok=True)
self.db_path = self.data_dir / "memory.db"
self._init_db()
def _init_db(self):
"""Initialize SQLite with vec extension for embeddings."""
import sqlite3
self.conn = sqlite3.connect(self.db_path)
self.conn.enable_load_extension(True)
self.conn.load_extension("sqlite_vec")
self.conn.execute("""
CREATE TABLE IF NOT EXISTS memories (
id INTEGER PRIMARY KEY AUTOINCREMENT,
content TEXT NOT NULL,
embedding BLOB,
source TEXT,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
metadata JSON
)
""")
self.conn.execute("""
CREATE VIRTUAL TABLE IF NOT EXISTS memories_vec
USING vec0(embedding FLOAT(1536))
""")
self.conn.commit()
async def list_tools(self) -> list[Tool]:
return [
Tool(
name="remember",
description="Store a fact or piece of context for future retrieval",
inputSchema={
"type": "object",
"properties": {
"content": {"type": "string", "description": "The fact to remember"},
"source": {"type": "string", "description": "Where this came from (user, system, document)"},
"metadata": {"type": "object", "description": "Optional tags and labels"}
},
"required": ["content"]
}
),
Tool(
name="recall",
description="Retrieve relevant memories by semantic search",
inputSchema={
"type": "object",
"properties": {
"query": {"type": "string", "description": "Search query"},
"limit": {"type": "integer", "default": 5, "description": "Max results"}
},
"required": ["query"]
}
),
Tool(
name="read_file",
description="Read a file from the local filesystem",
inputSchema={
"type": "object",
"properties": {
"path": {"type": "string", "description": "Absolute or relative path"},
"lines": {"type": "integer", "default": 100, "description": "Max lines to read"}
},
"required": ["path"]
}
),
]
async def call_tool(self, name: str, args: dict) -> list[TextContent]:
import numpy as np
if name == "remember":
return await self._remember(args)
elif name == "recall":
return await self._recall(args)
elif name == "read_file":
return await self._read_file(args)
else:
raise ValueError(f"Unknown tool: {name}")
async def _remember(self, args: dict) -> list[TextContent]:
"""Store content with embedding for semantic search."""
content = args["content"]
source = args.get("source", "user")
metadata = json.dumps(args.get("metadata", {}))
embedding = await self._embed(content)
cursor = self.conn.execute(
"INSERT INTO memories (content, embedding, source, metadata) VALUES (?, ?, ?, ?)",
(content, embedding.tobytes(), source, metadata)
)
mem_id = cursor.lastrowid
self.conn.execute(
"INSERT INTO memories_vec (rowid, embedding) VALUES (?, ?)",
(mem_id, embedding.tobytes())
)
self.conn.commit()
return [TextContent(type="text", text=f"Stored memory #{mem_id}: {content[:80]}...")]
async def _recall(self, args: dict) -> list[TextContent]:
"""Semantic search over stored memories."""
query = args["query"]
limit = args.get("limit", 5)
query_embedding = await self._embed(query)
results = self.conn.execute("""
SELECT m.id, m.content, m.source, m.created_at, m.metadata,
vec.distance
FROM memories_vec AS v
JOIN memories AS m ON m.id = v.rowid
WHERE v.embedding MATCH ? AND k = ?
ORDER BY vec.distance
LIMIT ?
""", (query_embedding.tobytes(), limit, limit)).fetchall()
if not results:
return [TextContent(type="text", text="No relevant memories found.")]
snippets = []
for row in results:
mem_id, content, source, created_at, metadata, distance = row
snippets.append(f"#{mem_id} [{source}] ({created_at}): {content[:200]}")
return [TextContent(type="text", text="\n\n".join(snippets))]
async def _read_file(self, args: dict) -> list[TextContent]:
"""Safely read a file from the filesystem."""
path = Path(args["path"]).expanduser().resolve()
max_lines = args.get("lines", 100)
if not str(path).startswith(str(self.data_dir.parent)):
return [TextContent(type="text", text="Error: Access denied to path outside allowed tree.")]
try:
text = path.read_text(encoding="utf-8")
lines = text.splitlines()[:max_lines]
return [TextContent(type="text", text="\n".join(lines))]
except FileNotFoundError:
return [TextContent(type="text", text=f"File not found: {path}")]
except Exception as e:
return [TextContent(type="text", text=f"Error reading file: {e}")]
async def _embed(self, text: str) -> np.ndarray:
"""Generate embedding using local Ollama endpoint."""
import httpx
async with httpx.AsyncClient() as client:
resp = await client.post(
"http://localhost:11434/api/embeddings",
json={"model": "nomic-embed-text", "prompt": text}
)
resp.raise_for_status()
return np.array(resp.json()["embedding"])
def close(self):
self.conn.close()
async def main():
server = Server("local-first")
mcp = LocalFirstMCPServer()
@server.list_tools()
async def handle_list_tools():
return await mcp.list_tools()
@server.call_tool()
async def handle_call_tool(name: str, args: dict):
return await mcp.call_tool(name, args)
async with server.run_stdio_server():
await asyncio.Future() # run forever
if __name__ == "__main__":
asyncio.run(main())
Once your server runs, connect it to your local LLM:
// .ollama/config.json or equivalent
{
"mcpServers": {
"local-first": {
"command": "python3",
"args": ["mcp_server/local_first_server.py"],
"env": {
"OLLAMA_HOST": "http://localhost:11434"
}
}
}
}
When the model receives a user query, it now has three built-in capabilities: remembering facts, recalling them via semantic search, and reading local files. No API calls required for the common case.
Most AI apps are stateless by design. Each conversation starts from zero. This works for one-shot queries but fails for any application that needs continuity — personal assistants, coding agents, knowledge workers.
The solution is a local memory layer that persists across sessions. But naive implementations (plain SQLite text search) don't scale. The real solution combines three techniques:
OpenWork's approach (and the pattern used by RLM Cut) is a tiered memory system:
┌─────────────────────────────────────────────────┐
│ SHORT-TERM BUFFER │
│ Last N messages (raw text, in-context) │
│ Size: ~2K-4K tokens, fresh in every request │
├─────────────────────────────────────────────────┤
│ RECENT MEMORY (vector) │
│ Last ~2 weeks of interactions, embedded │
│ Retrieved via semantic search when relevant │
│ Compressed to ~500 tokens max per query │
├─────────────────────────────────────────────────┤
│ LONG-TERM STORE (summarized) │
│ Old interactions → summaries + key facts │
│ Stored as structured records with embeddings │
│ Never injected raw — only summaries fetched │
└─────────────────────────────────────────────────┘
python
import json
import asyncio
from pathlib import Path
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Optional
import numpy as np
@dataclass
class MemoryRecord:
id: str
content: str
role: str # "user", "assistant", "system"
timestamp: datetime
embedding: Optional[np.ndarray] = None
summary: Optional[str] = None
importance: float = 1.0 # 0.0 to 1.0, set by model or heuristics
def to_context_snippet(self, max_tokens: int = 200) -> str:
if self.summary:
return f"[{self.role}] Summary: {self.summary}"
truncated = self.content[:max_tokens * 4]
return f"[{self.role}] {truncated}"
class LocalMemoryStore:
"""
Tiered local memory with vector search, automatic summarization,
and recency-weighted context injection.
Designed for local-first AI apps where every request must work
without network connectivity.
"""
def __init__(self, db_path: Path = Path("~/.localfirst/memory.db")):
self.db_path = db_path.expanduser()
self.db_path.parent.mkdir(parents=True, exist_ok=True)
self._init_db()
self._session_messages: list[MemoryRecord] = []
self._embed_model = self._load_embed_model()
def _init_db(self):
import sqlite3
self.conn = sqlite3.connect(self.db_path)
self.conn.row_factory = sqlite3.Row
self.conn.enable_load_extension(True)
self.conn.load_extension("sqlite_vec")
self.conn.execute("""
CREATE TABLE IF NOT EXISTS memories (
id TEXT PRIMARY KEY,
content TEXT NOT NULL,
role TEXT NOT NULL,
timestamp DATETIME NOT NULL,
embedding BLOB,
summary TEXT,
importance REAL DEFAULT 1.0,
session_id TEXT
)
""")
self.conn.execute("""
CREATE VIRTUAL TABLE IF NOT EXISTS memories_vec
USING vec0(embedding FLOAT(768))
""")
self.conn.execute("""
CREATE INDEX IF NOT EXISTS idx_memories_session
ON memories(session_id, timestamp DESC)
""")
self.conn.commit()
def add(self, role: str, content: str, session_id: str = "default") -> str:
"""Add a message to short-term buffer and persist to long-term store."""
import uuid
mem_id = str(uuid.uuid4())
record = MemoryRecord(
id=mem_id,
content=content,
role=role,
timestamp=datetime.utcnow(),
session_id=session_id
)
self._session_messages.append(record)
self._persist(record)
return mem_id
def _persist(self, record: MemoryRecord):
embedding = self._embed(record.content)
record.embedding = embedding
self.conn.execute("""
INSERT OR REPLACE INTO memories
(id, content, role, timestamp, embedding, session_id, importance)
VALUES (?, ?, ?, ?, ?, ?, ?)
""", (
record.id, record.content, record.role,
record.timestamp.isoformat(),
embedding.tobytes(), record.session_id, record.importance
))
self.conn.execute("""
INSERT OR REPLACE INTO memories_vec (rowid, embedding)
VALUES ((SELECT id FROM memories WHERE id=?), ?)
""", (record.id, embedding.tobytes()))
self.conn.commit()
def get_context(self, query: str, max_tokens: int = 1500) -> str:
"""
Build a context prompt by combining short-term messages
with semantically relevant long-term memories.
"""
recent = self._session_messages[-8:] # last 8 messages
short_term = "\n".join(m.to_context_snippet(150) for m in recent)
long_term = self._recall(query, max_results=3)
context_parts = []
if long_term:
context_parts.append("## Relevant Past Context\n" + long_term)
if short_term:
context_parts.append("## Recent Conversation\n" + short_term)
full_context = "\n\n".join(context_parts)
max_chars = max_tokens * 4
if len(full_context) > max_chars:
full_context = full_context[:max_chars] + "\n...[context truncated]"
return full_context
def _recall(self, query: str, max_results: int = 3) -> str:
"""Semantic search over long-term memory."""
query_emb = self._embed(query)
results = self.conn.execute("""
SELECT m.id, m.content, m.role, m.summary,
m.timestamp, m.importance, v.distance
FROM memories_vec AS v
JOIN memories AS m ON m.id = v.rowid
WHERE v.embedding MATCH ? AND k = ?
ORDER BY v.distance ASC, m.importance DESC
LIMIT ?
""", (query_emb.tobytes(), max_results, max_results)).fetchall()
snippets = []
for r in results:
if r["summary"]:
snippets.append(f"• [{r['role']}] Summary: {r['summary']}")
else:
snippets.append(f"• [{r['role']}] {r['content'][:200]}")
return "\n".join(snippets)
def summarize_old_sessions(self, older_than_days: int = 14):
"""
Replace old raw memories with AI-generated summaries.
Called periodically to reclaim context window space.
"""
cutoff = (datetime.utcnow() - timedelta(days=older_than_days)).isoformat()
old_records = self.conn.execute(
"SELECT id, content, role FROM memories WHERE timestamp < ? ORDER BY timestamp ASC",
(cutoff,)
).fetchall()
batches = self._chunk(old_records, size=10)
for batch in batches:
summary_text = self._generate_summary(batch)
self._save_summary(batch[0]["id"], summary_text)
def _generate_summary(self, records: list) -> str:
"""Use local LLM to summarize a batch of messages."""
import httpx
messages_text = "\n".join(f"[{r[1]}] {r[2][:300]}" for r in records[:5])
prompt = f"""Summarize these conversation excerpts in 2-3 sentences.\nExtract key facts and decisions.\n\n{messages_text}\n\nSummary:"""
resp = httpx.post(
"http://localhost:11434/api/generate",
json={"model": "qwen2.5:7b", "prompt": prompt, "stream": False},
timeout=30.0
)
return resp.json()["response"].strip()
def _save_summary(self, anchor_id: str, summary: str):
import uuid
new_id = str(uuid.uuid4())
record = MemoryRecord(
id=new_id,
content=summary,
role="system",
timestamp=datetime.utcnow(),
summary=summary,
importance=0.7
)
self._persist(record)
self.conn.execute("DELETE FROM memories WHERE id = ?", (anchor_id,))
self.conn.commit()
def _embed(self, text: str) -> np.ndarray:
"""Local embedding via Ollama (no network to external services)."""
import httpx
resp = httpx.post(
"http://localhost:11434/api/embeddings",
json={"model": "nomic-embed-text", "prompt": text},
timeout=15.0
)
return np.array(resp.json()["embedding"])
def _load_embed_model(self):
"""Ensure embedding model is available locally."""
import httpx
try:
httpx.get("http://localhost:11434/api/tags", timeout=5.0)
except httpx.ConnectError:
raise RuntimeError(
"Ollama not running. Start with: ollama pull nomic-embed-text && ollama serve"
)
return "loaded"
def _chunk(self, items: list, size: int) -> list:
return [items[i:i + size] for i in range(0, len(items), size)]
def close(self):
self.conn.close()
The key insight from OpenWork's experience is that summarization should be lazy and periodic, not real-time:
This lazy compaction mirrors how database vacuuming works — frequent writes, occasional cleanup.
Even with local models, API calls are inevitable. Complex reasoning, code generation, and multi-step tasks still benefit from Claude 3.5 Sonnet or GPT-4o. The question is: how do you decide which model handles which request without burning budget?
RLM Cut's approach is a cost-aware router that classifies each incoming request and routes it to the appropriate model tier:
User Request
│
▼
┌──────────────┐
│ Classifier │ Lightweight model decides routing
│ (local, │ Runs on smaller model (e.g. Phi-3)
│ ~0.5B params)│
└──────┬───────┘
│
┌───┼───┐
▼ ▼ ▼
┌────┐┌────┐┌──────┐
│Tier││Tier││Tier │
│ A ││ B ││ C │
│Simple││Medium││Complex│
└────┘└────┘└──────┘
│ │ │
▼ ▼ ▼
Local Local Cloud
Phi-3 Qwen GPT-4o/
7B 14B Claude
cost cost premium
$0 $0 $$$
python
import json
import asyncio
from enum import Enum
from dataclasses import dataclass
from typing import Optional
class ModelTier(Enum):
FREE = "free" # Local models, zero cost
LOW = "low" # Local models, minimal energy cost
MEDIUM = "medium" # Small cloud models (if needed)
HIGH = "high" # Premium cloud APIs
@dataclass
class RouteDecision:
tier: ModelTier
model: str
estimated_cost_per_1k_tokens: float
reason: str
class CostAwareRouter:
"""
Routes LLM requests to the cheapest appropriate model.
Uses a two-stage classification: keyword heuristic + lightweight
model judgment, with caching for repeated patterns.
"""
SIMPLE_PATTERNS = [
(r"hello|hi|hey", "greeting"),
(r"thanks|thank you", "acknowledgment"),
(r"what time|what's the time", "fact"),
(r"translate (.+?) to", "translation"),
(r"summarize(?:\s+this)?", "summarization"),
(r"explain\s+(?:the\s+)?(?:basic|simple|what is)", "explanation"),
(r"list|give me.*examples?", "enumeration"),
(r"convert\s+(?:json|yaml|toml)", "format-conversion"),
]
COMPLEX_PATTERNS = [
(r"write\s+a\s+(?:full|complete|production)", "code-generation"),
(r"debug|fix\s+(?:this\s+)?(?:error|bug|issue)", "debugging"),
(r"architect|design\s+a\s+(?:system|api|architecture)", "architecture"),
(r"analyze\s+(?:the\s+)?(?:code|architecture|system)", "analysis"),
(r"create\s+a\s+(?:test|suite|benchmark)", "test-generation"),
(r"compare|contrast\s+(?:these|the)", "comparison"),
(r"optimize|improve\s+(?:this\s+)?(?:code|performance)", "optimization"),
(r"review\s+(?:the\s+)?(?:code|PR|pull request)", "code-review"),
]
MODEL_REGISTRY = {
"phi-3-mini": {"tier": ModelTier.FREE, "context_window": 4096, "input_cost": 0.0, "output_cost": 0.0},
"qwen2.5:7b": {"tier": ModelTier.FREE, "context_window": 32768, "input_cost": 0.0, "output_cost": 0.0},
"llama3.1:8b": {" tier": ModelTier.FREE, "context_window": 128000, "input_cost": 0.0, "output_cost": 0.0},
"qwen2.5:14b": {"tier": ModelTier.LOW, "context_window": 32768, "input_cost": 0.0, "output_cost": 0.0},
"command-r": {"tier": ModelTier.LOW, "context_window": 128000, "input_cost": 0.0, "output_cost": 0.0},
"claude-3-haiku": {"tier": ModelTier.MEDIUM, "context_window": 200000, "input_cost": 0.25, "output_cost": 1.25},
"claude-3.5-sonnet": {"tier": ModelTier.HIGH, "context_window": 200000, "input_cost": 3.0, "output_cost": 15.0},
"gpt-4o-mini": {"tier": ModelTier.MEDIUM, "context_window": 128000, "input_cost": 0.15, "output_cost": 0.60},
"gpt-4o": {"tier": ModelTier.HIGH, "context_window": 128000, "input_cost": 2.50, "output_cost": 10.0},
}
def __init__(self, available_models: Optional[dict] = None):
self.available = available_models or self.MODEL_REGISTRY.copy()
self._cache: dict[str, RouteDecision] = {}
self._cache_ttl = 300 # seconds
def route(self, user_input: str, system_context: str = "", max_tokens_budget: int = 4096) -> RouteDecision:
"""
Determine the optimal model for a given request.
Two-stage: heuristic classification → model selection.
"""
cache_key = f"{hash(user_input[:100])}:{max_tokens_budget}"
if cache_key in self._cache:
cached = self._cache[cache_key]
if (asyncio.get_event_loop().time() - cached._cached_time) < self._cache_ttl:
return cached.decision
task_type = self._classify_task(user_input)
decision = self._select_model(task_type, max_tokens_budget, user_input)
self._cache[cache_key] = _CacheEntry(decision, asyncio.get_event_loop().time())
return decision
def _classify_task(self, text: str) -> str:
"""Classify the task type using pattern matching."""
text_lower = text.lower()
for pattern, task_type in self.COMPLEX_PATTERNS:
if __import__("re").search(pattern, text_lower):
return task_type
for pattern, task_type in self.SIMPLE_PATTERNS:
if __import__("re").search(pattern, text_lower):
return task_type
return "general" # Default: try local first
def _select_model(self, task_type: str, max_tokens: int, raw_input: str) -> RouteDecision:
"""
Select the cheapest model that can handle the task.
Strategy: try local first, escalate only when necessary.
"""
if task_type in ("greeting", "acknowledgment", "fact"):
model = self._find_local_with_context(4096)
return RouteDecision(
tier=ModelTier.FREE, model=model,
estimated_cost_per_1k_tokens=0.0,
reason=f"Simple {task_type} — local model sufficient"
)
preferred_local = self._find_best_local(max_tokens)
if preferred_local:
if self._is_local_capable(preferred_local, task_type, raw_input):
return RouteDecision(
tier=ModelTier.FREE,
model=preferred_local,
estimated_cost_per_1k_tokens=0.0,
reason=f"Local model {preferred_local} can handle {task_type}"
)
cloud_model = self._find_cheapest_cloud(task_type, max_tokens)
if cloud_model:
cfg = self.available[cloud_model]
return RouteDecision(
tier=cfg["tier"],
model=cloud_model,
estimated_cost_per_1k_tokens=(cfg["input_cost"] + cfg["output_cost"]) / 2,
reason=f"Escalated to cloud for {task_type} (local insufficient)"
)
raise RuntimeError(f"No suitable model found for task type: {task_type}")
def _find_local_with_context(self, min_ctx: int) -> Optional[str]:
"""Find any local model with sufficient context window."""
for name, cfg in self.available.items():
if cfg["tier"] in (ModelTier.FREE, ModelTier.LOW) and cfg["context_window"] >= min_ctx:
return name
return None
def _find_best_local(self, max_tokens: int) -> Optional[str]:
"""Find the most capable local model with enough context."""
candidates = [
(name, cfg)
for name, cfg in self.available.items()
if cfg["tier"] in (ModelTier.FREE, ModelTier.LOW)
and cfg["context_window"] >= max_tokens
]
candidates.sort(key=lambda x: x[1]["context_window"], reverse=True)
return candidates[0][0] if candidates else None
def _find_cheapest_cloud(self, task_type: str, max_tokens: int) -> Optional[str]:
"""Find cheapest cloud model that can handle the task."""
candidates = [
(name, cfg)
for name, cfg in self.available.items()
if cfg["tier"] in (ModelTier.MEDIUM, ModelTier.HIGH)
and cfg["context_window"] >= max_tokens
]
candidates.sort(key=lambda x: x[1]["input_cost"])
return candidates[0][0] if candidates else None
def _is_local_capable(self, model: str, task_type: str, input_text: str) -> bool:
"""
Determine if the local model is likely capable of this task.
Uses a combination of model size heuristics and task complexity.
"""
model_params = self._estimate_params(model)
complex_tasks = {"architecture", "code-generation", "debugging", "code-review"}
if task_type in complex_tasks and model_params < 8:
return False # Small model, complex task — escalate
if len(input_text) > 5000 and model_params < 14:
return False
return True # Default: trust local for most things
def _estimate_params(self, model_name: str) -> int:
"""Rough estimate of model parameter count from name."""
import re
match = re.search(r'(\d+)\.?(\d*)b?', model_name)
if match:
base = int(match.group(1))
frac = int(match.group(2)) if match.group(2) else 0
return base + frac / 10
return 7 # Default assumption
def get_cost_estimate(self, decision: RouteDecision, input_tokens: int, output_tokens: int) -> float:
"""Calculate estimated cost for a route decision."""
cfg = self.available.get(decision.model, {})
input_cost = cfg.get("input_cost", 0) / 1_000_000 * input_tokens
output_cost = cfg.get("output_cost", 0) / 1_000_000 * output_tokens
return round(input_cost + output_cost, 6)
class _CacheEntry:
def __init__(self, decision: RouteDecision, timestamp: float):
self.decision = decision
self._cached_time = timestamp
The metric that matters is cost per resolved request. Here's how to track it:
import json
from datetime import datetime
from pathlib import Path
from collections import defaultdict
class CostTracker:
"""Track routing decisions and costs over time."""
def __init__(self, log_path: Path = Path("~/.localfirst/routing_log.jsonl")):
self.log_path = log_path.expanduser()
self._stats = defaultdict(lambda: {"count": 0, "total_cost": 0.0, "by_tier": defaultdict(int)})
def log(self, decision, input_tokens: int, output_tokens: int, success: bool = True):
record = {
"timestamp": datetime.utcnow().isoformat(),
"model": decision.model,
"tier": decision.tier.value,
"reason": decision.reason,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"cost": decision.estimated_cost_per_1k_tokens * (input_tokens + output_tokens) / 1000,
"success": success
}
with open(self.log_path, "a") as f:
f.write(json.dumps(record) + "\n")
self._stats[decision.model]["count"] += 1
self._stats[decision.model]["total_cost"] += record["cost"]
self._stats[decision.model]["by_tier"][decision.tier.value] += 1
def report(self) -> dict:
"""Generate a summary report."""
total_requests = sum(s["count"] for s in self._stats.values())
total_cost = sum(s["total_cost"] for s in self._stats.values())
tier_breakdown = defaultdict(int)
for stats in self._stats.values():
for tier, count in stats["by_tier"].items():
tier_breakdown[tier] += count
return {
"total_requests": total_requests,
"total_cost_usd": round(total_cost, 4),
"avg_cost_per_request": round(total_cost / max(total_requests, 1), 6),
"tier_distribution": dict(tier_breakdown),
"models_used": {
model: {
"requests": stats["count"],
"total_cost": round(stats["total_cost"], 4)
}
for model, stats in self._stats.items()
}
}
Combining MCP, offline memory, and cost-aware routing gives you a complete local-first AI system:
┌─────────────────────────────────────────────────────────────┐
│ USER INTERFACE │
│ (Web, CLI, Desktop — any frontend) │
└─────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ APP ORCHESTRATOR │
│ ┌─────────────┐ ┌─────────────┐ ┌──────────────────┐ │
│ │ Memory │ │ Cost │ │ MCP │ │
│ │ Manager │ │ Router │ │ Client │ │
│ │ │ │ │ │ │ │
│ │ • Add ctx │ │ • Classify │ │ • Tool calls │ │
│ │ • Recall │ │ • Route │ │ • Resource read │ │
│ │ • Summarize │ │ • Estimate │ │ • Prompt templates│ │
│ └─────────────┘ └─────────────┘ └──────────────────┘ │
└─────────────────────────────────────────────────────────────┘
│ │ │
┌─────▼──────┐ ┌────▼─────┐ ┌───▼─────────┐
│ Local │ │ Local │ │ MCP Servers │
│ Memory │ │ LLM │ │ (stdio/SSE) │
│ Store │ │ (Ollama) │ │ │
│ (SQLite+ │ │ │ │ • Filesystem │
│ sqlite_ │ │ Primary │ │ • Search │
│ vec) │ │ Compute │ │ • Database │
│ │ │ │ │ • Knowledge │
│ │ │ Fallback│ │ Base │
│ │ │ (cloud) │ └──────────────┘
│ │ └──────────┘
└────────────┘
python
import asyncio
from pathlib import Path
from typing import Optional
from memory.store import LocalMemoryStore
from routing.cost_aware_router import CostAwareRouter, ModelTier
from monitoring.metrics import CostTracker
class LocalFirstApp:
"""
Complete local-first AI application.
All operations work offline. Cloud is fallback only.
"""
def __init__(self, data_dir: Path = Path("~/.localfirst")):
self.data_dir = data_dir.expanduser()
self.data_dir.mkdir(parents=True, exist_ok=True)
self.memory = LocalMemoryStore(self.data_dir / "memory.db")
self.router = CostAwareRouter()
self.tracker = CostTracker(self.data_dir / "routing_log.jsonl")
self.session_id = "default"
async def chat(self, user_input: str) -> dict:
"""
Process a user message through the full local-first pipeline.
Returns response + metadata about routing decision.
"""
context = self.memory.get_context(user_input, max_tokens=1500)
decision = self.router.route(user_input, system_context=context)
prompt = self._build_prompt(user_input, context)
result = await self._execute_with_model(prompt, decision)
self.memory.add("user", user_input, self.session_id)
self.memory.add("assistant", result["response"], self.session_id)
self.tracker.log(decision, result["input_tokens"], result["output_tokens"])
return {
"response": result["response"],
"routing": {
"model": decision.model,
"tier": decision.tier.value,
"reason": decision.reason,
"estimated_cost": self.tracker.get_cost_estimate(
decision, result["input_tokens"], result["output_tokens"]
)
}
}
def _build_prompt(self, user_input: str, context: str) -> str:
"""Construct the full prompt with context injection."""
system_prompt = """You are a helpful AI assistant running in a local-first environment.
You have access to tools (memory, file system, search) via MCP.
Be concise, accurate, and respect the user's privacy — all data stays local."""
parts = [f"System: {system_prompt}"]
if context:
parts.append(f"\n{context}")
parts.append(f"\nUser: {user_input}")
parts.append("\nAssistant:")
return "\n".join(parts)
async def _execute_with_model(self, prompt: str, decision) -> dict:
"""Execute the prompt using the routed model."""
import httpx
if decision.tier in (ModelTier.FREE, ModelTier.LOW):
async with httpx.AsyncClient(timeout=120.0) as client:
resp = await client.post(
"http://localhost:11434/api/generate",
json={
"model": decision.model,
"prompt": prompt,
"stream": False,
"options": {"num_ctx": 4096}
}
)
data = resp.json()
return {
"response": data["response"],
"input_tokens": data.get("prompt_eval_count", 0),
"output_tokens": data.get("eval_count", 0)
}
else:
async with httpx.AsyncClient(timeout=60.0) as client:
return await self._execute_with_model(prompt, decision)
def run_periodic_maintenance(self):
"""Run memory compaction and cache cleanup."""
self.memory.summarize_old_sessions(older_than_days=14)
self.router._cache.clear()
print(f"Maintenance complete. Costs so far: {self.tracker.report()}")
def close(self):
self.memory.close()
async def main():
app = LocalFirstApp()
print("Local-First AI App (type 'quit' to exit)\n")
while True:
try:
user_input = input("> ").strip()
if user_input.lower() in ("quit", "exit"):
break
if not user_input:
continue
result = await app.chat(user_input)
print(f"\n{result['response']}\n")
print(f"[Routed: {result['routing']['model']} ({result['routing']['tier']})]")
except KeyboardInterrupt:
break
except Exception as e:
print(f"Error: {e}")
app.run_periodic_maintenance()
app.close()
print(f"\nFinal report: {app.tracker.report()}")
if __name__ == "__main__":
asyncio.run(main())
version: "3.9"
services:
ollama:
image: ollama/ollama:latest
container_mode: true
ports:
- "11434:11434"
volumes:
- ollama_models:/root/.ollama
command: ["serve"]
volumes:
ollama_models:
OpenChatCut (a reference architecture for MCP-heavy apps) teaches one critical lesson: design your MCP interfaces before your UI. The structure of your tools and resources determines what the model can actually do. If your MCP server exposes only "read_file" and "write_file," the model will treat every request as a file operation. Expose richer abstractions — "remember," "recall," "search" — and the model's behavior changes fundamentally.
The key implementation detail: use schema validation strictly. OpenChatCut validates all MCP tool inputs server-side, not client-side. This prevents malformed queries from reaching the model and ensures the model learns correct patterns faster.
RLM Cut's most distinctive contribution is treating cost as a first-class system metric, not an afterthought. Their routing decisions are logged, their cost per tier is tracked, and their model selection is auditable. The pattern:
This is not just operational hygiene — it directly shapes product decisions. When you can see that 73% of requests are handled by local models at zero marginal cost, you invest in better local models. When you see that 15% are complex debugging tasks requiring cloud escalation, you consider fine-tuning a local model on debug patterns.
OpenWork popularized the tiered memory architecture described in Section 3. Their key insight: memory is not one size fits all. The same system that stores your last 8 messages verbatim should also compress last month's conversations into summaries. The bridge between these tiers is semantic search — the recall mechanism finds relevant compressed memories and expands them just enough for context.
Another OpenWork contribution: importance scoring. Not all memories are equal. User corrections, explicit instructions, and emotional context get higher importance scores, which biases the recall algorithm to surface them first. This is implemented as a lightweight scoring function applied at write time, not a post-hoc analysis.
Before shipping a local-first AI app, verify:
nomic-embed-text
, qwen2.5:7b
, qwen2.5:14b
)Q: Can I run this on a machine with only 8GB RAM?
Yes. Use phi-3-mini
(2.3GB) or qwen2.5:3b
(2GB) for the primary model. The memory store uses SQLite which is lightweight. The bottleneck will be response speed, not functionality. Consider command-r
(5.8GB) if you need better reasoning on medium tasks.
Q: How do I handle tasks the local model can't solve?
The cost-aware router escalates to cloud APIs automatically. In production, you'd configure fallback providers (OpenAI, Anthropic) with rate limits and cost caps. The router's _is_local_capable
heuristic can be refined based on your actual error rates — if your local model fails on 30% of debugging tasks, mark those as requires_cloud
in your routing config.
Q: Does the memory store grow unbounded?
No. The summarize_old_sessions
method compresses memories older than the configured threshold (default: 14 days) into summarized records. The vector index is rebuilt during this process. In practice, a monthly maintenance cycle keeps the store under 100K records for typical usage, with ~5MB of disk usage.
Building local-first AI isn't about rejecting the cloud — it's about making the cloud optional. When your app works perfectly without network, costs almost nothing to run, and remembers everything it needs to, you've built something that scales differently: not by spending more, but by thinking smarter. The patterns from OpenChatCut, RLM Cut, and OpenWork show this is production-viable today, not a research exercise.
source & further reading
dev.to — original article
Building a Secure Enterprise AI Assistant: A Complete Architecture & Usage Guide
A Detector That Only Ever Says "Clean" Proves Nothing
I reimplemented the RustDesk wire protocol in pure Go — here's what it took