Skip to content

feat: "fault tolerance" prevent agent execution error on failed retrieving #4812

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 6 commits into
base: main
Choose a base branch
from
Open

feat: "fault tolerance" prevent agent execution error on failed retrieving #4812

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
873273d
feat: "fault tolerance" prevent agent execution error on store statue…
ahmedrowaihi Jul 8, 2025
9037a50
refactor: enhance error handling in document store interactions
ahmedrowaihi Jul 9, 2025
3256f20
fix: update TypeScript configuration to target ES2022 and enhance err…
ahmedrowaihi Jul 9, 2025
9749b18
fix: enhance DocumentStoreError cause trace
ahmedrowaihi Jul 9, 2025
0534d18
Merge branch 'FlowiseAI:main' into main
ahmedrowaihi Jul 23, 2025
3aefbe8
Merge branch 'main' into main
ahmedrowaihi Aug 29, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
115 changes: 67 additions & 48 deletions packages/components/nodes/agentflow/Agent/Agent.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ import { Tool } from '@langchain/core/tools'
import { ARTIFACTS_PREFIX, SOURCE_DOCUMENTS_PREFIX, TOOL_ARGS_PREFIX } from '../../../src/agents'
import { flatten } from 'lodash'
import zodToJsonSchema from 'zod-to-json-schema'
import { getErrorMessage } from '../../../src/error'
import { DocumentStoreError, getErrorMessage } from '../../../src/error'
import { DataSource } from 'typeorm'
import {
getPastChatHistoryImageMessages,
Expand Down Expand Up @@ -573,64 +573,83 @@ class Agent_Agentflow implements INode {
const knowledgeBases = nodeData.inputs?.agentKnowledgeDocumentStores as IKnowledgeBase[]
if (knowledgeBases && knowledgeBases.length > 0) {
for (const knowledgeBase of knowledgeBases) {
const nodeInstanceFilePath = options.componentNodes['retrieverTool'].filePath as string
const nodeModule = await import(nodeInstanceFilePath)
const newRetrieverToolNodeInstance = new nodeModule.nodeClass()
const [storeId, storeName] = knowledgeBase.documentStore.split(':')

const docStoreVectorInstanceFilePath = options.componentNodes['documentStoreVS'].filePath as string
const docStoreVectorModule = await import(docStoreVectorInstanceFilePath)
const newDocStoreVectorInstance = new docStoreVectorModule.nodeClass()
const docStoreVectorInstance = await newDocStoreVectorInstance.init(
{
try {
const nodeInstanceFilePath = options.componentNodes['retrieverTool'].filePath as string
const nodeModule = await import(nodeInstanceFilePath)
const newRetrieverToolNodeInstance = new nodeModule.nodeClass()
const [storeId, storeName] = knowledgeBase.documentStore.split(':')

const docStoreVectorInstanceFilePath = options.componentNodes['documentStoreVS'].filePath as string
const docStoreVectorModule = await import(docStoreVectorInstanceFilePath)
const newDocStoreVectorInstance = new docStoreVectorModule.nodeClass()
let docStoreVectorInstance
try {
docStoreVectorInstance = await newDocStoreVectorInstance.init(
{
...nodeData,
inputs: {
...nodeData.inputs,
selectedStore: storeId
},
outputs: {
output: 'retriever'
}
},
'',
options
)
} catch (error) {
if (error instanceof DocumentStoreError) {
throw error
}
throw new DocumentStoreError(
error.message,
storeId,
error instanceof Error ? { cause: error.cause } : undefined
)
}
const newRetrieverToolNodeData = {
...nodeData,
inputs: {
...nodeData.inputs,
selectedStore: storeId
},
outputs: {
output: 'retriever'
name: storeName
.toLowerCase()
.replace(/ /g, '_')
.replace(/[^a-z0-9_-]/g, ''),
description: knowledgeBase.docStoreDescription,
retriever: docStoreVectorInstance,
returnSourceDocuments: knowledgeBase.returnSourceDocuments
}
},
'',
options
)

const newRetrieverToolNodeData = {
...nodeData,
inputs: {
...nodeData.inputs,
}
const retrieverToolInstance = await newRetrieverToolNodeInstance.init(newRetrieverToolNodeData, '', options)
toolsInstance.push(retrieverToolInstance as Tool)
const jsonSchema = zodToJsonSchema(retrieverToolInstance.schema)
if (jsonSchema.$schema) {
delete jsonSchema.$schema
}
const componentNode = options.componentNodes['retrieverTool']
availableTools.push({
name: storeName
.toLowerCase()
.replace(/ /g, '_')
.replace(/[^a-z0-9_-]/g, ''),
description: knowledgeBase.docStoreDescription,
retriever: docStoreVectorInstance,
returnSourceDocuments: knowledgeBase.returnSourceDocuments
schema: jsonSchema,
toolNode: {
label: componentNode?.label || retrieverToolInstance.name,
name: componentNode?.name || retrieverToolInstance.name
}
})
} catch (error) {
if (error instanceof DocumentStoreError) {
console.warn(
`Failed to initialize document store ${knowledgeBase.documentStore}, skipping:`,
getErrorMessage(error)
)
continue
}
throw error
}
const retrieverToolInstance = await newRetrieverToolNodeInstance.init(newRetrieverToolNodeData, '', options)

toolsInstance.push(retrieverToolInstance as Tool)

const jsonSchema = zodToJsonSchema(retrieverToolInstance.schema)
if (jsonSchema.$schema) {
delete jsonSchema.$schema
}
const componentNode = options.componentNodes['retrieverTool']

availableTools.push({
name: storeName
.toLowerCase()
.replace(/ /g, '_')
.replace(/[^a-z0-9_-]/g, ''),
description: knowledgeBase.docStoreDescription,
schema: jsonSchema,
toolNode: {
label: componentNode?.label || retrieverToolInstance.name,
name: componentNode?.name || retrieverToolInstance.name
}
})
}
}

Expand Down
60 changes: 40 additions & 20 deletions packages/components/nodes/agentflow/Retriever/Retriever.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import { updateFlowState } from '../utils'
import { DataSource } from 'typeorm'
import { BaseRetriever } from '@langchain/core/retrievers'
import { Document } from '@langchain/core/documents'
import { DocumentStoreError } from '../../../src/error'

interface IKnowledgeBase {
documentStore: string
Expand Down Expand Up @@ -152,27 +153,46 @@ class Retriever_Agentflow implements INode {
const knowledgeBases = nodeData.inputs?.retrieverKnowledgeDocumentStores as IKnowledgeBase[]
if (knowledgeBases && knowledgeBases.length > 0) {
for (const knowledgeBase of knowledgeBases) {
const [storeId, _] = knowledgeBase.documentStore.split(':')

const docStoreVectorInstanceFilePath = options.componentNodes['documentStoreVS'].filePath as string
const docStoreVectorModule = await import(docStoreVectorInstanceFilePath)
const newDocStoreVectorInstance = new docStoreVectorModule.nodeClass()
const docStoreVectorInstance = (await newDocStoreVectorInstance.init(
{
...nodeData,
inputs: {
...nodeData.inputs,
selectedStore: storeId
},
outputs: {
output: 'retriever'
try {
const [storeId, _] = knowledgeBase.documentStore.split(':')

const docStoreVectorInstanceFilePath = options.componentNodes['documentStoreVS'].filePath as string
const docStoreVectorModule = await import(docStoreVectorInstanceFilePath)
const newDocStoreVectorInstance = new docStoreVectorModule.nodeClass()
let docStoreVectorInstance
try {
docStoreVectorInstance = (await newDocStoreVectorInstance.init(
{
...nodeData,
inputs: {
...nodeData.inputs,
selectedStore: storeId
},
outputs: {
output: 'retriever'
}
},
'',
options
)) as BaseRetriever
} catch (error) {
if (error instanceof DocumentStoreError) {
throw error
}
},
'',
options
)) as BaseRetriever

docs = await docStoreVectorInstance.invoke(retrieverQuery || input, { signal: abortController?.signal })
throw new DocumentStoreError(error.message, storeId, error instanceof Error ? { cause: error.cause } : undefined)
}
const storeDocs = await docStoreVectorInstance.invoke(retrieverQuery || input, { signal: abortController?.signal })
docs.push(...storeDocs)
} catch (error) {
if (error instanceof DocumentStoreError) {
console.warn(
`Document store ${knowledgeBase.documentStore} unavailable, continuing with other stores:`,
error.message
)
continue
}
throw error
}
}
}

Expand Down
39 changes: 26 additions & 13 deletions packages/components/nodes/tools/RetrieverTool/RetrieverTool.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ import { getBaseClasses, resolveFlowObjValue } from '../../../src/utils'
import { SOURCE_DOCUMENTS_PREFIX } from '../../../src/agents'
import { RunnableConfig } from '@langchain/core/runnables'
import { VectorStoreRetriever } from '@langchain/core/vectorstores'
import { getErrorMessage } from '../../../src/error'

const howToUse = `Add additional filters to vector store. You can also filter with flow config, including the current "state":
- \`$flow.sessionId\`
Expand Down Expand Up @@ -193,20 +194,32 @@ class Retriever_Tools implements INode {
const flow = { chatflowId: options.chatflowid }

const func = async ({ input }: { input: string }, _?: CallbackManagerForToolRun, flowConfig?: IFlowConfig) => {
if (retrieverToolMetadataFilter) {
const flowObj = flowConfig

const metadatafilter =
typeof retrieverToolMetadataFilter === 'object' ? retrieverToolMetadataFilter : JSON.parse(retrieverToolMetadataFilter)
const newMetadataFilter = resolveFlowObjValue(metadatafilter, flowObj)

const vectorStore = (retriever as VectorStoreRetriever<any>).vectorStore
vectorStore.filter = newMetadataFilter
try {
if (retrieverToolMetadataFilter) {
const flowObj = flowConfig
const metadatafilter =
typeof retrieverToolMetadataFilter === 'object'
? retrieverToolMetadataFilter
: JSON.parse(retrieverToolMetadataFilter)
const newMetadataFilter = resolveFlowObjValue(metadatafilter, flowObj)
const vectorStore = (retriever as VectorStoreRetriever<any>).vectorStore
vectorStore.filter = newMetadataFilter
}
const docs = await retriever.invoke(input)
const content = docs.map((doc) => doc.pageContent).join('\n\n')
const sourceDocuments = JSON.stringify(docs)
return returnSourceDocuments ? content + SOURCE_DOCUMENTS_PREFIX + sourceDocuments : content
} catch (error) {
const errorMessage = getErrorMessage(error)
const isDocStoreError =
errorMessage &&
(errorMessage.includes('document store') || errorMessage.includes('vector store') || errorMessage.includes('retriever'))
if (isDocStoreError) {
console.warn('Document store retrieval failed, returning fallback response:', getErrorMessage(error))
return 'Knowledge base temporarily unavailable. Proceeding with general knowledge.'
}
throw error
}
const docs = await retriever.invoke(input)
const content = docs.map((doc) => doc.pageContent).join('\n\n')
const sourceDocuments = JSON.stringify(docs)
return returnSourceDocuments ? content + SOURCE_DOCUMENTS_PREFIX + sourceDocuments : content
}

const schema = z.object({
Expand Down
82 changes: 47 additions & 35 deletions packages/components/nodes/vectorstores/DocumentStoreVS/DocStoreVector.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import { ICommonObject, IDatabaseEntity, INode, INodeData, INodeOptionsValue, INodeOutputsValue, INodeParams } from '../../../src/Interface'
import { DataSource } from 'typeorm'
import { DocumentStoreError, getErrorMessage } from '../../../src/error'

class DocStore_VectorStores implements INode {
label: string
Expand Down Expand Up @@ -73,45 +74,56 @@ class DocStore_VectorStores implements INode {
}

async init(nodeData: INodeData, _: string, options: ICommonObject): Promise<any> {
const selectedStore = nodeData.inputs?.selectedStore as string
const appDataSource = options.appDataSource as DataSource
const databaseEntities = options.databaseEntities as IDatabaseEntity
const output = nodeData.outputs?.output as string

const entity = await appDataSource.getRepository(databaseEntities['DocumentStore']).findOneBy({ id: selectedStore })
if (!entity) {
return { error: 'Store not found' }
}
const data: ICommonObject = {}
data.output = output

// Prepare Embeddings Instance
const embeddingConfig = JSON.parse(entity.embeddingConfig)
data.embeddingName = embeddingConfig.name
data.embeddingConfig = embeddingConfig.config
let embeddingObj = await _createEmbeddingsObject(options.componentNodes, data, options)
if (!embeddingObj) {
return { error: 'Failed to create EmbeddingObj' }
}
try {
const selectedStore = nodeData.inputs?.selectedStore as string
const appDataSource = options.appDataSource as DataSource
const databaseEntities = options.databaseEntities as IDatabaseEntity
const output = nodeData.outputs?.output as string

// Prepare Vector Store Instance
const vsConfig = JSON.parse(entity.vectorStoreConfig)
data.vectorStoreName = vsConfig.name
data.vectorStoreConfig = vsConfig.config
if (data.inputs) {
data.vectorStoreConfig = { ...vsConfig.config, ...data.inputs }
}
const entity = await appDataSource.getRepository(databaseEntities['DocumentStore']).findOneBy({ id: selectedStore })
if (!entity) {
throw new DocumentStoreError('Store not found', selectedStore)
}
const data: ICommonObject = {}
data.output = output

// Prepare Embeddings Instance
const embeddingConfig = JSON.parse(entity.embeddingConfig)
data.embeddingName = embeddingConfig.name
data.embeddingConfig = embeddingConfig.config
let embeddingObj = await _createEmbeddingsObject(options.componentNodes, data, options)
if (!embeddingObj) {
throw new DocumentStoreError('Failed to create EmbeddingObj', selectedStore)
}

// Prepare Vector Store Instance
const vsConfig = JSON.parse(entity.vectorStoreConfig)
data.vectorStoreName = vsConfig.name
data.vectorStoreConfig = vsConfig.config
if (data.inputs) {
data.vectorStoreConfig = { ...vsConfig.config, ...data.inputs }
}

// Prepare Vector Store Node Data
const vStoreNodeData = _createVectorStoreNodeData(options.componentNodes, data, embeddingObj)
// Prepare Vector Store Node Data
const vStoreNodeData = _createVectorStoreNodeData(options.componentNodes, data, embeddingObj)

// Finally create the Vector Store or Retriever object (data.output)
const vectorStoreObj = await _createVectorStoreObject(options.componentNodes, data)
const retrieverOrVectorStore = await vectorStoreObj.init(vStoreNodeData, '', options)
if (!retrieverOrVectorStore) {
return { error: 'Failed to create vectorStore' }
// Finally create the Vector Store or Retriever object (data.output)
const vectorStoreObj = await _createVectorStoreObject(options.componentNodes, data)
const retrieverOrVectorStore = await vectorStoreObj.init(vStoreNodeData, '', options)
if (!retrieverOrVectorStore) {
throw new DocumentStoreError('Failed to create vectorStore', selectedStore)
}
return retrieverOrVectorStore
} catch (error) {
if (error instanceof DocumentStoreError) {
throw error
}
throw new DocumentStoreError(
getErrorMessage(error),
nodeData.inputs?.selectedStore,
error instanceof Error ? { cause: error.cause } : undefined
)
}
return retrieverOrVectorStore
}
}

Expand Down
9 changes: 9 additions & 0 deletions packages/components/src/error.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,3 +23,12 @@ const toErrorWithMessage = (maybeError: unknown): ErrorWithMessage => {
export const getErrorMessage = (error: unknown) => {
return toErrorWithMessage(error).message
}

export class DocumentStoreError extends Error {
public storeId?: string
constructor(message: string, storeId?: string, errorOptions?: ErrorOptions) {
super(`Document store error: ${message}`, errorOptions)
this.name = 'DocumentStoreError'
this.storeId = storeId
}
}
Loading