diff --git a/copilot/.gitignore b/copilot/.gitignore index f574addd89f..76cbb5e51cd 100644 --- a/copilot/.gitignore +++ b/copilot/.gitignore @@ -1,3 +1,4 @@ .workspace scripts/* !scripts/copilot-switch-app.sh +!scripts/copilot-auto-readme.md diff --git a/copilot/scripts/copilot-auto-readme.md b/copilot/scripts/copilot-auto-readme.md new file mode 100644 index 00000000000..0bce1bfca94 --- /dev/null +++ b/copilot/scripts/copilot-auto-readme.md @@ -0,0 +1,17 @@ +# Copilot Auto-Deploy Scripts + +Automated deployment scripts for AWS Copilot environments. + +## Install + +```bash +pnpm secure-run 'eval "$COPILOT_AUTO_INSTALL"' +``` + +## Usage + +```bash +pnpm copilot:auto +``` + +Guides you through environment selection, domain configuration and service deployment and ensures you're using the correct AWS credentials. diff --git a/packages-answers/ui/src/Admin/Chatflows/index.tsx b/packages-answers/ui/src/Admin/Chatflows/index.tsx index 3d2f5101a3d..f6b8ef50acd 100644 --- a/packages-answers/ui/src/Admin/Chatflows/index.tsx +++ b/packages-answers/ui/src/Admin/Chatflows/index.tsx @@ -1,6 +1,5 @@ 'use client' import { useEffect, useState } from 'react' -import { Link } from 'react-router-dom' import { Box, Skeleton, @@ -160,22 +159,8 @@ const AdminChatflows = () => { refreshChatflows() } - const getCanvasRoute = (chatflow: any) => { - // For react-router-dom Link component (relative paths) - console.log('🤖id', chatflow.id) - console.log('type', chatflow) - if (chatflow.type === 'AGENTFLOW') { - return `/v2/agentcanvas/${chatflow.id}` - } else if (chatflow.type === 'MULTIAGENT') { - return `/agentcanvas/${chatflow.id}` - } else { - // Default to regular chatflow canvas - return `/canvas/${chatflow.id}` - } - } - const getCanvasFullUrl = (chatflow: any) => { - // For window.open() (full URLs with base path) + // Generate full absolute URL with /sidekick-studio prefix for navigation if (chatflow.type === 'AGENTFLOW') { return `/sidekick-studio/v2/agentcanvas/${chatflow.id}` } else if (chatflow.type === 'MULTIAGENT') { @@ -330,8 +315,8 @@ const AdminChatflows = () => { setRollbackConfirmOpen(false) setSelectedVersionForRollback(null) handleCloseVersions() - // Refresh data - window.location.reload() + // Refresh data without full page reload + refreshChatflows() } catch (error) { console.error('Failed to rollback chatflow:', error) } @@ -361,7 +346,7 @@ const AdminChatflows = () => { return ( - @@ -645,7 +630,7 @@ const AdminChatflows = () => { window.open(getCanvasRoute(fullDefaultTemplate), '_blank')} + onClick={() => window.open(getCanvasFullUrl(fullDefaultTemplate), '_blank')} sx={{ color: 'rgba(255, 193, 7, 0.8)', bgcolor: 'rgba(255, 193, 7, 0.1)', @@ -1213,19 +1198,22 @@ const AdminChatflows = () => { - - {chatflow.name} - + {chatflow.name} {defaultTemplateData && chatflow.id === defaultTemplateData.id && ( { return ( + + + diff --git a/packages/components/nodes/agentflow/Agent/Agent.ts b/packages/components/nodes/agentflow/Agent/Agent.ts index 6dbafcc88c6..296fd1f361f 100644 --- a/packages/components/nodes/agentflow/Agent/Agent.ts +++ b/packages/components/nodes/agentflow/Agent/Agent.ts @@ -11,7 +11,7 @@ import { IUsedTool } from '../../../src/Interface' import { AIMessageChunk, BaseMessageLike, MessageContentText } from '@langchain/core/messages' -import { AnalyticHandler } from '../../../src/handler' +import { AnalyticHandler, additionalCallbacks } from '../../../src/handler' import { DEFAULT_SUMMARIZER_TEMPLATE } from '../prompt' import { ILLMMessage } from '../Interface.Agentflow' import { Tool } from '@langchain/core/tools' @@ -20,6 +20,7 @@ import { flatten } from 'lodash' import zodToJsonSchema from 'zod-to-json-schema' import { getErrorMessage } from '../../../src/error' import { DataSource } from 'typeorm' +import { LangfuseSpanClient } from 'langfuse' import { getPastChatHistoryImageMessages, getUniqueImageMessages, @@ -477,6 +478,19 @@ class Agent_Agentflow implements INode { throw new Error('Model is required') } + // Setup analytics tracing options and attach to options object for easy access + const callbacks = await additionalCallbacks(nodeData, { + ...options, + parentLangfuseTrace: options.parentLangfuseTrace, + parentLangfuseSpan: options.parentLangfuseSpan + }) + const llmCallOptions: ICommonObject = { signal: abortController?.signal } + if (callbacks && callbacks.length > 0) { + llmCallOptions.callbacks = callbacks + } + options._llmCallOptions = llmCallOptions + options._parentToolSpan = options.parentLangfuseSpan as LangfuseSpanClient | undefined + // Extract tools const tools = nodeData.inputs?.agentTools as ITool[] @@ -856,9 +870,9 @@ class Agent_Agentflow implements INode { } } else { if (isStreamable) { - response = await this.handleStreamingResponse(sseStreamer, llmNodeInstance, messages, chatId, abortController) + response = await this.handleStreamingResponse(sseStreamer, llmNodeInstance, messages, chatId, options) } else { - response = await llmNodeInstance.invoke(messages, { signal: abortController?.signal }) + response = await llmNodeInstance.invoke(messages, options._llmCallOptions || { signal: abortController?.signal }) } } @@ -1056,6 +1070,7 @@ class Agent_Agentflow implements INode { runtimeImageMessagesWithFileRef: BaseMessageLike[] pastImageMessagesWithFileRef: BaseMessageLike[] }): Promise { + const llmCallOptions = options._llmCallOptions || { signal: abortController?.signal } const { updatedPastMessages, transformedPastMessages } = await getPastChatHistoryImageMessages(pastChatHistory, options) pastChatHistory = updatedPastMessages pastImageMessagesWithFileRef.push(...transformedPastMessages) @@ -1102,12 +1117,12 @@ class Agent_Agentflow implements INode { ) } ], - { signal: abortController?.signal } + llmCallOptions ) messages.push({ role: 'assistant', content: summary.content as string }) } else if (memoryType === 'conversationSummaryBuffer') { // Summary buffer: Summarize messages that exceed token limit - await this.handleSummaryBuffer(messages, pastMessages, llmNodeInstance, nodeData, abortController) + await this.handleSummaryBuffer(messages, pastMessages, llmNodeInstance, nodeData, abortController, options) } else { // Default: Use all messages messages.push(...pastMessages) @@ -1131,8 +1146,10 @@ class Agent_Agentflow implements INode { pastMessages: BaseMessageLike[], llmNodeInstance: BaseChatModel, nodeData: INodeData, - abortController: AbortController + abortController: AbortController, + options: ICommonObject ): Promise { + const llmCallOptions = options._llmCallOptions || { signal: abortController?.signal } const maxTokenLimit = (nodeData.inputs?.agentMemoryMaxTokenLimit as number) || 2000 // Convert past messages to a format suitable for token counting @@ -1166,7 +1183,7 @@ class Agent_Agentflow implements INode { content: DEFAULT_SUMMARIZER_TEMPLATE.replace('{conversation}', messagesToSummarizeString) } ], - { signal: abortController?.signal } + llmCallOptions ) // Add summary as a system message at the beginning, then add remaining messages @@ -1186,12 +1203,13 @@ class Agent_Agentflow implements INode { llmNodeInstance: BaseChatModel, messages: BaseMessageLike[], chatId: string, - abortController: AbortController + options: ICommonObject ): Promise { + const llmCallOptions = options._llmCallOptions || { signal: options.abortController?.signal } let response = new AIMessageChunk('') try { - for await (const chunk of await llmNodeInstance.stream(messages, { signal: abortController?.signal })) { + for await (const chunk of await llmNodeInstance.stream(messages, llmCallOptions)) { if (sseStreamer) { let content = '' if (Array.isArray(chunk.content) && chunk.content.length > 0) { @@ -1394,6 +1412,10 @@ class Agent_Agentflow implements INode { totalTokens: number isWaitingForHumanInput?: boolean }> { + const llmCallOptions = options._llmCallOptions || { signal: abortController?.signal } + const parentToolSpan = options._parentToolSpan + let analyticHandlers = options.analyticHandlers as AnalyticHandler + // Track total tokens used throughout this process let totalTokens = response.usage_metadata?.total_tokens || 0 @@ -1445,6 +1467,12 @@ class Agent_Agentflow implements INode { return { response, usedTools, sourceDocuments, artifacts, totalTokens, isWaitingForHumanInput: true } } + let toolSpan = analyticHandlers?.createToolSpan(parentToolSpan, toolCall.name, toolCall.args, { + nodeId: nodeData.id, + nodeLabel: nodeData.label, + toolCallId: toolCall.id + }) + try { //@ts-ignore let toolOutput = await selectedTool.call(toolCall.args, { signal: abortController?.signal }, undefined, flowConfig) @@ -1491,12 +1519,29 @@ class Agent_Agentflow implements INode { toolInput: toolCall.args, toolOutput }) + + if (toolSpan && analyticHandlers) { + const spanPayload: ICommonObject = { + output: analyticHandlers.safeSerializeForTrace(toolOutput) + } + if (parsedDocs !== undefined) spanPayload.sourceDocuments = analyticHandlers.safeSerializeForTrace(parsedDocs) + if (parsedArtifacts !== undefined) spanPayload.artifacts = analyticHandlers.safeSerializeForTrace(parsedArtifacts) + analyticHandlers.finishToolSpan(toolSpan, 'FINISHED', spanPayload) + toolSpan = undefined + } } catch (e) { console.error('Error invoking tool:', e) // Enhanced error message for MCP connection issues const enhancedErrorMessage = this.enhanceMcpErrorMessage(e, toolCall.name, nodeData, options) + if (toolSpan && analyticHandlers) { + analyticHandlers.finishToolSpan(toolSpan, 'ERROR', { + error: analyticHandlers.safeSerializeForTrace(enhancedErrorMessage) + }) + toolSpan = undefined + } + usedTools.push({ tool: selectedTool.name, toolInput: toolCall.args, @@ -1534,9 +1579,9 @@ class Agent_Agentflow implements INode { let newResponse: AIMessageChunk if (isStreamable) { - newResponse = await this.handleStreamingResponse(sseStreamer, llmNodeInstance, messages, chatId, abortController) + newResponse = await this.handleStreamingResponse(sseStreamer, llmNodeInstance, messages, chatId, options) } else { - newResponse = await llmNodeInstance.invoke(messages, { signal: abortController?.signal }) + newResponse = await llmNodeInstance.invoke(messages, llmCallOptions) // Stream non-streaming response if this is the last node if (isLastNode && sseStreamer) { @@ -1629,6 +1674,9 @@ class Agent_Agentflow implements INode { totalTokens: number isWaitingForHumanInput?: boolean }> { + const llmCallOptions = options._llmCallOptions || { signal: abortController?.signal } + const parentToolSpan = options._parentToolSpan + const analyticHandlers = options.analyticHandlers as AnalyticHandler let llmNodeInstance = llmWithoutToolsBind const lastCheckpointMessages = humanInputAction?.data?.input?.messages ?? [] @@ -1688,8 +1736,17 @@ class Agent_Agentflow implements INode { if (humanInput.type === 'reject') { messages.pop() toolsInstance = toolsInstance.filter((tool) => tool.name !== toolCall.name) + continue } + if (humanInput.type === 'proceed') { + let toolSpan: LangfuseSpanClient | undefined + toolSpan = analyticHandlers?.createToolSpan(parentToolSpan, toolCall.name, toolCall.args, { + nodeId: nodeData.id, + nodeLabel: nodeData.label, + toolCallId: toolCall.id + }) + try { //@ts-ignore let toolOutput = await selectedTool.call(toolCall.args, { signal: abortController?.signal }, undefined, flowConfig) @@ -1736,12 +1793,30 @@ class Agent_Agentflow implements INode { toolInput: toolCall.args, toolOutput }) + + if (toolSpan && analyticHandlers) { + const spanPayload: ICommonObject = { + output: analyticHandlers.safeSerializeForTrace(toolOutput) + } + if (parsedDocs !== undefined) spanPayload.sourceDocuments = analyticHandlers.safeSerializeForTrace(parsedDocs) + if (parsedArtifacts !== undefined) + spanPayload.artifacts = analyticHandlers.safeSerializeForTrace(parsedArtifacts) + analyticHandlers.finishToolSpan(toolSpan, 'FINISHED', spanPayload) + toolSpan = undefined + } } catch (e) { console.error('Error invoking tool:', e) // Enhanced error message for MCP connection issues const enhancedErrorMessage = this.enhanceMcpErrorMessage(e, toolCall.name, nodeData, options) + if (toolSpan && analyticHandlers) { + analyticHandlers.finishToolSpan(toolSpan, 'ERROR', { + error: analyticHandlers.safeSerializeForTrace(enhancedErrorMessage) + }) + toolSpan = undefined + } + usedTools.push({ tool: selectedTool.name, toolInput: toolCall.args, @@ -1789,9 +1864,9 @@ class Agent_Agentflow implements INode { } if (isStreamable) { - newResponse = await this.handleStreamingResponse(sseStreamer, llmNodeInstance, messages, chatId, abortController) + newResponse = await this.handleStreamingResponse(sseStreamer, llmNodeInstance, messages, chatId, options) } else { - newResponse = await llmNodeInstance.invoke(messages, { signal: abortController?.signal }) + newResponse = await llmNodeInstance.invoke(messages, llmCallOptions) // Stream non-streaming response if this is the last node if (isLastNode && sseStreamer) { diff --git a/packages/components/nodes/agentflow/ConditionAgent/ConditionAgent.ts b/packages/components/nodes/agentflow/ConditionAgent/ConditionAgent.ts index 6ec809f9678..15fad11395e 100644 --- a/packages/components/nodes/agentflow/ConditionAgent/ConditionAgent.ts +++ b/packages/components/nodes/agentflow/ConditionAgent/ConditionAgent.ts @@ -1,4 +1,4 @@ -import { AnalyticHandler } from '../../../src/handler' +import { AnalyticHandler, additionalCallbacks } from '../../../src/handler' import { ICommonObject, INode, INodeData, INodeOptionsValue, INodeOutputsValue, INodeParams } from '../../../src/Interface' import { AIMessageChunk, BaseMessageLike } from '@langchain/core/messages' import { @@ -239,6 +239,17 @@ class ConditionAgent_Agentflow implements INode { if (!model) { throw new Error('Model is required') } + // Setup analytics tracing options and attach to options object for easy access + const callbacks = await additionalCallbacks(nodeData, { + ...options, + parentLangfuseTrace: options.parentLangfuseTrace, + parentLangfuseSpan: options.parentLangfuseSpan + }) + const llmCallOptions: ICommonObject = { signal: abortController?.signal } + if (callbacks && callbacks.length > 0) { + llmCallOptions.callbacks = callbacks + } + options._llmCallOptions = llmCallOptions const conditionAgentInput = nodeData.inputs?.conditionAgentInput as string let input = conditionAgentInput || question const conditionAgentInstructions = nodeData.inputs?.conditionAgentInstructions as string @@ -360,7 +371,7 @@ class ConditionAgent_Agentflow implements INode { // Track execution time const startTime = Date.now() - response = await llmNodeInstance.invoke(messages, { signal: abortController?.signal }) + response = await llmNodeInstance.invoke(messages, options._llmCallOptions || { signal: abortController?.signal }) // Calculate execution time const endTime = Date.now() @@ -482,6 +493,7 @@ class ConditionAgent_Agentflow implements INode { runtimeImageMessagesWithFileRef: BaseMessageLike[] pastImageMessagesWithFileRef: BaseMessageLike[] }): Promise { + const llmCallOptions = options._llmCallOptions || { signal: abortController?.signal } const { updatedPastMessages, transformedPastMessages } = await getPastChatHistoryImageMessages(pastChatHistory, options) pastChatHistory = updatedPastMessages pastImageMessagesWithFileRef.push(...transformedPastMessages) @@ -523,12 +535,12 @@ class ConditionAgent_Agentflow implements INode { ) } ], - { signal: abortController?.signal } + llmCallOptions ) messages.push({ role: 'assistant', content: summary.content as string }) } else if (memoryType === 'conversationSummaryBuffer') { // Summary buffer: Summarize messages that exceed token limit - await this.handleSummaryBuffer(messages, pastMessages, llmNodeInstance, nodeData, abortController) + await this.handleSummaryBuffer(messages, pastMessages, llmNodeInstance, nodeData, abortController, options) } else { // Default: Use all messages messages.push(...pastMessages) @@ -549,8 +561,10 @@ class ConditionAgent_Agentflow implements INode { pastMessages: BaseMessageLike[], llmNodeInstance: BaseChatModel, nodeData: INodeData, - abortController: AbortController + abortController: AbortController, + options: ICommonObject ): Promise { + const llmCallOptions = options._llmCallOptions || { signal: abortController?.signal } const maxTokenLimit = (nodeData.inputs?.conditionAgentMemoryMaxTokenLimit as number) || 2000 // Convert past messages to a format suitable for token counting @@ -584,7 +598,7 @@ class ConditionAgent_Agentflow implements INode { content: DEFAULT_SUMMARIZER_TEMPLATE.replace('{conversation}', messagesToSummarizeString) } ], - { signal: abortController?.signal } + llmCallOptions ) // Add summary as a system message at the beginning, then add remaining messages diff --git a/packages/components/nodes/agentflow/LLM/LLM.ts b/packages/components/nodes/agentflow/LLM/LLM.ts index 18f8d187d1a..cbc967e1bd3 100644 --- a/packages/components/nodes/agentflow/LLM/LLM.ts +++ b/packages/components/nodes/agentflow/LLM/LLM.ts @@ -3,7 +3,7 @@ import { ICommonObject, INode, INodeData, INodeOptionsValue, INodeParams, IServe import { AIMessageChunk, BaseMessageLike, MessageContentText } from '@langchain/core/messages' import { DEFAULT_SUMMARIZER_TEMPLATE } from '../prompt' import { z } from 'zod' -import { AnalyticHandler } from '../../../src/handler' +import { AnalyticHandler, additionalCallbacks } from '../../../src/handler' import { ILLMMessage, IStructuredOutput } from '../Interface.Agentflow' import { getPastChatHistoryImageMessages, @@ -346,6 +346,18 @@ class LLM_Agentflow implements INode { throw new Error('Model is required') } + // Setup analytics tracing options and attach to options object for easy access + const callbacks = await additionalCallbacks(nodeData, { + ...options, + parentLangfuseTrace: options.parentLangfuseTrace, + parentLangfuseSpan: options.parentLangfuseSpan + }) + const llmCallOptions: ICommonObject = { signal: abortController?.signal } + if (callbacks && callbacks.length > 0) { + llmCallOptions.callbacks = callbacks + } + options._llmCallOptions = llmCallOptions + // Extract memory and configuration options const enableMemory = nodeData.inputs?.llmEnableMemory as boolean const memoryType = nodeData.inputs?.llmMemoryType as string @@ -453,9 +465,9 @@ class LLM_Agentflow implements INode { const sseStreamer: IServerSideEventStreamer | undefined = options.sseStreamer if (isStreamable) { - response = await this.handleStreamingResponse(sseStreamer, llmNodeInstance, messages, chatId, abortController) + response = await this.handleStreamingResponse(sseStreamer, llmNodeInstance, messages, chatId, options) } else { - response = await llmNodeInstance.invoke(messages, { signal: abortController?.signal }) + response = await llmNodeInstance.invoke(messages, options._llmCallOptions || { signal: abortController?.signal }) // Stream whole response back to UI if this is the last node if (isLastNode && options.sseStreamer) { @@ -620,6 +632,7 @@ class LLM_Agentflow implements INode { runtimeImageMessagesWithFileRef: BaseMessageLike[] pastImageMessagesWithFileRef: BaseMessageLike[] }): Promise { + const llmCallOptions = options._llmCallOptions || { signal: abortController?.signal } const { updatedPastMessages, transformedPastMessages } = await getPastChatHistoryImageMessages(pastChatHistory, options) pastChatHistory = updatedPastMessages pastImageMessagesWithFileRef.push(...transformedPastMessages) @@ -666,12 +679,12 @@ class LLM_Agentflow implements INode { ) } ], - { signal: abortController?.signal } + llmCallOptions ) messages.push({ role: 'assistant', content: summary.content as string }) } else if (memoryType === 'conversationSummaryBuffer') { // Summary buffer: Summarize messages that exceed token limit - await this.handleSummaryBuffer(messages, pastMessages, llmNodeInstance, nodeData, abortController) + await this.handleSummaryBuffer(messages, pastMessages, llmNodeInstance, nodeData, abortController, options) } else { // Default: Use all messages messages.push(...pastMessages) @@ -695,8 +708,10 @@ class LLM_Agentflow implements INode { pastMessages: BaseMessageLike[], llmNodeInstance: BaseChatModel, nodeData: INodeData, - abortController: AbortController + abortController: AbortController, + options: ICommonObject ): Promise { + const llmCallOptions = options._llmCallOptions || { signal: abortController?.signal } const maxTokenLimit = (nodeData.inputs?.llmMemoryMaxTokenLimit as number) || 2000 // Convert past messages to a format suitable for token counting @@ -730,7 +745,7 @@ class LLM_Agentflow implements INode { content: DEFAULT_SUMMARIZER_TEMPLATE.replace('{conversation}', messagesToSummarizeString) } ], - { signal: abortController?.signal } + llmCallOptions ) // Add summary as a system message at the beginning, then add remaining messages @@ -803,12 +818,13 @@ class LLM_Agentflow implements INode { llmNodeInstance: BaseChatModel, messages: BaseMessageLike[], chatId: string, - abortController: AbortController + options: ICommonObject ): Promise { + const llmCallOptions = options._llmCallOptions || { signal: options.abortController?.signal } let response = new AIMessageChunk('') try { - for await (const chunk of await llmNodeInstance.stream(messages, { signal: abortController?.signal })) { + for await (const chunk of await llmNodeInstance.stream(messages, llmCallOptions)) { if (sseStreamer) { let content = '' if (Array.isArray(chunk.content) && chunk.content.length > 0) { diff --git a/packages/components/src/flowCredentialExtractor.ts b/packages/components/src/flowCredentialExtractor.ts index 1f57c391588..d67a8fde4b6 100644 --- a/packages/components/src/flowCredentialExtractor.ts +++ b/packages/components/src/flowCredentialExtractor.ts @@ -7,6 +7,7 @@ interface FlowNode { data: { id: string type: string + category?: string credential?: string inputs?: { modelName?: string @@ -37,6 +38,7 @@ interface CredentialInfo { nodeId: string nodeType: string credentialId: string + visibility?: string[] } /** @@ -54,6 +56,7 @@ interface ModelInfo { interface ExtractionResult { credentials: CredentialInfo[] models: ModelInfo[] + hasPlatformAINodes: boolean } /** @@ -68,7 +71,8 @@ function extractCredentialsAndModels(flowData: FlowData | string): ExtractionRes // Initialize result object const result: ExtractionResult = { credentials: [], - models: [] + models: [], + hasPlatformAINodes: false } // Check if flow has nodes @@ -76,6 +80,11 @@ function extractCredentialsAndModels(flowData: FlowData | string): ExtractionRes return result } + // Check for AAI platform AI nodes + result.hasPlatformAINodes = flow.nodes.some( + (node) => node.data?.type?.startsWith('AAI') && ['Chat Models', 'Embeddings'].includes(node.data.category || '') + ) + // Iterate through nodes flow.nodes.forEach((node) => { if (node.data) { diff --git a/packages/components/src/followUpPrompts.ts b/packages/components/src/followUpPrompts.ts index ecfcfe825a8..78feb1c8157 100644 --- a/packages/components/src/followUpPrompts.ts +++ b/packages/components/src/followUpPrompts.ts @@ -26,7 +26,18 @@ export const generateFollowUpPrompts = async ( const providerConfig = followUpPromptsConfig[followUpPromptsConfig.selectedProvider] if (!providerConfig) return undefined const credentialId = providerConfig.credentialId as string - const credentialData = await getCredentialData(credentialId ?? '', options) + + let credentialData: ICommonObject + try { + credentialData = await getCredentialData(credentialId ?? '', options) + if (!credentialData || Object.keys(credentialData).length === 0) { + console.warn('No credential data found for follow-up prompts, skipping generation') + return undefined + } + } catch (error) { + console.error('Error retrieving credential data for follow-up prompts:', error) + return undefined + } const followUpPromptsPrompt = providerConfig.prompt.replace('{history}', apiMessageContent) switch (followUpPromptsConfig.selectedProvider) { diff --git a/packages/components/src/handler.ts b/packages/components/src/handler.ts index 0fe1526a804..5963464fe69 100644 --- a/packages/components/src/handler.ts +++ b/packages/components/src/handler.ts @@ -27,8 +27,7 @@ import { LunaryHandler } from '@langchain/community/callbacks/handlers/lunary' import { getCredentialData, getCredentialParam, getEnvironmentVariable } from './utils' import { ICommonObject, IDatabaseEntity, INodeData, IServerSideEventStreamer } from './Interface' import { LangWatch, LangWatchSpan, LangWatchTrace, autoconvertTypedValues } from 'langwatch' -import { extractCredentialsAndModels } from './flowCredentialExtractor' -import { In } from 'typeorm' +import { CredentialInfo, extractCredentialsAndModels } from './flowCredentialExtractor' export interface TraceMetadata { stripeCustomerId: string subscriptionTier?: string @@ -37,12 +36,63 @@ export interface TraceMetadata { aiCredentialsOwnership: string [key: string]: any } -import { DataSource } from 'typeorm' +import { DataSource, In } from 'typeorm' import { ChatGenerationChunk } from '@langchain/core/outputs' import { AIMessageChunk, BaseMessageLike } from '@langchain/core/messages' import { Serialized } from '@langchain/core/load/serializable' import { JLINCTracer } from '@jlinc/langchain' +/** + * Apply environment variable overrides for analytics providers + * This allows global analytics configuration via env vars without UI setup + * + * @param analyticConfig - Existing analytics config (may be empty, string, or object) + * @returns Analytics config with env overrides applied + */ +function applyEnvAnalyticsOverrides(analyticConfig?: string | object): any { + let analytic: any = {} + + // Parse existing config if provided + if (typeof analyticConfig === 'string') { + try { + analytic = JSON.parse(analyticConfig) + } catch { + analytic = {} + } + } else if (analyticConfig && typeof analyticConfig === 'object') { + analytic = analyticConfig + } + + // Langfuse env override - takes precedence over UI configuration + if (process.env.LANGFUSE_SECRET_KEY) { + analytic.langFuse = { + status: true, + release: process.env.LANGFUSE_RELEASE ?? process.env.GIT_COMMIT_HASH, + secretKey: process.env.LANGFUSE_SECRET_KEY, + publicKey: process.env.LANGFUSE_PUBLIC_KEY, + endpoint: process.env.LANGFUSE_HOST ?? 'https://cloud.langfuse.com', + sdkIntegration: 'Flowise' + } + } + + // Future providers can be added here: + // if (process.env.LANGSMITH_API_KEY) { analytic.langSmith = { ... } } + // if (process.env.ARIZE_API_KEY) { analytic.arize = { ... } } + + return analytic +} + +/** + * Check if analytics is enabled via UI config or environment variables + * + * @param analyticConfig - Analytics config from chatflow/agentflow + * @returns True if any analytics provider is configured + */ +export function isAnalyticsEnabled(analyticConfig?: string | object): boolean { + const analytic = applyEnvAnalyticsOverrides(analyticConfig) + return Object.keys(analytic).length > 0 +} + interface AgentRun extends Run { actions: AgentAction[] } @@ -488,20 +538,13 @@ class ExtendedLunaryHandler extends LunaryHandler { export const additionalCallbacks = async (nodeData: INodeData, options: ICommonObject) => { try { - if (!options.analytic && !process.env.LANGFUSE_SECRET_KEY) return [] + const analytic = applyEnvAnalyticsOverrides(options.analytic) + if (Object.keys(analytic).length === 0) return [] - const analytic = options.analytic ? JSON.parse(options.analytic) : {} - if (process.env.LANGFUSE_SECRET_KEY) { - analytic.langFuse = { - status: true, - release: process.env.LANGFUSE_RELEASE ?? process.env.GIT_COMMIT_HASH, - secretKey: process.env.LANGFUSE_SECRET_KEY, - publicKey: process.env.LANGFUSE_PUBLIC_KEY, - endpoint: process.env.LANGFUSE_HOST ?? 'https://cloud.langfuse.com', - sdkIntegration: 'Flowise' - } - } const callbacks: any = [] + const parentLangfuseTrace = options.parentLangfuseTrace as LangfuseTraceClient | undefined + const parentLangfuseSpan = options.parentLangfuseSpan as LangfuseSpanClient | undefined + const analyticHandlersInstance = options.analyticHandlers as AnalyticHandler | undefined for (const provider in analytic) { const providerStatus = analytic[provider].status as boolean @@ -556,15 +599,22 @@ export const additionalCallbacks = async (nodeData: INodeData, options: ICommonO .getRepository(options.databaseEntities['ChatFlow']) .findOneBy({ id: options.chatflowid }) - const credentialIds = extractCredentialsAndModels(chatflow.flowData) - const credentials = await options.appDataSource - .getRepository(options.databaseEntities['Credential']) - .findBy({ id: In(credentialIds?.credentials?.map((credential) => credential.credentialId) ?? []) }) - - // console.debug('Credentials::::', credentials) + // Default to platform if no chatflow found let aiCredentialsOwnership = 'platform' - if (credentials.every((credential: { visibility: string[] }) => !credential.visibility?.includes('Platform'))) { - aiCredentialsOwnership = 'user' + + if (chatflow?.flowData) { + const extracted = extractCredentialsAndModels(chatflow.flowData) + + // Fast path: AAI nodes = platform + if (!extracted.hasPlatformAINodes && extracted.credentials.length > 0) { + const credentials = await options.appDataSource + .getRepository(options.databaseEntities['Credential']) + .findBy({ id: In(extracted.credentials.map((c) => c.credentialId)) }) + + aiCredentialsOwnership = credentials.every((c: any) => !c.visibility?.includes('Platform')) + ? 'user' + : 'platform' + } } let langFuseOptions = { @@ -607,7 +657,7 @@ export const additionalCallbacks = async (nodeData: INodeData, options: ICommonO // metadata: metadata // }) - const handler = new CallbackHandler({ + const handlerConfig: ICommonObject = { ...langFuseOptions, metadata: metadata, userId: options?.user?.id, @@ -619,7 +669,34 @@ export const additionalCallbacks = async (nodeData: INodeData, options: ICommonO // BUT Everything gets updatedon the root trace so the attributes and metadata are inconsistent // root: trace, // updateRoot: true - }) + } + + if (parentLangfuseSpan || parentLangfuseTrace) { + handlerConfig.root = parentLangfuseSpan ?? parentLangfuseTrace + handlerConfig.updateRoot = false + + try { + if (parentLangfuseTrace) { + parentLangfuseTrace.update({ + tags: [`Name:${chatflow.name}`], + metadata, + userId: options?.user?.id, + sessionId: options.sessionId, + version: chatflow.updatedDate + }) + } + } catch (err) { + if (process.env.DEBUG === 'true') { + console.error('Error updating Langfuse parent trace metadata:', err) + } + } + + analyticHandlersInstance?.setLangfuseCallbacksActive(true) + } else { + analyticHandlersInstance?.setLangfuseCallbacksActive(false) + } + + const handler = new CallbackHandler(handlerConfig) callbacks.push(handler) //console.debug('Handler added to callbacks.') @@ -756,13 +833,16 @@ export class AnalyticHandler { private analyticsConfig: string | undefined private chatId: string private createdAt: number + private langfuseCallbacksActive = false + private useNodeLevelLangfuseSpans = false private constructor(nodeData: INodeData, options: ICommonObject) { this.nodeData = nodeData this.options = options - this.analyticsConfig = options.analytic + this.analyticsConfig = JSON.stringify(applyEnvAnalyticsOverrides(options.analytic)) this.chatId = options.chatId this.createdAt = Date.now() + this.useNodeLevelLangfuseSpans = Boolean(options.useNodeLevelLangfuseSpans) } static getInstance(nodeData: INodeData, options: ICommonObject): AnalyticHandler { @@ -771,7 +851,8 @@ export class AnalyticHandler { // Reset instance if analytics config changed for this chat const instance = AnalyticHandler.instances.get(chatId) - if (instance?.analyticsConfig !== options.analytic) { + const currentProcessedConfig = JSON.stringify(applyEnvAnalyticsOverrides(options.analytic)) + if (instance?.analyticsConfig !== currentProcessedConfig) { AnalyticHandler.resetInstance(chatId) } @@ -799,9 +880,9 @@ export class AnalyticHandler { if (this.initialized) return try { - if (!this.options.analytic) return + const analytic = applyEnvAnalyticsOverrides(this.options.analytic) + if (Object.keys(analytic).length === 0) return - const analytic = JSON.parse(this.options.analytic) for (const provider in analytic) { const providerStatus = analytic[provider].status as boolean if (providerStatus) { @@ -821,6 +902,22 @@ export class AnalyticHandler { return this.handlers } + setLangfuseCallbacksActive(active: boolean) { + this.langfuseCallbacksActive = active + } + + isLangfuseCallbacksActive(): boolean { + return this.langfuseCallbacksActive + } + + getParentTraceClient(provider: 'langFuse', traceId: string): LangfuseTraceClient | undefined { + if (!traceId) return undefined + const providerHandler = this.handlers[provider] as ICommonObject | undefined + const traces = providerHandler?.trace as Record | undefined + if (!traces) return undefined + return traces[traceId] + } + async initializeProvider(provider: string, providerConfig: any, credentialData: any) { if (provider === 'langSmith') { const langSmithProject = providerConfig.projectName as string @@ -835,9 +932,9 @@ export class AnalyticHandler { this.handlers['langSmith'] = { client, langSmithProject } } else if (provider === 'langFuse') { const release = providerConfig.release as string - const langFuseSecretKey = getCredentialParam('langFuseSecretKey', credentialData, this.nodeData) - const langFusePublicKey = getCredentialParam('langFusePublicKey', credentialData, this.nodeData) - const langFuseEndpoint = getCredentialParam('langFuseEndpoint', credentialData, this.nodeData) + const langFuseSecretKey = providerConfig?.secretKey ?? getCredentialParam('langFuseSecretKey', credentialData, this.nodeData) + const langFusePublicKey = providerConfig?.publicKey ?? getCredentialParam('langFusePublicKey', credentialData, this.nodeData) + const langFuseEndpoint = providerConfig?.endpoint ?? getCredentialParam('langFuseEndpoint', credentialData, this.nodeData) const langfuse = new Langfuse({ secretKey: langFuseSecretKey, @@ -1365,15 +1462,21 @@ export class AnalyticHandler { } if (Object.prototype.hasOwnProperty.call(this.handlers, 'langFuse')) { - const trace: LangfuseTraceClient | undefined = this.handlers['langFuse'].trace[parentIds['langFuse'].trace] - if (trace) { - const generation = trace.generation({ - name, - input: input - }) - this.handlers['langFuse'].generation = { [generation.id]: generation } - returnIds['langFuse'].generation = generation.id - // console.log(`Langfuse generation created: ${generation.id}`) + if (this.langfuseCallbacksActive || this.useNodeLevelLangfuseSpans) { + if (parentIds?.['langFuse']?.trace) { + returnIds['langFuse'].trace = parentIds['langFuse'].trace + } + } else { + const trace: LangfuseTraceClient | undefined = this.handlers['langFuse'].trace[parentIds['langFuse'].trace] + if (trace) { + const generation = trace.generation({ + name, + input: input + }) + this.handlers['langFuse'].generation = { [generation.id]: generation } + returnIds['langFuse'].generation = generation.id + // console.log(`Langfuse generation created: ${generation.id}`) + } } } @@ -1480,12 +1583,16 @@ export class AnalyticHandler { } if (Object.prototype.hasOwnProperty.call(this.handlers, 'langFuse')) { - const generation: LangfuseGenerationClient | undefined = this.handlers['langFuse'].generation[returnIds['langFuse'].generation] - if (generation) { - generation.end({ - output: output - }) - // console.log(`Langfuse generation ended: ${generation.id}`) + if (!this.langfuseCallbacksActive && !this.useNodeLevelLangfuseSpans) { + const generationId = returnIds['langFuse'].generation + const generation: LangfuseGenerationClient | undefined = this.handlers['langFuse'].generation[generationId] + if (generation) { + generation.end({ + output: output + }) + delete this.handlers['langFuse'].generation[generationId] + // console.log(`Langfuse generation ended: ${generation.id}`) + } } } @@ -1555,12 +1662,16 @@ export class AnalyticHandler { } if (Object.prototype.hasOwnProperty.call(this.handlers, 'langFuse')) { - const generation: LangfuseGenerationClient | undefined = this.handlers['langFuse'].generation[returnIds['langFuse'].generation] - if (generation) { - generation.end({ - output: error - }) - // console.log(`Langfuse generation errored: ${generation.id}`) + if (!this.langfuseCallbacksActive && !this.useNodeLevelLangfuseSpans) { + const generationId = returnIds['langFuse'].generation + const generation: LangfuseGenerationClient | undefined = this.handlers['langFuse'].generation[generationId] + if (generation) { + generation.end({ + output: error + }) + delete this.handlers['langFuse'].generation[generationId] + // console.log(`Langfuse generation errored: ${generation.id}`) + } } } @@ -1895,6 +2006,74 @@ export class AnalyticHandler { } } } + + /** + * Safely serialize a value for Langfuse trace payload + * Handles circular references and unserializable values + */ + safeSerializeForTrace(value: any): any { + if (value === undefined || value === null) return value + if (typeof value === 'string' || typeof value === 'number' || typeof value === 'boolean') return value + try { + return JSON.parse(JSON.stringify(value)) + } catch (error) { + console.warn(`Failed to serialize payload for Langfuse trace: ${error instanceof Error ? error.message : String(error)}`) + return '[Unserializable]' + } + } + + /** + * Create a Langfuse tool span for tracing tool execution + * @param parentSpan - Parent Langfuse span to attach this tool span to + * @param toolName - Name of the tool being executed + * @param args - Tool arguments/input + * @param metadata - Additional metadata to attach to the span + * @returns LangfuseSpanClient if successful, undefined otherwise + */ + createToolSpan( + parentSpan: LangfuseSpanClient | undefined, + toolName: string, + args: any, + metadata?: ICommonObject + ): LangfuseSpanClient | undefined { + if (!parentSpan) return undefined + try { + return parentSpan.span({ + name: `Tool:${toolName}`, + metadata: { + toolName, + status: 'IN_PROGRESS', + ...(metadata ?? {}) + }, + input: { + args: this.safeSerializeForTrace(args) + } + }) + } catch (error) { + console.warn(`Failed to create Langfuse tool span for ${toolName}: ${error instanceof Error ? error.message : String(error)}`) + return undefined + } + } + + /** + * Finalize a Langfuse tool span with status and output + * @param span - The Langfuse span to finalize + * @param status - Final status ('FINISHED' or 'ERROR') + * @param payload - Output data to attach to the span + */ + finishToolSpan(span: LangfuseSpanClient | undefined, status: 'FINISHED' | 'ERROR', payload: ICommonObject): void { + if (!span) return + try { + span.update({ + metadata: { + status + } + }) + span.end({ output: payload }) + } catch (error) { + console.warn(`Failed to finalize Langfuse tool span: ${error instanceof Error ? error.message : String(error)}`) + } + } } /** diff --git a/packages/components/src/index.ts b/packages/components/src/index.ts index 470150d3248..56344f1a954 100644 --- a/packages/components/src/index.ts +++ b/packages/components/src/index.ts @@ -12,5 +12,5 @@ export * from './handler' export * from './followUpPrompts' export * from './validator' export * from './agentflowv2Generator' - export { ATLASSIAN_MCP_SERVER_URL } from './constants' +export * from './flowCredentialExtractor' diff --git a/packages/embed b/packages/embed index 09bc07d7450..fa836ede40e 160000 --- a/packages/embed +++ b/packages/embed @@ -1 +1 @@ -Subproject commit 09bc07d745059581ebfeef12dfcf3e42c918798e +Subproject commit fa836ede40e0e74fd7df445e367aeaaa2b3d8cfc diff --git a/packages/embed-react/package.json b/packages/embed-react/package.json index a4a487b4f00..ad8058c7832 100644 --- a/packages/embed-react/package.json +++ b/packages/embed-react/package.json @@ -1,6 +1,6 @@ { "name": "aai-embed-react", - "version": "1.0.7", + "version": "3.0.10", "description": "React library to display flowise chatbot on your website", "main": "dist/index.js", "types": "dist/index.d.ts", @@ -35,7 +35,7 @@ "typescript": "~5.5.4" }, "peerDependencies": { - "aai-embed": "^3.0.0", + "aai-embed": "^3.0.10", "react": "18.x" } } diff --git a/packages/server/src/Interface.ts b/packages/server/src/Interface.ts index 0a2fcd1bded..2262540da57 100644 --- a/packages/server/src/Interface.ts +++ b/packages/server/src/Interface.ts @@ -463,7 +463,7 @@ export interface IExecuteFlowParams extends IPredictionQueueAppServer { fileUploads?: IFileUpload[] uploadedFilesContent?: string isUpsert?: boolean - user: IUser + user?: IUser isRecursive?: boolean parentExecutionId?: string iterationContext?: ICommonObject diff --git a/packages/server/src/aai-utils/billing/config.ts b/packages/server/src/aai-utils/billing/config.ts index 4d8993f7947..5b3ea94434a 100644 --- a/packages/server/src/aai-utils/billing/config.ts +++ b/packages/server/src/aai-utils/billing/config.ts @@ -8,7 +8,8 @@ export const log = console as unknown as Logger export const DEFAULT_CUSTOMER_ID = process.env.BILLING_DEFAULT_STRIPE_CUSTOMER_ID // Load environment variables with defaults -const BILLING_CREDIT_PRICE_USD = parseFloat(process.env.BILLING_CREDIT_PRICE_USD || '0') +// Base rate: $20 for 500,000 credits = $0.00004 per credit +const BILLING_CREDIT_PRICE_USD = parseFloat(process.env.BILLING_CREDIT_PRICE_USD || '0.00004') const MARGIN_MULTIPLIER = parseFloat(process.env.BILLING_MARGIN_MULTIPLIER || '1') const BILLING_PRO_PLAN_CREDITS = parseInt(process.env.BILLING_PRO_PLAN_CREDITS || '500000') const BILLING_FREE_PLAN_CREDITS = parseInt(process.env.BILLING_FREE_PLAN_CREDITS || '10000') diff --git a/packages/server/src/aai-utils/billing/core/BillingService.ts b/packages/server/src/aai-utils/billing/core/BillingService.ts index 761a6645c4c..46d5bcb8273 100644 --- a/packages/server/src/aai-utils/billing/core/BillingService.ts +++ b/packages/server/src/aai-utils/billing/core/BillingService.ts @@ -20,7 +20,7 @@ import { } from './types' import { LangfuseProvider } from '../langfuse/LangfuseProvider' import { StripeProvider } from '../stripe/StripeProvider' -import { log } from '../config' +import { log, BILLING_CONFIG } from '../config' import Stripe from 'stripe' import { MeterEventSummary } from '../stripe/types' @@ -163,7 +163,7 @@ export class BillingService implements BillingProvider { currency: invoice.currency, dueDate: invoice.dueDate, // Calculate total credits used based on the invoice amount - totalCreditsUsed: Math.round(invoice.amount / (0.00004 * 100)) // Assuming $0.001 per credit + totalCreditsUsed: Math.round(invoice.amount / (BILLING_CONFIG.CREDIT_TO_USD * 100)) } } } catch (error) { diff --git a/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts b/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts index 064b5d69ce7..7cf14263da8 100644 --- a/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts +++ b/packages/server/src/aai-utils/billing/langfuse/LangfuseProvider.ts @@ -3,11 +3,16 @@ import { langfuse, log, DEFAULT_CUSTOMER_ID, BILLING_CONFIG } from '../config' import type { GetLangfuseTraceResponse, GetLangfuseTracesResponse } from 'langfuse-core' import { StripeProvider } from '../stripe/StripeProvider' import Stripe from 'stripe' +import { extractCredentialsAndModels } from 'flowise-components' +import { getRunningExpressApp } from '../../../utils/getRunningExpressApp' type Trace = GetLangfuseTracesResponse['data'][number] type FullTrace = GetLangfuseTraceResponse export class LangfuseProvider { + // Cache for flow platform status (resets each sync) + private platformNodeCache = new Map() + async syncUsageToStripe(traceId?: string): Promise { let traces: Trace[] = [] let creditsDataWithTraces: Array<{ creditsData: CreditsData; fullTrace: any }> = [] @@ -48,6 +53,9 @@ export class LangfuseProvider { } catch (error: any) { log.error('Error syncing usage data:', error) failedTraces = [{ traceId: traceId || 'unknown', error: error.message }] + } finally { + // Clear cache after sync + this.platformNodeCache.clear() } return { @@ -205,9 +213,13 @@ export class LangfuseProvider { return undefined } - const metadata = (trace.metadata || {}) as TraceMetadata + const metadata = { + ...((trace.metadata || {}) as TraceMetadata), + aiCredentialsOwnership: 'user' + } as TraceMetadata const { data: fullTrace } = await langfuse.fetchTrace(trace.id) const costs = await this.calculateCosts(fullTrace) + metadata.aiCredentialsOwnership = costs.aiCredentialsOwnership const credits = this.convertCostsToCredits(costs) const modelUsage = await this.getModelUsage(fullTrace as any) @@ -219,15 +231,51 @@ export class LangfuseProvider { } } + private async hasAAIPlatformNodes(chatflowId: string): Promise { + if (!chatflowId?.trim()) return false + + const cached = this.platformNodeCache.get(chatflowId) + if (cached !== undefined) return cached + + try { + const appServer = getRunningExpressApp() + const result = await appServer.AppDataSource.query('SELECT flow_data FROM chat_flow WHERE id = $1', [chatflowId]) + + if (result?.[0]?.flow_data) { + const { hasPlatformAINodes } = extractCredentialsAndModels(result[0].flow_data) + this.platformNodeCache.set(chatflowId, hasPlatformAINodes) + return hasPlatformAINodes + } + } catch (error) { + log.debug('Failed to check AAI nodes', { chatflowId, error }) + } + + this.platformNodeCache.set(chatflowId, false) + return false + } + private async calculateCosts(trace: FullTrace): Promise<{ ai: number compute: number storage: number total: number withMargin: number + aiCredentialsOwnership: string }> { const computeMinutes = trace.latency / (1000 * 60) - const aiCost = trace.metadata.aiCredentialsOwnership === 'platform' ? trace.totalCost : 0 + const metadata = ((trace.metadata || {}) as TraceMetadata) || ({} as TraceMetadata) + + // Validate ownership + let aiCredentialsOwnership = metadata.aiCredentialsOwnership || 'user' + + if (aiCredentialsOwnership !== 'platform' && metadata.chatflowid) { + const hasAAI = await this.hasAAIPlatformNodes(metadata.chatflowid) + if (hasAAI) { + aiCredentialsOwnership = 'platform' + } + } + + const aiCost = aiCredentialsOwnership === 'platform' ? trace.totalCost : 0 const computeCost = computeMinutes * 0.05 const storageCost = 0 const totalBase = aiCost + computeCost + storageCost @@ -238,7 +286,8 @@ export class LangfuseProvider { compute: computeCost, storage: storageCost, total: totalBase, - withMargin + withMargin, + aiCredentialsOwnership } } diff --git a/packages/server/src/services/predictions/index.ts b/packages/server/src/services/predictions/index.ts index b4149a17361..10f6fc95be5 100644 --- a/packages/server/src/services/predictions/index.ts +++ b/packages/server/src/services/predictions/index.ts @@ -9,7 +9,7 @@ const buildChatflow = async (fullRequest: Request) => { try { const { chatId, question: prompt, overrideConfig } = fullRequest.body const chatflowId = fullRequest.params.id - const user = fullRequest.user! + const user = fullRequest.user // Ensure overrideConfig.sessionId is a valid UUID v4 if (overrideConfig?.sessionId) { diff --git a/packages/server/src/utils/buildAgentflow.ts b/packages/server/src/utils/buildAgentflow.ts index 1caba870824..674f81de959 100644 --- a/packages/server/src/utils/buildAgentflow.ts +++ b/packages/server/src/utils/buildAgentflow.ts @@ -11,7 +11,8 @@ import { IMessage, IServerSideEventStreamer, convertChatHistoryToText, - generateFollowUpPrompts + generateFollowUpPrompts, + isAnalyticsEnabled } from 'flowise-components' import { IncomingAgentflowInput, @@ -54,6 +55,7 @@ import { utilAddChatMessage } from './addChatMesage' import { CachePool } from '../CachePool' import { ChatMessage } from '../database/entities/ChatMessage' import { Telemetry } from './telemetry' +import { LangfuseSpanClient, LangfuseTraceClient } from 'langfuse' interface IWaitingNode { nodeId: string @@ -90,7 +92,7 @@ interface IAgentFlowRuntime { } interface IExecuteNodeParams { - user: IUser + user?: IUser nodeId: string reactFlowNode: IReactFlowNode nodes: IReactFlowNode[] @@ -125,10 +127,14 @@ interface IExecuteNodeParams { parentExecutionId?: string isRecursive?: boolean iterationContext?: ICommonObject + parentLangfuseTrace?: LangfuseTraceClient + parentLangfuseSpan?: LangfuseSpanClient } interface IExecuteAgentFlowParams extends Omit { incomingInput: IncomingAgentflowInput + parentLangfuseTrace?: LangfuseTraceClient + parentLangfuseSpan?: LangfuseSpanClient } const MAX_LOOP_COUNT = process.env.MAX_LOOP_COUNT ? parseInt(process.env.MAX_LOOP_COUNT) : 10 @@ -800,12 +806,64 @@ const executeNode = async ({ analyticHandlers, isInternal, isRecursive, - iterationContext + iterationContext, + parentLangfuseTrace, + parentLangfuseSpan }: IExecuteNodeParams): Promise<{ result: any shouldStop?: boolean agentFlowExecutedData?: IAgentflowExecutedData[] }> => { + let langfuseSpan: LangfuseSpanClient | undefined + let spanName = nodeId + const safeSerialize = (value: any) => { + if (value === undefined || value === null) return value + if (typeof value === 'string' || typeof value === 'number' || typeof value === 'boolean') return value + try { + return JSON.parse(JSON.stringify(value)) + } catch (err) { + logger.debug(`Failed to serialize Langfuse span payload for node ${spanName}: ${getErrorMessage(err)}`) + return '[Unserializable]' + } + } + + const buildSpanOutput = (data: any): ICommonObject => { + const spanOutput: ICommonObject = {} + if (data) { + if (data.output !== undefined) spanOutput.output = safeSerialize(data.output) + if (data.state !== undefined) spanOutput.state = safeSerialize(data.state) + if (data.usedTools !== undefined) spanOutput.usedTools = safeSerialize(data.usedTools) + if (data.sourceDocuments !== undefined) spanOutput.sourceDocuments = safeSerialize(data.sourceDocuments) + if (!Object.keys(spanOutput).length) spanOutput.result = safeSerialize(data) + } + return spanOutput + } + + const reflectNodeName = reactFlowNode.data.name + const reflectNodeLabel = reactFlowNode.data.label ?? reflectNodeName + const reflectNodeType = reactFlowNode.data.type ?? reactFlowNode.data.name + + const finishLangfuseSpan = (status: string, payload: ICommonObject, duration?: number) => { + if (!langfuseSpan) return + try { + langfuseSpan.update({ + metadata: { + nodeId, + nodeName: reflectNodeName, + nodeLabel: reflectNodeLabel, + nodeType: reflectNodeType, + status, + spanName, + ...(duration !== undefined ? { durationMs: duration } : {}) + } + }) + langfuseSpan.end({ output: payload }) + } catch (err) { + logger.warn(`Failed to finalize Langfuse span for node ${spanName}: ${getErrorMessage(err)}`) + } + langfuseSpan = undefined + } + let nodeStartTime = 0 try { if (abortController?.signal?.aborted) { throw new Error('Aborted') @@ -930,12 +988,47 @@ const executeNode = async ({ abortController, analyticHandlers, parentTraceIds, + parentLangfuseTrace, + parentLangfuseSpan, humanInputAction, iterationContext } + const langfuseTrace = parentLangfuseTrace + const spanInput: ICommonObject = { + nodeId + } + if (finalInput !== undefined) spanInput.finalInput = safeSerialize(finalInput) + if (reactFlowNodeData.inputs !== undefined) spanInput.resolvedInputs = safeSerialize(reactFlowNodeData.inputs) + spanName = reactFlowNode?.data?.label || reactFlowNode.data.name + if (langfuseTrace) { + try { + langfuseSpan = langfuseTrace.span({ + name: spanName, + metadata: { + nodeId, + nodeName: reactFlowNode.data.name, + nodeType: reactFlowNode.data.type ?? reactFlowNode.data.name, + status: 'IN_PROGRESS' + }, + input: spanInput + }) + } catch (err) { + logger.warn(`Failed to create Langfuse span for node ${spanName}: ${getErrorMessage(err)}`) + } + } + + if (langfuseSpan) { + runParams.parentLangfuseSpan = langfuseSpan + } else if (parentLangfuseSpan) { + runParams.parentLangfuseSpan = parentLangfuseSpan + } + + nodeStartTime = Date.now() + // Execute node let results = await newNodeInstance.run(reactFlowNodeData, finalInput, runParams) + const nodeDuration = Date.now() - nodeStartTime // Handle iteration node with recursive execution if ( @@ -1006,7 +1099,9 @@ const executeNode = async ({ iterationContext: { ...iterationContext, agentflowRuntime - } + }, + parentLangfuseTrace, + parentLangfuseSpan: langfuseSpan }) // Store the result @@ -1122,6 +1217,7 @@ const executeNode = async ({ sseStreamer?.streamActionEvent(chatId, humanInputAction) + finishLangfuseSpan('STOPPED', buildSpanOutput(results), nodeDuration) return { result: results, shouldStop: true, agentFlowExecutedData } } @@ -1169,11 +1265,16 @@ const executeNode = async ({ sseStreamer?.streamActionEvent(chatId, humanInputAction) + finishLangfuseSpan('STOPPED', buildSpanOutput(results), nodeDuration) return { result: results, shouldStop: true, agentFlowExecutedData } } + finishLangfuseSpan('FINISHED', buildSpanOutput(results), nodeDuration) return { result: results, agentFlowExecutedData } } catch (error) { + const spanErrorStatus = getErrorMessage(error).includes('Aborted') ? 'TERMINATED' : 'ERROR' + const errorDuration = nodeStartTime ? Date.now() - nodeStartTime : undefined + finishLangfuseSpan(spanErrorStatus, { error: getErrorMessage(error) }, errorDuration) logger.error(`[server]: Error executing node ${nodeId}: ${getErrorMessage(error)}`) throw error } @@ -1215,10 +1316,23 @@ export const executeAgentFlow = async ({ parentExecutionId, iterationContext, isTool = false, - user + user, + parentLangfuseTrace: inheritedLangfuseTrace, + parentLangfuseSpan }: IExecuteAgentFlowParams) => { logger.debug('\n🚀 Starting flow execution') + const safeTraceSerialize = (value: any) => { + if (value === undefined || value === null) return value + if (typeof value === 'string' || typeof value === 'number' || typeof value === 'boolean') return value + try { + return JSON.parse(JSON.stringify(value)) + } catch (err) { + logger.debug(`Failed to serialize value for Langfuse trace output: ${getErrorMessage(err)}`) + return '[Unserializable]' + } + } + const question = incomingInput.question const form = incomingInput.form let overrideConfig = incomingInput.overrideConfig ?? {} @@ -1437,21 +1551,28 @@ export const executeAgentFlow = async ({ let analyticHandlers: AnalyticHandler | undefined let parentTraceIds: ICommonObject | undefined + let parentLangfuseTrace: LangfuseTraceClient | undefined = inheritedLangfuseTrace try { - if (chatflow.analytic) { + if (isAnalyticsEnabled(chatflow.analytic)) { analyticHandlers = AnalyticHandler.getInstance({ inputs: {} } as any, { appDataSource, databaseEntities, componentNodes, analytic: chatflow.analytic, - chatId + chatId, + useNodeLevelLangfuseSpans: true }) await analyticHandlers.init() parentTraceIds = await analyticHandlers.onChainStart( 'Agentflow', form && Object.keys(form).length > 0 ? JSON.stringify(form) : question || '' ) + + const langfuseTraceId = parentTraceIds?.langFuse?.trace + if (!parentLangfuseTrace && langfuseTraceId && analyticHandlers) { + parentLangfuseTrace = analyticHandlers.getParentTraceClient('langFuse', langfuseTraceId) + } } } catch (error) { logger.error(`[server]: Error initializing analytic handlers: ${getErrorMessage(error)}`) @@ -1486,7 +1607,7 @@ export const executeAgentFlow = async ({ // Execute current node const executionResult = await executeNode({ - user: user!, + user: user, nodeId: currentNode.nodeId, reactFlowNode, nodes, @@ -1519,6 +1640,8 @@ export const executeAgentFlow = async ({ abortController, parentTraceIds, analyticHandlers, + parentLangfuseTrace, + parentLangfuseSpan, isRecursive, iterationContext }) @@ -1676,6 +1799,23 @@ export const executeAgentFlow = async ({ await analyticHandlers.onChainEnd(parentTraceIds, content, true) } + if (parentLangfuseTrace) { + try { + parentLangfuseTrace.update({ + metadata: { + status, + nodeCount: agentFlowExecutedData.length + }, + output: { + finalOutput: safeTraceSerialize(lastNodeOutput), + status + } + }) + } catch (err) { + logger.warn(`Failed to update Langfuse trace metadata: ${getErrorMessage(err)}`) + } + } + if (isRecursive) { return { agentFlowExecutedData, diff --git a/packages/server/src/utils/buildChatflow.ts b/packages/server/src/utils/buildChatflow.ts index 6fd4a6b08f1..74b5e683c43 100644 --- a/packages/server/src/utils/buildChatflow.ts +++ b/packages/server/src/utils/buildChatflow.ts @@ -89,7 +89,7 @@ const initEndingNode = async ({ nodeOverrides, variableOverrides }: { - user: IUser + user?: IUser endingNodeIds: string[] componentNodes: IComponentNodes reactFlowNodes: IReactFlowNode[] @@ -928,7 +928,7 @@ export const utilBuildChatflow = async (req: Request, isInternal: boolean = fals telemetry: appServer.telemetry, cachePool: appServer.cachePool, componentNodes: appServer.nodesPool.componentNodes, - user: req.user!, + user: req.user, isTool // used to disable streaming if incoming request its from ChatflowTool } diff --git a/streaming-demo.html b/streaming-demo.html new file mode 100644 index 00000000000..b7d9634c60f --- /dev/null +++ b/streaming-demo.html @@ -0,0 +1,819 @@ + + + + + + AnswerAI Streaming Demo + + + +
+
+

🤖 AnswerAI Direct Streaming Demo

+

Connect directly to AnswerAI for real-time AI agent streaming (no local service required)

+
+ +
+ +
+

Configuration

+
+ + + Direct AnswerAI endpoint (not your local service) +
+
+ + + Your AnswerAI chatflow ID for domain research +
+
+ + + The domain or topic you want the AI to research +
+
+ + +
+
+ + + +
+
+ + + + + + + + + + + + +
+
+ + + +