System Architecture Blueprint

RAG-009: Hybrid Cloud-Edge RAG System — Engineering Blueprint

This architectural blueprint outlines the system topology, mathematical routing heuristics, edge-to-cloud security boundaries, and local-first execution codes for RAG-009 (Hybrid Cloud-Edge RAG System).

1. System Specifications & Performance SLA

Edge Ingestion & Retrieval SLA:

Local embedding generation: $< 15\text{ms}$ utilizing WebGPU or ONNX Runtime.

Local vector similarity search: $< 5\text{ms}$ on SQLite/local-RAM indexes.

Local synthesis: $< 800\text{ms}$ using quantized SLMs via Ollama.

Network & Resource Limits:

Maximum client RAM footprint allocation: $< 1.5\text{ GB}$.

Bandwidth optimization target: $> 75\%$ reduction in network egress/payload transmissions.

Privacy Target: 100% containment of classified, regulatory, or sensitive raw records within the physical boundaries of the local edge environment.

1. Distributed Hybrid Ingestion & Routing Topology

The architecture divides execution tasks. Rather than operating as a pass-through API client, the edge browser or local system acts as the primary gatekeeper, hosting local models and deciding whether to dispatch tasks to the cloud or execute them on-device.

                     [ User Query / Input Stream ]
                                  │
                                  ▼
                    [ Edge Routing Coordinator ]
                     • Scans for sensitive entities
                     • Determines processing weight
                                  │
       ┌──────────────────────────┴──────────────────────────┐
       ▼ (R(Q) >= Theta)                                     ▼ (R(Q) < Theta)

[ Local Edge Workspace ] [ Secure Cloud Route ] • Local Embedding (ONNX) • Encrypted SSL Channel • Local SQLite Vector Space • Cloud Vector DB (Pinecone/Qdrant) • Local SLM (Ollama) • Heavy LLM Synthesis (SaaS)

2.1 Edge & Cloud Task Division

Client/Edge Tier: Processes local private collections, handles user-specific context cached on the device, generates instant semantic embeddings (Transformers.js), and applies prompt filtering rules.

Cloud Infrastructure Tier: Processes heavy enterprise document repositories, maintains large-scale vector indexes (Qdrant/Pinecone), and executes advanced reasoning tasks across global data scopes.

1. Mathematical Workload Routing & Privacy Classification

To decide whether a query is processed locally or dispatched to cloud architectures, the edge engine evaluates a Workload Routing Heuristic based on privacy score ($P(Q)$) and query complexity ($C(Q)$).

Let $Q$ represent the input query.

3.1 Privacy Metric Formulation

The edge coordinator runs a lightweight, regex-based entity extraction model alongside a local named-entity recognition (NER) model to detect sensitive data classes (e.g., medical labels, credentials, bank routing targets). We formulate the privacy score $P(Q) \in [0.0, 1.0]$ as:

$$P(Q) = 1.0 - \prod_{e \in E(Q)} (1.0 - \omega_e)$$

Where:

$E(Q)$ is the set of detected entity classes in query $Q$.

$\omega_e \in [0.0, 1.0]$ is the sensitivity weight associated with class $e$ (e.g., $\omega_{\text{SSN}} = 1.0$, $\omega_{\text{IP_Address}} = 0.8$, $\omega_{\text{Casual_Topic}} = 0.0$).

3.2 Complexity Metric Formulation

The query complexity score $C(Q) \in [0.0, 1.0]$ evaluates the difficulty of answering the query based on length, comparative phrasing, and dependency terms:

$$C(Q) = \tanh\left( \gamma_1 \cdot \text{len}(Q) + \gamma_2 \cdot N_{\text{comparatives}} + \gamma_3 \cdot I_{\text{global_search}} \right)$$

Where:

$\text{len}(Q)$ is the word length of the query.

$N_{\text{comparatives}}$ is the count of analytical comparisons or logical operators (e.g., "compare", "versus", "across").

