DIY Private AI Agent Gateway: Building a Resilient Semantic Router for Local Coding Models

Executive Summary

As personal agentic workflows mature from single prompt-response interactions to multi-agent coding loops, relying on a single backend engine introduces brittle single-point-of-failure risks. When training workloads claim workstation GPUs or inference containers experience memory compaction pauses, background coding agents stall.

To guarantee continuous availability for autonomous developer tools, personal infrastructure requires an intelligent gateway layer. A self-hosted semantic router sits between local editor extensions (VSCode, Cursor, OpenHands) and underlying hardware backends. By dynamically categorizing incoming payloads into specialized execution paths (such as fast syntax edits, deep reasoning reviews, or offline background processing), the gateway maximizes accelerator utilization while providing sub-second failover.

flowchart TD
    Client["IDE / Autonomous Agent CLI"] --> Gateway["Personal Semantic Router (:11436)"]
    Gateway --> Classifier{"Prompt Intent Classifier"}
    Classifier -- "Code Synthesis & AST Edits" --> NodeA["Node A: High-Throughput Coder (vLLM :8000)"]
    Classifier -- "Deep Reasoning & Audit" --> NodeB["Node B: Multi-Stage Judge (vLLM :8000)"]
    Classifier -- "Fallback / Busy Drain" --> NodeC["Node C: Quantized Backup Engine (:11434)"]
    NodeA -. "Health Check Failed" .-> NodeC
    NodeB -. "Health Check Failed" .-> NodeC

Implementation

The core implementation utilizes an asynchronous Python reverse proxy built on FastAPI and httpx. The router continuously polls node health endpoints, measures rolling p95 latency, and transparently retries requests against alternate backends if a primary node drops a connection.

1. Resilient Model Router Service

Save this script as model_router.py:

import asyncio
import time
from typing import Dict, List, Optional
import httpx
from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse

app = FastAPI(title="Local Autonomous Model Router")

BACKENDS = [
    {"name": "primary-coder", "url": "http://127.0.0.1:8000/v1", "healthy": True, "consecutive_fails": 0},
    {"name": "secondary-evaluator", "url": "http://127.0.0.1:8001/v1", "healthy": True, "consecutive_fails": 0},
]

HEALTH_INTERVAL_SECONDS = 5.0

async def health_monitor_loop():
    """Background task monitoring node connectivity and queue depths."""
    async with httpx.AsyncClient(timeout=3.0) as client:
        while True:
            for b in BACKENDS:
                try:
                    resp = await client.get(f"{b['url']}/models")
                    if resp.status_code == 200:
                        b["healthy"] = True
                        b["consecutive_fails"] = 0
                    else:
                        b["healthy"] = False
                except Exception:
                    b["consecutive_fails"] += 1
                    if b["consecutive_fails"] >= 2:
                        b["healthy"] = False
            await asyncio.sleep(HEALTH_INTERVAL_SECONDS)

@app.on_event("startup")
async def startup_event():
    asyncio.create_task(health_monitor_loop())

@app.post("/v1/chat/completions")
async def route_completion(request: Request):
    body = await request.body()
    active_nodes = [b for b in BACKENDS if b["healthy"]]
    
    if not active_nodes:
        return JSONResponse({"error": "No healthy local backends available"}, status_code=503)

    # Simple round-robin with instantaneous fallback
    selected_node = active_nodes[0]
    async with httpx.AsyncClient(timeout=180.0) as client:
        try:
            resp = await client.post(
                f"{selected_node['url']}/chat/completions",
                content=body,
                headers={"Content-Type": "application/json"}
            )
            return Response(content=resp.content, status_code=resp.status_code, media_type="application/json")
        except Exception as exc:
            # Fallback to secondary node on network drop
            if len(active_nodes) > 1:
                fallback_node = active_nodes[1]
                resp = await client.post(
                    f"{fallback_node['url']}/chat/completions",
                    content=body,
                    headers={"Content-Type": "application/json"}
                )
                return Response(content=resp.content, status_code=resp.status_code, media_type="application/json")
            return JSONResponse({"error": str(exc)}, status_code=502)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=11436)

ShVaSa Intuition

In a self-hosted developer environment, separating the application interface from the physical inference engine is essential for operational stability. High-concurrency coding agents submit bursts of nested tool queries, file inspections, and AST diff evaluations. If an agent speaks directly to a raw inference engine, any crash or model swap during a training cycle immediately aborts the active editing task.

By inserting a stateless semantic routing proxy:

1. Zero-Downtime Model Swaps: You can reload a 70B model or upgrade vLLM kernels on one card while the proxy routes traffic to a companion node without breaking IDE editor sessions.

2. Deterministic Token Isolation: Requests can be inspected for context size and dispatched to engines configured with appropriate KV-cache memory parameters.

3. Audit and Air-Gap Enforcement: The proxy acts as a verifiable barrier, confirming that neither prompt source code nor proprietary repo context ever traverses outside your local network boundary.

This architecture forms the backbone of a self-sustaining personal lab: unmetered inference compute delivered with enterprise-grade resilience.

Canonical Source: https://blog.shvasa.us/posts/diy-private-ai-agent-gateway-semantic-router