Langfuse 和 JWT 用于 LLM：企业系统的代理解决方案

# 通过 JWT 将 Langfuse 与 LLM 集成：解决动态授权问题

Langfuse 是一个功能强大的工具，用于追踪 LLM 请求、管理提示词并评估模型。然而，在企业环境中，会出现一个关键问题：访问推理网关需要动态 JWT 令牌，而 Langfuse 只支持静态 API 密钥。这种不匹配使得直接集成变得不可能。解决方案是创建一个轻量级代理服务，将静态授权转换为动态授权。

我们来考察一种与 OpenAI 风格 API 兼容的架构方案。它在保持对 Langfuse 的透明性的同时，满足企业基础设施的安全要求。

代理架构：弥合静态与动态之间的差距

核心问题是两种范式的冲突：

• Langfuse 使用带有 API 密钥的固定连接配置
• 企业推理网关要求带有短暂 TTL 的短期 JWT 令牌

代理充当适配器，执行三个关键操作：

• 从 Langfuse 接收 OpenAI 兼容请求
• 通过 client_credentials 流程动态获取 JWT
• 透明地将请求转发到上游 API，并替换 Authorization 头部

这保留了标准的 Langfuse 工作流程，无需代码修改，同时符合内部基础设施安全策略。

技术实现：一步步构建

第一步：使用 FastAPI 的异步核心

代理是一个 I/O 密集型服务，主要负载来自网络调用。使用 FastAPI 和 httpx.AsyncClient 的异步架构提供：

• 非阻塞处理并行请求
• 高效连接管理
• 资源最小化下的最大吞吐量

项目结构：

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

示例生命周期管理实现：

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

第二步：路由 OpenAI 兼容端点

最小必需路由：

• /v1/chat/completions — 生成的主要端点
• /v1/models — 检查可用模型
• /healthz — 供编排器使用的健康检查

关键是要保留原始路径，以避免破坏与客户端库的兼容性。路由器实现：

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

第三步：通过 Pydantic 设置实现动态配置

使用 pydantic-settings 可以灵活管理跨环境参数。关键参数：

• LLM_BASE_URL — 推理网关地址
• SYSTEM_ID_HEADER_NAME — 客户端标识符头部
• JWT_ISSUER 和 JWT_SCOPE — OIDC 提供商参数
• 所有外部调用的超时时间

示例设置类：

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

第四步：带缓存的智能 JWT 提供程序

关键优化是在过期前缓存令牌。工作流程算法：

• 从传入请求中提取 client_id 和 client_secret
• 通过 OIDC 发现自动发现 token_endpoint
• 使用 client_credentials 流程请求令牌
• 根据 expires_in 缓存 access_token

没有缓存，高负载下认证服务会成为瓶颈。使用 LRU 缓存的实现：

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

第五步：透明请求代理

关键实现方面：

• 保留原始查询参数和请求体
• 过滤传输头部（Host、Content-Length）
• 用新鲜 JWT 替换 Authorization
• 不修改所有 HTTP 状态码，直接透传

绝不要修改负载——这确保了与 Langfuse 客户端库的兼容性。示例处理：

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

关键要点：核心关注事项

• 代理作为必需适配器：使用短期令牌时，无法直接集成 Langfuse——需要中间层
• 令牌缓存至关重要：否则认证服务在高负载下会成为瓶颈
• 保持透明性：不要修改负载或 HTTP 状态码——这确保兼容性
• 数据安全：绝不记录 Authorization 头部；限制代理访问
• 灵活配置：.env 参数化便于跨环境适配

Langfuse 设置：最后步骤

在 Langfuse 界面中的配置：

• 将 Base URL 设置为代理地址（例如 http://proxy-service/v1）
• 在 API Key 中输入 client_secret（作为 Bearer 凭证传递）
• 在 Custom Headers 中添加名称来自 SYSTEM_ID_HEADER_NAME 的头部

这形成了一个闭环：Langfuse 发送带有固定参数的请求 → 代理动态获取 JWT → 请求以所需格式到达推理网关。

重要测试场景：

• 令牌过期
• 认证服务错误
• 高负载（验证缓存效率）

此架构保留了 Langfuse 用于 LLM 请求监控的所有优势，同时满足严格的企业安全要求。具备基本 FastAPI 和 OIDC 协议技能的实现只需 1-2 天。

— Editorial Team