$I_{\text{global_search}} \in {0, 1}$ indicates the explicit request for global enterprise directories.

$\gamma_1, \gamma_2, \gamma_3$ are scaling constants (typically $0.05, 0.35, 0.8$).

3.3 Dynamic Routing Score

The final routing metric $R(Q)$ is a weighted linear combination:

$$R(Q) = w_p \cdot P(Q) - w_c \cdot C(Q)$$

Where $w_p, w_c > 0$ are design balance factors. The edge coordinator routes the query using the threshold $\theta$:

$$\text{Route} = \begin{cases} \text{EDGE (Local)}, & \text{if } R(Q) \ge \theta \ \text{CLOUD (Server)}, & \text{if } R(Q) < \theta \end{cases}$$

High $P(Q)$ pulls $R(Q)$ upward, keeping sensitive inputs on the edge regardless of complexity.

High $C(Q)$ pulls $R(Q)$ downward, routing complex, non-sensitive queries to the cloud.

1. Edge-Cloud Workload Dispatcher Implementation

This self-contained Python module simulates the client-side edge environment. It runs local embeddings using a lightweight model, parses query parameters to evaluate routing factors, executes local search on SQLite, and manages failovers to cloud vector databases.

import os import re import math import numpy as np from typing import Dict, Any, List, Tuple

Mocking local ONNX/Transformers.js embedding behaviour

class LocalONNXEmbeddingEngine: def init(self): # In a real environment, this loads a local quantized ONNX model self.dimension = 384 # MiniLM dimension size

def embed_query(self, text: str) -> List[float]:
    """Generates a mock vector representation locally on the CPU."""
    # Simple deterministic hashing representation for demonstration
    np.random.seed(hash(text) % (2**32 - 1))
    vector = np.random.randn(self.dimension)
    norm = np.linalg.norm(vector)
    if norm == 0:
        return vector.tolist()
    return (vector / norm).tolist()

class HybridCloudEdgeDispatcher: def init(self, cloud_db_endpoint: str, threshold: float = 0.30): self.local_embed_engine = LocalONNXEmbeddingEngine() self.threshold = threshold self.cloud_endpoint = cloud_db_endpoint

    # Simple simulated local document storage
    self.local_db = [
        {"id": "loc_1", "text": "Project Falcon uses local-first encryption standards.", "vector": []},
        {"id": "loc_2", "text": "This client device cache stores current session variables.", "vector": []}
    ]
    self._initialize_local_vectors()

def _initialize_local_vectors(self):
    for item in self.local_db:
        item["vector"] = self.local_embed_engine.embed_query(item["text"])

def _cosine_similarity(self, v1: List[float], v2: List[float]) -> float:
    a = np.array(v1)
    b = np.array(v2)
    dot = np.dot(a, b)
    norm_a = np.linalg.norm(a)
    norm_b = np.linalg.norm(b)
    if norm_a == 0 or norm_b == 0:
        return 0.0
    return float(dot / (norm_a * norm_b))

def evaluate_privacy_score(self, query: str) -> float:
    """Calculates P(Q) based on local regex entity mapping."""
    entity_weights = {
        r"\b\d{3}-\d{2}-\d{4}\b": 1.0,         # SSN
        r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b": 0.9, # Email
        r"\b\d{4}-\d{4}-\d{4}-\d{4}\b": 1.0,   # Credit Card
        r"0x[0-9a-fA-F]{40}": 0.85,             # Ethereum Wallet
        r"(?i)\b(password|secret|key|api_key)\b": 0.95
    }

    sensitivity_factors = []
    for pattern, weight in entity_weights.items():
        if re.search(pattern, query):
            sensitivity_factors.append(weight)

    if not sensitivity_factors:
        return 0.0

    # P(Q) = 1 - Prod(1 - w)
    product = 1.0
    for w in sensitivity_factors:
        product *= (1.0 - w)
    return 1.0 - product

