open-sse Architecture (Deutsch)

Warum ein separates Workspace-Paket?

open-sse/ ist aus mehreren Gründen ein eigenständiger Workspace im OmniRoute-Monorepo:

1. Wiederverwendbarkeit — open-sse wird als @omniroute/open-sse auf npm veröffentlicht, sodass andere Projekte es unabhängig verwenden können
2. Klare Abgrenzung — die Streaming-Engine ist von der OmniRoute-spezifischen UI-/DB-Schicht entkoppelt
3. Performance — die Engine hat keine Next.js-Abhängigkeiten, wodurch schnellere Kaltstarts in CLI-/Serverless-Kontexten möglich sind
4. Versionierung — open-sse kann nach einem eigenen Zeitplan veröffentlicht werden

package.json

"workspaces": ["open-sse"]

---

Struktur auf oberster Ebene

open-sse/
├── index.ts              # Öffentlicher Einstiegspunkt
├── types.d.ts            # Öffentliche Typ-Exporte
├── package.json          # @omniroute/open-sse
├── config/               # Provider-Konfigurationen, Konstanten, Registrierungen
├── executors/            # HTTP-Executors pro Provider (67 + base.ts/index.ts)
├── handlers/             # Anfrage-Handler (chatCore, responses usw.)
├── lib/                  # Interne Hilfsfunktionen
├── mcp-server/           # Model-Context-Protocol-Server
├── services/             # Etwa 298 Service-Module
├── transformer/          # Format-Transformer für die Responses API
├── translator/           # Formatübersetzung (OpenAI ↔ Claude ↔ Gemini)
└── utils/                # Gemeinsam genutzte Hilfsfunktionen (Logging, Fehler, Streams usw.)

Anzahl der Module

| Verzeichnis | Dateien | Zweck | | executors/ | 167 | HTTP-Executors pro Provider (vereinheitlicht über die DefaultExecutor-Factory) | | handlers/ | 157 | Anfrage-Einstiegspunkte (chatCore, responses, embeddings) | | services/ | ~536 | Routing, Caching, Ratenbegrenzung, Aktualisierung usw. | | translator/ | 56 | Formatkonvertierung (OpenAI ↔ Claude ↔ Gemini) | | mcp-server/ | 44 | MCP-Tools und -Transporte | | utils/ | ~108 | Übergreifende Hilfsfunktionen (Logging, Fehler, Streams) | | config/ | ~339 | Provider-Konfigurationen, Konstanten, Registrierungen |

---

Die Anfrage-Pipeline

Jede LLM-Anfrage durchläuft eine 5-stufige Pipeline:

                              ┌──────────────┐
   HTTP-Anfrage                │  1. ROUTING  │   Combo-Auflösung, Modellauswahl
   (Next.js-Route)             └──────┬───────┘
                                     │
                                     ▼
                              ┌──────────────┐
                              │2. ÜBERSETZUNG│   Formatkonvertierung (OpenAI ↔ Claude ↔ Gemini)
                              └──────┬───────┘
                                     │
                                     ▼
                              ┌──────────────┐
                              │3. AUSFÜHRUNG │   Provider-Executor, HTTP, Wiederholung, Schutzschalter
                              └──────┬───────┘
                                     │
                                     ▼
                              ┌──────────────┐
                              │ 4. STREAMING │   SSE-Transformation, Gegendruck
                              └──────┬───────┘
                                     │
                                     ▼
                              ┌──────────────┐
                              │5. ERFASSUNG  │   Nutzungserfassung, Aufrufprotokoll, Fehlerklassifizierung
                              └──────┬───────┘
                                     │
                                     ▼
                              HTTP-Antwort (SSE oder JSON)

Stufe 1: Routing (services/combo.ts)

Einstiegspunkt: handleComboChat() in services/combo.ts

Löst die Anfrage in ein konkretes Tupel aus (provider, model, account, credentials) auf:

• Combo anhand der ID suchen (oder eine virtuelle Combo für auto/*-Modelle erstellen)
• Routing-Strategie anwenden (Priorität, gewichtet, Round-Robin usw.)
• Fehlerhafte Provider herausfiltern (Schutzschalter)
• Das nächste geeignete Ziel auswählen

Für auto/*-Modelle führt diese Stufe außerdem Folgendes aus:

• Ausführung des 16-Faktoren-Bewertungsalgorithmus (services/autoCombo/)
• Auswahl eines provider+model-Paars basierend auf Zustand, Kosten, Latenz usw.

Stufe 2: Übersetzung (translator/)

Wenn sich das Quellformat (z. B. OpenAI) vom Zielformat (z. B. Claude) unterscheidet, wird die Anfrage übersetzt:

• System-Prompt → Systemnachricht
• Tool-Definitionen → providerspezifisches Tool-Format
• Reasoning-/Thinking-Parameter → providerspezifische Entsprechungen
• Normalisierung der Nachrichtenrollen (developer → system für andere Provider als OpenAI)

translator/index.ts stellt Folgendes bereit:

translateRequest(body, sourceFormat, targetFormat): TranslatedRequest
needsTranslation(source, target): boolean

Stufe 3: Ausführung (executors/)

Einstiegspunkt: getExecutor(providerId).execute(request, options)

Alle Provider verwenden DefaultExecutor (executors/default.ts) über den Fallback der getExecutor()-Factory. Der Executor:

• Erstellt die Upstream-URL (buildUrl())
• Fügt providerspezifische Header hinzu (buildHeaders())
• Transformiert den Anfrage-Body (transformRequest())
• Sendet die HTTP-Anfrage mit Wiederholungsversuchen und exponentiellem Backoff
• Verarbeitet bei Bedarf die Authentifizierungsaktualisierung (OAuth-Provider)

Alle Executors erweitern BaseExecutor (executors/base.ts, 1170 LOC), der Folgendes bereitstellt:

• Gemeinsame Wiederholungslogik
• Proxy-Integration
• Schutzschalter-Integration
• Hooks zur Nutzungserfassung

Stufe 4: Streaming (utils/stream.ts)

Für Streaming-Antworten gibt der Executor einen ReadableStream zurück. Der Handler:

• Leitet die Daten durch eine SSE-Transformation (createSSETransformStreamWithLogger)
• Verwendet Heartbeat-Pings, um unterbrochene Verbindungen zu erkennen
• Behandelt Client-Verbindungsabbrüche ordnungsgemäß (pipeWithDisconnect)
• Transformiert SSE → JSON für Clients ohne Streaming-Unterstützung

Für Nicht-Streaming-Antworten gibt der Executor ein geparstes JSON-Objekt zurück, das unverändert weitergeleitet wird.

Stufe 5: Erfassung (services/usage.ts)

Nach der Antwort (Erfolg oder Fehler) wird die Nutzung erfasst:

• prompt_tokens, completion_tokens, cached_tokens aus der Antwort
• cost_usd, berechnet anhand der Preisdaten
• latency_ms, status, error_class bei einem Fehler
• In der Tabelle usage_history gespeichert

Aufrufprotokoll-Artefakte werden (sofern aktiviert) unter ${DATA_DIR}/call_logs/ gespeichert.

---

Detaillierte Betrachtung wichtiger Dateien

chatCore.ts (5977 Zeilen)

Der zentrale Request-Handler. Trotz seiner Größe weist er eine klare Struktur auf:

// Pseudostruktur von chatCore.ts
export async function handleChat(request: NextRequest) {
  // 1. Authentifizierung + CORS
  await authenticateRequest(request);
  applyCorsHeaders(response);

  // 2. Validierung des Bodys
  const body = await parseRequestBody(request);

  // 3. Formaterkennung + Übersetzung
  const sourceFormat = detectFormat(request);
  const targetFormat = getTargetFormat(providerId);
  if (needsTranslation(sourceFormat, targetFormat)) {
    body = translateRequest(body, sourceFormat, targetFormat);
  }

  // 4. Combo-Routing
  const targets = await resolveComboTargets(comboId, body);
  for (const target of targets) {
    try {
      const result = await executeOnTarget(target, body);
      await recordUsage(result);
      return result;
    } catch (err) {
      // Mit dem nächsten Ziel fortfahren
    }
  }

  // 5. Notfall-Fallback
  return await emergencyFallback(body);
}

Obwohl es sich um eine einzige riesige Funktion handelt, ist sie in kommentierte Abschnitte unterteilt, die der fünfstufigen Pipeline entsprechen.

combo.ts (4456 Codezeilen)

Die Routing-Engine, die eine Combo in geordnete Ziele auflöst.

services/combo.ts

export async function handleComboChat(body, comboId): Promise<ChatResult> {
  const targets = await resolveComboTargets(comboId, body);
  for (const target of targets) {
    try {
      return await handleSingleModel(target, body);
    } catch (err) {
      log.warn("target failed, trying next", { target, err });
    }
  }
  throw new ComboExhaustedError("All targets failed");
}

Unterstützt 19 Routing-Strategien (siehe src/shared/constants/routingStrategies.ts):

Strategie	Verhalten
priority	Geordnete Liste mit dem ersten Ziel als Priorität
weighted	Probabilistische Auswahl anhand der Gewichtung jedes Ziels
round-robin	Ziele der Reihe nach zyklisch durchlaufen
context-relay	Kontext zwischen Zielen weiterreichen
fill-first	Kontingent ausschöpfen, bevor zum nächsten Ziel gewechselt wird
p2c	Auswahl aus zwei Optionen
random	Gleichmäßige Zufallsauswahl
least-used	Das Ziel mit den wenigsten kürzlichen Nutzungen auswählen
cost-optimized	Günstigstes funktionsfähiges Ziel zuerst
reset-aware	Berücksichtigt die Reset-Zeitfenster des Providers
reset-window	Routing auf Basis von Reset-Zeitfenstern
headroom	Ziel mit dem größten verbleibenden Kontingentspielraum zuerst
strict-random	Echte Gleichverteilung ohne Qualitätsgewichtung
auto	Bewertung anhand von 16 Faktoren verwenden (autoCombo/)
lkgp	Zuletzt als funktionsfähig bekannter Provider zuerst
context-optimized	Am besten für Requests mit langem Kontext geeignet
fusion	Parallel an ein Panel verteilen und anschließend über einen Judge synthetisieren (fusion.ts)

base.ts (1170 Codezeilen)

Der abstrakte Executor, den alle 107 Executoren erweitern. Er enthält:

• buildUrl() — standardmäßige URL-Konstruktion (Unterklassen überschreiben sie für benutzerdefinierte Anforderungen)
• buildHeaders() — Standard-Header (Authentifizierung, Inhaltstyp)
• transformRequest() — standardmäßig unveränderte Weitergabe
• execute() — die zentrale HTTP-Schleife mit Wiederholungsversuchen, Backoff und Circuit Breaker

open-sse/executors/default.ts

export class DefaultExecutor extends BaseExecutor {
  // Verarbeitet alle OpenAI-/Anthropic-kompatiblen Provider
  // Provider registrieren Konfigurationen (URL, Authentifizierung, Header), verwenden aber dieselbe Executor-Logik
}

Providerspezifisches Verhalten (Authentifizierungs-Header, Basis-URL, Versions-Header) wird über die Provider-Registry konfiguriert, nicht über separate Executor-Klassen.

---

## Dienste (117 Module)

Dienste sind **fokussierte Module mit jeweils einem einzigen Zweck**, die von Handlern kombiniert werden. Die wichtigsten Kategorien:

### Routing & Kombination

- `combo.ts` — Einstiegspunkt für über Kombinationen geroutete Anfragen
- `services/autoCombo/` — Bewertung anhand von 16 Faktoren, 8 automatische Routing-Strategien
- `wildcardRouter.ts` — gleicht Wildcard-Routen (`gpt-*`) ab
- `modelFamilyFallback.ts` — T5-Fallback innerhalb der Modellfamilie

### Ratenbegrenzung & Kontingente

- `rateLimitManager.ts` — Token-Bucket pro Schlüssel und Anbieter
- `usage.ts` — Erfassung der Nutzung
- `quotaCache.ts` — In-Memory-Momentaufnahmen der Kontingente

### Konto & Token

- `tokenRefresh.ts` — OAuth-Aktualisierung bei 401
- `accountFallback.ts` — Wechsel zu einem alternativen Konto
- `sessionManager.ts` — Sitzungsstatus für Dialoge mit mehreren Interaktionen

### Intelligenz

- `intentClassifier.ts` — klassifiziert die Absicht der Anfrage
- `taskAwareRouter.ts` — routet nach Aufgabentyp
- `thinkingBudget.ts` — weist Denk-Token zu
- `contextManager.ts` — fügt Routing-Kontext ein

### Ausfallsicherheit

- `resilience.ts` — Orchestrierung von Wiederholungsversuchen, Backoff und Circuit Breakern
- `emergencyFallback.ts` — Fallback als letzter Ausweg
- `modelDeprecation.ts` — automatisches Routing zu Nachfolgemodellen

### Zustand

- `signatureCache.ts` — Deduplizierung anhand der Anfragesignatur
- `volumeDetector.ts` — Lastabwurf
- `contextHandoff.ts` — Serialisierung von Sitzungen

### Komprimierung

- `compression/` (Unterverzeichnis) — vollständige Komprimierungspipeline
- 39 Dateien für Engines, Regelpakete und Adapter

### Skills

- (behandelt in [SKILLS.md](./SKILLS.md))

### Speicher

- (behandelt in [MEMORY.md](./MEMORY.md))

---

## Executors (mehr als 75 Dateien)

Eine Datei pro Anbieter. Sie erweitern alle `BaseExecutor` und überschreiben die jeweils abweichenden Teile.

### Allgemeine Muster

Anbieter werden über `getExecutor(providerId)` aufgelöst, das den konfigurierten Executor zurückgibt. OpenAI-/Anthropic-kompatible Anbieter verwenden `DefaultExecutor` (`executors/default.ts`). Anbieterspezifisches Verhalten (Basis-URL, Authentifizierungsheader, API-Version) wird in `open-sse/config/providers/` konfiguriert, während Transformationen des Anfragekörpers in `open-sse/translator/` verarbeitet werden.

Die **benutzerdefinierte URL** wird über die Anbieterkonfiguration festgelegt:

```ts
// Anbieterkonfiguration in open-sse/config/providers/
export default {
  id: "together",
  baseURL: "https://api.together.xyz/v1/chat/completions",
}

Die benutzerdefinierte Authentifizierung wird über die Authentifizierungskonfiguration der Anbieter-Registry gehandhabt (API-Schlüssel, OAuth, Header-Profile).

Transformationen eines benutzerdefinierten Anfragekörpers (z. B. die Trennung von system und messages bei Anthropic) werden pro Anbieter in open-sse/translator/ registriert.

### Die Executor-Factory

`executors/index.ts` exportiert `getExecutor(providerId)`:

```ts
import { getExecutor } from "@omniroute/open-sse/executors";

const executor = getExecutor("anthropic");
const result = await executor.execute({
  model: "claude-sonnet-4-5",
  messages: [...],
});

Die Auflösung erfolgt über die ExecutorRegistry (executors/registry.ts): Jeder spezialisierte Executor wird in der integrierten Tabelle von executors/index.ts deklariert und beim Laden des Moduls über registerExecutor(alias, instance) registriert; getExecutor() fragt die Registry ab und greift für jeden Anbieter ohne spezialisierten Eintrag auf einen memoisierten DefaultExecutor zurück. Die vollständige Zuordnung von Alias zu Executor wird durch den Golden-Test tests/unit/executor-map-golden.test.ts charakterisiert.

---

Übersetzer

Übersetzen zwischen 3 Formaten: OpenAI, Anthropic, Gemini sowie der neuen Responses API.

Wann eine Übersetzung erfolgt

import { needsTranslation, translateRequest } from "@omniroute/open-sse/translator";

if (needsTranslation(sourceFormat, targetFormat)) {
  body = translateRequest(body, sourceFormat, targetFormat);
}

Häufige Übersetzungen:

• OpenAI → Anthropic: separates system-Feld, x-api-key-Header
• OpenAI → Gemini: contents anstelle von messages, systemInstruction
• OpenAI → Responses API: input-Array, Zustand über previous_response_id

Behandelte Sonderfälle

• Rolle developer → system für Nicht-OpenAI-Anbieter
• Rolle system → für GLM/ERNIE mit der ersten Benutzernachricht zusammengeführt
• json_schema → Geminis responseMimeType + responseSchema
• tools → anbieterspezifisches Tool-Format
• Thinking-Parameter (o1, Claude) → anbieterspezifische Entsprechungen

---

MCP-Server

open-sse/mcp-server/ implementiert den Model Context Protocol-Server:

• 110 Tools (Anbieterverwaltung, Kombinationen, Speicher, Cache, Komprimierung, Proxy, Skills, Gamification, Plugins, Notion, Obsidian, lokaler Korpus)
• 3 Transportarten: stdio, SSE, Streamable HTTP
• 33 Berechtigungsbereiche für eine fein abgestufte Autorisierung

Tool-Registrierung

Tools werden als eigenständige Dateien in open-sse/mcp-server/tools/ registriert. Jede exportiert einen Namen, ein Schema, einen Handler und einen Berechtigungsbereich:

open-sse/mcp-server/tools/getHealth.ts

import { z } from "zod";
export default {
  name: "omniroute_get_health",
  description: "Get system health snapshot",
  scope: "read:health",
  inputSchema: z.object({}),
  handler: async (_args, ctx) => {
    return await getSystemHealth();
  },
};

Transportarten

// stdio (CLI-Nutzung)
startMcpStdio(server);

// SSE (HTTP-basiertes Streaming)
startMcpSse(server, port);

// Streamable HTTP (modernes MCP)
startMcpStreamable(server, port);

Autorisierung

Jeder Tool-Aufruf durchläuft Berechtigungsbereichsprüfungen (open-sse/mcp-server/auth/):

if (!hasScope(apiKey, "providers:read")) {
  throw new Error("Insufficient scope");
}

---

Transformatoren

open-sse/transformer/ konvertiert zwischen den Formaten Chat Completions und Responses API.

Warum ein separater Transformator?

Die Responses API ist das neue Format von OpenAI mit zustandsbehafteten Konversationen (previous_response_id). Wenn ein Client eine Responses-Anfrage sendet, führt OmniRoute folgende Schritte aus:

1. Interne Konvertierung von Responses → Chat Completions
2. Senden an den Anbieter (jeden Anbieter, der Chat Completions unterstützt)
3. Rückkonvertierung der Antwort in das Responses-Format
4. Streaming der konvertierten Antwort an den Client

Der Transformator (transformer/responsesTransformer.ts) stellt Folgendes bereit:

createResponsesApiTransformStream(): TransformStream

Dabei werden folgende Elemente verarbeitet:

• response.output_item.added-Ereignisse
• response.output_text.delta-Ereignisse
• response.completed-Ereignis
• Zuordnung von Tool-Aufrufen (function_call ↔ tool_calls)

---

Konfiguration

open-sse/config/ enthält die Konfigurationsschicht:

Datei	Zweck
providerRegistry.ts	Chatmodell-Registry für den Katalog mit 352 Anbietern
providerModels.ts	Modellaliase, Formatzuordnung
constants.ts	Zeitüberschreitungen, Grenzwerte, Statuscodes
defaultThinkingSignature.ts	Standardmäßige Thinking-Signatur von Claude
modelStrip.ts (in services)	Anbieterspezifisches Entfernen von Feldern

Schema der Anbieter-Registry

interface ProviderConfig {
  id: string;
  name: string;
  baseUrl: string;
  authType: "bearer" | "api-key" | "oauth" | "cookie";
  executorClass: string;
  defaultModel: string;
  capabilities: ProviderCapabilities;
  models: ModelDefinition[];
}

Die Zod-Validierung beim Laden des Moduls stellt sicher, dass alle Anbieterkonfigurationen gültig sind.

---

Performancebeschränkungen

Die Routing-Engine unterliegt strengen Performancevorgaben:

Operation	Ziel	Messung
Combo-Auflösung	<10ms	Für 50 Ziele
Ratenbegrenzungsprüfung	<1ms	In-Memory-Token-Bucket
Modellfamilien-Fallback	<5ms	Zwischengespeicherte Familiendefinitionen
Weiterleitung der Routing-Anfrage	<2ms	Kritischer Ausführungspfad
Keine blockierenden E/A-Vorgänge im kritischen Routing-Pfad	—	Vollständig asynchron

---

Anti-Patterns

❌ Synchrone DB-Aufrufe in combo.ts — vorab berechnen und zwischenspeichern ❌ Wiederholungslogik in Handlern — retry() des Resilienzservices verwenden ❌ Direkter Zugriff auf die Provider-Konfiguration — Getter von providerRegistry verwenden ❌ Hartcodierte Fallback-Ketten — in modelFamilyFallback.ts definieren ❌ Zustandsänderungen über gleichzeitige Anfragen hinweg — ausschließlich anfragebezogenen Kontext verwenden

---

Hinzufügen einer neuen Komponente

Hinzufügen eines neuen Services

1. open-sse/services/[serviceName].ts mit klar abgegrenzter Verantwortlichkeit erstellen
2. Haupt-Handler-Funktion und alle Konstanten exportieren
3. Unit-Tests in tests/unit/services/[serviceName].test.mjs hinzufügen
4. In die Anfrage-Pipeline in handlers/chatCore.ts integrieren (falls Routing-bezogen)
5. Routing-Logik in combo.ts aktualisieren, falls der Service die Zielauswahl beeinflusst
6. In dieser Datei dokumentieren

Hinzufügen eines neuen Executors

1. open-sse/executors/[provider].ts erstellen und BaseExecutor erweitern
2. In config/providerRegistry.ts registrieren
3. Zur Factory in executors/index.ts hinzufügen
4. Unit-Tests für den Executor hinzufügen
5. In docs/architecture/ARCHITECTURE.md dokumentieren

Hinzufügen eines neuen MCP-Tools

1. open-sse/mcp-server/tools/[category]Tools.ts erstellen oder aktualisieren
2. Zod-Schema für Eingaben definieren
3. Tool in mcp-server/index.ts registrieren
4. Zur Berechtigungsmatrix in mcp-server/auth/ hinzufügen
5. Unit-Tests hinzufügen

---

Siehe auch

• ARCHITECTURE.md — übergeordnete Architektur
• CODEBASE_DOCUMENTATION.md — technische Referenz
• REPOSITORY_MAP.md — Verzeichnisübersicht
• AUTO-COMBO.md — Bewertung anhand von 16 Faktoren
• MCP-SERVER.md — MCP-Server
• A2A-SERVER.md — A2A-Server
• Quellcode: open-sse/ (über 400 Dateien, ca. 143.000 Codezeilen)

---

OmniRoute-Quellcode (a58000c7685f)