From 9e7615f832876251741fdd111e5063e562ae734c Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Fri, 19 Jul 2024 08:06:05 -0400 Subject: [PATCH] =?UTF-8?q?=E2=9A=99=EF=B8=8F=20fix:=20Plugin=20Message=20?= =?UTF-8?q?Handling=20Errors=20(#3392)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add unique index for messageId and user in messageSchema - use `updateMessage` for updating the plugins message? - add better logging for updateMessage - prevents dupe_key or getKeyIndex error --- api/models/Message.js | 12 +++++-- api/models/schema/messageSchema.js | 1 + api/server/routes/ask/gptPlugins.js | 45 +++++++++++++------------ api/server/routes/edit/gptPlugins.js | 49 +++++++++++++++------------- 4 files changed, 62 insertions(+), 45 deletions(-) diff --git a/api/models/Message.js b/api/models/Message.js index 838d20f4c6..384b7e7d4d 100644 --- a/api/models/Message.js +++ b/api/models/Message.js @@ -108,6 +108,9 @@ ${JSON.stringify(params, null, 2)} return message.toObject(); } catch (err) { logger.error('Error saving message:', err); + if (metadata && metadata?.context) { + logger.info(`\`saveMessage\` context: ${metadata.context}`); + } throw err; } } @@ -178,7 +181,7 @@ async function recordMessage({ new: true, }); } catch (err) { - logger.error('Error saving message:', err); + logger.error('Error recording message:', err); throw err; } } @@ -217,10 +220,12 @@ async function updateMessageText(req, { messageId, text }) { * @param {boolean} [message.isCreatedByUser] - Indicates if the message was created by the user. * @param {string} [message.sender] - The identifier of the sender. * @param {number} [message.tokenCount] - The number of tokens in the message. + * @param {Object} [metadata] - The operation metadata + * @param {string} [metadata.context] - The operation metadata * @returns {Promise} The updated message document. * @throws {Error} If there is an error in updating the message or if the message is not found. */ -async function updateMessage(req, message) { +async function updateMessage(req, message, metadata) { try { const { messageId, ...update } = message; update.isEdited = true; @@ -248,6 +253,9 @@ async function updateMessage(req, message) { }; } catch (err) { logger.error('Error updating message:', err); + if (metadata && metadata?.context) { + logger.info(`\`updateMessage\` context: ${metadata.context}`); + } throw err; } } diff --git a/api/models/schema/messageSchema.js b/api/models/schema/messageSchema.js index 3e89a5caa9..282bc437f8 100644 --- a/api/models/schema/messageSchema.js +++ b/api/models/schema/messageSchema.js @@ -129,6 +129,7 @@ if (process.env.MEILI_HOST && process.env.MEILI_MASTER_KEY) { } messageSchema.index({ createdAt: 1 }); +messageSchema.index({ messageId: 1, user: 1 }, { unique: true }); const Message = mongoose.models.Message || mongoose.model('Message', messageSchema); diff --git a/api/server/routes/ask/gptPlugins.js b/api/server/routes/ask/gptPlugins.js index 602ff25086..8408e7a5d6 100644 --- a/api/server/routes/ask/gptPlugins.js +++ b/api/server/routes/ask/gptPlugins.js @@ -4,7 +4,7 @@ const { getResponseSender, Constants, CacheKeys, Time } = require('librechat-dat const { initializeClient } = require('~/server/services/Endpoints/gptPlugins'); const { sendMessage, createOnProgress } = require('~/server/utils'); const { addTitle } = require('~/server/services/Endpoints/openAI'); -const { saveMessage } = require('~/models'); +const { saveMessage, updateMessage } = require('~/models'); const { getLogStores } = require('~/cache'); const { handleAbort, @@ -73,7 +73,14 @@ router.post( }; const messageCache = getLogStores(CacheKeys.MESSAGES); - const throttledSetMessage = throttle(messageCache.set, 3000, { trailing: false }); + const throttledCacheSet = throttle( + (text) => { + messageCache.set(responseMessageId, text, Time.FIVE_MINUTES); + }, + 3000, + { trailing: false }, + ); + let streaming = null; let timer = null; @@ -87,21 +94,7 @@ router.post( clearTimeout(timer); } - /* - { - messageId: responseMessageId, - sender, - conversationId, - parentMessageId: overrideParentMessageId || userMessageId, - text: partialText, - model: endpointOption.modelOptions.model, - unfinished: true, - error: false, - plugins, - user, - } - */ - throttledSetMessage(responseMessageId, partialText, Time.FIVE_MINUTES); + throttledCacheSet(partialText); streaming = new Promise((resolve) => { timer = setTimeout(() => { @@ -175,7 +168,11 @@ router.post( const onChainEnd = () => { if (!client.skipSaveUserMessage) { - saveMessage(req, { ...userMessage, user }); + saveMessage( + req, + { ...userMessage, user }, + { context: 'api/server/routes/ask/gptPlugins.js - onChainEnd' }, + ); } sendIntermediateMessage(res, { plugins, @@ -212,9 +209,6 @@ router.post( logger.debug('[/ask/gptPlugins]', response); - response.plugins = plugins.map((p) => ({ ...p, loading: false })); - await saveMessage(req, { ...response, user }); - const { conversation = {} } = await client.responsePromise; conversation.title = conversation && !conversation.title ? null : conversation?.title || 'New Chat'; @@ -235,6 +229,15 @@ router.post( client, }); } + + response.plugins = plugins.map((p) => ({ ...p, loading: false })); + if (response.plugins?.length > 0) { + await updateMessage( + req, + { ...response, user }, + { context: 'api/server/routes/ask/gptPlugins.js - save plugins used' }, + ); + } } catch (error) { const partialText = getPartialText(); handleAbortError(res, req, error, { diff --git a/api/server/routes/edit/gptPlugins.js b/api/server/routes/edit/gptPlugins.js index 926c8e4f5f..cd4e575e83 100644 --- a/api/server/routes/edit/gptPlugins.js +++ b/api/server/routes/edit/gptPlugins.js @@ -13,7 +13,7 @@ const { } = require('~/server/middleware'); const { sendMessage, createOnProgress, formatSteps, formatAction } = require('~/server/utils'); const { initializeClient } = require('~/server/services/Endpoints/gptPlugins'); -const { saveMessage } = require('~/models'); +const { saveMessage, updateMessage } = require('~/models'); const { getLogStores } = require('~/cache'); const { validateTools } = require('~/app'); const { logger } = require('~/config'); @@ -81,7 +81,14 @@ router.post( }; const messageCache = getLogStores(CacheKeys.MESSAGES); - const throttledSetMessage = throttle(messageCache.set, 3000, { trailing: false }); + const throttledCacheSet = throttle( + (text) => { + messageCache.set(responseMessageId, text, Time.FIVE_MINUTES); + }, + 3000, + { trailing: false }, + ); + const { onProgress: progressCallback, sendIntermediateMessage, @@ -92,22 +99,7 @@ router.post( if (plugin.loading === true) { plugin.loading = false; } - - /* - { - messageId: responseMessageId, - sender, - conversationId, - parentMessageId: overrideParentMessageId || userMessageId, - text: partialText, - model: endpointOption.modelOptions.model, - unfinished: true, - isEdited: true, - error: false, - user, - } - */ - throttledSetMessage(responseMessageId, partialText, Time.FIVE_MINUTES); + throttledCacheSet(partialText); }, }); @@ -115,7 +107,11 @@ router.post( let { intermediateSteps: steps } = data; plugin.outputs = steps && steps[0].action ? formatSteps(steps) : 'An error occurred.'; plugin.loading = false; - saveMessage(req, { ...userMessage, user }); + saveMessage( + req, + { ...userMessage, user }, + { context: 'api/server/routes/ask/gptPlugins.js - onChainEnd' }, + ); sendIntermediateMessage(res, { plugin, parentMessageId: userMessage.messageId, @@ -146,7 +142,11 @@ router.post( plugin.inputs.push(formattedAction); plugin.latest = formattedAction.plugin; if (!start && !client.skipSaveUserMessage) { - saveMessage(req, { ...userMessage, user }); + saveMessage( + req, + { ...userMessage, user }, + { context: 'api/server/routes/ask/gptPlugins.js - onAgentAction' }, + ); } sendIntermediateMessage(res, { plugin, @@ -184,8 +184,6 @@ router.post( } logger.debug('[/edit/gptPlugins] CLIENT RESPONSE', response); - response.plugin = { ...plugin, loading: false }; - await saveMessage(req, { ...response, user }); const { conversation = {} } = await client.responsePromise; conversation.title = @@ -199,6 +197,13 @@ router.post( responseMessage: response, }); res.end(); + + response.plugin = { ...plugin, loading: false }; + await updateMessage( + req, + { ...response, user }, + { context: 'api/server/routes/edit/gptPlugins.js' }, + ); } catch (error) { const partialText = getPartialText(); handleAbortError(res, req, error, {