def evaluate_complexity_score(self, query: str) -> float:
    """Calculates C(Q) based on syntax weight metrics."""
    word_count = len(query.strip().split())

    # Detect comparative indicators
    comparative_indicators = ["compare", "versus", "vs", "difference", "benchmark", "historical"]
    comparative_count = sum(1 for word in query.lower().split() if word in comparative_indicators)

    # Explicit global search tags
    global_search = 1 if "enterprise" in query.lower() or "global" in query.lower() else 0

    # Apply C(Q) = tanh(gamma_1*len + gamma_2*comp + gamma_3*global)
    g1, g2, g3 = 0.05, 0.35, 0.8
    score_input = (g1 * word_count) + (g2 * comparative_count) + (g3 * global_search)
    return float(math.tanh(score_input))

def dispatch_and_route(self, query: str) -> Tuple[str, Dict[str, Any]]:
    """
    Determines the optimal routing destination based on Privacy and Complexity.
    Executes local/cloud actions accordingly.
    """
    p_score = self.evaluate_privacy_score(query)
    c_score = self.evaluate_complexity_score(query)

    # R(Q) = w_p * P(Q) - w_c * C(Q)
    w_p, w_c = 0.8, 0.4
    routing_score = (w_p * p_score) - (w_c * c_score)

    meta = {
        "privacy_score": p_score,
        "complexity_score": c_score,
        "routing_score": routing_score
    }

    if routing_score >= self.threshold:
        # Execute local execution loop
        local_results = self._execute_local_retrieval(query)
        return "EDGE", {**meta, "results": local_results, "summary": "Query processed entirely on edge device."}
    else:
        # Dispatch to cloud database endpoint
        cloud_results = self._execute_cloud_dispatch(query)
        return "CLOUD", {**meta, "results": cloud_results, "summary": f"Query offloaded to cloud cluster at: {self.cloud_endpoint}"}

def _execute_local_retrieval(self, query: str) -> List[Dict[str, Any]]:
    """Performs localized semantic search over the client-side vector index."""
    query_vector = self.local_embed_engine.embed_query(query)
    scored = []
    for item in self.local_db:
        score = self._cosine_similarity(query_vector, item["vector"])
        scored.append({"id": item["id"], "text": item["text"], "score": score})
    return sorted(scored, key=lambda x: x["score"], reverse=True)

def _execute_cloud_dispatch(self, query: str) -> List[Dict[str, Any]]:
    """Simulates secure, encrypted API call to Cloud Vector DB."""
    # Returns mocked cloud-only context records
    return [
        {"id": "cloud_992", "text": "Global Enterprise Balance Sheet for Q4.", "score": 0.89},
        {"id": "cloud_993", "text": "Historical comparative trend datasets across 15 regions.", "score": 0.84}
    ]

1. Verification & Validation Protocols

5.1 Workload Routing Test Suite

Assert that queries containing highly sensitive parameters (e.g., secret API keys, credit cards) remain isolated on-device, while complex structural analytical queries are successfully offloaded.

import pytest

def test_sensitive_query_remains_local(): dispatcher = HybridCloudEdgeDispatcher(cloud_db_endpoint="https://api.qdrant-cloud.com")

# Query contains cleartext credential patterns (high privacy score)
sensitive_query = "What is the access log for API_KEY 0x99A82B3C?"
route, details = dispatcher.dispatch_and_route(sensitive_query)

assert route == "EDGE"
assert details["privacy_score"] > 0.80
assert "results" in details
# The output must derive strictly from local data pools
assert any("loc_" in r["id"] for r in details["results"])

def test_complex_query_offloaded_to_cloud(): dispatcher = HybridCloudEdgeDispatcher(cloud_db_endpoint="https://api.qdrant-cloud.com")

# Query is highly complex and comparative, with zero sensitive tokens
complex_query = "Compare our global standard operating incomes versus regional compliance benchmarks across the last 10 historical quarters."
route, details = dispatcher.dispatch_and_route(complex_query)

