openrouter-reference-architecture

Design production architectures using OpenRouter as the LLM gateway. Use when planning system design, reviewing architecture, or scaling AI applications. Triggers: 'openrouter architecture', 'openrouter system design', 'openrouter at scale', 'llm gateway architecture'.

Zur Installation springen

Repository: jeremylongshore/claude-code-plugins-plus-skills
Letzte Quellaktivität: 17. Juli 2026 um 22:18
Erkannte Sprache von SKILL.md: Englisch
Sterne: 2.604
Forks: 378

Installationsoptionen

Standardmäßig ist der Prompt ausgewählt, der zuerst die Quelle prüft. Sie können zu einem direkten Befehl wechseln oder eine lokale Kopie herunterladen.

Quelldateien prüfen

Lesen Sie SKILL.md und alle von SkillsMP angezeigten Begleitdateien, bevor Sie sich für eine Installation entscheiden.

Zurück zur Suche Zurück zur Installation

name	openrouter-reference-architecture
description	Design production architectures using OpenRouter as the LLM gateway. Use when planning system design, reviewing architecture, or scaling AI applications. Triggers: 'openrouter architecture', 'openrouter system design', 'openrouter at scale', 'llm gateway architecture'.
allowed-tools	Read, Write, Edit, Grep, Bash(python3:*)
version	1.20.0
license	MIT
author	Jeremy Longshore <jeremy@intentsolutions.io>
tags	["saas","openrouter","architecture","system-design","scaling"]
compatibility	Designed for Claude Code, also compatible with Codex and OpenClaw

OpenRouter Reference Architecture

Overview

OpenRouter serves as a unified LLM gateway, abstracting provider complexity. A production architecture wraps it with caching, rate limiting, cost controls, observability, and async processing. This skill provides three reference architectures: simple (single service), standard (microservice), and enterprise (event-driven).

Prerequisites

An OpenRouter API key (sk-or-v1-...) exported as OPENROUTER_API_KEY — see the openrouter-install-auth skill for setup
Python 3.8+ with the OpenAI SDK; FastAPI + Pydantic for Architecture 2's AI service, and a Redis instance (with the redis package) for Architecture 2's cache and Architecture 3's queue/results store
SQLite or Postgres if you implement Architecture 2's budget enforcer
Your scale numbers — team size, requests/day, and latency needs drive the decision in Choosing an Architecture

Instructions

Score your system against the Choosing an Architecture table: team size, requests/day, latency needs, budget-tracking granularity, failure handling, observability.
Start with Architecture 1 (Simple): one shared client (max_retries=3, timeout=30.0) behind the logging complete() wrapper.
When you need task routing, caching, and per-user budgets, move to Architecture 2 (Standard): a FastAPI /v1/complete endpoint with the ROUTING_TABLE, cache-first lookup, budget check, and a fallback chain (models + route: "fallback").
At 100K+ requests/day or mixed sync/async workloads, adopt Architecture 3 (Enterprise): queue (Redis/SQS) → auto-scaling workers running worker_loop() → results store, with OTEL metrics feeding dashboards and alerts.
Whichever tier you land on, route every call through the same OpenRouter client wrapper per Enterprise Considerations — consistent logging, cost tracking, and no budget bypass.

Architecture 1: Simple (Single Service)

┌─────────────┐     ┌──────────────────────────┐     ┌──────────────┐
│  Your App   │────▶│  OpenRouter Client        │────▶│  OpenRouter  │
│             │     │  - Retry (SDK built-in)   │     │  /api/v1     │
│             │◀────│  - Cost tracking          │◀────│              │
│             │     │  - Structured logging     │     └──────────────┘
└─────────────┘     └──────────────────────────┘

import os, logging
from openai  OpenAI

log = logging.getLogger()

client = OpenAI(
    base_url=,
    api_key=os.environ[],
    max_retries=,
    timeout=,
    default_headers={: , : },
)

 ():
    kwargs.setdefault(, )
    response = client.chat.completions.create(
        model=model,
        messages=[{: , : prompt}],
        **kwargs,
    )
    log.info()
     response.choices[].message.content

┌─────────────┐     ┌─────────────────────┐     ┌──────────────┐
│  API Gateway│────▶│  AI Service          │────▶│  OpenRouter  │
│  (auth,     │     │  ┌─────────────┐    │     │  /api/v1     │
│   rate-limit│     │  │ Router      │    │     └──────────────┘
│   logging)  │     │  │ (task→model)│    │
└─────────────┘     │  └─────────────┘    │
                    │  ┌─────────────┐    │
                    │  │ Cache       │◀──▶│── Redis
                    │  │ (TTL-based) │    │
                    │  └─────────────┘    │
                    │  ┌─────────────┐    │
                    │  │ Budget      │◀──▶│── SQLite/Postgres
                    │  │ Enforcer    │    │
                    │  └─────────────┘    │
                    └─────────────────────┘

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel

