Langfuse i JWT dla LLM: rozwiązanie proxy dla korporacyjnych systemów

# Integracja Langfuse z LLM za pomocą JWT: rozwiązanie problemu dynamicznej autoryzacji

Langfuse to potężne narzędzie do śledzenia zapytań LLM, zarządzania promptami i oceny modeli. Jednak w środowiskach korporacyjnych pojawia się krytyczny problem: system wymaga dynamicznych tokenów JWT do dostępu do inference gateway, podczas gdy Langfuse obsługuje tylko statyczne klucze API. Ta niezgodność uniemożliwia bezpośrednią integrację. Rozwiązaniem jest stworzenie lekkiego serwisu proxy przekształcającego statyczną autoryzację w dynamiczną.

Przedstawiamy rozwiązanie architektoniczne zgodne z API w stylu OpenAI, które zachowuje przejrzystość dla Langfuse i spełnia wymagania bezpieczeństwa infrastruktury enterprise.

Architektura proxy: przezwyciężenie luki między statyką a dynamiką

Kluczowy problem polega na konflikcie paradygmatów:

• Langfuse używa stałych konfiguracji connection z kluczami API
• Korporacyjne inference gateway wymagają tymczasowych tokenów JWT z krótkim TTL

Proxy działa jako adapter wykonujący trzy kluczowe operacje:

• Odbieranie zapytań zgodnych z OpenAI od Langfuse
• Dynamiczne pobieranie JWT za pomocą client_credentials flow
• Przejrzyste przekazywanie zapytań do upstream API z zastąpieniem nagłówka autoryzacji

Dzięki temu zachowujemy standardowy workflow Langfuse bez zmian w jego kodzie źródłowym, jednocześnie przestrzegając polityk bezpieczeństwa wewnętrznej infrastruktury.

Implementacja techniczna: krok po kroku

Krok 1: Asynchroniczne jądro na FastAPI

Proxy to serwis I/O-bound, w którym główna obciążenie przypada na wywołania sieciowe. Asynchroniczna architektura z FastAPI i httpx.AsyncClient zapewnia:

• Nieblokującą obsługę równoległych zapytań
• Efektywne zarządzanie połączeniami
• Maksymalną przepustowość przy minimalnych zasobach

Struktura projektu:

app/
├── __init__.py
├── app.py          # Tochka input and lifecycle management
├── routers.py      # Routing zaprosov
├── proxy_service.py # Logic wytyczsirovaniya
├── jwt_provider.py # Receiving and keshirovanie tokenov
└── settings.py     # Konfiguratsiya cherez Pydantic

Przykład implementacji zarządzania cyklem życia:

@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: Routing endpointów zgodnych z OpenAI

Minimalnie niezbędne trasy:

• /v1/chat/completions — główny endpoint do generowania
• /v1/models — sprawdzenie dostępnych modeli
• /healthz — health-check dla orkiestratorów

Krytycznie ważne jest zachowanie oryginalnych ścieżek, aby nie naruszać kompatybilności z bibliotekami klienckimi. Implementacja routera:

@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: Dynamiczna konfiguracja za pomocą Pydantic Settings

Użycie pydantic-settings pozwala elastycznie zarządzać parametrami w różnych środowiskach. Kluczowe parametry:

• LLM_BASE_URL — adres inference gateway
• SYSTEM_ID_HEADER_NAME — nagłówek dla identyfikatora klienta
• JWT_ISSUER i JWT_SCOPE — parametry dostawcy OIDC
• Timeouty dla wszystkich zewnętrznych wywołań

Przykład klasy ustawień:

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: Inteligentny dostawca JWT z buforowaniem

Krytyczna optymalizacja to buforowanie tokenów do ich wygaśnięcia. Algorytm działania:

• Pobranie client_id i client_secret z przychodzącego zapytania
• Automatyczne wykrycie token_endpoint za pomocą OIDC discovery
• Żądanie tokena za pomocą client_credentials flow
• Buforowanie access_token z uwzględnieniem expires_in

Bez buforowania serwis autoryzacji staje się wąskim gardłem przy wysokiej obciążeniu. Implementacja za pomocą LRU-cache:

@lru_cache(maxsize=128)
async def get_token(
    self,
    client_id: str,
    client_secret: str
) -> str:
    # Logic polucheniya and keshirovaniya

Krok 5: Przejrzyste proxyowanie zapytań

Kluczowe aspekty implementacji:

• Zachowanie oryginalnych parametrów query i ciała zapytania
• Filtrowanie nagłówków transportowych (Host, Content-Length)
• Zastąpienie Authorization świeżym JWT
• Przekazanie wszystkich kodów statusu HTTP bez modyfikacji

Ważne jest, aby nie modyfikować payloadu — to gwarantuje kompatybilność z bibliotekami klienckimi Langfuse. Przykład obsługi:

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 najważniejsze: kluczowe wnioski

• Proxy jako niezbędny adapter: Przy użyciu tymczasowych tokenów bezpośrednia integracja Langfuse jest niemożliwa — potrzebna jest warstwa pośrednia
• Buforowanie tokenów jest kluczowe: Bez niego serwis autoryzacji staje się wąskim gardłem przy obciążeniu
• Zachowaj przejrzystość: Nie modyfikuj payloadu i kodów statusu HTTP — to gwarantuje kompatybilność
• Bezpieczeństwo danych: Nigdy nie loguj nagłówków Authorization, ogranicz dostęp do proxy
• Elastyczna konfiguracja: Parametryzacja za pomocą .env pozwala łatwo dostosować rozwiązanie do różnych środowisk

Konfiguracja Langfuse: ostatnie kroki

Konfiguracja w interfejsie Langfuse:

• W polu Base URL podaj adres proxy (np. http://proxy-service/v1)
• W API Key wpisz client_secret (przekazywany jako Bearer credential)
• W Custom Headers dodaj nagłówek z nazwą z SYSTEM_ID_HEADER_NAME

To tworzy zamknięty obieg: Langfuse wysyła zapytanie ze stałymi parametrami → proxy dynamicznie pobiera JWT → zapytanie dociera do inference gateway w wymaganym formacie.

Ważne jest przetestowanie scenariuszy:

• Wygaśnięcie tokena
• Błędy w serwisie autoryzacji
• Wysokie obciążenie (sprawdzenie efektywności buforowania)

Taka architektura zachowuje wszystkie zalety Langfuse do monitorowania zapytań LLM, jednocześnie przestrzegając surowych wymagań korporacyjnego bezpieczeństwa. Implementacja zajmuje 1-2 dni przy podstawowej znajomości FastAPI i protokołów OIDC.

— Editorial Team