From e8deeb25cff10b8d17393de3632b6febcee8c336 Mon Sep 17 00:00:00 2001 From: Anuj Verma <42962743+Ashes47@users.noreply.github.com> Date: Sun, 28 Jan 2024 19:45:19 +0530 Subject: [PATCH 1/5] Update MongoDBMemory.ts --- .../memory/MongoDBMemory/MongoDBMemory.ts | 145 +++++++++--------- 1 file changed, 76 insertions(+), 69 deletions(-) diff --git a/packages/components/nodes/memory/MongoDBMemory/MongoDBMemory.ts b/packages/components/nodes/memory/MongoDBMemory/MongoDBMemory.ts index c593c20d..b35de5ae 100644 --- a/packages/components/nodes/memory/MongoDBMemory/MongoDBMemory.ts +++ b/packages/components/nodes/memory/MongoDBMemory/MongoDBMemory.ts @@ -1,9 +1,19 @@ -import { MongoClient, Collection, Document } from 'mongodb' -import { MongoDBChatMessageHistory } from 'langchain/stores/message/mongodb' -import { BufferMemory, BufferMemoryInput } from 'langchain/memory' -import { mapStoredMessageToChatMessage, AIMessage, HumanMessage, BaseMessage } from 'langchain/schema' -import { convertBaseMessagetoIMessage, getBaseClasses, getCredentialData, getCredentialParam } from '../../../src/utils' -import { FlowiseMemory, ICommonObject, IMessage, INode, INodeData, INodeParams, MemoryMethods, MessageType } from '../../../src/Interface' +import { MongoClient, Collection, Document } from 'mongodb'; +import { MongoDBChatMessageHistory } from 'langchain/stores/message/mongodb'; +import { BufferMemory, BufferMemoryInput } from 'langchain/memory'; +import { mapStoredMessageToChatMessage, AIMessage, HumanMessage, BaseMessage } from 'langchain/schema'; +import { convertBaseMessagetoIMessage, getBaseClasses, getCredentialData, getCredentialParam } from '../../../src/utils'; +import { FlowiseMemory, ICommonObject, IMessage, INode, INodeData, INodeParams, MemoryMethods, MessageType } from '../../../src/Interface'; + +let mongoClientSingleton = null; + +const getMongoClient = async (mongoDBConnectUrl) => { + if (!mongoClientSingleton) { + mongoClientSingleton = new MongoClient(mongoDBConnectUrl, { useNewUrlParser: true, useUnifiedTopology: true }); + await mongoClientSingleton.connect(); + } + return mongoClientSingleton; +}; class MongoDB_Memory implements INode { label: string @@ -18,20 +28,20 @@ class MongoDB_Memory implements INode { inputs: INodeParams[] constructor() { - this.label = 'MongoDB Atlas Chat Memory' - this.name = 'MongoDBAtlasChatMemory' - this.version = 1.0 - this.type = 'MongoDBAtlasChatMemory' - this.icon = 'mongodb.svg' - this.category = 'Memory' - this.description = 'Stores the conversation in MongoDB Atlas' - this.baseClasses = [this.type, ...getBaseClasses(BufferMemory)] + this.label = 'MongoDB Atlas Chat Memory'; + this.name = 'MongoDBAtlasChatMemory'; + this.version = 1.0; + this.type = 'MongoDBAtlasChatMemory'; + this.icon = 'mongodb.svg'; + this.category = 'Memory'; + this.description = 'Stores the conversation in MongoDB Atlas'; + this.baseClasses = [this.type, ...getBaseClasses(BufferMemory)]; this.credential = { label: 'Connect Credential', name: 'credential', type: 'credential', credentialNames: ['mongoDBUrlApi'] - } + }; this.inputs = [ { label: 'Database', @@ -49,8 +59,7 @@ class MongoDB_Memory implements INode { label: 'Session Id', name: 'sessionId', type: 'string', - description: - 'If not specified, a random id will be used. Learn more', + description: 'If not specified, a random id will be used. Learn more', default: '', additionalParams: true, optional: true @@ -62,128 +71,126 @@ class MongoDB_Memory implements INode { default: 'chat_history', additionalParams: true } - ] + ]; } async init(nodeData: INodeData, _: string, options: ICommonObject): Promise { - return initializeMongoDB(nodeData, options) + return initializeMongoDB(nodeData, options); } } const initializeMongoDB = async (nodeData: INodeData, options: ICommonObject): Promise => { - const databaseName = nodeData.inputs?.databaseName as string - const collectionName = nodeData.inputs?.collectionName as string - const memoryKey = nodeData.inputs?.memoryKey as string - const sessionId = nodeData.inputs?.sessionId as string + const databaseName = nodeData.inputs?.databaseName as string; + const collectionName = nodeData.inputs?.collectionName as string; + const memoryKey = nodeData.inputs?.memoryKey as string; + const sessionId = nodeData.inputs?.sessionId as string; - const credentialData = await getCredentialData(nodeData.credential ?? '', options) - const mongoDBConnectUrl = getCredentialParam('mongoDBConnectUrl', credentialData, nodeData) + const credentialData = await getCredentialData(nodeData.credential ?? '', options); + const mongoDBConnectUrl = getCredentialParam('mongoDBConnectUrl', credentialData, nodeData); - const client = new MongoClient(mongoDBConnectUrl) - await client.connect() - - const collection = client.db(databaseName).collection(collectionName) + const client = await getMongoClient(mongoDBConnectUrl); + const collection = client.db(databaseName).collection(collectionName); const mongoDBChatMessageHistory = new MongoDBChatMessageHistory({ collection, sessionId - }) + }); mongoDBChatMessageHistory.getMessages = async (): Promise => { const document = await collection.findOne({ sessionId: (mongoDBChatMessageHistory as any).sessionId - }) - const messages = document?.messages || [] - return messages.map(mapStoredMessageToChatMessage) - } + }); + const messages = document?.messages || []; + return messages.map(mapStoredMessageToChatMessage); + }; mongoDBChatMessageHistory.addMessage = async (message: BaseMessage): Promise => { - const messages = [message].map((msg) => msg.toDict()) + const messages = [message].map((msg) => msg.toDict()); await collection.updateOne( { sessionId: (mongoDBChatMessageHistory as any).sessionId }, { $push: { messages: { $each: messages } } }, { upsert: true } - ) - } + ); + }; mongoDBChatMessageHistory.clear = async (): Promise => { - await collection.deleteOne({ sessionId: (mongoDBChatMessageHistory as any).sessionId }) - } + await collection.deleteOne({ sessionId: (mongoDBChatMessageHistory as any).sessionId }); + }; return new BufferMemoryExtended({ memoryKey: memoryKey ?? 'chat_history', chatHistory: mongoDBChatMessageHistory, sessionId, collection - }) -} + }); +}; interface BufferMemoryExtendedInput { - collection: Collection - sessionId: string + collection: Collection; + sessionId: string; } class BufferMemoryExtended extends FlowiseMemory implements MemoryMethods { - sessionId = '' - collection: Collection + sessionId = ''; + collection: Collection; constructor(fields: BufferMemoryInput & BufferMemoryExtendedInput) { - super(fields) - this.sessionId = fields.sessionId - this.collection = fields.collection + super(fields); + this.sessionId = fields.sessionId; + this.collection = fields.collection; } async getChatMessages(overrideSessionId = '', returnBaseMessages = false): Promise { - if (!this.collection) return [] + if (!this.collection) return []; - const id = overrideSessionId ?? this.sessionId - const document = await this.collection.findOne({ sessionId: id }) - const messages = document?.messages || [] - const baseMessages = messages.map(mapStoredMessageToChatMessage) - return returnBaseMessages ? baseMessages : convertBaseMessagetoIMessage(baseMessages) + const id = overrideSessionId ?? this.sessionId; + const document = await this.collection.findOne({ sessionId: id }); + const messages = document?.messages || []; + const baseMessages = messages.map(mapStoredMessageToChatMessage); + return returnBaseMessages ? baseMessages : convertBaseMessagetoIMessage(baseMessages); } async addChatMessages(msgArray: { text: string; type: MessageType }[], overrideSessionId = ''): Promise { - if (!this.collection) return + if (!this.collection) return; - const id = overrideSessionId ?? this.sessionId - const input = msgArray.find((msg) => msg.type === 'userMessage') - const output = msgArray.find((msg) => msg.type === 'apiMessage') + const id = overrideSessionId ?? this.sessionId; + const input = msgArray.find((msg) => msg.type === 'userMessage'); + const output = msgArray.find((msg) => msg.type === 'apiMessage'); if (input) { - const newInputMessage = new HumanMessage(input.text) - const messageToAdd = [newInputMessage].map((msg) => msg.toDict()) + const newInputMessage = new HumanMessage(input.text); + const messageToAdd = [newInputMessage].map((msg) => msg.toDict()); await this.collection.updateOne( { sessionId: id }, { $push: { messages: { $each: messageToAdd } } }, { upsert: true } - ) + ); } if (output) { - const newOutputMessage = new AIMessage(output.text) - const messageToAdd = [newOutputMessage].map((msg) => msg.toDict()) + const newOutputMessage = new AIMessage(output.text); + const messageToAdd = [newOutputMessage].map((msg) => msg.toDict()); await this.collection.updateOne( { sessionId: id }, { $push: { messages: { $each: messageToAdd } } }, { upsert: true } - ) + ); } } async clearChatMessages(overrideSessionId = ''): Promise { - if (!this.collection) return + if (!this.collection) return; - const id = overrideSessionId ?? this.sessionId - await this.collection.deleteOne({ sessionId: id }) - await this.clear() + const id = overrideSessionId ?? this.sessionId; + await this.collection.deleteOne({ sessionId: id }); + await this.clear(); } } -module.exports = { nodeClass: MongoDB_Memory } +module.exports = { nodeClass: MongoDB_Memory }; From e154461f1dac59415f867ed5acee7e0f6d195cdb Mon Sep 17 00:00:00 2001 From: Anuj Verma <42962743+Ashes47@users.noreply.github.com> Date: Sun, 28 Jan 2024 21:33:09 +0530 Subject: [PATCH 2/5] Update MongoDBMemory.ts --- .../memory/MongoDBMemory/MongoDBMemory.ts | 22 +++++++++---------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/packages/components/nodes/memory/MongoDBMemory/MongoDBMemory.ts b/packages/components/nodes/memory/MongoDBMemory/MongoDBMemory.ts index b35de5ae..c2c7ada6 100644 --- a/packages/components/nodes/memory/MongoDBMemory/MongoDBMemory.ts +++ b/packages/components/nodes/memory/MongoDBMemory/MongoDBMemory.ts @@ -1,19 +1,19 @@ -import { MongoClient, Collection, Document } from 'mongodb'; -import { MongoDBChatMessageHistory } from 'langchain/stores/message/mongodb'; -import { BufferMemory, BufferMemoryInput } from 'langchain/memory'; -import { mapStoredMessageToChatMessage, AIMessage, HumanMessage, BaseMessage } from 'langchain/schema'; -import { convertBaseMessagetoIMessage, getBaseClasses, getCredentialData, getCredentialParam } from '../../../src/utils'; -import { FlowiseMemory, ICommonObject, IMessage, INode, INodeData, INodeParams, MemoryMethods, MessageType } from '../../../src/Interface'; +import { MongoClient, Collection, Document } from 'mongodb' +import { MongoDBChatMessageHistory } from 'langchain/stores/message/mongodb' +import { BufferMemory, BufferMemoryInput } from 'langchain/memory' +import { mapStoredMessageToChatMessage, AIMessage, HumanMessage, BaseMessage } from 'langchain/schema' +import { convertBaseMessagetoIMessage, getBaseClasses, getCredentialData, getCredentialParam } from '../../../src/utils' +import { FlowiseMemory, ICommonObject, IMessage, INode, INodeData, INodeParams, MemoryMethods, MessageType } from '../../../src/Interface' -let mongoClientSingleton = null; +let mongoClientSingleton = null const getMongoClient = async (mongoDBConnectUrl) => { if (!mongoClientSingleton) { - mongoClientSingleton = new MongoClient(mongoDBConnectUrl, { useNewUrlParser: true, useUnifiedTopology: true }); - await mongoClientSingleton.connect(); + mongoClientSingleton = new MongoClient(mongoDBConnectUrl, { useNewUrlParser: true, useUnifiedTopology: true }) + await mongoClientSingleton.connect() } - return mongoClientSingleton; -}; + return mongoClientSingleton +} class MongoDB_Memory implements INode { label: string From 36ab1681ac455a5cea1482e98d9765eaf3042071 Mon Sep 17 00:00:00 2001 From: Ashes47 Date: Sun, 28 Jan 2024 21:39:22 +0530 Subject: [PATCH 3/5] lint --- .../memory/MongoDBMemory/MongoDBMemory.ts | 121 +++++++++--------- 1 file changed, 61 insertions(+), 60 deletions(-) diff --git a/packages/components/nodes/memory/MongoDBMemory/MongoDBMemory.ts b/packages/components/nodes/memory/MongoDBMemory/MongoDBMemory.ts index c2c7ada6..3c2903f3 100644 --- a/packages/components/nodes/memory/MongoDBMemory/MongoDBMemory.ts +++ b/packages/components/nodes/memory/MongoDBMemory/MongoDBMemory.ts @@ -28,20 +28,20 @@ class MongoDB_Memory implements INode { inputs: INodeParams[] constructor() { - this.label = 'MongoDB Atlas Chat Memory'; - this.name = 'MongoDBAtlasChatMemory'; - this.version = 1.0; - this.type = 'MongoDBAtlasChatMemory'; - this.icon = 'mongodb.svg'; - this.category = 'Memory'; - this.description = 'Stores the conversation in MongoDB Atlas'; - this.baseClasses = [this.type, ...getBaseClasses(BufferMemory)]; + this.label = 'MongoDB Atlas Chat Memory' + this.name = 'MongoDBAtlasChatMemory' + this.version = 1.0 + this.type = 'MongoDBAtlasChatMemory' + this.icon = 'mongodb.svg' + this.category = 'Memory' + this.description = 'Stores the conversation in MongoDB Atlas' + this.baseClasses = [this.type, ...getBaseClasses(BufferMemory)] this.credential = { label: 'Connect Credential', name: 'credential', type: 'credential', credentialNames: ['mongoDBUrlApi'] - }; + } this.inputs = [ { label: 'Database', @@ -59,7 +59,8 @@ class MongoDB_Memory implements INode { label: 'Session Id', name: 'sessionId', type: 'string', - description: 'If not specified, a random id will be used. Learn more', + description: + 'If not specified, a random id will be used. Learn more', default: '', additionalParams: true, optional: true @@ -71,126 +72,126 @@ class MongoDB_Memory implements INode { default: 'chat_history', additionalParams: true } - ]; + ] } async init(nodeData: INodeData, _: string, options: ICommonObject): Promise { - return initializeMongoDB(nodeData, options); + return initializeMongoDB(nodeData, options) } } const initializeMongoDB = async (nodeData: INodeData, options: ICommonObject): Promise => { - const databaseName = nodeData.inputs?.databaseName as string; - const collectionName = nodeData.inputs?.collectionName as string; - const memoryKey = nodeData.inputs?.memoryKey as string; - const sessionId = nodeData.inputs?.sessionId as string; + const databaseName = nodeData.inputs?.databaseName as string + const collectionName = nodeData.inputs?.collectionName as string + const memoryKey = nodeData.inputs?.memoryKey as string + const sessionId = nodeData.inputs?.sessionId as string - const credentialData = await getCredentialData(nodeData.credential ?? '', options); - const mongoDBConnectUrl = getCredentialParam('mongoDBConnectUrl', credentialData, nodeData); + const credentialData = await getCredentialData(nodeData.credential ?? '', options) + const mongoDBConnectUrl = getCredentialParam('mongoDBConnectUrl', credentialData, nodeData) - const client = await getMongoClient(mongoDBConnectUrl); - const collection = client.db(databaseName).collection(collectionName); + const client = await getMongoClient(mongoDBConnectUrl) + const collection = client.db(databaseName).collection(collectionName) const mongoDBChatMessageHistory = new MongoDBChatMessageHistory({ collection, sessionId - }); + }) mongoDBChatMessageHistory.getMessages = async (): Promise => { const document = await collection.findOne({ sessionId: (mongoDBChatMessageHistory as any).sessionId - }); - const messages = document?.messages || []; - return messages.map(mapStoredMessageToChatMessage); - }; + }) + const messages = document?.messages || [] + return messages.map(mapStoredMessageToChatMessage) + } mongoDBChatMessageHistory.addMessage = async (message: BaseMessage): Promise => { - const messages = [message].map((msg) => msg.toDict()); + const messages = [message].map((msg) => msg.toDict()) await collection.updateOne( { sessionId: (mongoDBChatMessageHistory as any).sessionId }, { $push: { messages: { $each: messages } } }, { upsert: true } - ); - }; + ) + } mongoDBChatMessageHistory.clear = async (): Promise => { - await collection.deleteOne({ sessionId: (mongoDBChatMessageHistory as any).sessionId }); - }; + await collection.deleteOne({ sessionId: (mongoDBChatMessageHistory as any).sessionId }) + } return new BufferMemoryExtended({ memoryKey: memoryKey ?? 'chat_history', chatHistory: mongoDBChatMessageHistory, sessionId, collection - }); -}; + }) +} interface BufferMemoryExtendedInput { - collection: Collection; - sessionId: string; + collection: Collection + sessionId: string } class BufferMemoryExtended extends FlowiseMemory implements MemoryMethods { - sessionId = ''; - collection: Collection; + sessionId = '' + collection: Collection constructor(fields: BufferMemoryInput & BufferMemoryExtendedInput) { - super(fields); - this.sessionId = fields.sessionId; - this.collection = fields.collection; + super(fields) + this.sessionId = fields.sessionId + this.collection = fields.collection } async getChatMessages(overrideSessionId = '', returnBaseMessages = false): Promise { - if (!this.collection) return []; + if (!this.collection) return [] - const id = overrideSessionId ?? this.sessionId; - const document = await this.collection.findOne({ sessionId: id }); - const messages = document?.messages || []; - const baseMessages = messages.map(mapStoredMessageToChatMessage); - return returnBaseMessages ? baseMessages : convertBaseMessagetoIMessage(baseMessages); + const id = overrideSessionId ?? this.sessionId + const document = await this.collection.findOne({ sessionId: id }) + const messages = document?.messages || [] + const baseMessages = messages.map(mapStoredMessageToChatMessage) + return returnBaseMessages ? baseMessages : convertBaseMessagetoIMessage(baseMessages) } async addChatMessages(msgArray: { text: string; type: MessageType }[], overrideSessionId = ''): Promise { - if (!this.collection) return; + if (!this.collection) return - const id = overrideSessionId ?? this.sessionId; - const input = msgArray.find((msg) => msg.type === 'userMessage'); - const output = msgArray.find((msg) => msg.type === 'apiMessage'); + const id = overrideSessionId ?? this.sessionId + const input = msgArray.find((msg) => msg.type === 'userMessage') + const output = msgArray.find((msg) => msg.type === 'apiMessage') if (input) { - const newInputMessage = new HumanMessage(input.text); - const messageToAdd = [newInputMessage].map((msg) => msg.toDict()); + const newInputMessage = new HumanMessage(input.text) + const messageToAdd = [newInputMessage].map((msg) => msg.toDict()) await this.collection.updateOne( { sessionId: id }, { $push: { messages: { $each: messageToAdd } } }, { upsert: true } - ); + ) } if (output) { - const newOutputMessage = new AIMessage(output.text); - const messageToAdd = [newOutputMessage].map((msg) => msg.toDict()); + const newOutputMessage = new AIMessage(output.text) + const messageToAdd = [newOutputMessage].map((msg) => msg.toDict()) await this.collection.updateOne( { sessionId: id }, { $push: { messages: { $each: messageToAdd } } }, { upsert: true } - ); + ) } } async clearChatMessages(overrideSessionId = ''): Promise { - if (!this.collection) return; + if (!this.collection) return - const id = overrideSessionId ?? this.sessionId; - await this.collection.deleteOne({ sessionId: id }); - await this.clear(); + const id = overrideSessionId ?? this.sessionId + await this.collection.deleteOne({ sessionId: id }) + await this.clear() } } -module.exports = { nodeClass: MongoDB_Memory }; +module.exports = { nodeClass: MongoDB_Memory } From 51388d50577f11b6d6757d1bbeb29e5d1f01db2f Mon Sep 17 00:00:00 2001 From: Ashes47 Date: Sun, 28 Jan 2024 21:48:15 +0530 Subject: [PATCH 4/5] update --- .../nodes/memory/MongoDBMemory/MongoDBMemory.ts | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/packages/components/nodes/memory/MongoDBMemory/MongoDBMemory.ts b/packages/components/nodes/memory/MongoDBMemory/MongoDBMemory.ts index 3c2903f3..7399ad81 100644 --- a/packages/components/nodes/memory/MongoDBMemory/MongoDBMemory.ts +++ b/packages/components/nodes/memory/MongoDBMemory/MongoDBMemory.ts @@ -5,12 +5,15 @@ import { mapStoredMessageToChatMessage, AIMessage, HumanMessage, BaseMessage } f import { convertBaseMessagetoIMessage, getBaseClasses, getCredentialData, getCredentialParam } from '../../../src/utils' import { FlowiseMemory, ICommonObject, IMessage, INode, INodeData, INodeParams, MemoryMethods, MessageType } from '../../../src/Interface' -let mongoClientSingleton = null +let mongoClientSingleton: MongoClient +let mongoUrl: string -const getMongoClient = async (mongoDBConnectUrl) => { - if (!mongoClientSingleton) { - mongoClientSingleton = new MongoClient(mongoDBConnectUrl, { useNewUrlParser: true, useUnifiedTopology: true }) +const getMongoClient = async (newMongoUrl: string) => { + if (!mongoClientSingleton || newMongoUrl !== mongoUrl) { + mongoClientSingleton = new MongoClient(newMongoUrl) + mongoUrl = newMongoUrl await mongoClientSingleton.connect() + return mongoClientSingleton } return mongoClientSingleton } From 1c108f35999a490c1d3727cdfb32736765d62587 Mon Sep 17 00:00:00 2001 From: Ashes47 Date: Sun, 28 Jan 2024 21:51:27 +0530 Subject: [PATCH 5/5] update --- .../nodes/memory/MongoDBMemory/MongoDBMemory.ts | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/packages/components/nodes/memory/MongoDBMemory/MongoDBMemory.ts b/packages/components/nodes/memory/MongoDBMemory/MongoDBMemory.ts index 7399ad81..b7309dcd 100644 --- a/packages/components/nodes/memory/MongoDBMemory/MongoDBMemory.ts +++ b/packages/components/nodes/memory/MongoDBMemory/MongoDBMemory.ts @@ -9,15 +9,20 @@ let mongoClientSingleton: MongoClient let mongoUrl: string const getMongoClient = async (newMongoUrl: string) => { - if (!mongoClientSingleton || newMongoUrl !== mongoUrl) { + if (!mongoClientSingleton) { + // if client doesn't exists + mongoClientSingleton = new MongoClient(newMongoUrl) + mongoUrl = newMongoUrl + return mongoClientSingleton + } else if (mongoClientSingleton && newMongoUrl !== mongoUrl) { + // if client exists but url changed + mongoClientSingleton.close() mongoClientSingleton = new MongoClient(newMongoUrl) mongoUrl = newMongoUrl - await mongoClientSingleton.connect() return mongoClientSingleton } return mongoClientSingleton } - class MongoDB_Memory implements INode { label: string name: string