app = FastAPI()

class CompletionRequest(BaseModel):
    prompt: str
    task_type: str = "general"  # classification, code, analysis, etc.
    max_tokens: int = 1024
    user_id: str = "anonymous"

ROUTING_TABLE = {
    "classification": "openai/gpt-4o-mini",
    "code": "anthropic/claude-3.5-sonnet",
    "analysis": "anthropic/claude-3.5-sonnet",
    "general": "openai/gpt-4o-mini",
    "budget": "meta-llama/llama-3.1-8b-instruct",
}

@app.post("/v1/complete")
async def complete(req: CompletionRequest):
    model = ROUTING_TABLE.get(req.task_type, "openai/gpt-4o-mini")

    # Check cache first (for deterministic requests)
    cached = cache.get(model, req.prompt)
    if cached:
        return {"content": cached, "cached": True}

    # Check budget
    budget.check(req.user_id, model, estimate_tokens(req.prompt), req.max_tokens)

    # Call OpenRouter
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": req.prompt}],
        max_tokens=req.max_tokens,
        extra_body={
            "models": [model, "openai/gpt-4o-mini"],  # Fallback
            "route": "fallback",
        },
    )

    # Record cost and cache
    budget.record(req.user_id, response.id)
    cache.set(model, req.prompt, response.choices[0].message.content)

    return {
        "content": response.choices[0].message.content,
        "model": response.model,
        "tokens": response.usage.prompt_tokens + response.usage.completion_tokens,
    }

┌──────────┐    ┌───────────┐    ┌──────────────┐    ┌──────────────┐
│  API     │───▶│  Queue    │───▶│  Workers     │───▶│  OpenRouter  │
│  Gateway │    │  (Redis/  │    │  (auto-scale) │    │  /api/v1     │
└──────────┘    │  SQS)     │    │  ┌──────────┐│    └──────────────┘
                └───────────┘    │  │ Router   ││
                     │           │  │ Cache    ││
                     ▼           │  │ Budget   ││
                ┌───────────┐    │  │ Audit    ││
                │  Results  │◀───│  └──────────┘│
                │  Store    │    └──────────────┘
                └───────────┘
                     │
                ┌───────────┐    ┌──────────────┐
                │  Metrics  │───▶│  Dashboard   │
                │  (OTEL)   │    │  Alerts      │
                └───────────┘    └──────────────┘

# Worker that processes queued AI requests
import json, redis

r = redis.Redis()

def worker_loop():
    """Process AI requests from the queue."""
    while True:
        _, raw = r.brpop("ai:requests")
        request = json.loads(raw)

        try:
            response = client.chat.completions.create(
                model=request["model"],
                messages=request["messages"],
                max_tokens=request.get("max_tokens", 1024),
                extra_body={
                    "models": [request["model"], "openai/gpt-4o-mini"],
                    "route": "fallback",
                },
            )
            result = {
                "id": request["id"],
                "content": response.choices[0].message.content,
                "model": response.model,
                "status": "complete",
            }
        except Exception as e:
            result = {"id": request["id"], "error": str(e), "status": "failed"}

        r.lpush(f"ai:results:{request['id']}", json.dumps(result))
        r.expire(f"ai:results:{request['id']}", 3600)

Factor	Simple	Standard	Enterprise
Team size	1-3	3-10	10+
Requests/day	<1K	1K-100K	100K+
Latency needs	Tolerant	Low	Mixed (sync+async)
Budget tracking	Basic	Per-user	Per-user + department
Failure handling	SDK retries	Fallback chain	Queue + retry + DLQ
Observability	Logging	Metrics + logging	Full OTEL tracing

# POST /v1/complete  (Architecture 2)
req = CompletionRequest(prompt="Refactor this function...", task_type="code", user_id="u42")
# ROUTING_TABLE maps "code" -> anthropic/claude-3.5-sonnet, with openai/gpt-4o-mini as fallback
# -> {"content": "...", "model": "anthropic/claude-3.5-sonnet", "tokens": 348}

Error	Cause	Fix
Single point of failure	No redundancy in AI service	Deploy 2+ instances behind load balancer
Queue backlog	Worker throughput < incoming rate	Auto-scale workers; implement backpressure
Cache stampede	Many requests for same uncached key	Use cache locking or singleflight pattern
Budget bypass	Direct calls skipping middleware	All calls must go through the AI service