feat: backend content aggregation for agents/bedrock

api/server/controllers/agents/callbacks.js

@@ -2,6 +2,7 @@ const { GraphEvents, ToolEndHandler, ChatModelStreamHandler } = require('@librec
 
 /** @typedef {import('@librechat/agents').EventHandler} EventHandler */
 /** @typedef {import('@librechat/agents').ChatModelStreamHandler} ChatModelStreamHandler */
+/** @typedef {import('@librechat/agents').ContentAggregatorResult['aggregateContent']} ContentAggregator */
 /** @typedef {import('@librechat/agents').GraphEvents} GraphEvents */
 
 /**
@@ -20,13 +21,17 @@ const sendEvent = (res, event) => {
 
 /**
  * Get default handlers for stream events.
- * @param {{ res?: ServerResponse }} options - The options object.
+ * @param {Object} options - The options object.
+ * @param {ServerResponse} options.res - The options object.
+ * @param {ContentAggregator} options.aggregateContent - The options object.
  * @returns {Record<string, t.EventHandler>} The default handlers.
  * @throws {Error} If the request is not found.
  */
-function getDefaultHandlers({ res }) {
-  if (!res) {
-    throw new Error('Request not found');
+function getDefaultHandlers({ res, aggregateContent }) {
+  if (!res || !aggregateContent) {
+    throw new Error(
+      `[getDefaultHandlers] Missing required options: res: ${!res}, aggregateContent: ${!aggregateContent}`,
+    );
   }
   const handlers = {
     // [GraphEvents.CHAT_MODEL_END]: new ModelEndHandler(),
@@ -40,6 +45,7 @@ function getDefaultHandlers({ res }) {
        */
       handle: (event, data) => {
         sendEvent(res, { event, data });
+        aggregateContent({ event, data });
       },
     },
     [GraphEvents.ON_RUN_STEP_DELTA]: {
@@ -50,6 +56,7 @@ function getDefaultHandlers({ res }) {
        */
       handle: (event, data) => {
         sendEvent(res, { event, data });
+        aggregateContent({ event, data });
       },
     },
     [GraphEvents.ON_RUN_STEP_COMPLETED]: {
@@ -60,6 +67,7 @@ function getDefaultHandlers({ res }) {
        */
       handle: (event, data) => {
         sendEvent(res, { event, data });
+        aggregateContent({ event, data });
       },
     },
     [GraphEvents.ON_MESSAGE_DELTA]: {
@@ -70,6 +78,7 @@ function getDefaultHandlers({ res }) {
        */
       handle: (event, data) => {
         sendEvent(res, { event, data });
+        aggregateContent({ event, data });
       },
     },
   };

api/server/controllers/agents/client.js

@@ -8,11 +8,7 @@
 // mapModelToAzureConfig,
 // } = require('librechat-data-provider');
 const { Callback } = require('@librechat/agents');
-const {
-  EModelEndpoint,
-  providerEndpointMap,
-  removeNullishValues,
-} = require('librechat-data-provider');
+const { providerEndpointMap, removeNullishValues } = require('librechat-data-provider');
 const {
   extractBaseURL,
   // constructAzureURL,
@@ -29,6 +25,8 @@ const BaseClient = require('~/app/clients/BaseClient');
 const { createRun } = require('./run');
 const { logger } = require('~/config');
 
+/** @typedef {import('@librechat/agents').MessageContentComplex} MessageContentComplex */
+
 class AgentClient extends BaseClient {
   constructor(options = {}) {
     super(options);
@@ -43,7 +41,9 @@ class AgentClient extends BaseClient {
 
     this.modelOptions = modelOptions;
     this.maxContextTokens = maxContextTokens;
-    this.options = Object.assign({ endpoint: EModelEndpoint.agents }, clientOptions);
+    /** @type {MessageContentComplex[]} */
+    this.contentParts = options.contentParts;
+    this.options = Object.assign({ endpoint: options.endpoint }, clientOptions);
   }
 
   setOptions(options) {
@@ -270,11 +270,12 @@ class AgentClient extends BaseClient {
   /** @type {sendCompletion} */
   async sendCompletion(payload, opts = {}) {
     this.modelOptions.user = this.user;
-    return await this.chatCompletion({
+    await this.chatCompletion({
       payload,
       onProgress: opts.onProgress,
       abortController: opts.abortController,
     });
+    return this.contentParts;
   }
 
   // async recordTokenUsage({ promptTokens, completionTokens, context = 'message' }) {
@@ -415,6 +416,7 @@ class AgentClient extends BaseClient {
           thread_id: this.conversationId,
         },
         run_id: this.responseMessageId,
+        signal: abortController.signal,
         streamMode: 'values',
         version: 'v2',
       };
@@ -424,7 +426,7 @@ class AgentClient extends BaseClient {
       }
 
       const messages = formatAgentMessages(payload);
-      const runMessages = await run.processStream({ messages }, config, {
+      await run.processStream({ messages }, config, {
         [Callback.TOOL_ERROR]: (graph, error, toolId) => {
           logger.error(
             '[api/server/controllers/agents/client.js #chatCompletion] Tool Error',
@@ -433,8 +435,7 @@ class AgentClient extends BaseClient {
           );
         },
       });
-      // console.dir(runMessages, { depth: null });
-      return runMessages;
+      logger.info(this.contentParts, { depth: null });
     } catch (err) {
       logger.error(
         '[api/server/controllers/agents/client.js #chatCompletion] Unhandled error type',

api/server/services/Endpoints/agents/initialize.js

@@ -11,6 +11,7 @@
 
 const { z } = require('zod');
 const { tool } = require('@langchain/core/tools');
+const { createContentAggregator } = require('@librechat/agents');
 const { EModelEndpoint, providerEndpointMap } = require('librechat-data-provider');
 const { getDefaultHandlers } = require('~/server/controllers/agents/callbacks');
 // for testing purposes
@@ -53,7 +54,8 @@ const initializeClient = async ({ req, res, endpointOption }) => {
   }
 
   // TODO: use endpointOption to determine options/modelOptions
-  const eventHandlers = getDefaultHandlers({ res });
+  const { contentParts, aggregateContent } = createContentAggregator();
+  const eventHandlers = getDefaultHandlers({ res, aggregateContent });
 
   // const tools = [createTavilySearchTool()];
   // const tools = [_getWeather];
@@ -106,6 +108,7 @@ const initializeClient = async ({ req, res, endpointOption }) => {
     agent,
     tools,
     toolMap,
+    contentParts,
     modelOptions,
     eventHandlers,
     endpoint: EModelEndpoint.agents,

api/server/services/Endpoints/bedrock/initialize.js

@@ -1,3 +1,4 @@
+const { createContentAggregator } = require('@librechat/agents');
 const { EModelEndpoint, providerEndpointMap } = require('librechat-data-provider');
 const { getDefaultHandlers } = require('~/server/controllers/agents/callbacks');
 // const { loadAgentTools } = require('~/server/services/ToolService');
@@ -10,8 +11,8 @@ const initializeClient = async ({ req, res, endpointOption }) => {
     throw new Error('Endpoint option not provided');
   }
 
-  // TODO: use endpointOption to determine options/modelOptions
-  const eventHandlers = getDefaultHandlers({ res });
+  const { contentParts, aggregateContent } = createContentAggregator();
+  const eventHandlers = getDefaultHandlers({ res, aggregateContent });
 
   // const tools = [createTavilySearchTool()];
 
@@ -45,6 +46,7 @@ const initializeClient = async ({ req, res, endpointOption }) => {
     // tools,
     // toolMap,
     modelOptions,
+    contentParts,
     eventHandlers,
     maxContextTokens,
     endpoint: EModelEndpoint.bedrock,