Langfuse y JWT para LLM: solución proxy para sistemas corporativos

# Integración de Langfuse con LLM mediante JWT: Solucionando el problema de autorización dinámica

Langfuse es una herramienta potente para rastrear solicitudes de LLM, gestionar prompts y evaluar modelos. Sin embargo, en entornos empresariales, surge un problema crítico: el sistema requiere tokens JWT dinámicos para acceder a la puerta de inferencia, mientras que Langfuse solo soporta claves API estáticas. Esta discrepancia hace imposible la integración directa. La solución es crear un servicio proxy ligero que convierta la autorización estática en dinámica.

Examinemos una solución arquitectónica compatible con APIs al estilo de OpenAI. Mantiene la transparencia para Langfuse mientras cumple con los requisitos de seguridad de la infraestructura empresarial.

Arquitectura del Proxy: Puente entre lo Estático y lo Dinámico

El problema central es un choque de paradigmas:

• Langfuse usa configuraciones de conexión fijas con claves API
• Las puertas de inferencia empresariales requieren tokens JWT de corta duración con TTL breves

El proxy actúa como un adaptador, realizando tres operaciones críticas:

• Recibir solicitudes compatibles con OpenAI de Langfuse
• Obtener JWT dinámicamente mediante el flujo client_credentials
• Reenviar de forma transparente las solicitudes a la API upstream con reemplazo del encabezado Authorization

Esto preserva el flujo de trabajo estándar de Langfuse sin cambios en el código, mientras cumple con las políticas de seguridad de la infraestructura interna.

Implementación Técnica: Construcción Paso a Paso

Paso 1: Núcleo Asíncrono con FastAPI

El proxy es un servicio limitado por E/S donde la carga principal proviene de llamadas de red. Una arquitectura asíncrona con FastAPI y httpx.AsyncClient ofrece:

• Manejo no bloqueante de solicitudes paralelas
• Gestión eficiente de conexiones
• Rendimiento máximo con recursos mínimos

Estructura del proyecto:

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

Ejemplo de implementación de gestión del ciclo de vida:

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

Paso 2: Enrutamiento de Endpoints Compatibles con OpenAI

Rutas mínimamente requeridas:

• /v1/chat/completions — endpoint principal para generación
• /v1/models — consultar modelos disponibles
• /healthz — verificación de salud para orquestadores

Es crucial preservar las rutas originales para evitar romper la compatibilidad con las bibliotecas de cliente. Implementación del router:

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

Paso 3: Configuración Dinámica mediante Configuraciones Pydantic

Usar pydantic-settings permite una gestión flexible de parámetros en distintos entornos. Parámetros clave:

• LLM_BASE_URL — dirección de la puerta de inferencia
• SYSTEM_ID_HEADER_NAME — encabezado para identificador de cliente
• JWT_ISSUER y JWT_SCOPE — parámetros del proveedor OIDC
• Timeouts para todas las llamadas externas

Clase de configuración de ejemplo:

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

Paso 4: Proveedor JWT Inteligente con Caché

Una optimización crítica es almacenar los tokens en caché hasta su expiración. Algoritmo del flujo de trabajo:

• Extraer client_id y client_secret de la solicitud entrante
• Descubrir automáticamente token_endpoint mediante descubrimiento OIDC
• Solicitar token usando el flujo client_credentials
• Almacenar access_token en caché considerando expires_in

Sin caché, el servicio de autenticación se convierte en un cuello de botella bajo alta carga. Implementación con caché LRU:

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

Paso 5: Proxy de Solicitudes Transparente

Aspectos clave de implementación:

• Preservar parámetros de consulta originales y cuerpo de la solicitud
• Filtrar encabezados de transporte (Host, Content-Length)
• Reemplazar Authorization con JWT fresco
• Pasar todos los códigos de estado HTTP sin modificarlos

Nunca modificar la carga — esto asegura compatibilidad con las bibliotecas de cliente de Langfuse. Ejemplo de manejo:

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

Lecciones Clave: Lo que Importa

• Proxy como Adaptador Esencial: Con tokens de corta duración, la integración directa de Langfuse es imposible — se necesita una capa intermediaria
• Caché de Tokens es Crítico: Sin él, el servicio de autenticación genera cuellos de botella bajo carga
• Mantener Transparencia: No modificar cargas ni códigos de estado HTTP — esto asegura compatibilidad
• Seguridad de Datos: Nunca registrar encabezados Authorization; restringir el acceso al proxy
• Configuración Flexible: La parametrización con .env facilita la adaptación en distintos entornos

Configuración de Langfuse: Pasos Finales

Configuración en la interfaz de Langfuse:

• Establecer Base URL en la dirección del proxy (p. ej., http://proxy-service/v1)
• Ingresar client_secret en API Key (pasado como credencial Bearer)
• Agregar encabezado con el nombre de SYSTEM_ID_HEADER_NAME en Custom Headers

Esto crea un bucle cerrado: Langfuse envía solicitudes con parámetros fijos → proxy obtiene JWT dinámicamente → la solicitud llega a la puerta de inferencia en el formato requerido.

Es importante probar escenarios:

• Expiración de tokens
• Errores del servicio de autenticación
• Alta carga (validar eficiencia del caché)

Esta arquitectura retiene todos los beneficios de Langfuse para el monitoreo de solicitudes de LLM mientras cumple con estrictos requisitos de seguridad empresarial. La implementación toma 1-2 días con conocimientos básicos de FastAPI y el protocolo OIDC.

— Editorial Team