📄 Local LLM Router Client Documentation (local_llm_client.py)

Overview and Purpose

The local_llm_client.py module implements a sophisticated Local Large Language Model (LLM) Router. Its primary function is to provide a transparent, high-performance layer for executing sub-tasks that require LLM capabilities (e.g., JSON parsing, data extraction, formatting).

This router prioritizes utilizing local GPU resources connected to specified hardware nodes (e.g., AMD Ryzen 7 5700X) to achieve zero API cost and minimal latency overhead associated with external network calls. It acts as a smart intermediary that intelligently routes requests based on availability and reliability, ensuring service continuity even when local infrastructure fails.

Key Architectural Goals:

1. Cost Optimization: Maximize the use of self-hosted, locally deployed models.
2. Performance: Minimize latency by keeping processing within the local network/machine boundary.
3. Resilience (Failover): Guarantee service availability by seamlessly transitioning to cloud APIs when local resources are unavailable or time out.

---

⚙️ Architecture and Mechanics

The client operates using a layered, prioritized execution model:

1. The Routing Mechanism (Local Priority)

When an API call is initiated, the router first attempts to connect to the configured list of local GPU endpoints.

• Target Nodes: Local machines/containers running optimized LLM inference engines (e.g., vLLM, llama.cpp).
• Routing Logic: The client sends the request payload (prompt, task type) to the designated local endpoint(s).
• Task Specialization: The router supports routing based on the type of sub-task required:
  • json_parser: For structured output extraction.
  • data_formatter: For transforming raw text into specific formats (e.g., Markdown, XML).
  • general_llm: For general conversational or reasoning tasks.

2. The Fallback Mechanism (Cloud Resilience)

The core strength of the router is its robust fallback logic. If a local attempt fails due to network issues, resource exhaustion, or exceeding a predefined timeout threshold, the client automatically and transparently switches the request to the configured cloud API endpoint.

Failure Condition	Local Action Taken	Fallback Target	Cloud Provider Used
Timeout (Local Node unresponsive)	Retry attempt on next node/endpoint.	Gemini 3.7 Flash via Google AI SDK.	Google Cloud API
Offline Status (Node unreachable)	Skip local endpoint and proceed to fallback.	Gemini 3.7 Flash via Google AI SDK.	Google Cloud API
Rate Limit Exceeded (Local Node overload)	Fallback immediately, logging the rate limit error.	Gemini 3.7 Flash via Google AI SDK.	Google Cloud API

---

💻 Code Examples and Implementation Details

A. Initialization and Configuration

The client requires configuration for both local endpoints and cloud credentials.

from local_llm_client import LLMRouter
import os

# --- Configuration ---
LOCAL_NODES = [
    "http://192.168.1.10:8000",  # Primary GPU Node (r7-5700x)
    "http://192.168.1.11:8000"   # Secondary GPU Node
]

# Ensure GEMINI_API_KEY is set in environment variables
CLOUD_FALLBACK = {
    'model': 'gemini-3.7-flash',
    'api_key': os.environ.get("GEMINI_API_KEY") 
}

try:
    router = LLMRouter(
        local_endpoints=LOCAL_NODES,
        fallback_config=CLOUD_FALLBACK,
        timeout_seconds=5  # Max time to wait for local response
    )
except Exception as e:
    print(f"Failed to initialize router: {e}")
    exit()

B. Usage Example: JSON Parsing (Task Routing)

This example demonstrates routing a request that requires structured output, allowing the client to automatically handle failure and switch to Gemini if the local node times out.

def extract_data(prompt: str):
    """Routes a prompt for JSON extraction."""
    print("--- Attempting Local LLM Routing (JSON Parser) ---")
    try:
        # The router handles the internal logic of checking nodes and falling back
        result = router.route_task(
            task_type="json_parser", 
            prompt=prompt, 
            max_retries=3 # How many times to try local nodes before failing over
        )
        return result

    except Exception as e:
        print(f"\n[CRITICAL FAILURE] All attempts failed. Error: {e}")
        # At this point, the router has exhausted all options (local and cloud).
        return None


# Example Prompt for extraction
extraction_prompt = "Analyze the following text and extract the name, date, and location in JSON format:\n'The meeting was held on October 25th, 2024, at the Grand Hyatt Hotel in New York.'"

extracted_json = extract_data(extraction_prompt)

if extracted_json:
    print("\n✅ Success! Received structured data.")
    # Assuming the result is a dictionary or JSON string
    import json
    print(json.dumps(extracted_json, indent=2))
else:
    print("\n❌ Failed to retrieve data from any source.")

C. Task Type Mapping (Internal Logic)

The route_task method internally maps the requested task type to the appropriate local endpoint or internal logic handler:

task_type	Description	Local Endpoint Priority	Fallback Behavior
"json_parser"	Structured data extraction (e.g., Pydantic schema adherence).	Dedicated JSON Parser Node	Gemini 3.7 Flash (with structured output mode)
"data_formatter"	Transforming raw text into specific formats (Markdown, YAML).	General LLM Endpoint	Gemini 3.7 Flash
`"general_ll