assert route == "CLOUD"
assert details["complexity_score"] > 0.50
assert any("cloud_" in r["id"] for r in details["results"])

5.2 Local Air-Gapped Verification

Verify that if all external sockets are closed, the edge routing engine catches connection timeout exceptions gracefully, locks downstream offloading, and executes query recovery loops locally.

def test_network_failure_edge_fallback(): # Force dispatcher cloud endpoint to be unreachable dispatcher = HybridCloudEdgeDispatcher(cloud_db_endpoint="https://unreachable-endpoint-fail.com")

# This query would normally route to the cloud (low privacy, high complexity)
query = "Compare global historical datasets."

try:
    # Simulate network timeout block
    route, details = dispatcher.dispatch_and_route(query)
    if route == "CLOUD":
        # Simulate network fault
        raise ConnectionError("Timeout attempting to reach cloud gateway.")
except ConnectionError:
    # Graceful fallback: run local extraction
    local_results = dispatcher._execute_local_retrieval(query)
    route = "EDGE_FALLBACK"
    details = {"results": local_results, "summary": "Cloud unreachable. Gracefully fallback to local storage."}

assert route == "EDGE_FALLBACK"
assert len(details["results"]) > 0
assert any("loc_" in r["id"] for r in details["results"])

Required Technical Skills

Advanced Engineering Skills Inventory: Hybrid Cloud-Edge RAG (RAG-009)

This matrix details the specialized distributed systems engineering, browser-native model execution, cryptographic vaulting, and dynamic mathematical routing capabilities required to design, implement, and maintain secure, edge-first hybrid RAG architectures.

1. Client-Side Model Serving & WebGPU Acceleration

Traditional RAG architectures rely entirely on server-side GPU instances. Running semantic extraction on-device requires deep familiarity with compiling, quantizing, and executing lightweight language models inside highly constrained client environments.

               [ User Query on Client Device ]
                             │
               ┌─────────────┴─────────────┐
               ▼                           ▼
     [ WebGPU Accelerated ]       [ WebAssembly (Wasm) ]
     • Direct GPU pipeline binding • SIMD Multi-threading fallbacks
     • Transformers.js Execution   • Heap Memory capped <1.5 GB

Quantized Model Compilation:

Quantizing open-source embedding models (such as nomic-embed-text-v1.5 or all-MiniLM-L6-v2) and Small Language Models (such as llama3.2:3b or gemma2:2b) down to int4 or int8 precision.

Compiling model layers into highly optimized ONNX runtimes compatible with browser-native JavaScript environments.

Browser Accelerator Bindings:

Leveraging Transformers.js to run in-browser inference loops.

Binding model execution pipelines directly to the client's GPU using WebGPU API endpoints.

Implementing graceful execution fallbacks to WebAssembly (Wasm) with Single Instruction Multiple Data (SIMD) multi-threading enabled when WebGPU is disabled.

Enforcing strict heap memory overhead boundaries ($< 1.5\text{ GB}$) to prevent browser tab crashes.

1. Local Sandboxed Storage & Web Cryptography

Executing RAG on local hardware requires managing persistent vector and text indices inside sandbox containers, using robust encryption keys to guard client records.

Client-Side Relational Databases:

Compiling and deploying SQLite to WebAssembly (Wasm) to run structured SQL engines inside browser threads.

Structuring schemas in IndexedDB or SQLite Wasm to manage local transaction arrays, session vectors, and persistent document fragments.

Client-Native Cryptography:

Utilizing the native browser WebCrypto container to encrypt and decrypt client databases using the AES-GCM-256 standard.

Mapping decryption keys securely to client-native key chains (e.g., macOS Keychain, Windows Credential Manager) via secure Electron or native browser sandboxes.

1. Dynamic Workload Routing Heuristics

