Langfuse und JWT für LLM: Proxy-Lösung für Unternehmenssysteme

# Langfuse mit LLM über JWT integrieren: Das Problem der dynamischen Autorisierung lösen

Langfuse ist ein leistungsstarkes Tool zum Nachverfolgen von LLM-Anfragen, Verwalten von Prompts und Bewerten von Modellen. In Unternehmensumgebungen taucht jedoch ein kritisches Problem auf: Das System erfordert dynamische JWT-Token für den Zugriff auf das Inferenz-Gateway, während Langfuse nur statische API-Keys unterstützt. Dieser Widerspruch macht eine direkte Integration unmöglich. Die Lösung ist ein schlanker Proxy-Dienst, der statische Autorisierung in dynamische umwandelt.

Lassen Sie uns eine architektonische Lösung betrachten, die mit OpenAI-ähnlichen APIs kompatibel ist. Sie bewahrt die Transparenz für Langfuse und erfüllt gleichzeitig die Sicherheitsanforderungen der Unternehmensinfrastruktur.

Proxy-Architektur: Die Lücke zwischen Statisch und Dynamisch überbrücken

Das Kernproblem ist ein Paradigmenkonflikt:

• Langfuse verwendet feste Verbindungs-Konfigurationen mit API-Keys
• Unternehmens-Inferenz-Gateways erfordern kurzlebige JWT-Token mit kurzen TTLs

Der Proxy fungiert als Adapter und führt drei kritische Operationen durch:

• Empfangen von OpenAI-kompatiblen Anfragen von Langfuse
• Dynamisches Abrufen von JWTs über den client_credentials-Flow
• Transparentes Weiterleiten der Anfragen an die Upstream-API mit Ersetzung des Authorization-Headers

Dies erhält den Standard-Workflow von Langfuse ohne Code-Änderungen und entspricht den internen Sicherheitsrichtlinien der Infrastruktur.

Technische Umsetzung: Schritt-für-Schritt-Anleitung

Schritt 1: Asynchroner Kern mit FastAPI

Der Proxy ist ein I/O-gebundener Dienst, bei dem die Hauptlast von Netzwerkaufrufen kommt. Eine asynchrone Architektur mit FastAPI und httpx.AsyncClient liefert:

• Nicht-blockierende Behandlung paralleler Anfragen
• Effizientes Verbindungsmanagement
• Maximale Durchsatzrate bei minimalen Ressourcen

Projektstruktur:

app/
├── __init__.py
├── app.py          # Entry point and lifecycle management
├── routers.py      # Request routing
├── proxy_service.py # Proxy logic
├── jwt_provider.py # Token retrieval and caching
└── settings.py     # Pydantic-based configuration

Beispiel für Lifecycle-Management:

@asynccontextmanager
async def lifespan(app: FastAPI):
    configure_logging()
    logger = get_logger()
    logger.info("Microservice is starting...")

    proxy_service = LLMProxyService()
    app.state.proxy_service = proxy_service

    try:
        yield
    finally:
        await proxy_service.aclose()
        logger.info("Microservice is shutting down...")

Schritt 2: Routing von OpenAI-kompatiblen Endpunkten

Minimale erforderliche Routen:

• /v1/chat/completions — Haupteindpunkt für Generierung
• /v1/models — Überprüfung verfügbarer Modelle
• /healthz — Health-Check für Orchestratoren

Es ist entscheidend, die ursprünglichen Pfade beizubehalten, um die Kompatibilität mit Client-Bibliotheken nicht zu brechen. Router-Implementierung:

@root_router.post("/v1/chat/completions")
async def proxy_chat_completions(request: Request):
    proxy_service = _get_proxy_service(request)
    return await proxy_service.proxy_chat_completions(request)

Schritt 3: Dynamische Konfiguration über Pydantic Settings

Mit pydantic-settings ist eine flexible Parameterverwaltung über Umgebungen möglich. Wichtige Parameter:

• LLM_BASE_URL — Adresse des Inferenz-Gateways
• SYSTEM_ID_HEADER_NAME — Header für Client-Identifier
• JWT_ISSUER und JWT_SCOPE — OIDC-Provider-Parameter
• Timeouts für alle externen Aufrufe

Beispiel-Klasse für Settings:

class Settings(BaseSettings):
    llm_base_url: str = Field(default="", alias="LLM_BASE_URL")
    system_id_header_name: str = Field(
        default="systemId", alias="SYSTEM_ID_HEADER_NAME")
    jwt_issuer: str = Field(default="", alias="JWT_ISSUER")

Schritt 4: Intelligenter JWT-Provider mit Caching

Eine entscheidende Optimierung ist das Cachen von Token bis zum Ablauf. Workflow-Algorithmus:

• Extrahieren von client_id und client_secret aus der eingehenden Anfrage
• Automatische Ermittlung des token_endpoint über OIDC-Discovery
• Anfordern des Tokens über client_credentials-Flow
• Cachen des access_token unter Berücksichtigung von expires_in

Ohne Caching wird der Auth-Dienst unter hoher Last zum Engpass. Implementierung mit LRU-Cache:

@lru_cache(maxsize=128)
async def get_token(
    self,
    client_id: str,
    client_secret: str
) -> str:
    # Token retrieval and caching logic

Schritt 5: Transparentes Proxying von Anfragen

Wichtige Implementierungsaspekte:

• Ursprüngliche Query-Parameter und Request-Body beibehalten
• Transport-Header filtern (Host, Content-Length)
• Authorization durch frisches JWT ersetzen
• Alle HTTP-Statuscodes unverändert durchreichen

Niemals die Payload modifizieren – das gewährleistet Kompatibilität mit Langfuse-Client-Bibliotheken. Beispiel-Behandlung:

async def _forward(self, request: Request, path: str) -> Response:
    client_id = request.headers.get(self._settings.system_id_header_name)
    authorization = request.headers.get("authorization")
    _, _, client_secret = authorization.partition(" ")

    token = await self._token_provider.get_token(
        client_id=client_id,
        client_secret=client_secret,
    )

    return await self._send(
        request=request,
        url=f"{self._settings.llm_base_url}{path}",
        token=token,
    )

Wichtige Erkenntnisse: Was zählt

• Proxy als essenzieller Adapter: Bei kurzlebigen Token ist eine direkte Langfuse-Integration unmöglich – eine Zwischenschicht ist erforderlich
• Token-Caching ist entscheidend: Ohne es wird der Auth-Dienst unter Last zum Engpass
• Transparenz wahren: Payloads oder HTTP-Statuscodes nicht verändern – das sichert Kompatibilität
• Datensicherheit: Authorization-Header niemals loggen; Proxy-Zugriff einschränken
• Flexible Konfiguration: .env-Parametrisierung erleichtert Anpassung über Umgebungen

Langfuse-Setup: Abschließende Schritte

Konfiguration in der Langfuse-Oberfläche:

• Base URL auf Proxy-Adresse setzen (z. B. http://proxy-service/v1)
• client_secret in API Key eingeben (als Bearer-Credential übermittelt)
• Header mit Namen aus SYSTEM_ID_HEADER_NAME in Custom Headers hinzufügen

Dies schafft einen geschlossenen Kreislauf: Langfuse sendet Anfragen mit festen Parametern → Proxy holt dynamisch JWT → Anfrage erreicht Inferenz-Gateway im geforderten Format.

Wichtige Test-Szenarien:

• Token-Ablauf
• Auth-Dienst-Fehler
• Hohe Last (Caching-Effizienz validieren)

Diese Architektur behält alle Vorteile von Langfuse für die Überwachung von LLM-Anfragen bei Einhaltung strenger Unternehmenssicherheitsanforderungen. Die Implementierung dauert bei grundlegenden FastAPI- und OIDC-Kenntnissen 1–2 Tage.

— Editorial Team