| name | local-llm-router |
| description | Route AI coding queries to local LLMs in air-gapped networks. Integrates Serena MCP for semantic code understanding. Use when working offline, with local models (Ollama, LM Studio, Jan, OpenWebUI), or in secure/closed environments. Triggers on local LLM, Ollama, LM Studio, Jan, air-gapped, offline AI, Serena, local inference, closed network, model routing, defense network, secure coding. |
Local LLM Router for Air-Gapped Networks
Intelligent routing of AI coding queries to local LLMs with Serena LSP integration for secure, offline-capable development environments.
Prerequisites (CRITICAL)
Before using this skill, ensure:
- Serena MCP Server installed and running (PRIMARY TOOL)
- At least one local LLM service running (Ollama, LM Studio, Jan, etc.)
pip install serena
uvx --from git+https://github.com/oraios/serena serena start-mcp-server
curl http://localhost:11434/api/version
curl http://localhost:1234/v1/models
curl http://localhost:1337/v1/models
Quick Start
import httpx
import asyncio
from dataclasses import dataclass
from enum import Enum
from typing import Optional
class TaskCategory(Enum):
CODING = "coding"
REASONING = "reasoning"
ANALYSIS = "analysis"
DOCUMENTATION = "documentation"
@dataclass
class RouterConfig:
"""Local LLM Router configuration."""
ollama_url: str = "http://localhost:11434"
lmstudio_url: str = "http://localhost:1234"
jan_url: str = "http://localhost:1337"
serena_enabled: bool = True
timeout: int = 30
async def quick_route(query: str, config: RouterConfig = RouterConfig()):
"""Quick routing example - detects services and routes query."""
services = await discover_services(config)
if not services:
raise RuntimeError("No local LLM services available")
category = classify_task(query)
model = select_model(category, services)
return await execute_query(query, model, services[0])
async def main():
response = await quick_route("Write a function to parse JSON safely")
print(response)
asyncio.run(main())
Serena Integration (PRIMARY TOOL)
CRITICAL: Serena MCP MUST be invoked FIRST for all code-related tasks. This provides semantic understanding of the codebase before routing to an LLM.
Why Serena First?
- Token Efficiency: Serena extracts only relevant code context
- Accuracy: Symbol-level operations vs grep-style searches
- Codebase Awareness: Understands types, references, call hierarchies
- Edit Precision: Applies changes at symbol level, not string matching
Serena MCP Setup
import subprocess
import json
from typing import Any
class SerenaMCP:
"""Serena MCP client for code intelligence."""
def __init__(self, workspace_root: str):
self.workspace = workspace_root
self.process = None
async def start(self):
"""Start Serena MCP server."""
self.process = subprocess.Popen(
["serena", "start-mcp-server", "--workspace", self.workspace],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
async def call(self, method: str, params: dict) -> Any:
"""Call Serena MCP method."""
request = {
"jsonrpc": "2.0",
"id": 1,
"method": method,
"params": params
}
self.process.stdin.write(json.dumps(request).encode() + b"\n")
self.process.stdin.flush()
response = self.process.stdout.readline()
return json.loads(response)
() -> :
.call(, {: name})
() -> :
.call(, {
: file,
: line,
: char
})
() -> :
.call(, {
: file,
: line,
: char
})
() -> :
.call(, {: file})
() -> :
.call(, {: file, : edits})
SERENA_TOOLS = {
: {: , : [, ]},
: {: , : [, ]},
: {: , : [, ]},
: {: , : []},
: {: , : []},
: {: , : []},
: {: , : []},
: {: , : []},
: {: , : []},
: {: , : []},
: {: , : []},
: {: , : [, ]},
: {: , : []},
}
Serena-First Request Handler
async def handle_code_request(
query: str,
file_context: Optional[dict] = None,
serena: SerenaMCP = None,
router: "LLMRouter" = None
):
"""
Handle code request with Serena-first pattern.
CRITICAL: Serena is ALWAYS invoked first for code tasks.
"""
category = classify_task(query)
serena_context = {}
if serena and file_context:
if file_context.get("file") and file_context.get("position"):
file = file_context["file"]
line = file_context["position"]["line"]
char = file_context["position"]["character"]
serena_context["hover"] = await serena.get_hover_info(file, line, char)
if category in [TaskCategory.ANALYSIS, TaskCategory.CODING]:
if "refactor" in query.lower() or "rename" in query.lower():
serena_context["references"] = await serena.get_references(
file, line, char
)
serena_context[] = serena.get_diagnostics(file)
enriched_query = build_enriched_query(query, serena_context)
model = router.select_model(category)
response = router.execute(enriched_query, model)
serena contains_code_edit(response):
edits = parse_code_edits(response)
serena.apply_edit(file_context[], edits)
response
() -> :
parts = [query]
serena_context.get():
hover = serena_context[]
parts.append()
serena_context.get():
refs = serena_context[]
parts.append()
ref refs[:]:
parts.append()
serena_context.get():
diags = serena_context[]
diags:
parts.append()
diag diags[:]:
parts.append()
.join(parts)
Service Discovery
Supported Services
| Service | Default Endpoint | Health Check | Models Endpoint | Chat Endpoint | API Style |
|---|
| Ollama | localhost:11434 | /api/version | /api/tags | /api/chat | Native |
| LM Studio | localhost:1234 | /v1/models | /v1/models | /v1/chat/completions | OpenAI |
| Jan | localhost:1337 | /v1/models | /v1/models | /v1/chat/completions | OpenAI |
| OpenWebUI | localhost:3000 | /api/health | /api/models | /api/chat | Custom |
| LocalAI | localhost:8080 | /readyz | /v1/models | /v1/chat/completions | OpenAI |
| vLLM | localhost:8000 | /health | /v1/models | /v1/chat/completions | OpenAI |
| llama.cpp | localhost:8080 | /health | /v1/models | /v1/chat/completions | OpenAI |
| Kobold.cpp | localhost:5001 | /api/v1/info | /api/v1/models | /api/v1/generate | Custom |
| GPT4All | localhost:4891 | /v1/models | /v1/models | /v1/chat/completions | OpenAI |
| text-generation-webui |
OS Detection
import sys
import os
import platform
from dataclasses import dataclass
@dataclass
class OSInfo:
platform: str
release: str
arch: str
is_wsl: bool
is_container: bool
def detect_os() -> OSInfo:
"""Detect operating system and environment."""
plat = sys.platform
if plat == 'win32':
plat = 'windows'
elif plat == 'darwin':
plat = 'darwin'
else:
plat = 'linux'
is_wsl = False
if plat == 'linux':
try:
with open('/proc/version', 'r') as f:
is_wsl = 'microsoft' in f.read().lower()
except FileNotFoundError:
pass
is_wsl = is_wsl or os.environ.get('WSL_DISTRO_NAME') is not None
is_container = (
os.path.exists()
os.environ.get()
)
is_container plat == :
:
(, ) f:
is_container = f.read() f.read()
FileNotFoundError:
OSInfo(
platform=plat,
release=platform.release(),
arch=platform.machine(),
is_wsl=is_wsl,
is_container=is_container
)
() -> :
os_info.is_wsl os_info.is_container:
endpoint.replace(, )
endpoint
Service Discovery Implementation
import httpx
import asyncio
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
@dataclass
class DiscoveredModel:
id: str
name: str
size: int = 0
family: Optional[str] = None
context_length: int = 4096
quantization: Optional[str] = None
@dataclass
class LLMService:
name: str
type: str
endpoint: str
status: str = 'unknown'
models: list = field(default_factory=list)
last_checked: datetime = None
api_style: str = 'openai'
health_path: str = '/v1/models'
models_path: str = '/v1/models'
chat_path: str = '/v1/chat/completions'
SERVICE_DEFAULTS = {
: LLMService(
name=,
=,
endpoint=,
health_path=,
models_path=,
chat_path=,
api_style=
),
: LLMService(
name=,
=,
endpoint=,
health_path=,
models_path=,
chat_path=,
api_style=
),
: LLMService(
name=,
=,
endpoint=,
health_path=,
models_path=,
chat_path=,
api_style=
),
: LLMService(
name=,
=,
endpoint=,
health_path=,
models_path=,
chat_path=,
api_style=
),
: LLMService(
name=,
=,
endpoint=,
health_path=,
models_path=,
chat_path=,
api_style=
),
: LLMService(
name=,
=,
endpoint=,
health_path=,
models_path=,
chat_path=,
api_style=
),
: LLMService(
name=,
=,
endpoint=,
health_path=,
models_path=,
chat_path=,
api_style=
),
: LLMService(
name=,
=,
endpoint=,
health_path=,
models_path=,
chat_path=,
api_style=
),
: LLMService(
name=,
=,
endpoint=,
health_path=,
models_path=,
chat_path=,
api_style=
),
}
:
():
.services: [, LLMService] = {}
.os_info = detect_os()
.custom_endpoints = custom_endpoints []
._client = httpx.AsyncClient(timeout=)
() -> [LLMService]:
discovered = []
tasks = []
key, default SERVICE_DEFAULTS.items():
service = LLMService(
name=default.name,
=default.,
endpoint=adjust_endpoint_for_os(default.endpoint, .os_info),
health_path=default.health_path,
models_path=default.models_path,
chat_path=default.chat_path,
api_style=default.api_style
)
tasks.append(._check_service(service))
custom .custom_endpoints:
service = LLMService(
name=custom.get(, ),
=,
endpoint=custom[],
health_path=custom.get(, ),
models_path=custom.get(, ),
chat_path=custom.get(, ),
api_style=custom.get(, )
)
tasks.append(._check_service(service))
results = asyncio.gather(*tasks, return_exceptions=)
result results:
(result, LLMService) result.status == :
discovered.append(result)
.services[result.] = result
discovered
() -> LLMService:
:
response = ._client.get(
)
response.status_code == :
service.status =
service.last_checked = datetime.now()
service.models = ._discover_models(service)
:
service.status =
(httpx.ConnectError, httpx.TimeoutException):
service.status =
service
() -> [DiscoveredModel]:
:
response = ._client.get(
)
data = response.json()
service. == :
[
DiscoveredModel(
=m[],
name=m[],
size=m.get(, ),
family=m.get(, {}).get(),
context_length=._infer_context_length(m[])
)
m data.get(, [])
]
:
[
DiscoveredModel(
=m[],
name=m[],
context_length=m.get(, )
)
m data.get(, [])
]
Exception:
[]
() -> :
name_lower = model_name.lower()
name_lower name_lower:
name_lower:
name_lower:
name_lower:
name_lower:
name_lower:
name_lower name_lower:
name_lower:
name_lower:
Task Classification
Classification System
import re
from enum import Enum
from dataclasses import dataclass
class TaskCategory(Enum):
CODING = "coding"
REASONING = "reasoning"
ANALYSIS = "analysis"
DOCUMENTATION = "documentation"
@dataclass
class ClassificationResult:
category: TaskCategory
confidence: float
requires_serena: bool
keywords_matched: list[str]
TASK_PATTERNS = {
TaskCategory.CODING: [
r"(?:write|create|implement|code|generate)\s+(?:a\s+)?(?:function|class|method|component)",
r"(?:fix|debug|solve)\s+(?:this|the)\s+(?:bug|error|issue)",
r"refactor\s+(?:this|the)",
r"add\s+(?:error\s+handling|validation|logging|tests?)",
r"complete\s+(?:this|the)\s+code",
r"(?:convert|translate)\s+(?:this|the)\s+code",
r"(?:optimize|improve)\s+(?:this|the)\s+(?:function|code|performance)",
],
TaskCategory.REASONING: [
r"(?:design|architect|plan)\s+(?:a|the)\s+(?:system|architecture|solution)",
r"how\s+should\s+(?:I|we)\s+(?:approach|structure|implement)",
r"what\s+(?:is|would\s+be)\s+the\s+best\s+(?:way|approach|pattern)",
r"explain\s+the\s+(?:logic|reasoning|algorithm)",
r"compare\s+(?:and\s+contrast|between)",
r"(?:recommend|suggest)\s+(?:an?\s+)?(?:approach|solution|pattern)",
r"trade-?offs?\s+(?:between|of)",
],
TaskCategory.ANALYSIS: [
r"(?:review|analyze|audit)\s+(?:this|the)\s+code",
,
,
,
,
,
],
TaskCategory.DOCUMENTATION: [
,
,
,
,
,
,
],
}
KEYWORD_WEIGHTS = {
: (TaskCategory.CODING, ),
: (TaskCategory.CODING, ),
: (TaskCategory.CODING, ),
: (TaskCategory.CODING, ),
: (TaskCategory.CODING, ),
: (TaskCategory.CODING, ),
: (TaskCategory.CODING, ),
: (TaskCategory.CODING, ),
: (TaskCategory.REASONING, ),
: (TaskCategory.REASONING, ),
: (TaskCategory.REASONING, ),
: (TaskCategory.REASONING, ),
: (TaskCategory.REASONING, ),
: (TaskCategory.REASONING, ),
: (TaskCategory.REASONING, ),
: (TaskCategory.ANALYSIS, ),
: (TaskCategory.ANALYSIS, ),
: (TaskCategory.ANALYSIS, ),
: (TaskCategory.ANALYSIS, ),
: (TaskCategory.ANALYSIS, ),
: (TaskCategory.ANALYSIS, ),
: (TaskCategory.DOCUMENTATION, ),
: (TaskCategory.DOCUMENTATION, ),
: (TaskCategory.DOCUMENTATION, ),
: (TaskCategory.DOCUMENTATION, ),
: (TaskCategory.DOCUMENTATION, ),
}
() -> ClassificationResult:
query_lower = query.lower()
scores = {cat: cat TaskCategory}
matched_keywords = []
category, patterns TASK_PATTERNS.items():
pattern patterns:
re.search(pattern, query_lower):
scores[category] +=
words = re.findall(, query_lower)
word words:
word KEYWORD_WEIGHTS:
category, weight = KEYWORD_WEIGHTS[word]
scores[category] += weight *
matched_keywords.append(word)
best_category = (scores, key=scores.get)
confidence = (scores[best_category], )
confidence < :
best_category = TaskCategory.CODING
confidence =
requires_serena = (
best_category == TaskCategory.ANALYSIS
(kw query_lower kw [
, , , ,
, , ,
])
)
ClassificationResult(
category=best_category,
confidence=confidence,
requires_serena=requires_serena,
keywords_matched=matched_keywords
)
Model Selection
Model Capability Matrix
from dataclasses import dataclass
from typing import Optional
@dataclass
class ModelCapability:
id: str
family: str
context_window: int
vram_gb: float
categories: list[TaskCategory]
performance_scores: dict[TaskCategory, int]
tier: int
quantization: Optional[str] = None
MODEL_DATABASE: dict[str, ModelCapability] = {
"deepseek-v3": ModelCapability(
id="deepseek-v3",
family="deepseek",
context_window=128000,
vram_gb=48,
categories=[TaskCategory.CODING, TaskCategory.REASONING, TaskCategory.ANALYSIS],
performance_scores={
TaskCategory.CODING: 99,
TaskCategory.REASONING: 97,
TaskCategory.ANALYSIS: 96,
TaskCategory.DOCUMENTATION: 92
},
tier=1
),
"qwen2.5-coder-32b": ModelCapability(
id="qwen2.5-coder-32b",
family="qwen",
context_window=131072,
vram_gb=,
categories=[TaskCategory.CODING, TaskCategory.ANALYSIS],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.CODING, TaskCategory.ANALYSIS, TaskCategory.REASONING],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.CODING],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.CODING],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.CODING],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.CODING],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.CODING],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.CODING],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.REASONING, TaskCategory.CODING],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.REASONING, TaskCategory.CODING],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.REASONING, TaskCategory.CODING],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.REASONING, TaskCategory.CODING],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.REASONING, TaskCategory.CODING],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.REASONING, TaskCategory.CODING],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.REASONING, TaskCategory.DOCUMENTATION],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.REASONING, TaskCategory.CODING],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.REASONING],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.REASONING, TaskCategory.DOCUMENTATION],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.REASONING, TaskCategory.CODING],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.REASONING],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.REASONING, TaskCategory.CODING],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.REASONING],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.ANALYSIS],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.DOCUMENTATION],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.DOCUMENTATION],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.CODING, TaskCategory.REASONING],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.CODING, TaskCategory.REASONING],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.REASONING, TaskCategory.DOCUMENTATION],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.REASONING, TaskCategory.DOCUMENTATION],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.CODING],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.CODING],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.CODING, TaskCategory.REASONING],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.REASONING],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
: ModelCapability(
=,
family=,
context_window=,
vram_gb=,
categories=[TaskCategory.REASONING, TaskCategory.DOCUMENTATION],
performance_scores={
TaskCategory.CODING: ,
TaskCategory.REASONING: ,
TaskCategory.ANALYSIS: ,
TaskCategory.DOCUMENTATION:
},
tier=
),
}
TASK_MODEL_PRIORITY = {
TaskCategory.CODING: [
, , ,
, , ,
, ,
, ,
],
TaskCategory.REASONING: [
, , ,
, ,
, , ,
, ,
, ,
],
TaskCategory.ANALYSIS: [
, , ,
,
],
TaskCategory.DOCUMENTATION: [
, , ,
, ,
],
}
Model Selection Logic
from typing import Optional
class ModelSelector:
"""Select optimal model for task based on availability and requirements."""
def __init__(self, available_models: list[str]):
self.available = set(m.lower() for m in available_models)
def select(
self,
category: TaskCategory,
required_context: int = 0,
max_vram_gb: Optional[float] = None
) -> Optional[str]:
"""Select best available model for task category."""
priority_list = TASK_MODEL_PRIORITY.get(category, [])
for model_id in priority_list:
if not self._is_available(model_id):
continue
capability = MODEL_DATABASE.get(model_id)
if not capability:
continue
if required_context > 0 and capability.context_window < required_context:
continue
max_vram_gb capability.vram_gb > max_vram_gb:
model_id
model_id, capability MODEL_DATABASE.items():
._is_available(model_id):
model_id
() -> :
model_lower = model_id.lower()
model_lower .available:
avail .available:
model_lower avail avail model_lower:
() -> []:
priority_list = TASK_MODEL_PRIORITY.get(category, [])
available_in_priority = [
m m priority_list ._is_available(m)
]
fallbacks = []
model_id available_in_priority:
capability = MODEL_DATABASE.get(model_id)
capability capability.tier >= :
fallbacks.append(model_id)
fallbacks
Context Management
Token Counting
from abc import ABC, abstractmethod
import re
class TokenCounter(ABC):
"""Base class for token counting."""
@abstractmethod
def count(self, text: str) -> int:
pass
class EstimationCounter(TokenCounter):
"""Estimation-based token counter (no external dependencies)."""
def __init__(self, chars_per_token: float = 4.0):
self.chars_per_token = chars_per_token
def count(self, text: str) -> int:
return int(len(text) / self.chars_per_token)
class QwenCounter(TokenCounter):
"""Token counter for Qwen models."""
def count(self, text: str) -> int:
return int(len(text) / 3.5)
class LlamaCounter():
() -> :
((text) / )
TOKEN_COUNTERS = {
: QwenCounter(),
: EstimationCounter(),
: LlamaCounter(),
: EstimationCounter(),
: EstimationCounter(),
: EstimationCounter(),
}
() -> TokenCounter:
capability = MODEL_DATABASE.get(model_id)
capability:
TOKEN_COUNTERS.get(capability.family, TOKEN_COUNTERS[])
TOKEN_COUNTERS[]
Context Manager
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
@dataclass
class Message:
role: str
content: str
timestamp: datetime = field(default_factory=datetime.now)
token_count: int = 0
metadata: dict = field(default_factory=dict)
@dataclass
class ConversationContext:
session_id: str
messages: list[Message] = field(default_factory=list)
total_tokens: int = 0
system_prompt: str = ""
system_prompt_tokens: int = 0
active_model: str = ""
model_history: list[str] = field(default_factory=list)
compaction_count: int = 0
class ContextManager:
"""Manage conversation context with compaction support."""
def __init__(
self,
session_id: str,
system_prompt: str = "",
compaction_threshold: float = 0.8,
compaction_target: = ,
preserve_recent: =
):
.context = ConversationContext(
session_id=session_id,
system_prompt=system_prompt
)
.compaction_threshold = compaction_threshold
.compaction_target = compaction_target
.preserve_recent = preserve_recent
._counter: [TokenCounter] =
():
.context.active_model:
.context.model_history.append(.context.active_model)
.context.active_model = model_id
._counter = get_token_counter(model_id)
._recount_tokens()
():
token_count = ._counter.count(content) ._counter
message = Message(
role=role,
content=content,
token_count=token_count,
metadata=metadata {}
)
.context.messages.append(message)
.context.total_tokens += token_count
() -> :
threshold = (max_tokens * .compaction_threshold)
.context.total_tokens > threshold:
._compact(max_tokens)
():
target = (max_tokens * .compaction_target)
msg .context.messages:
msg.role == msg.token_count > :
original = msg.token_count
msg.content =
msg.token_count = ._counter.count(msg.content)
msg.metadata[] =
msg.metadata[] = original
._recalculate_total()
.context.total_tokens <= target:
(.context.messages) > .preserve_recent:
older = .context.messages[:-.preserve_recent]
recent = .context.messages[-.preserve_recent:]
summary = ._create_summary(older)
summary_msg = Message(
role=,
content=,
token_count=._counter.count(summary),
metadata={: }
)
.context.messages = [summary_msg] + recent
.context.compaction_count +=
._recalculate_total()
() -> :
key_points = []
msg messages:
msg.role == :
first_sentence = msg.content.split()[][:]
key_points.append()
msg.role == (key_points) < :
msg.content.lower() msg.content.lower():
first_sentence = msg.content.split()[][:]
key_points.append()
.join(key_points[:])
():
._counter:
.context.system_prompt_tokens = ._counter.count(.context.system_prompt)
msg .context.messages:
msg.token_count = ._counter.count(msg.content)
._recalculate_total()
():
.context.total_tokens = (
.context.system_prompt_tokens +
(m.token_count m .context.messages)
)
() -> []:
messages = []
.context.system_prompt:
messages.append({
: ,
: .context.system_prompt
})
msg .context.messages:
messages.append({
: msg.role,
: msg.content
})
messages
() -> :
.set_model(new_model)
Configuration
Inline Configuration Schema
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class ServiceConfig:
"""Configuration for a single LLM service."""
enabled: bool = True
endpoint: str = ""
priority: int = 1
timeout: int = 30000
max_retries: int = 3
api_style: str = "openai"
@dataclass
class TaskRoutingConfig:
"""Configuration for task routing."""
primary_models: list[str] = field(default_factory=list)
fallback_models: list[str] = field(default_factory=list)
min_context: int = 8192
require_serena: bool = False
@dataclass
class SecurityConfig:
"""Security configuration for air-gapped networks."""
allow_external: bool = False
allowed_hosts: list[str] = field(default_factory=lambda: [
"localhost", "127.0.0.1", "host.docker.internal"
])
allowed_cidrs: list[] = field(default_factory=: [
, ,
])
audit_enabled: =
audit_log_path: =
log_queries: =
log_responses: =
verify_checksums: =
:
compaction_threshold: =
compaction_target: =
preserve_recent_messages: =
preserve_recent_tool_calls: =
max_tool_output_tokens: =
:
ollama: ServiceConfig = field(default_factory=: ServiceConfig(
endpoint=,
priority=
))
lmstudio: ServiceConfig = field(default_factory=: ServiceConfig(
endpoint=,
priority=
))
jan: ServiceConfig = field(default_factory=: ServiceConfig(
endpoint=,
priority=
))
custom_endpoints: [] = field(default_factory=)
coding: TaskRoutingConfig = field(default_factory=: TaskRoutingConfig(
primary_models=[, , ],
fallback_models=[, , ],
min_context=
))
reasoning: TaskRoutingConfig = field(default_factory=: TaskRoutingConfig(
primary_models=[, , ],
fallback_models=[, ],
min_context=
))
analysis: TaskRoutingConfig = field(default_factory=: TaskRoutingConfig(
primary_models=[, ],
fallback_models=[, ],
min_context=,
require_serena=
))
documentation: TaskRoutingConfig = field(default_factory=: TaskRoutingConfig(
primary_models=[, ],
fallback_models=[, ],
min_context=
))
serena_enabled: =
serena_priority: =
context: ContextConfig = field(default_factory=ContextConfig)
security: SecurityConfig = field(default_factory=SecurityConfig)
DEFAULT_CONFIG = RouterConfig()
() -> RouterConfig:
config = RouterConfig()
data:
service_name, service_data data[].items():
(config, service_name):
(config, service_name, ServiceConfig(**service_data))
category [, , , ]:
category data.get(, {}):
(config, category, TaskRoutingConfig(**data[][category]))
data:
config.security = SecurityConfig(**data[])
config
Example YAML Configuration (for reference)
version: "1.0"
environment: "air-gapped"
services:
ollama:
enabled: true
endpoint: "http://localhost:11434"
priority: 1
timeout: 30000
lmstudio:
enabled: true
endpoint: "http://localhost:1234"
priority: 2
jan:
enabled: false
endpoint: "http://localhost:1337"
priority: 3
custom_endpoints:
- name: "internal-gpu-server"
endpoint: "http://192.168.1.100:8000"
priority: 0
api_style: "openai"
task_routing:
coding:
primary_models:
- "deepseek-v3"
- "qwen2.5-coder-32b"
- "deepseek-coder-v2"
fallback_models:
- "codellama-34b"
-
Fallback Strategy
Graceful Degradation
from enum import IntEnum
from dataclasses import dataclass
from typing import Optional, Any
class FallbackLevel(IntEnum):
PRIMARY = 0
FALLBACK_MODELS = 1
REDUCED_CONTEXT = 2
SMALLEST_MODEL = 3
FAILED = 4
@dataclass
class ExecutionResult:
success: bool
model: Optional[str] = None
service: Optional[str] = None
response: Any = None
fallback_level: FallbackLevel = FallbackLevel.PRIMARY
error: Optional[str] = None
class FallbackExecutor:
"""Execute queries with multi-level fallback."""
def __init__(
self,
discovery: ServiceDiscovery,
context_manager: ContextManager,
config: RouterConfig
):
self.discovery = discovery
self.context = context_manager
self.config = config
async def execute_with_fallback(
self,
query: str,
category: TaskCategory
) -> ExecutionResult:
"""Execute query with fallback strategy."""
task_config = (.config, category.value)
primary_models = task_config.primary_models
fallback_models = task_config.fallback_models
model primary_models:
result = ._try_model(model, query)
result.success:
result.fallback_level = FallbackLevel.PRIMARY
result
model fallback_models:
result = ._try_model(model, query)
result.success:
result.fallback_level = FallbackLevel.FALLBACK_MODELS
result
.context._compact(task_config.min_context)
model primary_models + fallback_models:
result = ._try_model(model, query)
result.success:
result.fallback_level = FallbackLevel.REDUCED_CONTEXT
result
smallest = ._find_smallest_model()
smallest:
result = ._try_model(smallest, query)
result.success:
result.fallback_level = FallbackLevel.SMALLEST_MODEL
result
ExecutionResult(
success=,
fallback_level=FallbackLevel.FAILED,
error=
)
() -> ExecutionResult:
service = ._find_service_with_model(model_id)
service:
ExecutionResult(
success=,
error=
)
:
response = ._execute_on_service(service, model_id, query)
ExecutionResult(
success=,
model=model_id,
service=service.name,
response=response
)
Exception e:
ExecutionResult(
success=,
error=(e)
)
() -> [LLMService]:
services = (.discovery.services.values())
services.sort(key= s: (.config, s., ServiceConfig()).priority)
service services:
model service.models:
model_id.lower() model..lower() model..lower() model_id.lower():
service
() -> []:
smallest =
smallest_vram = ()
service .discovery.services.values():
model service.models:
capability = MODEL_DATABASE.get(model.)
capability capability.vram_gb < smallest_vram:
smallest = model.
smallest_vram = capability.vram_gb
smallest
() -> :
httpx
messages = .context.export_for_api()
messages.append({: , : query})
httpx.AsyncClient() client:
service.api_style == service. == :
response = client.post(
,
json={
: model_id,
: messages,
:
},
timeout=.config.ollama.timeout /
)
data = response.json()
data.get(, {}).get(, )
:
response = client.post(
,
json={
: model_id,
: messages,
:
},
timeout=
)
data = response.json()
data.get(, [{}])[].get(, {}).get(, )
Security (Air-Gapped)
Network Isolation
import hashlib
import json
from datetime import datetime
from dataclasses import dataclass
from typing import Optional
import ipaddress
import logging
@dataclass
class AuditLogEntry:
timestamp: str
event_type: str
session_id: Optional[str] = None
model: Optional[str] = None
service: Optional[str] = None
query_hash: Optional[str] = None
tokens_in: int = 0
tokens_out: int = 0
success: bool = True
error: Optional[str] = None
class SecurityModule:
"""Security enforcement for air-gapped networks."""
def __init__(self, config: SecurityConfig):
self.config = config
self._allowed_ips = self._parse_allowed_networks()
self._logger = self._setup_audit_logger()
() -> :
networks = []
host .config.allowed_hosts:
host:
networks.append(ipaddress.ip_network(host, strict=))
:
:
ip = ipaddress.ip_address(host)
networks.append(ipaddress.ip_network())
ValueError:
host == :
networks.append(ipaddress.ip_network())
host == :
networks.append(ipaddress.ip_network())
cidr .config.allowed_cidrs:
networks.append(ipaddress.ip_network(cidr, strict=))
networks
() -> logging.Logger:
logger = logging.getLogger()
logger.setLevel(logging.INFO)
.config.audit_enabled:
handler = logging.FileHandler(.config.audit_log_path)
handler.setFormatter(logging.Formatter())
logger.addHandler(handler)
logger
() -> :
.config.allow_external:
:
urllib.parse urlparse
parsed = urlparse(url)
host = parsed.hostname
host [, , ]:
:
ip = ipaddress.ip_address(host)
network ._allowed_ips:
ip network:
ValueError:
host [, ]
Exception:
():
.config.audit_enabled:
entry = AuditLogEntry(
timestamp=datetime.now().isoformat(),
event_type=,
session_id=session_id,
model=model,
service=service,
query_hash=._hash_content(query) .config.log_queries ,
tokens_in=tokens_in,
tokens_out=tokens_out,
success=success,
error=error
)
._logger.info(json.dumps(entry.__dict__))
():
.config.audit_enabled:
entry = {
: datetime.now().isoformat(),
: ,
**details
}
._logger.warning(json.dumps(entry))
() -> :
hashlib.sha256(content.encode()).hexdigest()[:]
AIR_GAPPED_CHECKLIST =
Coding Agent Detection
Detect Active Coding Agent
import os
import sys
from dataclasses import dataclass
from typing import Optional
@dataclass
class CodingAgentInfo:
name: str
type: str
version: Optional[str] = None
config_path: Optional[str] = None
AGENT_ENV_MARKERS = {
'QWEN_CLI_VERSION': ('qwen-cli', 'cli'),
'OPENCODE_SESSION': ('opencode', 'cli'),
'AIDER_SESSION': ('aider', 'cli'),
'CODEX_SESSION': ('codex', 'cli'),
'GEMINI_CLI_SESSION': ('gemini-cli', 'cli'),
'CONTINUE_SESSION': ('continue', 'ide'),
'CLINE_SESSION': ('cline', 'ide'),
'ROO_CODE_SESSION': ('roo-code', 'ide'),
'CURSOR_SESSION': ('cursor', 'ide'),
'OPENWEBUI_SESSION': (, ),
: (, ),
: (, ),
: (, ),
}
() -> CodingAgentInfo:
env_var, (name, agent_type) AGENT_ENV_MARKERS.items():
value = os.environ.get(env_var)
value:
CodingAgentInfo(
name=name,
=agent_type,
version=value value !=
)
:
psutil
parent = psutil.Process(os.getppid())
parent_name = parent.name().lower()
agent_process_names = {
: ,
: ,
: ,
: ,
: ,
}
proc_name, agent_name agent_process_names.items():
proc_name parent_name:
CodingAgentInfo(name=agent_name, =)
ImportError:
os.environ.get():
CodingAgentInfo(
name=os.environ.get(, ),
=
)
CodingAgentInfo(name=, =)
() -> :
configs = {
: {
: ,
: ,
},
: {
: ,
: ,
},
: {
: ,
: ,
},
: {
: ,
: ,
},
}
configs.get(agent.name, {})
Complete Router Implementation
class LocalLLMRouter:
"""
Complete Local LLM Router with Serena integration.
Usage:
router = LocalLLMRouter(workspace="/path/to/project")
await router.initialize()
response = await router.route("Implement a binary search function")
print(response)
"""
def __init__(
self,
workspace: str,
config: RouterConfig = None,
session_id: str = None
):
self.workspace = workspace
self.config = config or DEFAULT_CONFIG
self.session_id = session_id or self._generate_session_id()
self.serena: Optional[SerenaMCP] = None
self.discovery: Optional[ServiceDiscovery] = None
self.context: Optional[ContextManager] = None
self.security: Optional[SecurityModule] = None
self.selector: Optional[ModelSelector] = None
self.fallback: Optional[FallbackExecutor] = None
self.os_info = detect_os()
self.coding_agent = detect_coding_agent()
self._initialized = False
async ():
.security = SecurityModule(.config.security)
.discovery = ServiceDiscovery(.config.custom_endpoints)
services = .discovery.discover_all()
services:
RuntimeError()
all_models = []
service services:
all_models.extend(m. m service.models)
.selector = ModelSelector(all_models)
.context = ContextManager(
session_id=.session_id,
system_prompt=._build_system_prompt(),
compaction_threshold=.config.context.compaction_threshold,
compaction_target=.config.context.compaction_target,
preserve_recent=.config.context.preserve_recent_messages
)
.config.serena_enabled:
.serena = SerenaMCP(.workspace)
:
.serena.start()
Exception e:
logging.warning()
.serena =
.fallback = FallbackExecutor(
.discovery,
.context,
.config
)
._initialized =
() -> :
._initialized:
.initialize()
classification = classify_task(query)
serena_context = {}
.serena (classification.requires_serena file_context):
serena_context = ._gather_serena_context(
query, file_context, classification
)
enriched_query = ._build_enriched_query(query, serena_context)
model = .selector.select(
classification.category,
required_context=.context.context.total_tokens + (query) //
)
model:
RuntimeError()
.context.set_model(model)
model_capability = MODEL_DATABASE.get(model)
model_capability:
.context.check_and_compact(model_capability.context_window)
result = .fallback.execute_with_fallback(
enriched_query,
classification.category
)
.security.log_query(
session_id=.session_id,
model=result.model model,
service=result.service ,
query=query,
tokens_in=(query) // ,
tokens_out=(result.response ) // ,
success=result.success,
error=result.error
)
result.success:
RuntimeError()
.context.add_message(, query)
.context.add_message(, result.response)
.serena file_context contains_code_edit(result.response):
._apply_serena_edits(result.response, file_context)
result.response
() -> :
context = {}
file_context:
context
file = file_context.get()
position = file_context.get(, {})
line = position.get(, )
char = position.get(, )
:
context[] = .serena.get_hover_info(file, line, char)
query.lower() query.lower():
context[] = .serena.get_references(file, line, char)
classification.category == TaskCategory.ANALYSIS:
context[] = .serena.get_diagnostics(file)
Exception e:
logging.warning()
context
() -> :
build_enriched_query(query, serena_context)
():
edits = parse_code_edits(response)
edits:
.serena.apply_edit(file_context[], edits)
() -> :
() -> :
uuid
(uuid.uuid4())[:]
() -> :
markers = [, , , , , , ]
(marker response marker markers)
() -> :
re
code_blocks = re.findall(, response, re.DOTALL)
[{: block.strip()} block code_blocks]
Resources
