From 24a90392f5c4ae23c31b0d0c645e4998de64de66 Mon Sep 17 00:00:00 2001 From: Ravi Kumar L Date: Sun, 21 Jun 2026 01:40:13 +0200 Subject: [PATCH] feat: fan out Langfuse feedback scores --- api/server/routes/messages.js | 89 +++---- config/seed-langfuse-fanout.js | 7 +- packages/api/src/langfuse/config.ts | 5 +- packages/api/src/langfuse/destinations.ts | 110 +++++++++ packages/api/src/langfuse/feedback.spec.ts | 264 +++++++++++++++++++++ packages/api/src/langfuse/feedback.ts | 173 +++++++++----- packages/api/src/langfuse/utils.ts | 7 + 7 files changed, 548 insertions(+), 107 deletions(-) create mode 100644 packages/api/src/langfuse/destinations.ts create mode 100644 packages/api/src/langfuse/utils.ts diff --git a/api/server/routes/messages.js b/api/server/routes/messages.js index 17e740c515..e8b0a68c6e 100644 --- a/api/server/routes/messages.js +++ b/api/server/routes/messages.js @@ -9,7 +9,7 @@ const { traceIdForMessage, } = require('@librechat/api'); const { findAllArtifacts, replaceArtifactContent } = require('~/server/services/Artifacts/update'); -const { requireJwtAuth, validateMessageReq } = require('~/server/middleware'); +const { requireJwtAuth, validateMessageReq, configMiddleware } = require('~/server/middleware'); const db = require('~/models'); const router = express.Router(); @@ -382,49 +382,56 @@ router.put('/:conversationId/:messageId', validateMessageReq, async (req, res) = } }); -router.put('/:conversationId/:messageId/feedback', validateMessageReq, async (req, res) => { - try { - const { conversationId, messageId } = req.params; - const { feedback } = req.body; +router.put( + '/:conversationId/:messageId/feedback', + validateMessageReq, + configMiddleware, + async (req, res) => { + try { + const { conversationId, messageId } = req.params; + const { feedback } = req.body; - const updatedMessage = await db.updateMessage( - req?.user?.id, - { - messageId, - feedback: feedback || null, - }, - { context: 'updateFeedback' }, - ); - - // Best-effort: Assistants messages do not have deterministic AgentRun traces. - if (!isAssistantsEndpoint(updatedMessage.endpoint)) { - sendFeedbackScore({ - traceId: traceIdForMessage(messageId), - feedback: updatedMessage.feedback, - metadata: { - messageId: updatedMessage.messageId ?? messageId, - parentMessageId: updatedMessage.parentMessageId, - conversationId: updatedMessage.conversationId ?? conversationId, - sessionId: updatedMessage.conversationId ?? conversationId, - userId: req?.user?.id, - endpoint: updatedMessage.endpoint, - sender: updatedMessage.sender, - isCreatedByUser: updatedMessage.isCreatedByUser, - tokenCount: updatedMessage.tokenCount, + const updatedMessage = await db.updateMessage( + req?.user?.id, + { + messageId, + feedback: feedback || null, }, - }).catch((err) => logger.error('[langfuse] feedback score failed:', err)); - } + { context: 'updateFeedback' }, + ); - res.json({ - messageId, - conversationId, - feedback: updatedMessage.feedback, - }); - } catch (error) { - logger.error('Error updating message feedback:', error); - res.status(500).json({ error: 'Failed to update feedback' }); - } -}); + // Best-effort: Assistants messages do not have deterministic AgentRun traces. + if (!isAssistantsEndpoint(updatedMessage.endpoint)) { + sendFeedbackScore({ + traceId: traceIdForMessage(messageId), + feedback: updatedMessage.feedback, + appConfig: req.config, + metadata: { + messageId: updatedMessage.messageId ?? messageId, + parentMessageId: updatedMessage.parentMessageId, + conversationId: updatedMessage.conversationId ?? conversationId, + sessionId: updatedMessage.conversationId ?? conversationId, + userId: req?.user?.id, + tenantId: req?.user?.tenantId, + endpoint: updatedMessage.endpoint, + sender: updatedMessage.sender, + isCreatedByUser: updatedMessage.isCreatedByUser, + tokenCount: updatedMessage.tokenCount, + }, + }).catch((err) => logger.error('[langfuse] feedback score failed:', err)); + } + + res.json({ + messageId, + conversationId, + feedback: updatedMessage.feedback, + }); + } catch (error) { + logger.error('Error updating message feedback:', error); + res.status(500).json({ error: 'Failed to update feedback' }); + } + }, +); router.delete('/:conversationId/:messageId', validateMessageReq, async (req, res) => { try { diff --git a/config/seed-langfuse-fanout.js b/config/seed-langfuse-fanout.js index 8e437e02f6..79eabb5a6f 100644 --- a/config/seed-langfuse-fanout.js +++ b/config/seed-langfuse-fanout.js @@ -42,6 +42,7 @@ function normalizeTenantConfigs(value) { publicKey, secretKey, baseUrl: normalizeString(config.baseUrl ?? config.base_url), + fanoutBaseUrl: normalizeString(config.fanoutBaseUrl ?? config.fanout_base_url), }; }); } @@ -70,6 +71,7 @@ async function patchTenantLangfuseConfig({ tenantId, langfuse, patchConfigFields } mongoose = require('mongoose'); + require('@librechat/data-schemas').createModels(mongoose); const connect = require('./connect'); const { patchConfigFields } = require('~/models'); @@ -80,14 +82,14 @@ async function patchTenantLangfuseConfig({ tenantId, langfuse, patchConfigFields 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 } : {}), + ...(config.baseUrl ? { baseUrl: config.baseUrl } : {}), fanout: { enabled: config.enabled !== false, - baseUrl: fanoutBaseUrl, + baseUrl: config.fanoutBaseUrl ?? defaultFanoutBaseUrl, }, }; await patchTenantLangfuseConfig({ tenantId: config.tenantId, langfuse, patchConfigFields }); @@ -95,6 +97,7 @@ async function patchTenantLangfuseConfig({ tenantId, langfuse, patchConfigFields } await mongoose.disconnect(); + process.exit(0); })().catch(async (error) => { console.error(error); if (mongoose?.connection?.readyState) { diff --git a/packages/api/src/langfuse/config.ts b/packages/api/src/langfuse/config.ts index f122e576e4..4be6993791 100644 --- a/packages/api/src/langfuse/config.ts +++ b/packages/api/src/langfuse/config.ts @@ -1,12 +1,9 @@ import type { AppConfig } from '@librechat/data-schemas'; import type { RunConfig } from '@librechat/agents'; +import { normalizeString } from './utils'; 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, diff --git a/packages/api/src/langfuse/destinations.ts b/packages/api/src/langfuse/destinations.ts new file mode 100644 index 0000000000..4ac0be98d4 --- /dev/null +++ b/packages/api/src/langfuse/destinations.ts @@ -0,0 +1,110 @@ +import type { AppConfig } from '@librechat/data-schemas'; +import { normalizeString, toBasicAuthorization } from './utils'; + +const DEFAULT_BASE_URL = 'https://cloud.langfuse.com'; + +export type LangfuseScoreDestination = { + name: 'central' | 'tenant'; + baseUrl: string; + authorization: string; +}; + +function isFalseEnv(value?: string): boolean { + return value != null && ['0', 'false', 'no', 'off'].includes(value.trim().toLowerCase()); +} + +function isSampleRateEnabled(value?: string): boolean { + if (value == null || value.trim() === '') { + return true; + } + const parsed = Number(value); + return !Number.isFinite(parsed) || parsed !== 0; +} + +function isTracingEnabled(): boolean { + return ( + !isFalseEnv(process.env.LANGFUSE_TRACING_ENABLED) && + isSampleRateEnabled(process.env.LANGFUSE_SAMPLE_RATE) + ); +} + +function getLegacyBaseUrl(): string { + return ( + normalizeString(process.env.LANGFUSE_BASE_URL) ?? + normalizeString(process.env.LANGFUSE_HOST) ?? + normalizeString(process.env.LANGFUSE_BASEURL) ?? + DEFAULT_BASE_URL + ); +} + +function getCentralScoreDestination(): LangfuseScoreDestination | undefined { + if (!isTracingEnabled()) { + return undefined; + } + + const fanoutAuthorization = normalizeString(process.env.LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER); + if (fanoutAuthorization) { + return { + name: 'central', + baseUrl: normalizeString(process.env.LANGFUSE_FANOUT_CENTRAL_BASE_URL) ?? DEFAULT_BASE_URL, + authorization: fanoutAuthorization, + }; + } + + const publicKey = normalizeString(process.env.LANGFUSE_PUBLIC_KEY); + const secretKey = normalizeString(process.env.LANGFUSE_SECRET_KEY); + if (!publicKey || !secretKey) { + return undefined; + } + + return { + name: 'central', + baseUrl: getLegacyBaseUrl(), + authorization: toBasicAuthorization(publicKey, secretKey), + }; +} + +function getTenantScoreDestination(appConfig?: AppConfig): LangfuseScoreDestination | undefined { + if (!isTracingEnabled()) { + return undefined; + } + + const config = appConfig?.langfuse; + if (config?.enabled === false) { + return undefined; + } + + const publicKey = normalizeString(config?.publicKey); + const secretKey = normalizeString(config?.secretKey); + if (!publicKey || !secretKey) { + return undefined; + } + + return { + name: 'tenant', + baseUrl: + normalizeString(config?.baseUrl) ?? + normalizeString(process.env.LANGFUSE_FANOUT_TENANT_BASE_URL) ?? + DEFAULT_BASE_URL, + authorization: toBasicAuthorization(publicKey, secretKey), + }; +} + +/** + * Score fanout uses Langfuse's direct REST API. Trace fanout may use the OTLP + * collector via appConfig.langfuse.fanout/base LANGFUSE_FANOUT_BASE_URL. + */ +export function getScoreDestinations(appConfig?: AppConfig): LangfuseScoreDestination[] { + const destinations = [getCentralScoreDestination(), getTenantScoreDestination(appConfig)].filter( + (destination): destination is LangfuseScoreDestination => Boolean(destination), + ); + const seen = new Set(); + return destinations.filter((destination) => { + const key = `${destination.baseUrl}\n${destination.authorization}`; + if (seen.has(key)) { + return false; + } + seen.add(key); + return true; + }); +} diff --git a/packages/api/src/langfuse/feedback.spec.ts b/packages/api/src/langfuse/feedback.spec.ts index 3de2f727a4..573e172976 100644 --- a/packages/api/src/langfuse/feedback.spec.ts +++ b/packages/api/src/langfuse/feedback.spec.ts @@ -1,8 +1,11 @@ +import type { AppConfig } from '@librechat/data-schemas'; + jest.mock( '@librechat/data-schemas', () => ({ logger: { debug: jest.fn(), + error: jest.fn(), }, }), { virtual: true }, @@ -17,6 +20,9 @@ const langfuseEnvKeys = [ 'LANGFUSE_TRACING_ENABLED', 'LANGFUSE_SAMPLE_RATE', 'LANGFUSE_TRACING_ENVIRONMENT', + 'LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER', + 'LANGFUSE_FANOUT_CENTRAL_BASE_URL', + 'LANGFUSE_FANOUT_TENANT_BASE_URL', ]; let fetchMock: jest.SpiedFunction; @@ -40,6 +46,17 @@ function getFetchMock(): jest.SpiedFunction { return fetchMock; } +function getTenantAuthorization( + publicKey = 'tenant-public-key', + secretKey = 'tenant-secret-key', +): string { + return `Basic ${Buffer.from(`${publicKey}:${secretKey}`).toString('base64')}`; +} + +function appConfigWithLangfuse(langfuse: AppConfig['langfuse']): AppConfig { + return { langfuse } as AppConfig; +} + describe('Langfuse feedback scores', () => { beforeEach(() => { clearLangfuseEnv(); @@ -120,6 +137,253 @@ describe('Langfuse feedback scores', () => { ); }); + it('posts feedback scores to central fanout and tenant Langfuse projects', async () => { + process.env.LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER = 'Basic central-auth'; + process.env.LANGFUSE_FANOUT_CENTRAL_BASE_URL = 'http://central-langfuse:3000'; + process.env.LANGFUSE_FANOUT_TENANT_BASE_URL = 'http://tenant-langfuse:3000'; + const { sendFeedbackScore } = await loadFeedback(); + + await sendFeedbackScore({ + traceId: 'trace-id', + feedback: { rating: 'thumbsDown', tag: 'wrong' }, + metadata: { tenantId: 'tenant-a' }, + appConfig: { + langfuse: { + publicKey: 'tenant-public-key', + secretKey: 'tenant-secret-key', + }, + } as AppConfig, + }); + + expect(getFetchMock()).toHaveBeenCalledTimes(2); + expect(getFetchMock()).toHaveBeenNthCalledWith( + 1, + 'http://central-langfuse:3000/api/public/scores', + expect.objectContaining({ + method: 'POST', + headers: expect.objectContaining({ + Authorization: 'Basic central-auth', + }), + }), + ); + expect(getFetchMock()).toHaveBeenNthCalledWith( + 2, + 'http://tenant-langfuse:3000/api/public/scores', + expect.objectContaining({ + method: 'POST', + headers: expect.objectContaining({ + Authorization: getTenantAuthorization(), + }), + }), + ); + const [, tenantInit] = getFetchMock().mock.calls[1]; + expect(JSON.parse(tenantInit?.body as string)).toMatchObject({ + id: 'feedback-trace-id', + traceId: 'trace-id', + name: 'user-feedback', + value: 0, + metadata: { + rating: 'thumbsDown', + tag: 'wrong', + tenantId: 'tenant-a', + }, + }); + }); + + it('deletes feedback scores from central and tenant Langfuse projects', async () => { + process.env.LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER = 'Basic central-auth'; + process.env.LANGFUSE_FANOUT_CENTRAL_BASE_URL = 'http://central-langfuse:3000'; + process.env.LANGFUSE_FANOUT_TENANT_BASE_URL = 'http://tenant-langfuse:3000'; + const { sendFeedbackScore } = await loadFeedback(); + + await sendFeedbackScore({ + traceId: 'trace-id', + feedback: null, + appConfig: { + langfuse: { + publicKey: 'tenant-public-key', + secretKey: 'tenant-secret-key', + }, + } as AppConfig, + }); + + expect(getFetchMock()).toHaveBeenCalledTimes(2); + expect(getFetchMock()).toHaveBeenNthCalledWith( + 1, + 'http://central-langfuse:3000/api/public/scores/feedback-trace-id', + expect.objectContaining({ + method: 'DELETE', + headers: { Authorization: 'Basic central-auth' }, + }), + ); + expect(getFetchMock()).toHaveBeenNthCalledWith( + 2, + 'http://tenant-langfuse:3000/api/public/scores/feedback-trace-id', + expect.objectContaining({ + method: 'DELETE', + headers: { + Authorization: getTenantAuthorization(), + }, + }), + ); + }); + + it('posts feedback scores to tenant Langfuse when no central destination is configured', async () => { + delete process.env.LANGFUSE_PUBLIC_KEY; + delete process.env.LANGFUSE_SECRET_KEY; + process.env.LANGFUSE_FANOUT_TENANT_BASE_URL = 'http://tenant-langfuse:3000'; + const { sendFeedbackScore } = await loadFeedback(); + + await sendFeedbackScore({ + traceId: 'trace-id', + feedback: { rating: 'thumbsUp' }, + appConfig: appConfigWithLangfuse({ + publicKey: 'tenant-public-key', + secretKey: 'tenant-secret-key', + }), + }); + + expect(getFetchMock()).toHaveBeenCalledTimes(1); + expect(getFetchMock()).toHaveBeenCalledWith( + 'http://tenant-langfuse:3000/api/public/scores', + expect.objectContaining({ + method: 'POST', + headers: expect.objectContaining({ Authorization: getTenantAuthorization() }), + }), + ); + }); + + it('skips tenant scores when tenant Langfuse is disabled but keeps central scores', async () => { + process.env.LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER = 'Basic central-auth'; + process.env.LANGFUSE_FANOUT_CENTRAL_BASE_URL = 'http://central-langfuse:3000'; + const { sendFeedbackScore } = await loadFeedback(); + + await sendFeedbackScore({ + traceId: 'trace-id', + feedback: { rating: 'thumbsUp' }, + appConfig: appConfigWithLangfuse({ + enabled: false, + publicKey: 'tenant-public-key', + secretKey: 'tenant-secret-key', + }), + }); + + expect(getFetchMock()).toHaveBeenCalledTimes(1); + expect(getFetchMock()).toHaveBeenCalledWith( + 'http://central-langfuse:3000/api/public/scores', + expect.objectContaining({ + method: 'POST', + headers: expect.objectContaining({ Authorization: 'Basic central-auth' }), + }), + ); + }); + + it('deduplicates matching central and tenant score destinations', async () => { + process.env.LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER = getTenantAuthorization(); + process.env.LANGFUSE_FANOUT_CENTRAL_BASE_URL = 'http://shared-langfuse:3000'; + const { sendFeedbackScore } = await loadFeedback(); + + await sendFeedbackScore({ + traceId: 'trace-id', + feedback: { rating: 'thumbsUp' }, + appConfig: appConfigWithLangfuse({ + publicKey: 'tenant-public-key', + secretKey: 'tenant-secret-key', + baseUrl: 'http://shared-langfuse:3000', + }), + }); + + expect(getFetchMock()).toHaveBeenCalledTimes(1); + expect(getFetchMock()).toHaveBeenCalledWith( + 'http://shared-langfuse:3000/api/public/scores', + expect.objectContaining({ method: 'POST' }), + ); + }); + + it('attempts every destination and reports partial feedback score failures', async () => { + process.env.LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER = 'Basic central-auth'; + process.env.LANGFUSE_FANOUT_CENTRAL_BASE_URL = 'http://central-langfuse:3000'; + process.env.LANGFUSE_FANOUT_TENANT_BASE_URL = 'http://tenant-langfuse:3000'; + fetchMock + .mockResolvedValueOnce(new Response('central down', { status: 500 })) + .mockResolvedValueOnce(new Response(null, { status: 200 })); + const { sendFeedbackScore } = await loadFeedback(); + const { logger } = await import('@librechat/data-schemas'); + + await expect( + sendFeedbackScore({ + traceId: 'trace-id', + feedback: { rating: 'thumbsUp' }, + appConfig: appConfigWithLangfuse({ + publicKey: 'tenant-public-key', + secretKey: 'tenant-secret-key', + }), + }), + ).rejects.toThrow('langfuse central score create failed: score create 500: central down'); + + expect(getFetchMock()).toHaveBeenCalledTimes(2); + expect(logger.error).toHaveBeenCalledWith( + expect.stringContaining('[langfuse] central feedback score send failed'), + expect.any(Error), + ); + }); + + it('reports tenant feedback score failures after central succeeds', async () => { + process.env.LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER = 'Basic central-auth'; + process.env.LANGFUSE_FANOUT_CENTRAL_BASE_URL = 'http://central-langfuse:3000'; + process.env.LANGFUSE_FANOUT_TENANT_BASE_URL = 'http://tenant-langfuse:3000'; + fetchMock + .mockResolvedValueOnce(new Response(null, { status: 200 })) + .mockResolvedValueOnce(new Response('tenant down', { status: 503 })); + const { sendFeedbackScore } = await loadFeedback(); + const { logger } = await import('@librechat/data-schemas'); + + await expect( + sendFeedbackScore({ + traceId: 'trace-id', + feedback: { rating: 'thumbsUp' }, + appConfig: appConfigWithLangfuse({ + publicKey: 'tenant-public-key', + secretKey: 'tenant-secret-key', + }), + }), + ).rejects.toThrow('langfuse tenant score create failed: score create 503: tenant down'); + + expect(getFetchMock()).toHaveBeenCalledTimes(2); + expect(logger.debug).toHaveBeenCalledWith( + expect.stringContaining('[langfuse] central feedback score sent'), + ); + expect(logger.error).toHaveBeenCalledWith( + expect.stringContaining('[langfuse] tenant feedback score send failed'), + expect.any(Error), + ); + }); + + it('aggregates feedback score failures when every destination fails', async () => { + process.env.LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER = 'Basic central-auth'; + process.env.LANGFUSE_FANOUT_CENTRAL_BASE_URL = 'http://central-langfuse:3000'; + process.env.LANGFUSE_FANOUT_TENANT_BASE_URL = 'http://tenant-langfuse:3000'; + fetchMock + .mockResolvedValueOnce(new Response('central down', { status: 500 })) + .mockResolvedValueOnce(new Response('tenant down', { status: 503 })); + const { sendFeedbackScore } = await loadFeedback(); + + await expect( + sendFeedbackScore({ + traceId: 'trace-id', + feedback: { rating: 'thumbsUp' }, + appConfig: appConfigWithLangfuse({ + publicKey: 'tenant-public-key', + secretKey: 'tenant-secret-key', + }), + }), + ).rejects.toThrow( + 'langfuse central score create failed: score create 500: central down; langfuse tenant score create failed: score create 503: tenant down', + ); + + expect(getFetchMock()).toHaveBeenCalledTimes(2); + }); + it('skips scores when Langfuse tracing is disabled', async () => { process.env.LANGFUSE_TRACING_ENABLED = 'false'; const { sendFeedbackScore } = await loadFeedback(); diff --git a/packages/api/src/langfuse/feedback.ts b/packages/api/src/langfuse/feedback.ts index 14d80df048..0e58ade588 100644 --- a/packages/api/src/langfuse/feedback.ts +++ b/packages/api/src/langfuse/feedback.ts @@ -1,4 +1,6 @@ import { logger } from '@librechat/data-schemas'; +import type { AppConfig } from '@librechat/data-schemas'; +import { getScoreDestinations, type LangfuseScoreDestination } from './destinations'; export type LangfuseFeedback = { rating?: 'thumbsUp' | 'thumbsDown'; @@ -13,39 +15,23 @@ export type SendFeedbackScoreParams = { feedback?: LangfuseFeedback | null; metadata?: LangfuseFeedbackMetadata; observationId?: string; + appConfig?: AppConfig; }; -const DEFAULT_BASE_URL = 'https://cloud.langfuse.com'; -const BASE = - process.env.LANGFUSE_BASE_URL ?? - process.env.LANGFUSE_HOST ?? - process.env.LANGFUSE_BASEURL ?? - DEFAULT_BASE_URL; - -function isFalseEnv(value?: string): boolean { - return value != null && ['0', 'false', 'no', 'off'].includes(value.trim().toLowerCase()); -} - -function isSampleRateEnabled(value?: string): boolean { - if (value == null || value.trim() === '') { - return true; - } - const parsed = Number(value); - return !Number.isFinite(parsed) || parsed !== 0; -} - -const ENABLED = - Boolean(process.env.LANGFUSE_PUBLIC_KEY && process.env.LANGFUSE_SECRET_KEY) && - !isFalseEnv(process.env.LANGFUSE_TRACING_ENABLED) && - isSampleRateEnabled(process.env.LANGFUSE_SAMPLE_RATE); -const AUTHORIZATION = ENABLED - ? 'Basic ' + - Buffer.from(`${process.env.LANGFUSE_PUBLIC_KEY}:${process.env.LANGFUSE_SECRET_KEY}`).toString( - 'base64', - ) - : undefined; const ENVIRONMENT = process.env.LANGFUSE_TRACING_ENVIRONMENT; +type LangfuseScorePayload = { + id: string; + traceId: string; + name: 'user-feedback'; + value: number; + dataType: 'BOOLEAN'; + comment?: string; + metadata: Record; + observationId?: string; + environment?: string; +}; + function cleanMetadata( metadata: LangfuseFeedbackMetadata, ): Record { @@ -61,30 +47,47 @@ function cleanMetadata( ); } -export async function sendFeedbackScore({ +async function deleteScore(destination: LangfuseScoreDestination, scoreId: string): Promise { + const res = await fetch( + `${destination.baseUrl}/api/public/scores/${encodeURIComponent(scoreId)}`, + { + method: 'DELETE', + headers: { Authorization: destination.authorization }, + }, + ); + if (!res.ok && res.status !== 404) { + throw new Error(`score delete ${res.status}: ${await res.text()}`); + } +} + +async function createScore( + destination: LangfuseScoreDestination, + payload: LangfuseScorePayload, +): Promise { + const res = await fetch(`${destination.baseUrl}/api/public/scores`, { + method: 'POST', + headers: { Authorization: destination.authorization, 'Content-Type': 'application/json' }, + body: JSON.stringify(payload), + }); + if (!res.ok) { + throw new Error(`score create ${res.status}: ${await res.text()}`); + } +} + +function buildScorePayload({ + scoreId, traceId, feedback, - metadata = {}, + metadata, observationId, -}: SendFeedbackScoreParams): Promise { - if (!ENABLED || !AUTHORIZATION || !traceId) { - return; - } - - const scoreId = `feedback-${traceId}`; - - if (!feedback?.rating) { - const res = await fetch(`${BASE}/api/public/scores/${encodeURIComponent(scoreId)}`, { - method: 'DELETE', - headers: { Authorization: AUTHORIZATION }, - }); - if (!res.ok && res.status !== 404) { - throw new Error(`langfuse score delete ${res.status}: ${await res.text()}`); - } - return; - } - - const body = { +}: { + scoreId: string; + traceId: string; + feedback: LangfuseFeedback; + metadata: LangfuseFeedbackMetadata; + observationId?: string; +}): LangfuseScorePayload { + return { id: scoreId, traceId, name: 'user-feedback', @@ -95,14 +98,64 @@ export async function sendFeedbackScore({ ...(observationId ? { observationId } : {}), ...(ENVIRONMENT ? { environment: ENVIRONMENT } : {}), }; - - const res = await fetch(`${BASE}/api/public/scores`, { - method: 'POST', - headers: { Authorization: AUTHORIZATION, 'Content-Type': 'application/json' }, - body: JSON.stringify(body), - }); - if (!res.ok) { - throw new Error(`langfuse score create ${res.status}: ${await res.text()}`); - } - logger.debug(`[langfuse] feedback score sent for trace ${traceId} (${feedback.rating})`); +} + +export async function sendFeedbackScore({ + traceId, + feedback, + metadata = {}, + observationId, + appConfig, +}: SendFeedbackScoreParams): Promise { + if (!traceId) { + return; + } + + const destinations = getScoreDestinations(appConfig); + if (destinations.length === 0) { + return; + } + + const scoreId = `feedback-${traceId}`; + const payload = feedback?.rating + ? buildScorePayload({ scoreId, traceId, feedback, metadata, observationId }) + : undefined; + + const results = await Promise.allSettled( + destinations.map((destination) => + payload ? createScore(destination, payload) : deleteScore(destination, scoreId), + ), + ); + const failures: string[] = []; + + results.forEach((result, index) => { + const destination = destinations[index]; + if (!destination) { + return; + } + if (result.status === 'fulfilled') { + logger.debug( + `[langfuse] ${destination.name} feedback score ${ + payload ? 'sent' : 'deleted' + } for trace ${traceId} (${feedback?.rating ?? 'none'})`, + ); + return; + } + + logger.error( + `[langfuse] ${destination.name} feedback score ${ + payload ? 'send' : 'delete' + } failed for trace ${traceId}:`, + result.reason, + ); + failures.push( + `langfuse ${destination.name} score ${payload ? 'create' : 'delete'} failed: ${ + result.reason instanceof Error ? result.reason.message : String(result.reason) + }`, + ); + }); + + if (failures.length > 0) { + throw new Error(failures.join('; ')); + } } diff --git a/packages/api/src/langfuse/utils.ts b/packages/api/src/langfuse/utils.ts new file mode 100644 index 0000000000..14ac26aad0 --- /dev/null +++ b/packages/api/src/langfuse/utils.ts @@ -0,0 +1,7 @@ +export function normalizeString(value: unknown): string | undefined { + return typeof value === 'string' && value.trim() !== '' ? value.trim() : undefined; +} + +export function toBasicAuthorization(publicKey: string, secretKey: string): string { + return `Basic ${Buffer.from(`${publicKey}:${secretKey}`).toString('base64')}`; +}