Langfuse a JWT pro LLM: proxy řešení pro firemní systémy

Integrace Langfuse s LLM přes JWT: řešení problému dynamické autorizace

Langfuse je výkonný nástroj pro sledování požadavků LLM, správu promptů a hodnocení modelů. Nicméně v korporátních prostředích vzniká kritický problém: systém vyžaduje dynamické JWT tokeny pro přístup k inference gateway, zatímco Langfuse podporuje pouze statické API klíče. Tento nesoulad činí přímou integraci nemožnou. Řešením je vytvoření tenkého proxy servisu, který převádí statickou autorizaci na dynamickou.

Prozkoumáme architektonické řešení kompatibilní s OpenAI-podobnými API, které zachovává průhlednost pro Langfuse a zároveň splňuje požadavky na bezpečnost enterprise infrastruktury.

Architektura proxy: překlenutí propasti mezi statikou a dynamikou

Klíčový problém spočívá v konfliktu paradigm:

• Langfuse používá fixní connection konfigurace s API klíči
• Korporátní inference gateway vyžadují dočasné JWT tokeny s krátkým TTL

Proxy slouží jako adaptér, který provádí tři klíčové operace:

• Přijímání OpenAI-kompatibilních požadavků od Langfuse
• Dynamické získávání JWT přes client_credentials flow
• Průhledné předávání požadavků do upstream API s nahrazením hlavičky autorizace

To umožňuje zachovat standardní workflow Langfuse bez změny jeho kódu a zároveň dodržovat bezpečnostní politiky vnitřní infrastruktury.

Technická realizace: krok za krokem

Krok 1: Asynchronní jádro na FastAPI

Proxy je I/O-bound služba, kde hlavní zátěž připadá na síťové volání. Asynchronní architektura s FastAPI a httpx.AsyncClient zajišťuje:

• Neblokující zpracování paralelních požadavků
• Efektivní správu spojení
• Maximální propustnost při minimální spotřebě zdrojů

Struktura projektu:

app/
├── __init__.py
├── app.py          # Bod vstupu a správa životního cyklu
├── routers.py      # Směrování požadavků
├── proxy_service.py # Logika proxyování
├── jwt_provider.py # Získávání a cachování tokenů
└── settings.py     # Konfigurace přes Pydantic

Příklad realizace správy životního cyklu:

@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...")

Krok 2: Směrování OpenAI-kompatibilních endpointů

Minimálně nezbytné trasy:

• /v1/chat/completions — hlavní endpoint pro generování
• /v1/models — kontrola dostupných modelů
• /healthz — health-check pro orchestrátory

Je klíčové zachovat původní cesty, aby nedošlo k porušení kompatibility s klientskými knihovnami. Realizace routeru:

@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)

Krok 3: Dynamická konfigurace přes Pydantic Settings

Použití pydantic-settings umožňuje flexibilní řízení parametrů v různých prostředích. Klíčové parametry:

• LLM_BASE_URL — adresa inference gateway
• SYSTEM_ID_HEADER_NAME — hlavička pro identifikátor klienta
• JWT_ISSUER a JWT_SCOPE — parametry OIDC providera
• Timeouty pro všechny externí volání

Příklad třídy nastavení:

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")

Krok 4: Chytrý JWT Provider s cachováním

Kritická optimalizace — cachování tokenů do jejich vypršení. Algoritmus práce:

• Získání client_id a client_secret z příchozího požadavku
• Automatické objevení token_endpoint přes OIDC discovery
• Žádost o token přes client_credentials flow
• Cachování access_token s ohledem na expires_in

Bez cachování se auth služba stává úzkým hrdlem při vysoké zátěži. Realizace přes LRU cache:

@lru_cache(maxsize=128)
async def get_token(
    self,
    client_id: str,
    client_secret: str
) -> str:
    # Logika získávání a cachování

Krok 5: Průhledné proxyování požadavků

Klíčové aspekty realizace:

• Zachování původních query parametrů a těla požadavku
• Filtrování transportních hlaviček (Host, Content-Length)
• Nahrazení Authorization čerstvým JWT
• Předání všech HTTP stavů bez úprav

Důležité je nemodifikovat payload — to zaručuje kompatibilitu s klientskými knihovnami Langfuse. Příklad zpracování:

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,
    )

Co je důležité: klíčové závěry

• Proxy jako nezbytný adaptér: Při použití dočasných tokenů není přímá integrace Langfuse možná — vyžaduje se mezivrstva
• Cachování tokenů je kritické: Bez něj se auth služba stává úzkým hrdlem při zátěži
• Dodržujte průhlednost: Nemodifikujte payload a HTTP stavy — to zaručuje kompatibilitu
• Bezpečnost dat: Nikdy nelogujte Authorization hlavičky, omezte přístup k proxy
• Flexibilní konfigurace: Parametrizace přes .env umožňuje snadnou adaptaci na různá prostředí

Nastavení Langfuse: závěrečné kroky

Konfigurace v rozhraní Langfuse:

• Do pole Base URL zadejte adresu proxy (např. http://proxy-service/v1)
• Do API Key vložte client_secret (přenáší se jako Bearer credential)
• Do Custom Headers přidejte hlavičku s názvem z SYSTEM_ID_HEADER_NAME

Tím vznikne uzavřený okruh: Langfuse odesílá požadavek s fixními parametry → proxy dynamicky získá JWT → požadavek dorazí k inference gateway ve požadovaném formátu.

Důležité je otestovat scénáře:

• Vypršení platnosti tokenu
• Chyby v auth službě
• Vysokou zátěž (kontrola efektivity cachování)

Tato architektura zachovává všechny výhody Langfuse pro monitorování požadavků LLM a zároveň splňuje přísné požadavky korporátní bezpečnosti. Realizace zabere 1–2 dny při základních znalostech FastAPI a OIDC protokolů.

— Editorial Team