From f9477718ccf96ea6b6224a149948959d665937f1 Mon Sep 17 00:00:00 2001 From: Xpert Agent Date: Mon, 16 Mar 2026 06:45:59 +0000 Subject: [PATCH] fix(knowledgebase): fetch copilot when visionModel has copilotId but copilot is not loaded When a vision model is configured with a copilotId but the copilot object is not loaded (e.g., when passed from workflow node entity), the getVisionModel method now fetches the copilot using CopilotGetOneQuery before throwing the 'KBReqVisionModel' error. This fixes the issue where the knowledge base pipeline reports 'The knowledge base requires a visual model to process images' even when a visual model with copilotId is properly configured. --- .../knowledgebase/knowledgebase.service.ts | 1401 +++++++++-------- 1 file changed, 704 insertions(+), 697 deletions(-) diff --git a/packages/server-ai/src/knowledgebase/knowledgebase.service.ts b/packages/server-ai/src/knowledgebase/knowledgebase.service.ts index 8fdae3e9c3..bf6b42c89d 100644 --- a/packages/server-ai/src/knowledgebase/knowledgebase.service.ts +++ b/packages/server-ai/src/knowledgebase/knowledgebase.service.ts @@ -3,37 +3,37 @@ import { Embeddings } from '@langchain/core/embeddings' import { BaseChatModel } from '@langchain/core/language_models/chat_models' import { VectorStore } from '@langchain/core/vectorstores' import { - AiBusinessRole, - channelName, - DocumentMetadata, - genPipelineKnowledgeBaseKey, - genPipelineSourceKey, - IKnowledgebase, - IKnowledgebaseTask, - IKnowledgeDocument, - IWFNKnowledgeBase, - IWFNProcessor, - IWFNSource, - IXpert, - KBDocumentStatusEnum, - KnowledgebaseChannel, - KnowledgebasePermission, - KnowledgebaseTypeEnum, - KnowledgeProviderEnum, - KNOWLEDGE_SOURCES_NAME, - KnowledgeTask, - mapTranslationLanguage, - STATE_VARIABLE_HUMAN, - WorkflowNodeTypeEnum, - XpertTypeEnum, - genXpertTriggerKey, - IWFNTrigger, - KnowledgeStructureEnum, - XpertAgentExecutionStatusEnum, - classificateDocumentCategory, - TCopilotModel, - KnowledgeDocumentMetadata, - IUser + AiBusinessRole, + channelName, + DocumentMetadata, + genPipelineKnowledgeBaseKey, + genPipelineSourceKey, + IKnowledgebase, + IKnowledgebaseTask, + IKnowledgeDocument, + IWFNKnowledgeBase, + IWFNProcessor, + IWFNSource, + IXpert, + KBDocumentStatusEnum, + KnowledgebaseChannel, + KnowledgebasePermission, + KnowledgebaseTypeEnum, + KnowledgeProviderEnum, + KNOWLEDGE_SOURCES_NAME, + KnowledgeTask, + mapTranslationLanguage, + STATE_VARIABLE_HUMAN, + WorkflowNodeTypeEnum, + XpertTypeEnum, + genXpertTriggerKey, + IWFNTrigger, + KnowledgeStructureEnum, + XpertAgentExecutionStatusEnum, + classificateDocumentCategory, + TCopilotModel, + KnowledgeDocumentMetadata, + IUser } from '@metad/contracts' import { getErrorMessage, shortuuid } from '@metad/server-common' import { IntegrationService, PaginationParams, RequestContext } from '@metad/server-core' @@ -42,22 +42,23 @@ import { QueryFailedError } from 'typeorm' import { OnEvent } from '@nestjs/event-emitter' import { InjectRepository } from '@nestjs/typeorm' import { - DocumentSourceRegistry, - DocumentTransformerRegistry, - ImageUnderstandingRegistry, - IRerank, - KnowledgeStrategyRegistry, - TextSplitterRegistry, + DocumentSourceRegistry, + DocumentTransformerRegistry, + ImageUnderstandingRegistry, + IRerank, + KnowledgeStrategyRegistry, + TextSplitterRegistry, } from '@xpert-ai/plugin-sdk' import { t } from 'i18next' import { assign, sortBy } from 'lodash' import { I18nService } from 'nestjs-i18n' import { FindOptionsWhere, In, IsNull, Not, Repository } from 'typeorm' import { - CopilotModelGetChatModelQuery, - CopilotModelGetEmbeddingsQuery, - CopilotModelGetRerankQuery + CopilotModelGetChatModelQuery, + CopilotModelGetEmbeddingsQuery, + CopilotModelGetRerankQuery } from '../copilot-model/queries/index' +import { CopilotGetOneQuery } from '../copilot' import { AiModelNotFoundException, CopilotModelNotFoundException, CopilotNotFoundException } from '../core/errors' import { RagCreateVStoreCommand } from '../rag-vstore' import { XpertWorkspaceBaseService } from '../xpert-workspace' @@ -76,661 +77,667 @@ import { XpertEnqueueTriggerDispatchCommand } from '../xpert/commands' @Injectable() export class KnowledgebaseService extends XpertWorkspaceBaseService { - readonly #logger = new Logger(KnowledgebaseService.name) - - @Inject(I18nService) - private readonly i18nService: I18nService - - @Inject(KnowledgeDocumentService) - private readonly documentService: KnowledgeDocumentService - - @Inject(TextSplitterRegistry) - private readonly textSplitterRegistry: TextSplitterRegistry - - @Inject(DocumentTransformerRegistry) - private readonly docTransformerRegistry: DocumentTransformerRegistry - - @Inject(ImageUnderstandingRegistry) - private readonly understandingRegistry: ImageUnderstandingRegistry - - @Inject(DocumentSourceRegistry) - private readonly docSourceRegistry: DocumentSourceRegistry - - @Inject(KnowledgeStrategyRegistry) - private readonly knowledgeStrategyRegistry: KnowledgeStrategyRegistry - - @Inject(XpertService) - private readonly xpertService: XpertService - - constructor( - @InjectRepository(Knowledgebase) - repository: Repository, - private readonly integrationService: IntegrationService, - private readonly taskService: KnowledgebaseTaskService, - ) { - super(repository) - } - - /** - * Override getAllByWorkspace to include knowledgebases with Public/Organization permissions - * from other workspaces in the same organization - */ - async getAllByWorkspace(workspaceId: string, data: PaginationParams, published: boolean, user: IUser) { - const { relations, order, take } = data ?? {} - let { where } = data ?? {} - where = where ?? {} - const organizationId = RequestContext.getOrganizationId() - - if (workspaceId === 'null' || workspaceId === 'undefined' || !workspaceId) { - where = { - ...(>where), - workspaceId: IsNull(), - createdById: user.id - } - if (published) { - where.publishAt = Not(IsNull()) - } - return this.findAll({ - where, - relations, - order, - take - }) - } else { - const workspace = await this.queryBus.execute(new GetXpertWorkspaceQuery(user, { id: workspaceId })) - if (!workspace) { - throw new NotFoundException(`Not found or no auth for xpert workspace '${workspaceId}'`) - } - - // Build where conditions array to include: - // 1. Knowledgebases that belong to this workspace - // 2. Public knowledgebases from any workspace in the same organization - // 3. Organization knowledgebases from other workspaces in the same organization - const whereConditions: FindOptionsWhere[] = [ - { - ...(>where), - workspaceId: workspaceId - } - ] - - // Add Public knowledgebases from any workspace (excluding those already in this workspace) - // Note: Using Not(In([workspaceId])) to exclude the current workspace - whereConditions.push({ - ...(>where), - permission: KnowledgebasePermission.Public, - organizationId: organizationId, - workspaceId: Not(In([workspaceId])) - }) - - // Add Organization knowledgebases from other workspaces in the same organization - whereConditions.push({ - ...(>where), - permission: KnowledgebasePermission.Organization, - organizationId: organizationId, - workspaceId: Not(In([workspaceId])) - }) - - // Apply published filter if needed - if (published) { - whereConditions.forEach(condition => { - condition.publishAt = Not(IsNull()) - }) - } - - return this.findAll({ - where: whereConditions, - relations, - order, - take - }) - } - } - - async create(entity: Partial) { - // Check name - const exist = await super.findOneOrFailByOptions({ - where: { name: entity.name } - }) - if (exist.success) { - throw new BadRequestException( - this.i18nService.t('xpert.Error.NameExists', { - lang: mapTranslationLanguage(RequestContext.getLanguageCode()) - }) - ) - } - - return await super.create(entity) - } - - async createExternal(entity: Partial) { - // Test external integration - if (!entity.integrationId) { - throw new BadRequestException( - await this.i18nService.t('xpert.Error.ExternalIntegrationRequired', { - lang: mapTranslationLanguage(RequestContext.getLanguageCode()) - }) - ) - } - - await this.searchExternalKnowledgebase(entity, 'test', 1, {}) - - return this.create({ - ...entity, - type: KnowledgebaseTypeEnum.External - }) - } - - async searchExternalKnowledgebase( - entity: Partial, - query: string, - k: number, - filter?: Record - ) { - const integration = await this.integrationService.findOne(entity.integrationId) - const knowledgeStrategy = this.knowledgeStrategyRegistry.get( - integration.provider as unknown as KnowledgeProviderEnum - ) - if (!knowledgeStrategy) { - throw new BadRequestException( - await this.i18nService.t('xpert.Error.KnowledgeStrategyNotFound', { - lang: mapTranslationLanguage(RequestContext.getLanguageCode()), - args: { - provider: integration.provider - } - }) - ) - } - return await knowledgeStrategy.execute(integration, { - query, - k, - filter, - options: { knowledgebaseId: entity.extKnowledgebaseId } - }) - } - - /** - * To solve the problem that Update cannot create OneToOne relation, it is uncertain whether using save to update might pose risks - */ - async update(id: string, entity: Partial) { - const _entity = await super.findOne(id) - - // Check name uniqueness if name is being changed - if (entity.name && entity.name !== _entity.name) { - const tenantId = RequestContext.currentTenantId() - const organizationId = RequestContext.getOrganizationId() - - // Check if another knowledgebase with the same name exists (excluding current one) - const exist = await this.repository.findOne({ - where: { - tenantId, - organizationId, - name: entity.name, - id: Not(id) // Exclude current knowledgebase - } - }) - - if (exist) { - throw new BadRequestException( - this.i18nService.t('xpert.Error.NameExists', { - lang: mapTranslationLanguage(RequestContext.getLanguageCode()) - }) - ) - } - } - - try { - assign(_entity, entity) - return await this.repository.save(_entity) - } catch (error) { - // Catch database unique constraint errors as a fallback - // PostgreSQL error code for unique violation: 23505 - // MySQL error code for duplicate entry: 1062 - if (error instanceof QueryFailedError) { - const driverError = (error as any).driverError - const errorCode = driverError?.code || driverError?.errno - if (errorCode === '23505' || errorCode === 1062 || - error.message?.includes('duplicate key') || - error.message?.includes('UNIQUE constraint') || - error.message?.includes('Duplicate entry')) { - throw new BadRequestException( - this.i18nService.t('xpert.Error.NameExists', { - lang: mapTranslationLanguage(RequestContext.getLanguageCode()) - }) - ) - } - } - // Re-throw other errors - throw error - } - } - - async getTextSplitterStrategies() { - return this.textSplitterRegistry.list().map((strategy) => strategy.meta) - } - - async getDocumentTransformerStrategies() { - return this.docTransformerRegistry.list().map((strategy) => { - return { - meta: strategy.meta, - integration: strategy.permissions?.find((permission) => permission.type === 'integration'), - } - }) - } - - async getUnderstandingStrategies() { - return this.understandingRegistry - .list() - .map((strategy) => ({ - meta: strategy.meta, - requireVisionModel: strategy.permissions?.some((permission) => permission.type === 'llm') - })) - } - - async getDocumentSourceStrategies() { - return this.docSourceRegistry - .list() - .map((strategy) => ({ - meta: strategy.meta, - integration: strategy.permissions?.find((permission) => permission.type === 'integration') - })) - } - - /** - * Test the hitting effect of the Knowledge based on the given query text. - * - * @param id Knowledgebase ID - * @param options Query options - * @returns Document chunks - */ - async test(id: string, options: { query: string; k?: number; filter?: KnowledgeDocumentMetadata }) { - const knowledgebase = await this.findOne(id) - const tenantId = RequestContext.currentTenantId() - const organizationId = RequestContext.getOrganizationId() - - const results = await this.queryBus.execute[]>( - new KnowledgeSearchQuery({ - tenantId, - organizationId, - knowledgebases: [knowledgebase.id], - source: 'hit_testing', - ...options - }) - ) - - return results - } - - /** - * Init a pipeline (xpert) for knowledgebase - * - * @param id - * @returns - */ - async createPipeline(id: string) { - const knowledgebase = await this.findOne(id) - const sourceKey = genPipelineSourceKey() - const knowledgebaseKey = genPipelineKnowledgeBaseKey() - const triggerKey = genXpertTriggerKey() - return await this.xpertService.create({ - name: `${knowledgebase.name} Pipeline - ${shortuuid()}`, - workspaceId: knowledgebase.workspaceId, - type: XpertTypeEnum.Knowledge, - latest: true, - knowledgebase: { - id: knowledgebase.id - }, - agent: { - key: shortuuid(), - options: { - hidden: true - } - }, - draft: { - nodes: [ - { - key: triggerKey, - type: 'workflow', - position: { x: 20, y: 320 }, - entity: { - key: triggerKey, - title: 'Trigger', - type: WorkflowNodeTypeEnum.TRIGGER, - from: 'chat' - } as IWFNTrigger - }, - { - key: sourceKey, - type: 'workflow', - position: { x: 300, y: 320 }, - entity: { - key: sourceKey, - title: 'Documents Source', - type: WorkflowNodeTypeEnum.SOURCE, - provider: 'local-file' - } as IWFNSource - }, - { - key: knowledgebaseKey, - type: 'workflow', - position: { x: 680, y: 320 }, - entity: { - key: knowledgebaseKey, - title: 'Knowledge Base', - type: WorkflowNodeTypeEnum.KNOWLEDGE_BASE, - structure: KnowledgeStructureEnum.General, - } as IWFNKnowledgeBase - } - ], - connections: [ - { - type: 'edge', - key: `${triggerKey}/${sourceKey}`, - from: triggerKey, - to: sourceKey - } - ] - } - }) - } - - async getVisionModel(knowledgebaseId: string, visionModel: TCopilotModel) { - if (!visionModel) { - const knowledgebase = await this.findOne(knowledgebaseId, { relations: ['visionModel', 'visionModel.copilot'] }) - visionModel = knowledgebase.visionModel - } - const copilot = visionModel?.copilot - if (!copilot) { - throw new BadRequestException(t('server-ai:Error.KBReqVisionModel')) - } - const chatModel = await this.queryBus.execute( - new CopilotModelGetChatModelQuery(copilot, visionModel, { - usageCallback: (token) => { - // execution.tokens += (token ?? 0) - } - }) - ) - - return chatModel - } - - async getVectorStore(knowledgebaseId: IKnowledgebase | string, requiredEmbeddings = false) { - let knowledgebase: IKnowledgebase - if (typeof knowledgebaseId === 'string') { - if (requiredEmbeddings) { - knowledgebase = await this.findOne(knowledgebaseId, { - relations: [ - 'rerankModel', - 'rerankModel.copilot', - 'rerankModel.copilot.modelProvider', - 'copilotModel', - 'copilotModel.copilot', - 'copilotModel.copilot.modelProvider', - 'documents' - ] - }) - } else { - knowledgebase = await this.findOne(knowledgebaseId, { relations: ['copilotModel'] }) - } - } else { - knowledgebase = knowledgebaseId - } - - const copilotModel = knowledgebase.copilotModel - if (requiredEmbeddings && !copilotModel) { - throw new CopilotModelNotFoundException( - await this.i18nService.t('rag.Error.KnowledgebaseNoModel', { - lang: mapTranslationLanguage(RequestContext.getLanguageCode()), - args: { - knowledgebase: knowledgebase.name - } - }) - ) - } - const copilot = copilotModel?.copilot - if (requiredEmbeddings && !copilot) { - throw new CopilotNotFoundException(`Copilot not set for knowledgebase '${knowledgebase.name}'`) - } - - let embeddings = null - if (copilotModel && copilot?.modelProvider) { - embeddings = await this.queryBus.execute( - new CopilotModelGetEmbeddingsQuery(copilot, copilotModel, { - tokenCallback: (token) => { - // execution.tokens += (token ?? 0) - } - }) - ) - } - - if (requiredEmbeddings && !embeddings) { - throw new AiModelNotFoundException( - `Embeddings model '${copilotModel.model || copilot?.copilotModel?.model}' not found for knowledgebase '${knowledgebase.name}'` - ) - } - - let rerankModel: IRerank = null - if (knowledgebase.rerankModel) { - rerankModel = await this.queryBus.execute( - new CopilotModelGetRerankQuery(knowledgebase.rerankModel.copilot, knowledgebase.rerankModel, { - tokenCallback: (token) => { - // execution.tokens += (token ?? 0) - } - }) - ) - if (!rerankModel) { - throw new AiModelNotFoundException( - `Rerank model '${knowledgebase.rerankModel.model || knowledgebase.rerankModel.copilot?.copilotModel?.model}' not found for knowledgebase '${knowledgebase.name}'` - ) - } - } - - const store = await this.commandBus.execute( - new RagCreateVStoreCommand(embeddings, { - collectionName: knowledgebase.id - }) - ) - const vStore = new KnowledgeDocumentStore(knowledgebase, store, rerankModel) - - // const vectorStore = new KnowledgeDocumentVectorStore(knowledgebase, this.pgPool, embeddings, rerankModel) - - // // Create table for vector store if not exist - // await vectorStore.ensureTableInDatabase() - - return vStore - } - - async similaritySearch( - query: string, - options?: { - k?: number - filter?: VectorStore['FilterType'] - score?: number - tenantId?: string - organizationId?: string - knowledgebases?: string[] - } - ) { - const { knowledgebases, k, score, filter } = options ?? {} - const tenantId = options?.tenantId ?? RequestContext.currentTenantId() - const organizationId = options?.organizationId ?? RequestContext.getOrganizationId() - const result = await this.findAll({ - where: { - tenantId, - organizationId, - id: knowledgebases ? In(knowledgebases) : Not(IsNull()) - } - }) - const _knowledgebases = result.items - - const documents: { doc: DocumentInterface>; score: number }[] = [] - const kbs = await Promise.all( - _knowledgebases.map((kb) => { - return this.getVectorStore(kb.id, true).then((vectorStore) => { - return vectorStore.similaritySearchWithScore(query, k, filter) - }) - }) - ) - - kbs.forEach((kb) => { - kb.forEach(([doc, score]) => { - documents.push({ doc, score }) - }) - }) - - return sortBy(documents, 'score', 'desc') - .filter(({ score: _score }) => 1 - _score >= (score ?? 0.1)) - .slice(0, k) - .map(({ doc }) => doc) - } - - async maxMarginalRelevanceSearch( - query: string, - options?: { - role?: AiBusinessRole - k: number - filter: Record - tenantId?: string - organizationId?: string - } - ) { - // - } - - /** - * Handle knowledgebase related xpert published event, update knowledgebase config from knowledge pipeline - * - * @param xpert The knowledgebase related xpert - * @returns - */ - @OnEvent(EventName_XpertPublished) - async handle(xpert: IXpert) { - if (xpert.type !== XpertTypeEnum.Knowledge || !xpert.knowledgebase?.id) return - - const knowledgebaseNode = xpert.graph.nodes.find( - (node) => node.type === 'workflow' && node.entity.type === WorkflowNodeTypeEnum.KNOWLEDGE_BASE - ) - if (!knowledgebaseNode) return - - const knowledgebaseEntity = knowledgebaseNode.entity as IWFNKnowledgeBase - - await this.update(xpert.knowledgebase.id, { - copilotModel: knowledgebaseEntity.copilotModel, - rerankModel: knowledgebaseEntity.rerankModel, - visionModel: knowledgebaseEntity.visionModel - }) - } - - // Pipeline - - /** - * Create a new task for a knowledgebase. - * If the task status is running, start immediately. - */ - async createTask(knowledgebaseId: string, task: Partial) { - const {id} = await this.taskService.createTask(knowledgebaseId, task) - const _task = await this.taskService.findOne(id, { relations: ['documents'] }) - if (task.status === 'running') { - _task.documents.forEach((doc) => { - doc.status = KBDocumentStatusEnum.WAITING - doc.processMsg = null - }) - // Update task status to running - await this.documentService.save(_task.documents) - // Start immediately - await this.processTask(knowledgebaseId, _task.id, { - sources: _task.documents?.reduce((obj, doc) => ({ - ...obj, - [doc.sourceConfig.key]: { - documents: [...(obj[doc.sourceConfig.key]?.documents ?? []), doc.id] - } - }), {}), - stage: 'prod' - }) - } - return _task - } - - async getTask(knowledgebaseId: string, taskId: string, params?: PaginationParams) { - const where = { ...(params?.where ?? {}), id: taskId, knowledgebaseId } as FindOptionsWhere - - return this.taskService.findOneByOptions({ - ...(params ?? {}), - where - }) - } - - /** - * Process a task, start the knowledge ingestion pipeline - */ - async processTask(knowledgebaseId: string, taskId: string, inputs: { sources?: { [key: string]: { documents: string[] } }; stage: 'preview' | 'prod'; options?: any; isDraft?: boolean }) { - const kb = await this.findOne(knowledgebaseId, { relations: ['pipeline'] }) - const execution = await this.commandBus.execute( - new XpertAgentExecutionUpsertCommand({ - // threadId: conversation.threadId, - status: XpertAgentExecutionStatusEnum.RUNNING - }) - ) - await this.taskService.update(taskId, { status: 'running', executionId: execution.id }) - const sources = inputs.sources ? Object.keys(inputs.sources) : null - - await this.commandBus.execute( - new XpertEnqueueTriggerDispatchCommand(kb.pipelineId, RequestContext.currentUserId(), { - [STATE_VARIABLE_HUMAN]: { - input: 'Process knowledges pipeline', - }, - [KnowledgebaseChannel]: { - knowledgebaseId: knowledgebaseId, - [KnowledgeTask]: taskId, - [KNOWLEDGE_SOURCES_NAME]: sources, - stage: inputs.stage - }, - ...(sources ?? []).reduce((obj, key) => ({ ...obj, [channelName(key)]: { documents: inputs.sources[key].documents } }), {}) - }, { - isDraft: inputs.isDraft, - from: 'knowledge', - executionId: execution.id - }) - ) - } - - async previewFile(id: string, filePath: string) { - const extension = filePath.split('.').pop().toLowerCase() - try { - const results = await this.transformDocuments(id, {provider: 'default', config: {}} as IWFNProcessor, false, [ - { - filePath, - name: filePath.split('/').pop(), - type: extension, - category: classificateDocumentCategory({type: extension}) - } - ]) - return results[0].chunks - } catch (error) { - throw new InternalServerErrorException(getErrorMessage(error)) - } - } - - async transformDocuments(knowledgebaseId: string, entity: IWFNProcessor, isDraft: boolean, input: Partial>[]) { - const strategy = this.docTransformerRegistry.get(entity.provider) - - const permissions = await this.commandBus.execute(new PluginPermissionsCommand(strategy.permissions, { - knowledgebaseId: knowledgebaseId, - integrationId: entity.integrationId, - folder: '' - })) - - const results = await strategy.transformDocuments(input, { - ...(entity.config ?? {}), - stage: isDraft ? 'test' : 'prod', - tempDir: new VolumeClient({ - tenantId: RequestContext.currentTenantId(), - catalog: 'knowledges', - userId: RequestContext.currentUserId(), - knowledgeId: knowledgebaseId - }).getVolumePath('tmp'), - permissions - }) - - return results - } + readonly #logger = new Logger(KnowledgebaseService.name) + + @Inject(I18nService) + private readonly i18nService: I18nService + + @Inject(KnowledgeDocumentService) + private readonly documentService: KnowledgeDocumentService + + @Inject(TextSplitterRegistry) + private readonly textSplitterRegistry: TextSplitterRegistry + + @Inject(DocumentTransformerRegistry) + private readonly docTransformerRegistry: DocumentTransformerRegistry + + @Inject(ImageUnderstandingRegistry) + private readonly understandingRegistry: ImageUnderstandingRegistry + + @Inject(DocumentSourceRegistry) + private readonly docSourceRegistry: DocumentSourceRegistry + + @Inject(KnowledgeStrategyRegistry) + private readonly knowledgeStrategyRegistry: KnowledgeStrategyRegistry + + @Inject(XpertService) + private readonly xpertService: XpertService + + constructor( + @InjectRepository(Knowledgebase) + repository: Repository, + private readonly integrationService: IntegrationService, + private readonly taskService: KnowledgebaseTaskService, + ) { + super(repository) + } + + /** + * Override getAllByWorkspace to include knowledgebases with Public/Organization permissions + * from other workspaces in the same organization + */ + async getAllByWorkspace(workspaceId: string, data: PaginationParams, published: boolean, user: IUser) { + const { relations, order, take } = data ?? {} + let { where } = data ?? {} + where = where ?? {} + const organizationId = RequestContext.getOrganizationId() + + if (workspaceId === 'null' || workspaceId === 'undefined' || !workspaceId) { + where = { + ...(>where), + workspaceId: IsNull(), + createdById: user.id + } + if (published) { + where.publishAt = Not(IsNull()) + } + return this.findAll({ + where, + relations, + order, + take + }) + } else { + const workspace = await this.queryBus.execute(new GetXpertWorkspaceQuery(user, { id: workspaceId })) + if (!workspace) { + throw new NotFoundException(`Not found or no auth for xpert workspace '${workspaceId}'`) + } + + // Build where conditions array to include: + // 1. Knowledgebases that belong to this workspace + // 2. Public knowledgebases from any workspace in the same organization + // 3. Organization knowledgebases from other workspaces in the same organization + const whereConditions: FindOptionsWhere[] = [ + { + ...(>where), + workspaceId: workspaceId + } + ] + + // Add Public knowledgebases from any workspace (excluding those already in this workspace) + // Note: Using Not(In([workspaceId])) to exclude the current workspace + whereConditions.push({ + ...(>where), + permission: KnowledgebasePermission.Public, + organizationId: organizationId, + workspaceId: Not(In([workspaceId])) + }) + + // Add Organization knowledgebases from other workspaces in the same organization + whereConditions.push({ + ...(>where), + permission: KnowledgebasePermission.Organization, + organizationId: organizationId, + workspaceId: Not(In([workspaceId])) + }) + + // Apply published filter if needed + if (published) { + whereConditions.forEach(condition => { + condition.publishAt = Not(IsNull()) + }) + } + + return this.findAll({ + where: whereConditions, + relations, + order, + take + }) + } + } + + async create(entity: Partial) { + // Check name + const exist = await super.findOneOrFailByOptions({ + where: { name: entity.name } + }) + if (exist.success) { + throw new BadRequestException( + this.i18nService.t('xpert.Error.NameExists', { + lang: mapTranslationLanguage(RequestContext.getLanguageCode()) + }) + ) + } + + return await super.create(entity) + } + + async createExternal(entity: Partial) { + // Test external integration + if (!entity.integrationId) { + throw new BadRequestException( + await this.i18nService.t('xpert.Error.ExternalIntegrationRequired', { + lang: mapTranslationLanguage(RequestContext.getLanguageCode()) + }) + ) + } + + await this.searchExternalKnowledgebase(entity, 'test', 1, {}) + + return this.create({ + ...entity, + type: KnowledgebaseTypeEnum.External + }) + } + + async searchExternalKnowledgebase( + entity: Partial, + query: string, + k: number, + filter?: Record + ) { + const integration = await this.integrationService.findOne(entity.integrationId) + const knowledgeStrategy = this.knowledgeStrategyRegistry.get( + integration.provider as unknown as KnowledgeProviderEnum + ) + if (!knowledgeStrategy) { + throw new BadRequestException( + await this.i18nService.t('xpert.Error.KnowledgeStrategyNotFound', { + lang: mapTranslationLanguage(RequestContext.getLanguageCode()), + args: { + provider: integration.provider + } + }) + ) + } + return await knowledgeStrategy.execute(integration, { + query, + k, + filter, + options: { knowledgebaseId: entity.extKnowledgebaseId } + }) + } + + /** + * To solve the problem that Update cannot create OneToOne relation, it is uncertain whether using save to update might pose risks + */ + async update(id: string, entity: Partial) { + const _entity = await super.findOne(id) + + // Check name uniqueness if name is being changed + if (entity.name && entity.name !== _entity.name) { + const tenantId = RequestContext.currentTenantId() + const organizationId = RequestContext.getOrganizationId() + + // Check if another knowledgebase with the same name exists (excluding current one) + const exist = await this.repository.findOne({ + where: { + tenantId, + organizationId, + name: entity.name, + id: Not(id) // Exclude current knowledgebase + } + }) + + if (exist) { + throw new BadRequestException( + this.i18nService.t('xpert.Error.NameExists', { + lang: mapTranslationLanguage(RequestContext.getLanguageCode()) + }) + ) + } + } + + try { + assign(_entity, entity) + return await this.repository.save(_entity) + } catch (error) { + // Catch database unique constraint errors as a fallback + // PostgreSQL error code for unique violation: 23505 + // MySQL error code for duplicate entry: 1062 + if (error instanceof QueryFailedError) { + const driverError = (error as any).driverError + const errorCode = driverError?.code || driverError?.errno + if (errorCode === '23505' || errorCode === 1062 || + error.message?.includes('duplicate key') || + error.message?.includes('UNIQUE constraint') || + error.message?.includes('Duplicate entry')) { + throw new BadRequestException( + this.i18nService.t('xpert.Error.NameExists', { + lang: mapTranslationLanguage(RequestContext.getLanguageCode()) + }) + ) + } + } + // Re-throw other errors + throw error + } + } + + async getTextSplitterStrategies() { + return this.textSplitterRegistry.list().map((strategy) => strategy.meta) + } + + async getDocumentTransformerStrategies() { + return this.docTransformerRegistry.list().map((strategy) => { + return { + meta: strategy.meta, + integration: strategy.permissions?.find((permission) => permission.type === 'integration'), + } + }) + } + + async getUnderstandingStrategies() { + return this.understandingRegistry + .list() + .map((strategy) => ({ + meta: strategy.meta, + requireVisionModel: strategy.permissions?.some((permission) => permission.type === 'llm') + })) + } + + async getDocumentSourceStrategies() { + return this.docSourceRegistry + .list() + .map((strategy) => ({ + meta: strategy.meta, + integration: strategy.permissions?.find((permission) => permission.type === 'integration') + })) + } + + /** + * Test the hitting effect of the Knowledge based on the given query text. + * + * @param id Knowledgebase ID + * @param options Query options + * @returns Document chunks + */ + async test(id: string, options: { query: string; k?: number; filter?: KnowledgeDocumentMetadata }) { + const knowledgebase = await this.findOne(id) + const tenantId = RequestContext.currentTenantId() + const organizationId = RequestContext.getOrganizationId() + + const results = await this.queryBus.execute[]>( + new KnowledgeSearchQuery({ + tenantId, + organizationId, + knowledgebases: [knowledgebase.id], + source: 'hit_testing', + ...options + }) + ) + + return results + } + + /** + * Init a pipeline (xpert) for knowledgebase + * + * @param id + * @returns + */ + async createPipeline(id: string) { + const knowledgebase = await this.findOne(id) + const sourceKey = genPipelineSourceKey() + const knowledgebaseKey = genPipelineKnowledgeBaseKey() + const triggerKey = genXpertTriggerKey() + return await this.xpertService.create({ + name: `${knowledgebase.name} Pipeline - ${shortuuid()}`, + workspaceId: knowledgebase.workspaceId, + type: XpertTypeEnum.Knowledge, + latest: true, + knowledgebase: { + id: knowledgebase.id + }, + agent: { + key: shortuuid(), + options: { + hidden: true + } + }, + draft: { + nodes: [ + { + key: triggerKey, + type: 'workflow', + position: { x: 20, y: 320 }, + entity: { + key: triggerKey, + title: 'Trigger', + type: WorkflowNodeTypeEnum.TRIGGER, + from: 'chat' + } as IWFNTrigger + }, + { + key: sourceKey, + type: 'workflow', + position: { x: 300, y: 320 }, + entity: { + key: sourceKey, + title: 'Documents Source', + type: WorkflowNodeTypeEnum.SOURCE, + provider: 'local-file' + } as IWFNSource + }, + { + key: knowledgebaseKey, + type: 'workflow', + position: { x: 680, y: 320 }, + entity: { + key: knowledgebaseKey, + title: 'Knowledge Base', + type: WorkflowNodeTypeEnum.KNOWLEDGE_BASE, + structure: KnowledgeStructureEnum.General, + } as IWFNKnowledgeBase + } + ], + connections: [ + { + type: 'edge', + key: `${triggerKey}/${sourceKey}`, + from: triggerKey, + to: sourceKey + } + ] + } + }) + } + + async getVisionModel(knowledgebaseId: string, visionModel: TCopilotModel) { + if (!visionModel) { + const knowledgebase = await this.findOne(knowledgebaseId, { relations: ['visionModel', 'visionModel.copilot'] }) + visionModel = knowledgebase.visionModel + } + + // If copilotId is set but copilot object is not loaded, fetch it + let copilot = visionModel?.copilot + if (!copilot && visionModel?.copilotId) { + copilot = await this.queryBus.execute(new CopilotGetOneQuery(RequestContext.currentTenantId(), visionModel.copilotId, ['modelProvider'])) + } + + if (!copilot) { + throw new BadRequestException(t('server-ai:Error.KBReqVisionModel')) + } + const chatModel = await this.queryBus.execute( + new CopilotModelGetChatModelQuery(copilot, visionModel, { + usageCallback: (token) => { + // execution.tokens += (token ?? 0) + } + }) + ) + + return chatModel + } + + async getVectorStore(knowledgebaseId: IKnowledgebase | string, requiredEmbeddings = false) { + let knowledgebase: IKnowledgebase + if (typeof knowledgebaseId === 'string') { + if (requiredEmbeddings) { + knowledgebase = await this.findOne(knowledgebaseId, { + relations: [ + 'rerankModel', + 'rerankModel.copilot', + 'rerankModel.copilot.modelProvider', + 'copilotModel', + 'copilotModel.copilot', + 'copilotModel.copilot.modelProvider', + 'documents' + ] + }) + } else { + knowledgebase = await this.findOne(knowledgebaseId, { relations: ['copilotModel'] }) + } + } else { + knowledgebase = knowledgebaseId + } + + const copilotModel = knowledgebase.copilotModel + if (requiredEmbeddings && !copilotModel) { + throw new CopilotModelNotFoundException( + await this.i18nService.t('rag.Error.KnowledgebaseNoModel', { + lang: mapTranslationLanguage(RequestContext.getLanguageCode()), + args: { + knowledgebase: knowledgebase.name + } + }) + ) + } + const copilot = copilotModel?.copilot + if (requiredEmbeddings && !copilot) { + throw new CopilotNotFoundException(`Copilot not set for knowledgebase '${knowledgebase.name}'`) + } + + let embeddings = null + if (copilotModel && copilot?.modelProvider) { + embeddings = await this.queryBus.execute( + new CopilotModelGetEmbeddingsQuery(copilot, copilotModel, { + tokenCallback: (token) => { + // execution.tokens += (token ?? 0) + } + }) + ) + } + + if (requiredEmbeddings && !embeddings) { + throw new AiModelNotFoundException( + `Embeddings model '${copilotModel.model || copilot?.copilotModel?.model}' not found for knowledgebase '${knowledgebase.name}'` + ) + } + + let rerankModel: IRerank = null + if (knowledgebase.rerankModel) { + rerankModel = await this.queryBus.execute( + new CopilotModelGetRerankQuery(knowledgebase.rerankModel.copilot, knowledgebase.rerankModel, { + tokenCallback: (token) => { + // execution.tokens += (token ?? 0) + } + }) + ) + if (!rerankModel) { + throw new AiModelNotFoundException( + `Rerank model '${knowledgebase.rerankModel.model || knowledgebase.rerankModel.copilot?.copilotModel?.model}' not found for knowledgebase '${knowledgebase.name}'` + ) + } + } + + const store = await this.commandBus.execute( + new RagCreateVStoreCommand(embeddings, { + collectionName: knowledgebase.id + }) + ) + const vStore = new KnowledgeDocumentStore(knowledgebase, store, rerankModel) + + // const vectorStore = new KnowledgeDocumentVectorStore(knowledgebase, this.pgPool, embeddings, rerankModel) + + // // Create table for vector store if not exist + // await vectorStore.ensureTableInDatabase() + + return vStore + } + + async similaritySearch( + query: string, + options?: { + k?: number + filter?: VectorStore['FilterType'] + score?: number + tenantId?: string + organizationId?: string + knowledgebases?: string[] + } + ) { + const { knowledgebases, k, score, filter } = options ?? {} + const tenantId = options?.tenantId ?? RequestContext.currentTenantId() + const organizationId = options?.organizationId ?? RequestContext.getOrganizationId() + const result = await this.findAll({ + where: { + tenantId, + organizationId, + id: knowledgebases ? In(knowledgebases) : Not(IsNull()) + } + }) + const _knowledgebases = result.items + + const documents: { doc: DocumentInterface>; score: number }[] = [] + const kbs = await Promise.all( + _knowledgebases.map((kb) => { + return this.getVectorStore(kb.id, true).then((vectorStore) => { + return vectorStore.similaritySearchWithScore(query, k, filter) + }) + }) + ) + + kbs.forEach((kb) => { + kb.forEach(([doc, score]) => { + documents.push({ doc, score }) + }) + }) + + return sortBy(documents, 'score', 'desc') + .filter(({ score: _score }) => 1 - _score >= (score ?? 0.1)) + .slice(0, k) + .map(({ doc }) => doc) + } + + async maxMarginalRelevanceSearch( + query: string, + options?: { + role?: AiBusinessRole + k: number + filter: Record + tenantId?: string + organizationId?: string + } + ) { + // + } + + /** + * Handle knowledgebase related xpert published event, update knowledgebase config from knowledge pipeline + * + * @param xpert The knowledgebase related xpert + * @returns + */ + @OnEvent(EventName_XpertPublished) + async handle(xpert: IXpert) { + if (xpert.type !== XpertTypeEnum.Knowledge || !xpert.knowledgebase?.id) return + + const knowledgebaseNode = xpert.graph.nodes.find( + (node) => node.type === 'workflow' && node.entity.type === WorkflowNodeTypeEnum.KNOWLEDGE_BASE + ) + if (!knowledgebaseNode) return + + const knowledgebaseEntity = knowledgebaseNode.entity as IWFNKnowledgeBase + + await this.update(xpert.knowledgebase.id, { + copilotModel: knowledgebaseEntity.copilotModel, + rerankModel: knowledgebaseEntity.rerankModel, + visionModel: knowledgebaseEntity.visionModel + }) + } + + // Pipeline + + /** + * Create a new task for a knowledgebase. + * If the task status is running, start immediately. + */ + async createTask(knowledgebaseId: string, task: Partial) { + const {id} = await this.taskService.createTask(knowledgebaseId, task) + const _task = await this.taskService.findOne(id, { relations: ['documents'] }) + if (task.status === 'running') { + _task.documents.forEach((doc) => { + doc.status = KBDocumentStatusEnum.WAITING + doc.processMsg = null + }) + // Update task status to running + await this.documentService.save(_task.documents) + // Start immediately + await this.processTask(knowledgebaseId, _task.id, { + sources: _task.documents?.reduce((obj, doc) => ({ + ...obj, + [doc.sourceConfig.key]: { + documents: [...(obj[doc.sourceConfig.key]?.documents ?? []), doc.id] + } + }), {}), + stage: 'prod' + }) + } + return _task + } + + async getTask(knowledgebaseId: string, taskId: string, params?: PaginationParams) { + const where = { ...(params?.where ?? {}), id: taskId, knowledgebaseId } as FindOptionsWhere + + return this.taskService.findOneByOptions({ + ...(params ?? {}), + where + }) + } + + /** + * Process a task, start the knowledge ingestion pipeline + */ + async processTask(knowledgebaseId: string, taskId: string, inputs: { sources?: { [key: string]: { documents: string[] } }; stage: 'preview' | 'prod'; options?: any; isDraft?: boolean }) { + const kb = await this.findOne(knowledgebaseId, { relations: ['pipeline'] }) + const execution = await this.commandBus.execute( + new XpertAgentExecutionUpsertCommand({ + // threadId: conversation.threadId, + status: XpertAgentExecutionStatusEnum.RUNNING + }) + ) + await this.taskService.update(taskId, { status: 'running', executionId: execution.id }) + const sources = inputs.sources ? Object.keys(inputs.sources) : null + + await this.commandBus.execute( + new XpertEnqueueTriggerDispatchCommand(kb.pipelineId, RequestContext.currentUserId(), { + [STATE_VARIABLE_HUMAN]: { + input: 'Process knowledges pipeline', + }, + [KnowledgebaseChannel]: { + knowledgebaseId: knowledgebaseId, + [KnowledgeTask]: taskId, + [KNOWLEDGE_SOURCES_NAME]: sources, + stage: inputs.stage + }, + ...(sources ?? []).reduce((obj, key) => ({ ...obj, [channelName(key)]: { documents: inputs.sources[key].documents } }), {}) + }, { + isDraft: inputs.isDraft, + from: 'knowledge', + executionId: execution.id + }) + ) + } + + async previewFile(id: string, filePath: string) { + const extension = filePath.split('.').pop().toLowerCase() + try { + const results = await this.transformDocuments(id, {provider: 'default', config: {}} as IWFNProcessor, false, [ + { + filePath, + name: filePath.split('/').pop(), + type: extension, + category: classificateDocumentCategory({type: extension}) + } + ]) + return results[0].chunks + } catch (error) { + throw new InternalServerErrorException(getErrorMessage(error)) + } + } + + async transformDocuments(knowledgebaseId: string, entity: IWFNProcessor, isDraft: boolean, input: Partial>[]) { + const strategy = this.docTransformerRegistry.get(entity.provider) + + const permissions = await this.commandBus.execute(new PluginPermissionsCommand(strategy.permissions, { + knowledgebaseId: knowledgebaseId, + integrationId: entity.integrationId, + folder: '' + })) + + const results = await strategy.transformDocuments(input, { + ...(entity.config ?? {}), + stage: isDraft ? 'test' : 'prod', + tempDir: new VolumeClient({ + tenantId: RequestContext.currentTenantId(), + catalog: 'knowledges', + userId: RequestContext.currentUserId(), + knowledgeId: knowledgebaseId + }).getVolumePath('tmp'), + permissions + }) + + return results + } }