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.
1. Why Local-First, Not Just "Offline"
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.
2. MCP Integration: The Structured Bridge Between Apps and Models
2.1 What MCP Actually Is
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:
- Tools — functions the model can call (search, read files, run commands)
- Resources — addressable data sources the model can read
- Prompts — reusable template patterns for common interactions
The client (your app) registers these with the LLM runtime, and the model learns to invoke them through structured JSON.
2.2 Architecture: Where MCP Lives in Your Stack
┌─────────────────────────────────────────────────────┐
│ 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.
└──────────────┘
2.3 Implementing an MCP Server in Python
Here's a production-grade MCP server that provides memory and search capabilities — the two most common needs for local-first apps:
# mcp_server/local_first_server.py
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", {}))
# Generate embedding using local model (e.g., via local Ollama endpoint)
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
# Sync to vector index
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())
2.4 Registering the MCP Server with an LLM Runtime
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.
3. Offline Memory: Persistent Context Without the Cloud
3.1 The Memory Problem in Local-First Apps
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:
- Vector embeddings for semantic retrieval
- Summarization to compress long histories
- Recency-weighted injection to prioritize fresh context
3.2 The Hybrid Memory Architecture
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 │
└─────────────────────────────────────────────────┘
3.3 Implementation: A Production Memory Store
# memory/store.py
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.
"""
# 1. Short-term: recent conversation history
recent = self._session_messages[-8:] # last 8 messages
short_term = "\n".join(m.to_context_snippet(150) for m in recent)
# 2. Long-term: semantic recall
long_term = self._recall(query, max_results=3)
# 3. Combine with priority ordering
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)
# 4. Trim to token budget (rough char-based estimate)
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()
# Group by session and summarize in batches
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:"""
# Call local Ollama for summarization
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)
# Remove old records
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()
3.4 When to Summarize vs. Keep Raw
The key insight from OpenWork's experience is that summarization should be lazy and periodic, not real-time:
- Keep raw for the last 7 days (high recall value, fits in context)
- Summarize anything older (saves tokens, preserves essence)
- Trigger summarization when the memory store exceeds a token budget (e.g., 50K tokens worth of records)
- Never summarize high-importance records (user explicitly marked them)
This lazy compaction mirrors how database vacuuming works — frequent writes, occasional cleanup.
4. Cost Optimization: The Hybrid Model Router
4.1 The Cost Problem
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?
4.2 The Router Architecture
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 $$$
4.3 Implementation: The Cost-Aware Router
# routing/cost_aware_router.py
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.
"""
# Keyword-based tier assignments (fast path)
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 with costs (per 1M tokens input/output)
MODEL_REGISTRY = {
# Local models — effectively free
"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},
# Cloud models with costs
"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
# Stage 1: Keyword heuristic classification
task_type = self._classify_task(user_input)
# Stage 2: Select model based on task type + budget
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.
"""
# Prefer local models for known simple tasks
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"
)
# For complex tasks, try local first, escalate on failure
preferred_local = self._find_best_local(max_tokens)
if preferred_local:
# Test local model capability with a lightweight probe
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}"
)
# Fall back to cloud — pick cheapest capable model
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
]
# Prefer larger, more capable models first
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.
"""
# Larger local models handle more complex tasks
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
# Check input length — very long inputs may exceed local context
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
4.4 Measuring the Impact
The metric that matters is cost per resolved request. Here's how to track it:
# monitoring/metrics.py
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()
}
}
5. Putting It All Together: The Complete Local-First App
5.1 System Architecture
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) │ └──────────────┘
│ │ └──────────┘
└────────────┘
5.2 The Main Application Loop
# app/orchestrator.py
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.
"""
# 1. Build context from memory
context = self.memory.get_context(user_input, max_tokens=1500)
# 2. Classify and route
decision = self.router.route(user_input, system_context=context)
# 3. Build the prompt
prompt = self._build_prompt(user_input, context)
# 4. Execute via chosen model
result = await self._execute_with_model(prompt, decision)
# 5. Store the exchange in memory
self.memory.add("user", user_input, self.session_id)
self.memory.add("assistant", result["response"], self.session_id)
# 6. Track costs
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):
# Local execution via Ollama
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:
# Cloud fallback
async with httpx.AsyncClient(timeout=60.0) as client:
# This would call OpenAI/Anthropic in production
# For now, simulate with local as fallback
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)
# Clear stale routing cache
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())
5.3 Docker Compose for Local Infrastructure
# docker-compose.yml
version: "3.9"
services:
ollama:
image: ollama/ollama:latest
container_mode: true
ports:
- "11434:11434"
volumes:
- ollama_models:/root/.ollama
command: ["serve"]
# MCP servers as separate containers (optional for isolation)
# Uncomment for production-grade isolation
# mcp-files:
# build: ./mcp-servers/files
# volumes:
# - ./data:/data
# environment:
# - DATA_DIR=/data
volumes:
ollama_models:
6. Lessons from OpenChatCut, RLM Cut, and OpenWork
6.1 OpenChatCut: The MCP-First Design
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.
6.2 RLM Cut: Cost Awareness as a First-Class Concern
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:
- Every request is classified before execution
- The classification rationale is stored
- Monthly reports show cost distribution by tier
- Thresholds trigger alerts (e.g., "cloud usage exceeded 20% this month")
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.
6.3 OpenWork: The Hybrid Memory Pattern
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.
7. Production Checklist
Before shipping a local-first AI app, verify:
- [ ] Ollama is running with all required models pulled (
nomic-embed-text,qwen2.5:7b,qwen2.5:14b) - [ ] MCP servers start successfully and pass health checks
- [ ] Memory store initializes without errors (SQLite + sqlite-vec)
- [ ] Routing logic falls back gracefully when local models fail
- [ ] Periodic maintenance runs (cron job or scheduled task for summarization)
- [ ] Cost tracking is active and logs are rotated
- [ ] Privacy audit complete — no data leaks to external endpoints unless explicitly routed
- [ ] Offline mode verified — all core features work with network disabled
- [ ] Context window management — long conversations don't exhaust local model capacity
- [ ] Embedding consistency — same inputs produce same embeddings (deterministic model loading)
Frequently Asked Questions
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.
Top comments (0)