From 45ec1450292e05bbf3ba805df107509c05fac35b Mon Sep 17 00:00:00 2001 From: Ravi Kumar L Date: Sun, 21 Jun 2026 00:23:08 +0200 Subject: [PATCH] feat: add opt-in Langfuse fanout collector --- .env.example | 13 ++ config/seed-langfuse-fanout.js | 104 ++++++++++++++++ deploy-compose.langfuse-fanout.yml | 33 +++++ docker-compose.langfuse-fanout.yml | 33 +++++ helm/librechat/templates/_helpers.tpl | 6 + helm/librechat/templates/configmap-env.yaml | 6 + .../templates/langfuse-fanout-configmap.yaml | 56 +++++++++ .../templates/langfuse-fanout-deployment.yaml | 65 ++++++++++ .../templates/langfuse-fanout-service.yaml | 19 +++ helm/librechat/values.yaml | 24 ++++ otel/langfuse-fanout/otelcol.yaml | 46 +++++++ package.json | 3 + .../__tests__/run-summarization.test.ts | 114 ++++++++++++++++++ .../api/src/agents/openai/service.spec.ts | 28 ++++- packages/api/src/agents/openai/service.ts | 9 +- packages/api/src/agents/run.ts | 14 +-- packages/api/src/langfuse/config.ts | 75 ++++++++++++ .../specs/config-schemas.spec.ts | 29 +++++ packages/data-provider/src/config.ts | 16 +++ packages/data-schemas/src/app/service.ts | 18 +++ packages/data-schemas/src/types/app.ts | 2 + 21 files changed, 697 insertions(+), 16 deletions(-) create mode 100644 config/seed-langfuse-fanout.js create mode 100644 deploy-compose.langfuse-fanout.yml create mode 100644 docker-compose.langfuse-fanout.yml create mode 100644 helm/librechat/templates/langfuse-fanout-configmap.yaml create mode 100644 helm/librechat/templates/langfuse-fanout-deployment.yaml create mode 100644 helm/librechat/templates/langfuse-fanout-service.yaml create mode 100644 otel/langfuse-fanout/otelcol.yaml create mode 100644 packages/api/src/langfuse/config.ts diff --git a/.env.example b/.env.example index 5726655396..aa482a13de 100644 --- a/.env.example +++ b/.env.example @@ -122,6 +122,19 @@ NODE_MAX_OLD_SPACE_SIZE=6144 # LANGFUSE_SECRET_KEY= # LANGFUSE_BASE_URL= +# Optional Langfuse fanout collector for tenant-scoped Langfuse projects. +# Enable by explicitly adding the fanout compose/Helm deployment; the collector +# is not started by default. Tenant public/secret keys are expected to come from +# the admin config data source. +# LANGFUSE_FANOUT_ENABLED=false +# LANGFUSE_FANOUT_BASE_URL=http://langfuse-fanout-collector:4318 +# LANGFUSE_FANOUT_CENTRAL_BASE_URL=https://cloud.langfuse.com +# LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER=Basic +# LANGFUSE_FANOUT_TENANT_BASE_URL=https://cloud.langfuse.com +# LANGFUSE_FANOUT_MEMORY_LIMIT_MIB=256 +# LANGFUSE_FANOUT_MEMORY_SPIKE_LIMIT_MIB=64 +# LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT=1000 + #=======================# # OpenTelemetry Tracing # #=======================# diff --git a/config/seed-langfuse-fanout.js b/config/seed-langfuse-fanout.js new file mode 100644 index 0000000000..8e437e02f6 --- /dev/null +++ b/config/seed-langfuse-fanout.js @@ -0,0 +1,104 @@ +const path = require('path'); +require('module-alias')({ base: path.resolve(__dirname, '..', 'api') }); + +const { BASE_CONFIG_PRINCIPAL_ID, tenantStorage } = require('@librechat/data-schemas'); +const { PrincipalType, PrincipalModel } = require('librechat-data-provider'); + +let mongoose; + +function readJsonEnv(name) { + const raw = process.env[name]; + if (!raw || raw.trim() === '') { + return undefined; + } + try { + return JSON.parse(raw); + } catch (error) { + throw new Error(`${name} must be valid JSON: ${error.message}`); + } +} + +function normalizeString(value) { + return typeof value === 'string' && value.trim() !== '' ? value.trim() : undefined; +} + +function normalizeTenantConfigs(value) { + if (value == null || typeof value !== 'object' || Array.isArray(value)) { + throw new Error('LANGFUSE_FANOUT_SEED_TENANT_CONFIGS must be an object keyed by tenant id'); + } + + return Object.entries(value).map(([tenantId, config]) => { + if (config == null || typeof config !== 'object' || Array.isArray(config)) { + throw new Error(`Tenant ${tenantId} config must be an object`); + } + const publicKey = normalizeString(config.publicKey ?? config.public_key); + const secretKey = normalizeString(config.secretKey ?? config.secret_key); + if (config.enabled !== false && (!publicKey || !secretKey)) { + throw new Error(`Tenant ${tenantId} requires publicKey and secretKey unless enabled=false`); + } + return { + tenantId, + enabled: config.enabled, + publicKey, + secretKey, + baseUrl: normalizeString(config.baseUrl ?? config.base_url), + }; + }); +} + +async function patchTenantLangfuseConfig({ tenantId, langfuse, patchConfigFields }) { + return tenantStorage.run({ tenantId }, () => + patchConfigFields( + PrincipalType.ROLE, + BASE_CONFIG_PRINCIPAL_ID, + PrincipalModel.ROLE, + { langfuse }, + 0, + ), + ); +} + +(async () => { + const tenantConfigs = normalizeTenantConfigs( + readJsonEnv('LANGFUSE_FANOUT_SEED_TENANT_CONFIGS') ?? {}, + ); + if (tenantConfigs.length === 0) { + console.log( + 'No tenant configs found. Set LANGFUSE_FANOUT_SEED_TENANT_CONFIGS to seed tenant Langfuse configs.', + ); + process.exit(0); + } + + mongoose = require('mongoose'); + const connect = require('./connect'); + const { patchConfigFields } = require('~/models'); + + const defaultFanoutBaseUrl = + normalizeString(process.env.LANGFUSE_FANOUT_BASE_URL) ?? + 'http://langfuse-fanout-collector:4318'; + + await connect(); + + for (const config of tenantConfigs) { + const fanoutBaseUrl = config.baseUrl ?? defaultFanoutBaseUrl; + const langfuse = { + ...(config.enabled === false ? { enabled: false } : {}), + ...(config.publicKey ? { publicKey: config.publicKey } : {}), + ...(config.secretKey ? { secretKey: config.secretKey } : {}), + fanout: { + enabled: config.enabled !== false, + baseUrl: fanoutBaseUrl, + }, + }; + await patchTenantLangfuseConfig({ tenantId: config.tenantId, langfuse, patchConfigFields }); + console.purple(`Seeded Langfuse fanout config for tenant ${config.tenantId}`); + } + + await mongoose.disconnect(); +})().catch(async (error) => { + console.error(error); + if (mongoose?.connection?.readyState) { + await mongoose.disconnect().catch(() => undefined); + } + process.exit(1); +}); diff --git a/deploy-compose.langfuse-fanout.yml b/deploy-compose.langfuse-fanout.yml new file mode 100644 index 0000000000..dda21b9ad4 --- /dev/null +++ b/deploy-compose.langfuse-fanout.yml @@ -0,0 +1,33 @@ +services: + api: + depends_on: + - langfuse-fanout-collector + environment: + - LANGFUSE_FANOUT_ENABLED=true + - LANGFUSE_FANOUT_BASE_URL=http://langfuse-fanout-collector:4318 + networks: + - default + - langfuse-fanout + + langfuse-fanout-collector: + image: otel/opentelemetry-collector-contrib:0.143.0 + restart: always + command: ['--config=/etc/otelcol/otelcol.yaml'] + env_file: + - .env + environment: + - LANGFUSE_FANOUT_CENTRAL_BASE_URL=${LANGFUSE_FANOUT_CENTRAL_BASE_URL:-https://cloud.langfuse.com} + - LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER=${LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER:?Set LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER to central Langfuse Basic auth header} + - LANGFUSE_FANOUT_TENANT_BASE_URL=${LANGFUSE_FANOUT_TENANT_BASE_URL:-https://cloud.langfuse.com} + - LANGFUSE_FANOUT_MEMORY_LIMIT_MIB=${LANGFUSE_FANOUT_MEMORY_LIMIT_MIB:-256} + - LANGFUSE_FANOUT_MEMORY_SPIKE_LIMIT_MIB=${LANGFUSE_FANOUT_MEMORY_SPIKE_LIMIT_MIB:-64} + - LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT=${LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT:-1000} + expose: + - '4318' + volumes: + - ./otel/langfuse-fanout/otelcol.yaml:/etc/otelcol/otelcol.yaml:ro + networks: + - langfuse-fanout + +networks: + langfuse-fanout: diff --git a/docker-compose.langfuse-fanout.yml b/docker-compose.langfuse-fanout.yml new file mode 100644 index 0000000000..dda21b9ad4 --- /dev/null +++ b/docker-compose.langfuse-fanout.yml @@ -0,0 +1,33 @@ +services: + api: + depends_on: + - langfuse-fanout-collector + environment: + - LANGFUSE_FANOUT_ENABLED=true + - LANGFUSE_FANOUT_BASE_URL=http://langfuse-fanout-collector:4318 + networks: + - default + - langfuse-fanout + + langfuse-fanout-collector: + image: otel/opentelemetry-collector-contrib:0.143.0 + restart: always + command: ['--config=/etc/otelcol/otelcol.yaml'] + env_file: + - .env + environment: + - LANGFUSE_FANOUT_CENTRAL_BASE_URL=${LANGFUSE_FANOUT_CENTRAL_BASE_URL:-https://cloud.langfuse.com} + - LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER=${LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER:?Set LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER to central Langfuse Basic auth header} + - LANGFUSE_FANOUT_TENANT_BASE_URL=${LANGFUSE_FANOUT_TENANT_BASE_URL:-https://cloud.langfuse.com} + - LANGFUSE_FANOUT_MEMORY_LIMIT_MIB=${LANGFUSE_FANOUT_MEMORY_LIMIT_MIB:-256} + - LANGFUSE_FANOUT_MEMORY_SPIKE_LIMIT_MIB=${LANGFUSE_FANOUT_MEMORY_SPIKE_LIMIT_MIB:-64} + - LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT=${LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT:-1000} + expose: + - '4318' + volumes: + - ./otel/langfuse-fanout/otelcol.yaml:/etc/otelcol/otelcol.yaml:ro + networks: + - langfuse-fanout + +networks: + langfuse-fanout: diff --git a/helm/librechat/templates/_helpers.tpl b/helm/librechat/templates/_helpers.tpl index 4c242d9582..6e8691d7f1 100755 --- a/helm/librechat/templates/_helpers.tpl +++ b/helm/librechat/templates/_helpers.tpl @@ -44,6 +44,12 @@ app.kubernetes.io/name: {{ include "librechat.fullname" . }} app.kubernetes.io/instance: {{ .Release.Name }} {{- end }} +{{/* +Langfuse fanout collector service name. +*/}} +{{- define "librechat.langfuseFanout.fullname" -}} +{{- printf "%s-langfuse-fanout" (include "librechat.fullname" .) | trunc 63 | trimSuffix "-" }} +{{- end }} {{/* RAG Selector labels diff --git a/helm/librechat/templates/configmap-env.yaml b/helm/librechat/templates/configmap-env.yaml index 5fd43940d2..2b7b807a52 100755 --- a/helm/librechat/templates/configmap-env.yaml +++ b/helm/librechat/templates/configmap-env.yaml @@ -24,6 +24,12 @@ data: {{- if and $adminPanelUrl (not $configAdminPanelUrl) }} ADMIN_PANEL_URL: {{ $adminPanelUrl | quote }} {{- end }} + {{- if and .Values.langfuseFanout.enabled (not (hasKey $configEnv "LANGFUSE_FANOUT_ENABLED")) }} + LANGFUSE_FANOUT_ENABLED: "true" + {{- end }} + {{- if and .Values.langfuseFanout.enabled (not (hasKey $configEnv "LANGFUSE_FANOUT_BASE_URL")) }} + LANGFUSE_FANOUT_BASE_URL: http://{{ include "librechat.langfuseFanout.fullname" . }}.{{ .Release.Namespace | lower }}.svc.cluster.local:{{ .Values.langfuseFanout.service.port }} + {{- end }} {{- if $configEnv }} {{- $renderedConfigEnv := $configEnv }} {{- if and $adminPanelUrl (hasKey $configEnv "ADMIN_PANEL_URL") (not $configAdminPanelUrl) }} diff --git a/helm/librechat/templates/langfuse-fanout-configmap.yaml b/helm/librechat/templates/langfuse-fanout-configmap.yaml new file mode 100644 index 0000000000..85062726da --- /dev/null +++ b/helm/librechat/templates/langfuse-fanout-configmap.yaml @@ -0,0 +1,56 @@ +{{- if .Values.langfuseFanout.enabled }} +apiVersion: v1 +kind: ConfigMap +metadata: + name: {{ include "librechat.langfuseFanout.fullname" . }}-config + labels: + {{- include "librechat.labels" . | nindent 4 }} +data: + otelcol.yaml: | + extensions: + headers_setter/tenant_passthrough: + headers: + - action: upsert + key: Authorization + from_context: authorization + + receivers: + otlp: + protocols: + http: + endpoint: 0.0.0.0:4318 + include_metadata: true + traces_url_path: /api/public/otel/v1/traces + + processors: + memory_limiter: + check_interval: 1s + limit_mib: ${env:LANGFUSE_FANOUT_MEMORY_LIMIT_MIB} + spike_limit_mib: ${env:LANGFUSE_FANOUT_MEMORY_SPIKE_LIMIT_MIB} + batch/by_auth: + timeout: 1s + send_batch_size: 128 + metadata_keys: [authorization] + metadata_cardinality_limit: ${env:LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT} + + exporters: + otlphttp/central: + endpoint: "${env:LANGFUSE_FANOUT_CENTRAL_BASE_URL}/api/public/otel" + headers: + Authorization: "${env:LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER}" + x-langfuse-ingestion-version: "4" + otlphttp/tenant: + endpoint: "${env:LANGFUSE_FANOUT_TENANT_BASE_URL}/api/public/otel" + auth: + authenticator: headers_setter/tenant_passthrough + headers: + x-langfuse-ingestion-version: "4" + + service: + extensions: [headers_setter/tenant_passthrough] + pipelines: + traces: + receivers: [otlp] + processors: [memory_limiter, batch/by_auth] + exporters: [otlphttp/central, otlphttp/tenant] +{{- end }} diff --git a/helm/librechat/templates/langfuse-fanout-deployment.yaml b/helm/librechat/templates/langfuse-fanout-deployment.yaml new file mode 100644 index 0000000000..4f4cea7c97 --- /dev/null +++ b/helm/librechat/templates/langfuse-fanout-deployment.yaml @@ -0,0 +1,65 @@ +{{- if .Values.langfuseFanout.enabled }} +apiVersion: apps/v1 +kind: Deployment +metadata: + name: {{ include "librechat.langfuseFanout.fullname" . }} + labels: + {{- include "librechat.labels" . | nindent 4 }} + app.kubernetes.io/component: langfuse-fanout +spec: + replicas: {{ .Values.langfuseFanout.replicaCount }} + selector: + matchLabels: + {{- include "librechat.selectorLabels" . | nindent 6 }} + app.kubernetes.io/component: langfuse-fanout + template: + metadata: + annotations: + checksum/langfuse-fanout-config: {{ include (print $.Template.BasePath "/langfuse-fanout-configmap.yaml") . | sha256sum }} + {{- with .Values.langfuseFanout.podAnnotations }} + {{- toYaml . | nindent 8 }} + {{- end }} + labels: + {{- include "librechat.selectorLabels" . | nindent 8 }} + app.kubernetes.io/component: langfuse-fanout + {{- with .Values.langfuseFanout.podLabels }} + {{- toYaml . | nindent 8 }} + {{- end }} + spec: + containers: + - name: otelcol + image: "{{ .Values.langfuseFanout.image.repository }}:{{ .Values.langfuseFanout.image.tag }}" + imagePullPolicy: {{ .Values.langfuseFanout.image.pullPolicy }} + args: ["--config=/etc/otelcol/otelcol.yaml"] + ports: + - name: otlp-http + containerPort: 4318 + protocol: TCP + env: + - name: LANGFUSE_FANOUT_CENTRAL_BASE_URL + value: {{ .Values.langfuseFanout.central.baseUrl | quote }} + - name: LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER + valueFrom: + secretKeyRef: + name: {{ required "langfuseFanout.central.authHeaderSecret.name is required when langfuseFanout.enabled=true" .Values.langfuseFanout.central.authHeaderSecret.name | quote }} + key: {{ .Values.langfuseFanout.central.authHeaderSecret.key | quote }} + - name: LANGFUSE_FANOUT_TENANT_BASE_URL + value: {{ .Values.langfuseFanout.tenant.baseUrl | quote }} + - name: LANGFUSE_FANOUT_MEMORY_LIMIT_MIB + value: {{ .Values.langfuseFanout.memoryLimitMiB | quote }} + - name: LANGFUSE_FANOUT_MEMORY_SPIKE_LIMIT_MIB + value: {{ .Values.langfuseFanout.memorySpikeLimitMiB | quote }} + - name: LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT + value: {{ .Values.langfuseFanout.metadataCardinalityLimit | quote }} + volumeMounts: + - name: config + mountPath: /etc/otelcol/otelcol.yaml + subPath: otelcol.yaml + readOnly: true + resources: + {{- toYaml .Values.langfuseFanout.resources | nindent 12 }} + volumes: + - name: config + configMap: + name: {{ include "librechat.langfuseFanout.fullname" . }}-config +{{- end }} diff --git a/helm/librechat/templates/langfuse-fanout-service.yaml b/helm/librechat/templates/langfuse-fanout-service.yaml new file mode 100644 index 0000000000..e329da97c6 --- /dev/null +++ b/helm/librechat/templates/langfuse-fanout-service.yaml @@ -0,0 +1,19 @@ +{{- if .Values.langfuseFanout.enabled }} +apiVersion: v1 +kind: Service +metadata: + name: {{ include "librechat.langfuseFanout.fullname" . }} + labels: + {{- include "librechat.labels" . | nindent 4 }} + app.kubernetes.io/component: langfuse-fanout +spec: + type: {{ .Values.langfuseFanout.service.type }} + ports: + - name: otlp-http + port: {{ .Values.langfuseFanout.service.port }} + targetPort: otlp-http + protocol: TCP + selector: + {{- include "librechat.selectorLabels" . | nindent 4 }} + app.kubernetes.io/component: langfuse-fanout +{{- end }} diff --git a/helm/librechat/values.yaml b/helm/librechat/values.yaml index e5321c2a93..0dbf91e77d 100755 --- a/helm/librechat/values.yaml +++ b/helm/librechat/values.yaml @@ -284,6 +284,30 @@ dnsConfig: {} updateStrategy: type: RollingUpdate +langfuseFanout: + enabled: false + replicaCount: 1 + image: + repository: otel/opentelemetry-collector-contrib + tag: "0.143.0" + pullPolicy: IfNotPresent + service: + type: ClusterIP + port: 4318 + central: + baseUrl: https://cloud.langfuse.com + authHeaderSecret: + name: "" + key: LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER + tenant: + baseUrl: https://cloud.langfuse.com + memoryLimitMiB: 256 + memorySpikeLimitMiB: 64 + metadataCardinalityLimit: 1000 + resources: {} + podAnnotations: {} + podLabels: {} + # Extra ConfigMaps to be created alongside the main ones additionalConfigMaps: {} # custom: # suffix of the ConfigMap name diff --git a/otel/langfuse-fanout/otelcol.yaml b/otel/langfuse-fanout/otelcol.yaml new file mode 100644 index 0000000000..b240cc0dec --- /dev/null +++ b/otel/langfuse-fanout/otelcol.yaml @@ -0,0 +1,46 @@ +extensions: + headers_setter/tenant_passthrough: + headers: + - action: upsert + key: Authorization + from_context: authorization + +receivers: + otlp: + protocols: + http: + endpoint: 0.0.0.0:4318 + include_metadata: true + traces_url_path: /api/public/otel/v1/traces + +processors: + memory_limiter: + check_interval: 1s + limit_mib: ${env:LANGFUSE_FANOUT_MEMORY_LIMIT_MIB} + spike_limit_mib: ${env:LANGFUSE_FANOUT_MEMORY_SPIKE_LIMIT_MIB} + batch/by_auth: + timeout: 1s + send_batch_size: 128 + metadata_keys: [authorization] + metadata_cardinality_limit: ${env:LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT} + +exporters: + otlphttp/central: + endpoint: '${env:LANGFUSE_FANOUT_CENTRAL_BASE_URL}/api/public/otel' + headers: + Authorization: '${env:LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER}' + x-langfuse-ingestion-version: '4' + otlphttp/tenant: + endpoint: '${env:LANGFUSE_FANOUT_TENANT_BASE_URL}/api/public/otel' + auth: + authenticator: headers_setter/tenant_passthrough + headers: + x-langfuse-ingestion-version: '4' + +service: + extensions: [headers_setter/tenant_passthrough] + pipelines: + traces: + receivers: [otlp] + processors: [memory_limiter, batch/by_auth] + exporters: [otlphttp/central, otlphttp/tenant] diff --git a/package.json b/package.json index a4a004728a..d1dc1df4e2 100644 --- a/package.json +++ b/package.json @@ -26,7 +26,9 @@ "update:deployed": "node config/deployed-update.js", "rebase:deployed": "node config/deployed-update.js --rebase", "start:deployed": "docker compose -f ./deploy-compose.yml up -d || docker-compose -f ./deploy-compose.yml up -d", + "start:deployed:langfuse-fanout": "docker compose -f ./deploy-compose.yml -f ./deploy-compose.langfuse-fanout.yml up -d || docker-compose -f ./deploy-compose.yml -f ./deploy-compose.langfuse-fanout.yml up -d", "stop:deployed": "docker compose -f ./deploy-compose.yml down || docker-compose -f ./deploy-compose.yml down", + "stop:deployed:langfuse-fanout": "docker compose -f ./deploy-compose.yml -f ./deploy-compose.langfuse-fanout.yml down || docker-compose -f ./deploy-compose.yml -f ./deploy-compose.langfuse-fanout.yml down", "upgrade": "node config/upgrade.js", "create-user": "node config/create-user.js", "invite-user": "node config/invite-user.js", @@ -35,6 +37,7 @@ "ban-user": "node config/ban-user.js", "delete-user": "node config/delete-user.js", "reset-meili-sync": "node config/reset-meili-sync.js", + "seed-langfuse-fanout": "node config/seed-langfuse-fanout.js", "update-banner": "node config/update-banner.js", "delete-banner": "node config/delete-banner.js", "backend": "cross-env NODE_ENV=production node api/server/index.js", diff --git a/packages/api/src/agents/__tests__/run-summarization.test.ts b/packages/api/src/agents/__tests__/run-summarization.test.ts index 868010b1e8..cf8312ad11 100644 --- a/packages/api/src/agents/__tests__/run-summarization.test.ts +++ b/packages/api/src/agents/__tests__/run-summarization.test.ts @@ -203,6 +203,8 @@ function makeAppConfig(customEndpoints: TestCustomEndpoint[]): AppConfig { beforeEach(() => { jest.clearAllMocks(); + delete process.env.LANGFUSE_FANOUT_ENABLED; + delete process.env.LANGFUSE_FANOUT_BASE_URL; }); // --------------------------------------------------------------------------- @@ -1111,10 +1113,12 @@ async function callAndCaptureRunConfig({ overrides, user, tenantId, + appConfig, }: { overrides?: Record; user?: Record; tenantId?: string; + appConfig?: AppConfig; } = {}): Promise> { const agents = [makeAgent(overrides)]; const signal = new AbortController().signal; @@ -1126,6 +1130,7 @@ async function callAndCaptureRunConfig({ streamUsage: true, user: user as never, tenantId, + appConfig, }); const createMock = Run.create as jest.Mock; @@ -1168,6 +1173,115 @@ describe('Langfuse run config', () => { tags: ['tenant:tenant-2'], }); }); + + it('adds tenant Langfuse credentials from tenant-scoped app config', async () => { + const callArgs = await callAndCaptureRunConfig({ + tenantId: 'tenant-1', + appConfig: { + langfuse: { + publicKey: 'pk-tenant-1', + secretKey: 'sk-tenant-1', + fanout: { + enabled: true, + baseUrl: 'http://langfuse-fanout-collector:4318', + }, + }, + } as AppConfig, + }); + + expect(callArgs.langfuse).toEqual({ + deterministicTraceId: true, + publicKey: 'pk-tenant-1', + secretKey: 'sk-tenant-1', + baseUrl: 'http://langfuse-fanout-collector:4318', + metadata: { 'librechat.tenant.id': 'tenant-1' }, + tags: ['tenant:tenant-1'], + }); + }); + + it('uses deployment fanout base URL when tenant app config only contains keys', async () => { + process.env.LANGFUSE_FANOUT_ENABLED = 'true'; + process.env.LANGFUSE_FANOUT_BASE_URL = 'http://collector-from-env:4318'; + + const callArgs = await callAndCaptureRunConfig({ + tenantId: 'tenant-1', + appConfig: { + langfuse: { + publicKey: 'pk-tenant-1', + secretKey: 'sk-tenant-1', + }, + } as AppConfig, + }); + + expect(callArgs.langfuse).toMatchObject({ + publicKey: 'pk-tenant-1', + secretKey: 'sk-tenant-1', + baseUrl: 'http://collector-from-env:4318', + }); + }); + + it('does not route to fanout when deployment fanout is enabled without tenant keys', async () => { + process.env.LANGFUSE_FANOUT_ENABLED = 'true'; + process.env.LANGFUSE_FANOUT_BASE_URL = 'http://collector-from-env:4318'; + + const callArgs = await callAndCaptureRunConfig({ + tenantId: 'tenant-1', + appConfig: { + langfuse: {}, + } as AppConfig, + }); + + expect(callArgs.langfuse).toEqual({ + deterministicTraceId: true, + metadata: { 'librechat.tenant.id': 'tenant-1' }, + tags: ['tenant:tenant-1'], + }); + }); + + it('lets tenant fanout.enabled=false override deployment fanout env', async () => { + process.env.LANGFUSE_FANOUT_ENABLED = 'true'; + process.env.LANGFUSE_FANOUT_BASE_URL = 'http://collector-from-env:4318'; + + const callArgs = await callAndCaptureRunConfig({ + tenantId: 'tenant-1', + appConfig: { + langfuse: { + publicKey: 'pk-tenant-1', + secretKey: 'sk-tenant-1', + baseUrl: 'https://cloud.langfuse.com', + fanout: { + enabled: false, + }, + }, + } as AppConfig, + }); + + expect(callArgs.langfuse).toMatchObject({ + publicKey: 'pk-tenant-1', + secretKey: 'sk-tenant-1', + baseUrl: 'https://cloud.langfuse.com', + }); + }); + + it('honors tenant Langfuse enabled=false as a tracing opt-out', async () => { + const callArgs = await callAndCaptureRunConfig({ + tenantId: 'tenant-1', + appConfig: { + langfuse: { + enabled: false, + publicKey: 'pk-tenant-1', + secretKey: 'sk-tenant-1', + }, + } as AppConfig, + }); + + expect(callArgs.langfuse).toEqual({ + deterministicTraceId: true, + enabled: false, + metadata: { 'librechat.tenant.id': 'tenant-1' }, + tags: ['tenant:tenant-1'], + }); + }); }); // --------------------------------------------------------------------------- diff --git a/packages/api/src/agents/openai/service.spec.ts b/packages/api/src/agents/openai/service.spec.ts index 542ea54b6a..ccd816bf3f 100644 --- a/packages/api/src/agents/openai/service.spec.ts +++ b/packages/api/src/agents/openai/service.spec.ts @@ -10,7 +10,11 @@ jest.mock('@librechat/data-schemas', () => ({ }, })); -type CreateRunArgs = { user?: Record }; +type CreateRunArgs = { + user?: Record; + tenantId?: string; + appConfig?: Record; +}; type ProcessStreamConfig = { configurable?: Record }; function createMockReq(user?: Record) { @@ -104,4 +108,26 @@ describe('createAgentChatCompletion - MCP permission user propagation', () => { expect(streamConfig.configurable?.user).toEqual({ id: 'api-user' }); expect(streamConfig.configurable?.user).not.toHaveProperty('role'); }); + + it('forwards appConfig and tenantId to createRun', async () => { + const appConfig = { + langfuse: { + publicKey: 'pk-tenant-1', + secretKey: 'sk-tenant-1', + }, + }; + deps.appConfig = appConfig as never; + const req = createMockReq({ + id: 'user-123', + tenantId: 'tenant-1', + role: 'USER', + }); + + await createAgentChatCompletion(req, createMockRes(), deps); + + expect(createRun).toHaveBeenCalledTimes(1); + const runArgs = createRun.mock.calls[0][0] as CreateRunArgs; + expect(runArgs.tenantId).toBe('tenant-1'); + expect(runArgs.appConfig).toBe(appConfig); + }); }); diff --git a/packages/api/src/agents/openai/service.ts b/packages/api/src/agents/openai/service.ts index 22b06226dd..d1fa256772 100644 --- a/packages/api/src/agents/openai/service.ts +++ b/packages/api/src/agents/openai/service.ts @@ -69,15 +69,17 @@ export interface ChatCompletionDependencies { /** Create agent run */ createRun?: CreateRunFn; /** - * App config. Optional, but required for agents with `execute_code` in - * their tools: the helper derives `codeEnvAvailable` from + * App config. Optional for basic chat, but required for tenant-scoped + * Langfuse fanout and for agents with `execute_code` in their tools: + * tenant Langfuse keys are forwarded to `createRun`, and the helper derives + * `codeEnvAvailable` from * `appConfig?.endpoints?.agents?.capabilities` and forwards it into * `deps.initializeAgent`. When `appConfig` is omitted, the resolved * `codeEnvAvailable` is `undefined`, so `initializeAgent` skips the * `execute_code` → `bash_tool` + `read_file` expansion entirely and * code-requesting agents silently lose sandbox tools. Pass `appConfig` * (even a minimal shape with just `endpoints.agents.capabilities`) to - * keep code execution working. + * keep tenant tracing and code execution working. */ appConfig?: AppConfig; /** Tool execute options for event-driven tool execution */ @@ -526,6 +528,7 @@ export async function createAgentChatCompletion( }, user: safeUser, tenantId: typeof reqUser?.tenantId === 'string' ? reqUser.tenantId : undefined, + appConfig: deps.appConfig, }); if (run) { diff --git a/packages/api/src/agents/run.ts b/packages/api/src/agents/run.ts index 0ecfe2d50a..c0b6cbe17c 100644 --- a/packages/api/src/agents/run.ts +++ b/packages/api/src/agents/run.ts @@ -42,6 +42,7 @@ import { getOpenAIConfig } from '~/endpoints/openai/config'; import { resolveConfigHeaders } from '~/utils/headers'; import { applyTestRunHook } from '~/agents/testHook'; import { isUserProvided } from '~/utils/common'; +import { buildLangfuseConfig } from '~/langfuse/config'; /** Expected shape of JSON tool search results */ interface ToolSearchJsonResult { @@ -794,17 +795,6 @@ function buildSubagentConfigs( return configs; } -function buildLangfuseConfig(tenantIdInput?: unknown) { - const tenantId = typeof tenantIdInput === 'string' ? tenantIdInput.trim() : ''; - return { - deterministicTraceId: true, - ...(tenantId !== '' && { - metadata: { 'librechat.tenant.id': tenantId }, - tags: [`tenant:${tenantId}`], - }), - }; -} - /** * Creates a new Run instance with custom handlers and configuration. * @@ -1110,7 +1100,7 @@ export async function createRun({ // feedback can be scored against the trace without a lookup (see the // feedback route in api/server/routes/messages.js). No-op unless Langfuse // tracing is enabled. Requires @librechat/agents >= 3.2.21. - langfuse: buildLangfuseConfig(tenantId ?? user?.tenantId), + langfuse: buildLangfuseConfig({ appConfig, tenantId: tenantId ?? user?.tenantId }), ...(enableToolOutputReferences && { toolOutputReferences: { enabled: true }, }), diff --git a/packages/api/src/langfuse/config.ts b/packages/api/src/langfuse/config.ts new file mode 100644 index 0000000000..f122e576e4 --- /dev/null +++ b/packages/api/src/langfuse/config.ts @@ -0,0 +1,75 @@ +import type { AppConfig } from '@librechat/data-schemas'; +import type { RunConfig } from '@librechat/agents'; + +type LangfuseRunConfig = NonNullable; + +function normalizeString(value: unknown): string | undefined { + return typeof value === 'string' && value.trim() !== '' ? value.trim() : undefined; +} + +function mergeTraceMetadata( + base: LangfuseRunConfig['metadata'], + tenantId?: string, +): LangfuseRunConfig['metadata'] | undefined { + if (!tenantId) { + return base; + } + return { + ...(base ?? {}), + 'librechat.tenant.id': tenantId, + }; +} + +function mergeTags(tags: string[] | undefined, tenantId?: string): string[] | undefined { + if (!tenantId) { + return tags; + } + return [...new Set([...(tags ?? []), `tenant:${tenantId}`])]; +} + +export function buildLangfuseConfig({ + appConfig, + tenantId, +}: { + appConfig?: AppConfig; + tenantId?: string; +} = {}): LangfuseRunConfig { + const normalizedTenantId = normalizeString(tenantId); + const config = appConfig?.langfuse; + + const langfuse: LangfuseRunConfig = { + deterministicTraceId: true, + metadata: mergeTraceMetadata(undefined, normalizedTenantId), + tags: mergeTags(undefined, normalizedTenantId), + }; + + if (config?.enabled === false) { + return { + ...langfuse, + enabled: false, + }; + } + + const publicKey = normalizeString(config?.publicKey); + const secretKey = normalizeString(config?.secretKey); + const hasTenantCredentials = Boolean(publicKey && secretKey); + if (hasTenantCredentials) { + langfuse.publicKey = publicKey; + langfuse.secretKey = secretKey; + + const fanout = config?.fanout; + const fanoutEnabled = + fanout?.enabled !== false && + (fanout?.enabled === true || process.env.LANGFUSE_FANOUT_ENABLED === 'true'); + const fanoutBaseUrl = + normalizeString(fanout?.baseUrl) ?? normalizeString(process.env.LANGFUSE_FANOUT_BASE_URL); + const baseUrl = + fanoutEnabled && fanoutBaseUrl ? fanoutBaseUrl : normalizeString(config?.baseUrl); + + if (baseUrl) { + langfuse.baseUrl = baseUrl; + } + } + + return langfuse; +} diff --git a/packages/data-provider/specs/config-schemas.spec.ts b/packages/data-provider/specs/config-schemas.spec.ts index 236295483b..995ec768ea 100644 --- a/packages/data-provider/specs/config-schemas.spec.ts +++ b/packages/data-provider/specs/config-schemas.spec.ts @@ -1178,3 +1178,32 @@ describe('specsConfigSchema', () => { expect(result.success).toBe(false); }); }); + +describe('configSchema langfuse', () => { + it('accepts tenant Langfuse fanout config', () => { + const result = configSchema.safeParse({ + version: '1.3.7', + langfuse: { + publicKey: 'pk-lf-tenant', + secretKey: 'sk-lf-tenant', + fanout: { + enabled: true, + baseUrl: 'http://langfuse-fanout-collector:4318', + }, + }, + }); + + expect(result.success).toBe(true); + }); + + it('accepts an explicit tenant Langfuse opt-out', () => { + const result = configSchema.safeParse({ + version: '1.3.7', + langfuse: { + enabled: false, + }, + }); + + expect(result.success).toBe(true); + }); +}); diff --git a/packages/data-provider/src/config.ts b/packages/data-provider/src/config.ts index c10597f53f..f21bc08b0a 100644 --- a/packages/data-provider/src/config.ts +++ b/packages/data-provider/src/config.ts @@ -1684,11 +1684,27 @@ export const messageFilterSchema = z.object({ export type MessageFilterConfig = z.infer; +export const langfuseConfigSchema = z.object({ + enabled: z.boolean().optional(), + publicKey: z.string().optional(), + secretKey: z.string().optional(), + baseUrl: z.string().optional(), + fanout: z + .object({ + enabled: z.boolean().optional(), + baseUrl: z.string().optional(), + }) + .optional(), +}); + +export type LangfuseConfig = z.infer; + export const configSchema = z.object({ version: z.string(), cache: z.boolean().default(true), ocr: ocrSchema.optional(), webSearch: webSearchSchema.optional(), + langfuse: langfuseConfigSchema.optional(), memory: memorySchema.optional(), summarization: summarizationConfigSchema.optional(), skillSync: skillSyncConfigSchema, diff --git a/packages/data-schemas/src/app/service.ts b/packages/data-schemas/src/app/service.ts index fe3ae1b007..19517945a6 100644 --- a/packages/data-schemas/src/app/service.ts +++ b/packages/data-schemas/src/app/service.ts @@ -1,6 +1,7 @@ import { EModelEndpoint, getConfigDefaults, + langfuseConfigSchema, skillSyncConfigSchema, summarizationConfigSchema, } from 'librechat-data-provider'; @@ -67,6 +68,21 @@ export function loadSkillSyncConfig(config: DeepPartial): AppConf return parsed.data; } +export function loadLangfuseConfig(config: DeepPartial): AppConfig['langfuse'] { + const raw = config.langfuse; + if (!raw || typeof raw !== 'object') { + return undefined; + } + + const parsed = langfuseConfigSchema.safeParse(raw); + if (!parsed.success) { + logger.warn('[AppService] Invalid Langfuse config', parsed.error.flatten()); + return undefined; + } + + return parsed.data; +} + export type Paths = { root: string; uploads: string; @@ -128,6 +144,7 @@ export const AppService = async (params?: { const turnstileConfig = loadTurnstileConfig(config, configDefaults); const speech = config.speech; const messageFilter = config.messageFilter; + const langfuse = loadLangfuseConfig(config); const defaultConfig = { ocr, @@ -145,6 +162,7 @@ export const AppService = async (params?: { transactions, filteredTools, includedTools, + langfuse, messageFilter, summarization, availableTools, diff --git a/packages/data-schemas/src/types/app.ts b/packages/data-schemas/src/types/app.ts index 0eb0f4d60b..4940261f5f 100644 --- a/packages/data-schemas/src/types/app.ts +++ b/packages/data-schemas/src/types/app.ts @@ -65,6 +65,8 @@ export interface AppConfig { webSearch?: TCustomConfig['webSearch']; /** Message filter configuration (PII and future filter types) */ messageFilter?: TCustomConfig['messageFilter']; + /** Langfuse tracing configuration */ + langfuse?: TCustomConfig['langfuse']; /** Skill sync configuration */ skillSync?: SkillSyncConfig; /** File storage strategy ('local', 's3', 'firebase', 'azure_blob', 'cloudfront') */