🤖 refactor: Improve Agents Memory Usage, Bump Keyv, Grok 3

api/app/clients/AnthropicClient.js

@@ -9,7 +9,7 @@ const {
   getResponseSender,
   validateVisionModel,
 } = require('librechat-data-provider');
-const { SplitStreamHandler: _Handler, GraphEvents } = require('@librechat/agents');
+const { SplitStreamHandler: _Handler } = require('@librechat/agents');
 const {
   truncateText,
   formatMessage,
@@ -26,10 +26,11 @@ const {
 const { getModelMaxTokens, getModelMaxOutputTokens, matchModelName } = require('~/utils');
 const { spendTokens, spendStructuredTokens } = require('~/models/spendTokens');
 const { encodeAndFormat } = require('~/server/services/Files/images/encode');
+const { createFetch, createStreamEventHandlers } = require('./generators');
 const Tokenizer = require('~/server/services/Tokenizer');
-const { logger, sendEvent } = require('~/config');
 const { sleep } = require('~/server/utils');
 const BaseClient = require('./BaseClient');
+const { logger } = require('~/config');
 
 const HUMAN_PROMPT = '\n\nHuman:';
 const AI_PROMPT = '\n\nAssistant:';
@@ -184,7 +185,10 @@ class AnthropicClient extends BaseClient {
   getClient(requestOptions) {
     /** @type {Anthropic.ClientOptions} */
     const options = {
-      fetch: this.fetch,
+      fetch: createFetch({
+        directEndpoint: this.options.directEndpoint,
+        reverseProxyUrl: this.options.reverseProxyUrl,
+      }),
       apiKey: this.apiKey,
     };
 
@@ -795,14 +799,11 @@ class AnthropicClient extends BaseClient {
     }
 
     logger.debug('[AnthropicClient]', { ...requestOptions });
+    const handlers = createStreamEventHandlers(this.options.res);
     this.streamHandler = new SplitStreamHandler({
       accumulate: true,
       runId: this.responseMessageId,
-      handlers: {
-        [GraphEvents.ON_RUN_STEP]: (event) => sendEvent(this.options.res, event),
-        [GraphEvents.ON_MESSAGE_DELTA]: (event) => sendEvent(this.options.res, event),
-        [GraphEvents.ON_REASONING_DELTA]: (event) => sendEvent(this.options.res, event),
-      },
+      handlers,
     });
 
     let intermediateReply = this.streamHandler.tokens;

api/app/clients/BaseClient.js

@@ -28,15 +28,10 @@ class BaseClient {
       month: 'long',
       day: 'numeric',
     });
-    this.fetch = this.fetch.bind(this);
     /** @type {boolean} */
     this.skipSaveConvo = false;
     /** @type {boolean} */
     this.skipSaveUserMessage = false;
-    /** @type {ClientDatabaseSavePromise} */
-    this.userMessagePromise;
-    /** @type {ClientDatabaseSavePromise} */
-    this.responsePromise;
     /** @type {string} */
     this.user;
     /** @type {string} */
@@ -564,6 +559,8 @@ class BaseClient {
   }
 
   async sendMessage(message, opts = {}) {
+    /** @type {Promise<TMessage>} */
+    let userMessagePromise;
     const { user, head, isEdited, conversationId, responseMessageId, saveOptions, userMessage } =
       await this.handleStartMethods(message, opts);
 
@@ -625,11 +622,11 @@ class BaseClient {
     }
 
     if (!isEdited && !this.skipSaveUserMessage) {
-      this.userMessagePromise = this.saveMessageToDatabase(userMessage, saveOptions, user);
+      userMessagePromise = this.saveMessageToDatabase(userMessage, saveOptions, user);
       this.savedMessageIds.add(userMessage.messageId);
       if (typeof opts?.getReqData === 'function') {
         opts.getReqData({
-          userMessagePromise: this.userMessagePromise,
+          userMessagePromise,
         });
       }
     }
@@ -655,7 +652,9 @@ class BaseClient {
 
     /** @type {string|string[]|undefined} */
     const completion = await this.sendCompletion(payload, opts);
-    this.abortController.requestCompleted = true;
+    if (this.abortController) {
+      this.abortController.requestCompleted = true;
+    }
 
     /** @type {TMessage} */
     const responseMessage = {
@@ -703,7 +702,13 @@ class BaseClient {
       if (usage != null && Number(usage[this.outputTokensKey]) > 0) {
         responseMessage.tokenCount = usage[this.outputTokensKey];
         completionTokens = responseMessage.tokenCount;
-        await this.updateUserMessageTokenCount({ usage, tokenCountMap, userMessage, opts });
+        await this.updateUserMessageTokenCount({
+          usage,
+          tokenCountMap,
+          userMessage,
+          userMessagePromise,
+          opts,
+        });
       } else {
         responseMessage.tokenCount = this.getTokenCountForResponse(responseMessage);
         completionTokens = responseMessage.tokenCount;
@@ -712,8 +717,8 @@ class BaseClient {
       await this.recordTokenUsage({ promptTokens, completionTokens, usage });
     }
 
-    if (this.userMessagePromise) {
-      await this.userMessagePromise;
+    if (userMessagePromise) {
+      await userMessagePromise;
     }
 
     if (this.artifactPromises) {
@@ -728,7 +733,11 @@ class BaseClient {
       }
     }
 
-    this.responsePromise = this.saveMessageToDatabase(responseMessage, saveOptions, user);
+    responseMessage.databasePromise = this.saveMessageToDatabase(
+      responseMessage,
+      saveOptions,
+      user,
+    );
     this.savedMessageIds.add(responseMessage.messageId);
     delete responseMessage.tokenCount;
     return responseMessage;
@@ -749,9 +758,16 @@ class BaseClient {
    * @param {StreamUsage} params.usage
    * @param {Record<string, number>} params.tokenCountMap
    * @param {TMessage} params.userMessage
+   * @param {Promise<TMessage>} params.userMessagePromise
    * @param {object} params.opts
    */
-  async updateUserMessageTokenCount({ usage, tokenCountMap, userMessage, opts }) {
+  async updateUserMessageTokenCount({
+    usage,
+    tokenCountMap,
+    userMessage,
+    userMessagePromise,
+    opts,
+  }) {
     /** @type {boolean} */
     const shouldUpdateCount =
       this.calculateCurrentTokenCount != null &&
@@ -787,7 +803,7 @@ class BaseClient {
       Note: we update the user message to be sure it gets the calculated token count;
       though `AskController` saves the user message, EditController does not
     */
-    await this.userMessagePromise;
+    await userMessagePromise;
     await this.updateMessageInDatabase({
       messageId: userMessage.messageId,
       tokenCount: userMessageTokenCount,
@@ -853,7 +869,7 @@ class BaseClient {
     }
 
     const savedMessage = await saveMessage(
-      this.options.req,
+      this.options?.req,
       {
         ...message,
         endpoint: this.options.endpoint,
@@ -877,7 +893,7 @@ class BaseClient {
     const existingConvo =
       this.fetchedConvo === true
         ? null
-        : await getConvo(this.options.req?.user?.id, message.conversationId);
+        : await getConvo(this.options?.req?.user?.id, message.conversationId);
 
     const unsetFields = {};
     const exceptions = new Set(['spec', 'iconURL']);
@@ -897,7 +913,7 @@ class BaseClient {
       }
     }
 
-    const conversation = await saveConvo(this.options.req, fieldsToKeep, {
+    const conversation = await saveConvo(this.options?.req, fieldsToKeep, {
       context: 'api/app/clients/BaseClient.js - saveMessageToDatabase #saveConvo',
       unsetFields,
     });

api/app/clients/ChatGPTClient.js

@@ -1,4 +1,4 @@
-const Keyv = require('keyv');
+const { Keyv } = require('keyv');
 const crypto = require('crypto');
 const { CohereClient } = require('cohere-ai');
 const { fetchEventSource } = require('@waylaidwanderer/fetch-event-source');
@@ -339,7 +339,7 @@ class ChatGPTClient extends BaseClient {
     opts.body = JSON.stringify(modelOptions);
 
     if (modelOptions.stream) {
-      // eslint-disable-next-line no-async-promise-executor
+
       return new Promise(async (resolve, reject) => {
         try {
           let done = false;

api/app/clients/OpenAIClient.js

@@ -1,7 +1,7 @@
 const OpenAI = require('openai');
 const { OllamaClient } = require('./OllamaClient');
 const { HttpsProxyAgent } = require('https-proxy-agent');
-const { SplitStreamHandler, GraphEvents } = require('@librechat/agents');
+const { SplitStreamHandler } = require('@librechat/agents');
 const {
   Constants,
   ImageDetail,
@@ -32,17 +32,18 @@ const {
   createContextHandlers,
 } = require('./prompts');
 const { encodeAndFormat } = require('~/server/services/Files/images/encode');
+const { createFetch, createStreamEventHandlers } = require('./generators');
 const { addSpaceIfNeeded, isEnabled, sleep } = require('~/server/utils');
 const Tokenizer = require('~/server/services/Tokenizer');
 const { spendTokens } = require('~/models/spendTokens');
 const { handleOpenAIErrors } = require('./tools/util');
 const { createLLM, RunManager } = require('./llm');
-const { logger, sendEvent } = require('~/config');
 const ChatGPTClient = require('./ChatGPTClient');
 const { summaryBuffer } = require('./memory');
 const { runTitleChain } = require('./chains');
 const { tokenSplit } = require('./document');
 const BaseClient = require('./BaseClient');
+const { logger } = require('~/config');
 
 class OpenAIClient extends BaseClient {
   constructor(apiKey, options = {}) {
@@ -609,7 +610,7 @@ class OpenAIClient extends BaseClient {
         return result.trim();
       }
 
-      logger.debug('[OpenAIClient] sendCompletion: result', result);
+      logger.debug('[OpenAIClient] sendCompletion: result', { ...result });
 
       if (this.isChatCompletion) {
         reply = result.choices[0].message.content;
@@ -818,7 +819,7 @@ ${convo}
 
         const completionTokens = this.getTokenCount(title);
 
-        this.recordTokenUsage({ promptTokens, completionTokens, context: 'title' });
+        await this.recordTokenUsage({ promptTokens, completionTokens, context: 'title' });
       } catch (e) {
         logger.error(
           '[OpenAIClient] There was an issue generating the title with the completion method',
@@ -1245,7 +1246,10 @@ ${convo}
       let chatCompletion;
       /** @type {OpenAI} */
       const openai = new OpenAI({
-        fetch: this.fetch,
+        fetch: createFetch({
+          directEndpoint: this.options.directEndpoint,
+          reverseProxyUrl: this.options.reverseProxyUrl,
+        }),
         apiKey: this.apiKey,
         ...opts,
       });
@@ -1275,12 +1279,13 @@ ${convo}
       }
 
       if (this.options.addParams && typeof this.options.addParams === 'object') {
+        const addParams = { ...this.options.addParams };
         modelOptions = {
           ...modelOptions,
-          ...this.options.addParams,
+          ...addParams,
         };
         logger.debug('[OpenAIClient] chatCompletion: added params', {
-          addParams: this.options.addParams,
+          addParams: addParams,
           modelOptions,
         });
       }
@@ -1309,11 +1314,12 @@ ${convo}
       }
 
       if (this.options.dropParams && Array.isArray(this.options.dropParams)) {
-        this.options.dropParams.forEach((param) => {
+        const dropParams = [...this.options.dropParams];
+        dropParams.forEach((param) => {
           delete modelOptions[param];
         });
         logger.debug('[OpenAIClient] chatCompletion: dropped params', {
-          dropParams: this.options.dropParams,
+          dropParams: dropParams,
           modelOptions,
         });
       }
@@ -1355,15 +1361,12 @@ ${convo}
         delete modelOptions.reasoning_effort;
       }
 
+      const handlers = createStreamEventHandlers(this.options.res);
       this.streamHandler = new SplitStreamHandler({
         reasoningKey,
         accumulate: true,
         runId: this.responseMessageId,
-        handlers: {
-          [GraphEvents.ON_RUN_STEP]: (event) => sendEvent(this.options.res, event),
-          [GraphEvents.ON_MESSAGE_DELTA]: (event) => sendEvent(this.options.res, event),
-          [GraphEvents.ON_REASONING_DELTA]: (event) => sendEvent(this.options.res, event),
-        },
+        handlers,
       });
 
       intermediateReply = this.streamHandler.tokens;

api/app/clients/PluginsClient.js

@@ -252,12 +252,14 @@ class PluginsClient extends OpenAIClient {
       await this.recordTokenUsage(responseMessage);
     }
 
-    this.responsePromise = this.saveMessageToDatabase(responseMessage, saveOptions, user);
+    const databasePromise = this.saveMessageToDatabase(responseMessage, saveOptions, user);
     delete responseMessage.tokenCount;
-    return { ...responseMessage, ...result };
+    return { ...responseMessage, ...result, databasePromise };
   }
 
   async sendMessage(message, opts = {}) {
+    /** @type {Promise<TMessage>} */
+    let userMessagePromise;
     /** @type {{ filteredTools: string[], includedTools: string[] }} */
     const { filteredTools = [], includedTools = [] } = this.options.req.app.locals;
 
@@ -327,10 +329,10 @@ class PluginsClient extends OpenAIClient {
     }
 
     if (!this.skipSaveUserMessage) {
-      this.userMessagePromise = this.saveMessageToDatabase(userMessage, saveOptions, user);
+      userMessagePromise = this.saveMessageToDatabase(userMessage, saveOptions, user);
       if (typeof opts?.getReqData === 'function') {
         opts.getReqData({
-          userMessagePromise: this.userMessagePromise,
+          userMessagePromise,
         });
       }
     }