diff --git a/api/app/clients/BaseClient.js b/api/app/clients/BaseClient.js index 58d111a32a..602c7a680c 100644 --- a/api/app/clients/BaseClient.js +++ b/api/app/clients/BaseClient.js @@ -13,9 +13,7 @@ const { encodeAndFormatVideos, getTransactionsConfig, encodeAndFormatDocuments, - getLangfuseTraceDestinationIds, - isLangfuseTraceSampled, - traceIdForMessage, + getLangfuseTraceMessageFields, } = require('@librechat/api'); const { Constants, @@ -726,10 +724,11 @@ class BaseClient { this.abortController.requestCompleted = true; } - const isAgentResponse = isAgentsEndpoint(this.options.endpoint); - const langfuseTraceId = isAgentResponse ? traceIdForMessage(responseMessageId) : undefined; - const langfuseSampled = - langfuseTraceId != null ? isLangfuseTraceSampled(langfuseTraceId) : undefined; + const isAgentResponse = + this.clientName === EModelEndpoint.agents || isAgentsEndpoint(this.options.endpoint); + const langfuseTraceFields = isAgentResponse + ? await getLangfuseTraceMessageFields(appConfig, responseMessageId) + : undefined; /** @type {TMessage} */ const responseMessage = { @@ -737,14 +736,7 @@ class BaseClient { conversationId, parentMessageId: userMessage.messageId, isCreatedByUser: false, - ...(isAgentResponse && { - langfuseSampled, - langfuseDestinationIds: await getLangfuseTraceDestinationIds( - appConfig, - langfuseTraceId, - langfuseSampled, - ), - }), + ...(langfuseTraceFields ?? {}), isEdited, model: this.getResponseModel(), sender: this.sender, diff --git a/api/app/clients/specs/BaseClient.test.js b/api/app/clients/specs/BaseClient.test.js index 70f0a91c17..f1ff51099a 100644 --- a/api/app/clients/specs/BaseClient.test.js +++ b/api/app/clients/specs/BaseClient.test.js @@ -877,6 +877,40 @@ describe('BaseClient', () => { } }); + test('persists the Langfuse sampling decision for agent clients using a provider endpoint', async () => { + const previousSampleRate = process.env.LANGFUSE_SAMPLE_RATE; + const previousClientName = TestClient.clientName; + const previousEndpoint = TestClient.options.endpoint; + process.env.LANGFUSE_SAMPLE_RATE = '0'; + TestClient.clientName = 'agents'; + TestClient.options.endpoint = 'bedrock'; + const saveSpy = jest.spyOn(TestClient, 'saveMessageToDatabase'); + + try { + const response = await TestClient.sendMessage('Hello, world!', { user: {} }); + + expect(response.langfuseSampled).toBe(false); + expect(response.langfuseDestinationIds).toEqual([]); + expect(saveSpy).toHaveBeenCalledWith( + expect.objectContaining({ + endpoint: 'bedrock', + langfuseSampled: false, + langfuseDestinationIds: [], + }), + expect.any(Object), + expect.any(Object), + ); + } finally { + if (previousSampleRate == null) { + delete process.env.LANGFUSE_SAMPLE_RATE; + } else { + process.env.LANGFUSE_SAMPLE_RATE = previousSampleRate; + } + TestClient.clientName = previousClientName; + TestClient.options.endpoint = previousEndpoint; + } + }); + test('persists no Langfuse destination when a sampled trace has no configured export', async () => { const envKeys = [ 'LANGFUSE_PUBLIC_KEY', diff --git a/api/server/controllers/agents/__tests__/responses.unit.spec.js b/api/server/controllers/agents/__tests__/responses.unit.spec.js index 35f25e3758..5c5c49db0d 100644 --- a/api/server/controllers/agents/__tests__/responses.unit.spec.js +++ b/api/server/controllers/agents/__tests__/responses.unit.spec.js @@ -152,6 +152,10 @@ jest.mock('@librechat/api', () => ({ getTransactionsConfig: mockGetTransactionsConfig, recordCollectedUsage: mockRecordCollectedUsage, createSubagentUsageSink: jest.fn().mockReturnValue(jest.fn()), + getLangfuseTraceMessageFields: jest.fn().mockResolvedValue({ + langfuseSampled: true, + langfuseDestinationIds: ['destination-1'], + }), extractManualSkills: jest.fn().mockReturnValue(undefined), injectSkillPrimes: jest.fn().mockReturnValue({ initialMessages: [], @@ -407,6 +411,28 @@ describe('createResponse controller', () => { ); }); + it('stores Langfuse trace markers with a persisted response', async () => { + const api = require('@librechat/api'); + const { saveMessage } = require('~/models'); + api.validateResponseRequest.mockReturnValueOnce({ + request: { ...req.body, store: true }, + }); + + await createResponse(req, res); + + expect(api.getLangfuseTraceMessageFields).toHaveBeenCalledWith(req.config, 'resp_mock-123'); + expect(saveMessage).toHaveBeenCalledWith( + req, + expect.objectContaining({ + messageId: 'resp_mock-123', + isCreatedByUser: false, + langfuseSampled: true, + langfuseDestinationIds: ['destination-1'], + }), + { context: 'Responses API - save assistant response' }, + ); + }); + describe('execution envelope', () => { it('creates the portable run input before agent initialization', async () => { req.user = { diff --git a/api/server/controllers/agents/responses.js b/api/server/controllers/agents/responses.js index 4c0ae894dd..b104800949 100644 --- a/api/server/controllers/agents/responses.js +++ b/api/server/controllers/agents/responses.js @@ -48,6 +48,7 @@ const { sendResponsesErrorResponse, createResponsesEventHandlers, createAggregatorEventHandlers, + getLangfuseTraceMessageFields, stripActivityLabelParts, } = require('@librechat/api'); const { @@ -223,6 +224,8 @@ async function saveResponseOutput(req, conversationId, responseId, response, age } } + const langfuseTraceFields = await getLangfuseTraceMessageFields(req.config, responseId); + // Save the assistant message await db.saveMessage( req, @@ -231,6 +234,7 @@ async function saveResponseOutput(req, conversationId, responseId, response, age conversationId, parentMessageId: null, isCreatedByUser: false, + ...langfuseTraceFields, text: responseText, sender: 'Agent', endpoint: EModelEndpoint.agents, diff --git a/packages/api/src/langfuse/destinations.ts b/packages/api/src/langfuse/destinations.ts index d682347a69..10dc1adf2d 100644 --- a/packages/api/src/langfuse/destinations.ts +++ b/packages/api/src/langfuse/destinations.ts @@ -11,6 +11,7 @@ import { import { normalizeBoolean, resolveTenantCredentials, toBasicAuthorization } from './utils'; import { resolveLangfuseTenantDestination } from './tenantDestinations'; import { normalizeString } from '~/utils/text'; +import { traceIdForMessage } from './trace'; const DEFAULT_BASE_URL = 'https://cloud.langfuse.com'; const PROJECT_LOOKUP_TIMEOUT_MS = 10_000; @@ -260,6 +261,22 @@ export async function getLangfuseTraceDestinationIds( return destinations.map(({ id }) => id as string); } +export async function getLangfuseTraceMessageFields( + appConfig: AppConfig | undefined, + messageId: string, +): Promise<{ langfuseSampled: boolean; langfuseDestinationIds?: string[] }> { + const traceId = traceIdForMessage(messageId); + const langfuseSampled = isLangfuseTraceSampled(traceId); + return { + langfuseSampled, + langfuseDestinationIds: await getLangfuseTraceDestinationIds( + appConfig, + traceId, + langfuseSampled, + ), + }; +} + const centralPublicKey = normalizeString(process.env.LANGFUSE_PUBLIC_KEY); const centralSecretKey = normalizeString(process.env.LANGFUSE_SECRET_KEY); if (centralPublicKey && centralSecretKey) {