LangGraph + HolySheep Kosten-aware LLM Router: Dynamische Multi-Model-Orchestrierung in der Produktion

Wer ein produktives LLM-System betreibt, kennt das Spannungsfeld: Latenz, Kosten und Antwortqualität stehen in einem harten Dreiecksverhältnis. Ein einziges Modell für alle Anfragen zu verwenden ist bequem – aber teuer. In diesem Artikel zeige ich, wie wir mit LangGraph und der HolySheep-Aggregator-API (Jetzt registrieren) einen kostenbewussten Router gebaut haben, der pro Anfrage entscheidet, welches Modell den Job übernimmt – basierend auf Prompt-Komplexität, Budget und Latenz-SLA.

Alle Benchmarks und Code-Snippets stammen aus einem realen Kundensystem, das wir im Q1 2026 in Produktion gebracht haben (≈ 1,4 Mio. Requests/Monat, 12 Mitarbeiter, deutschsprachige Enterprise-Suche).

1. Architektur-Überblick

Der Router ist ein stateful graph in LangGraph mit drei Kernknoten: classify, route und execute. Die zentrale Idee: Wir klassifizieren jeden eingehenden Prompt in eine von vier Schwierigkeitsstufen und routen entsprechend.

# router_graph.py — Top-Level-Definition
from typing import Literal, TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
import os, time

HolySheep ist OpenAI-kompatibel — keine Vendor-Lock-in
HS_BASE = "https://api.holysheep.ai/v1"
HS_KEY  = os.environ["HOLYSHEEP_API_KEY"]

class RouterState(TypedDict):
    prompt: str
    tier: Literal["trivial", "simple", "medium", "hard"]
    budget_cents: float
    chosen_model: str
    response: str
    latency_ms: int
    cost_cents: float

def classify(state: RouterState) -> RouterState:
    """Heuristische Klassifikation — kein LLM-Call nötig."""
    p = state["prompt"]
    n_tokens = len(p.split())
    has_code = "```" in p or "def " in p or "class " in p
    if n_tokens < 12 and not has_code:
        state["tier"] = "trivial"
    elif n_tokens < 80:
        state["tier"] = "simple"
    elif has_code or n_tokens > 250:
        state["tier"] = "hard"
    else:
        state["tier"] = "medium"
    return state

Tier → Modell-Mapping (siehe Tabelle unten)
TIER_MAP = {
    "trivial": "deepseek-v3.2",     # $0.42/MTok
    "simple":  "gemini-2.5-flash",  # $2.50/MTok
    "medium":  "gpt-4.1-mini",
    "hard":    "claude-sonnet-4.5", # $15/MTok
}

def route(state: RouterState) -> RouterState:
    state["chosen_model"] = TIER_MAP[state["tier"]]
    return state

def execute(state: RouterState) -> RouterState:
    llm = ChatOpenAI(
        model=state["chosen_model"],
        base_url=HS_BASE,
        api_key=HS_KEY,
        timeout=8,
    )
    t0 = time.perf_counter()
    resp = llm.invoke(state["prompt"])
    dt = (time.perf_counter() - t0) * 1000
    state["response"] = resp.content
    state["latency_ms"] = int(dt)
    # Kostenschätzung — Output-Preis × Tokens / 1e6 × 100 = Cents
    out_tokens = resp.response_metadata.get("token_usage", {}).get("completion_tokens", 0)
    PRICE = {"deepseek-v3.2": 0.042, "gemini-2.5-flash": 0.25,
             "gpt-4.1-mini": 1.6, "claude-sonnet-4.5": 1.5}
    state["cost_cents"] = (PRICE[state["chosen_model"]] / 100) * out_tokens
    return state

g = StateGraph(RouterState)
g.add_node("classify", classify)
g.add_node("route", route)
g.add_node("execute", execute)
g.add_edge("classify", "route")
g.add_edge("route", "execute")
g.add_edge("execute", END)
g.set_entry_point("classify")
app = g.compile()

# budget_guard.py — adaptive Downgrades
BUDGET_CEILINGS_CENTS = {
    "trivial": 0.05, "simple": 0.20,
    "medium":  0.80, "hard":   5.00,
}

DOWNGRADE_CHAIN = ["hard", "medium", "simple", "trivial"]

def enforce_budget(state: RouterState) -> RouterState:
    tier = state["tier"]
    while (state["cost_cents"] > BUDGET_CEILINGS_CENTS[tier]
           and tier in DOWNGRADE_CHAIN[1:]):
        tier = DOWNGRADE_CHAIN[DOWNGRADE_CHAIN.index(tier) - 1]
        state["tier"] = tier
        state["chosen_model"] = TIER_MAP[tier]
    return state

Im Graph zwischen route und execute einhängen:
g.add_node("budget", enforce_budget)
g.add_edge("route", "budget")
g.add_edge("budget", "execute")

# async_router.py — concurrency + retry
import asyncio, random
from tenacity import retry, stop_after_attempt, wait_exponential_jitter

sem = asyncio.Semaphore(40)   # max 40 parallele HolySheep-Calls

@retry(stop=stop_after_attempt(3),
       wait=wait_exponential_jitter(initial=0.4, max=4))
async def execute_async(state):
    async with sem:
        llm = ChatOpenAI(model=state["chosen_model"],
                         base_url=HS_BASE, api_key=HS_KEY)
        return await llm.ainvoke(state["prompt"])

async def stream(prompts):
    tasks = [execute_async({"prompt": p, "tier": "simple",
                            "chosen_model": "gemini-2.5-flash",
                            "budget_cents": 0.2,
                            "response": "", "latency_ms": 0,
                            "cost_cents": 0.0}) for p in prompts]
    return await asyncio.gather(*tasks, return_exceptions=True)

import os
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"   # nicht openai.com!
os.environ["OPENAI_API_KEY"]  = os.environ["HOLYSHEEP_API_KEY"]

from tenacity import retry, wait_exponential_jitter, stop_after_attempt

@retry(wait=wait_exponential_jitter(initial=0.5, max=8),
       stop=stop_after_attempt(4))
def safe_call(llm, prompt):
    return llm.invoke(prompt)

def cap_output(state):
    """Hartes Output-Limit vor dem Call."""
    MAX = {"trivial": 256, "simple": 512,
           "medium": 1500, "hard": 4000}
    state["max_tokens"] = MAX[state["tier"]]
    return state