To balance computational tasks across client-server lines, you must write edge classifiers that can evaluate query characteristics instantly before dispatching transactions.

                       [ Incoming Query ]
                               │
          ┌────────────────────┴────────────────────┐
          ▼                                         ▼

[ Privacy Classifier P(Q) ] [ Complexity Classifier C(Q) ] • Scans PII / Credentials • Word counts & Comparatives • Multi-Tenant Hashing Salts • Explicit Global Target Identifiers │ │ └────────────────────┬────────────────────┘ ▼ [ Routing Score: R(Q) vs. Theta ]

Privacy Classification Model Engineering:

Constructing lightweight, local, regex-based named entity recognition (NER) checkers to scan queries for sensitive patterns (such as Social Security Numbers, credentials, API keys, wallet addresses, and email logs).

Formulating dynamic joint privacy scores $P(Q)$ utilizing independent risk weight parameters:

$$P(Q) = 1.0 - \prod_{e \in E(Q)} (1.0 - \omega_e)$$

Complexity Estimation:

Writing syntactic classifiers to compute query complexity $C(Q)$ based on word lengths, comparative syntax structures (e.g., "versus", "trend"), and global enterprise target identifiers.

Synthesizing parameters mathematically to calculate dynamic routing thresholds:

$$R(Q) = w_p \cdot P(Q) - w_c \cdot C(Q)$$

Routing queries dynamically to EDGE or CLOUD targets based on boundary factors ($\theta$).

1. Hybrid Synchronization & Cloud Multi-Tenancy

When local edge datasets need to interface with massive enterprise indices, you must secure the boundary lanes with logical or physical separation controls.

Dynamic Scope JWT Extraction:

Wrapping offloaded queries in secured HTTPS channels carrying cryptographically signed JSON Web Tokens (JWT).

Writing API gateway filters to decrypt JWT properties, mapping incoming requests to corresponding isolated tenant collections inside cloud vector engines (such as Qdrant or Pinecone).

Enterprise Cloud Ingress Protections:

Placing centralized database clusters within isolated Virtual Private Clouds (VPC).

Restricting network access limits using ingress rules mapped strictly to authenticated gateway endpoints.

1. Offline Resiliency & Distributed Cost Optimization

Edge-first systems are built to be highly resilient under extreme network outages while structurally shielding the host SaaS from high cloud hosting expenses.

[ Local Network Disconnect ] ──► [ Connection Timeout Interceptor ] ──► [ Auto-EDGE Lock ] • Bypass threshold checks • Run local SQLite / Ollama

Graceful Network Degradation:

Writing non-blocking connection timeout interceptors to catch offline or network-failure exceptions immediately ($< 10\text{ms}$).

Implementing Auto-EDGE Lock protocols to bypass routing calculations during network failure, running queries strictly on local SQLite/Ollama databases to avoid application crashes.

Financial Token Optimization:

Engineering comparative margin optimization equations to calculate cost benefits of client-side processing:

$$\text{Savings} = N_{\text{queries}} \cdot (1.0 - \text{Offload}\%) \cdot \text{Inference Cost}$$

Designing sliding-scale query dispatch rules to balance cost margins against user latency targets dynamically.

🚀 Quick Installation & Setup Guide

To run this project locally, follow these simple steps designed for both developers and users with no prior programming experience.

1. Download Python: Make sure you have Python 3.10+ installed on your computer. Download it from the official Python Website. During installation, make sure to check the box "Add Python to PATH".
2. Extract / Setup Code: Open the terminal (Command Prompt on Windows) and navigate to the project directory, or download/extract the project files.
3. Install Dependencies (pip): In Command Prompt, run the following command to install all necessary packages automatically:

   pip install -r requirements.txt

4. Setup API Keys / Environment: Open the .env file in a text editor (like Notepad) and paste your API keys:

   GEMINI_API_KEY=your_gemini_api_key_here

5. Launch the Application: Double-click the run_project.bat or run_pito.bat file, or run the following command:

   python app.py

   (or app_gui.py)