feat(metaSearchAgent): eliminate runnables

src/lib/search/metaSearchAgent.ts

@@ -6,11 +6,6 @@ import {
   MessagesPlaceholder,
   PromptTemplate,
 } from '@langchain/core/prompts';
-import {
-  RunnableLambda,
-  RunnableMap,
-  RunnableSequence,
-} from '@langchain/core/runnables';
 import { BaseMessage } from '@langchain/core/messages';
 import { StringOutputParser } from '@langchain/core/output_parsers';
 import LineListOutputParser from '../outputParsers/listLineOutputParser';
@@ -24,6 +19,7 @@ import computeSimilarity from '../utils/computeSimilarity';
 import formatChatHistoryAsString from '../utils/formatHistory';
 import eventEmitter from 'events';
 import { StreamEvent } from '@langchain/core/tracers/log_stream';
+import { EventEmitter } from 'node:stream';
 
 export interface MetaSearchAgentType {
   searchAndAnswer: (
@@ -46,7 +42,7 @@ interface Config {
   activeEngines: string[];
 }
 
-type BasicChainInput = {
+type SearchInput = {
   chat_history: BaseMessage[];
   query: string;
 };
@@ -59,14 +55,25 @@ class MetaSearchAgent implements MetaSearchAgentType {
     this.config = config;
   }
 
-  private async createSearchRetrieverChain(llm: BaseChatModel) {
+  private async searchSources(
+    llm: BaseChatModel,
+    input: SearchInput,
+    emitter: EventEmitter,
+  ) {
     (llm as unknown as ChatOpenAI).temperature = 0;
 
-    return RunnableSequence.from([
-      PromptTemplate.fromTemplate(this.config.queryGeneratorPrompt),
-      llm,
-      this.strParser,
-      RunnableLambda.from(async (input: string) => {
+    const chatPrompt = PromptTemplate.fromTemplate(
+      this.config.queryGeneratorPrompt,
+    );
+
+    const processedChatPrompt = await chatPrompt.invoke({
+      chat_history: formatChatHistoryAsString(input.chat_history),
+      query: input.query,
+    });
+
+    const llmRes = await llm.invoke(processedChatPrompt);
+    const messageStr = await this.strParser.invoke(llmRes);
+
     const linksOutputParser = new LineListOutputParser({
       key: 'links',
     });
@@ -75,10 +82,10 @@ class MetaSearchAgent implements MetaSearchAgentType {
       key: 'question',
     });
 
-        const links = await linksOutputParser.parse(input);
+    const links = await linksOutputParser.parse(messageStr);
     let question = this.config.summarizer
-          ? await questionOutputParser.parse(input)
-          : input;
+      ? await questionOutputParser.parse(messageStr)
+      : messageStr;
 
     if (question === 'not_needed') {
       return { query: '', docs: [] };
@@ -98,8 +105,7 @@ class MetaSearchAgent implements MetaSearchAgentType {
       linkDocs.map((doc) => {
         const URLDocExists = docGroups.find(
           (d) =>
-                d.metadata.url === doc.metadata.url &&
-                d.metadata.totalDocs < 10,
+            d.metadata.url === doc.metadata.url && d.metadata.totalDocs < 10,
         );
 
         if (!URLDocExists) {
@@ -114,8 +120,7 @@ class MetaSearchAgent implements MetaSearchAgentType {
 
         const docIndex = docGroups.findIndex(
           (d) =>
-                d.metadata.url === doc.metadata.url &&
-                d.metadata.totalDocs < 10,
+            d.metadata.url === doc.metadata.url && d.metadata.totalDocs < 10,
         );
 
         if (docIndex !== -1) {
@@ -227,40 +232,30 @@ class MetaSearchAgent implements MetaSearchAgentType {
 
       return { query: question, docs: documents };
     }
-      }),
-    ]);
   }
 
-  private async createAnsweringChain(
+  private async streamAnswer(
     llm: BaseChatModel,
     fileIds: string[],
     embeddings: Embeddings,
     optimizationMode: 'speed' | 'balanced' | 'quality',
+    input: SearchInput,
+    emitter: EventEmitter,
   ) {
-    return RunnableSequence.from([
-      RunnableMap.from({
-        query: (input: BasicChainInput) => input.query,
-        chat_history: (input: BasicChainInput) => input.chat_history,
-        date: () => new Date().toISOString(),
-        context: RunnableLambda.from(async (input: BasicChainInput) => {
-          const processedHistory = formatChatHistoryAsString(
-            input.chat_history,
-          );
+    const chatPrompt = ChatPromptTemplate.fromMessages([
+      ['system', this.config.responsePrompt],
+      new MessagesPlaceholder('chat_history'),
+      ['user', '{query}'],
+    ]);
 
     let docs: Document[] | null = null;
     let query = input.query;
 
     if (this.config.searchWeb) {
-            const searchRetrieverChain =
-              await this.createSearchRetrieverChain(llm);
+      const searchResults = await this.searchSources(llm, input, emitter);
 
-            const searchRetrieverResult = await searchRetrieverChain.invoke({
-              chat_history: processedHistory,
-              query,
-            });
-
-            query = searchRetrieverResult.query;
-            docs = searchRetrieverResult.docs;
+      query = searchResults.query;
+      docs = searchResults.docs;
     }
 
     const sortedDocs = await this.rerankDocs(
@@ -271,23 +266,29 @@ class MetaSearchAgent implements MetaSearchAgentType {
       optimizationMode,
     );
 
-          return sortedDocs;
-        })
-          .withConfig({
-            runName: 'FinalSourceRetriever',
-          })
-          .pipe(this.processDocs),
-      }),
-      ChatPromptTemplate.fromMessages([
-        ['system', this.config.responsePrompt],
-        new MessagesPlaceholder('chat_history'),
-        ['user', '{query}'],
-      ]),
-      llm,
-      this.strParser,
-    ]).withConfig({
-      runName: 'FinalResponseGenerator',
+    emitter.emit('data', JSON.stringify({ type: 'sources', data: sortedDocs }));
+
+    const context = this.processDocs(sortedDocs);
+
+    const formattedChatPrompt = await chatPrompt.invoke({
+      query: input.query,
+      chat_history: input.chat_history,
+      date: new Date().toISOString(),
+      context: context,
     });
+
+    const llmRes = await llm.stream(formattedChatPrompt);
+
+    for await (const data of llmRes) {
+      const messageStr = await this.strParser.invoke(data);
+
+      emitter.emit(
+        'data',
+        JSON.stringify({ type: 'response', data: messageStr }),
+      );
+    }
+
+    emitter.emit('end');
   }
 
   private async rerankDocs(
@@ -428,39 +429,6 @@ class MetaSearchAgent implements MetaSearchAgentType {
       .join('\n');
   }
 
-  private async handleStream(
-    stream: AsyncGenerator<StreamEvent, any, any>,
-    emitter: eventEmitter,
-  ) {
-    for await (const event of stream) {
-      if (
-        event.event === 'on_chain_end' &&
-        event.name === 'FinalSourceRetriever'
-      ) {
-        ``;
-        emitter.emit(
-          'data',
-          JSON.stringify({ type: 'sources', data: event.data.output }),
-        );
-      }
-      if (
-        event.event === 'on_chain_stream' &&
-        event.name === 'FinalResponseGenerator'
-      ) {
-        emitter.emit(
-          'data',
-          JSON.stringify({ type: 'response', data: event.data.chunk }),
-        );
-      }
-      if (
-        event.event === 'on_chain_end' &&
-        event.name === 'FinalResponseGenerator'
-      ) {
-        emitter.emit('end');
-      }
-    }
-  }
-
   async searchAndAnswer(
     message: string,
     history: BaseMessage[],
@@ -471,25 +439,18 @@ class MetaSearchAgent implements MetaSearchAgentType {
   ) {
     const emitter = new eventEmitter();
 
-    const answeringChain = await this.createAnsweringChain(
+    this.streamAnswer(
       llm,
       fileIds,
       embeddings,
       optimizationMode,
-    );
-
-    const stream = answeringChain.streamEvents(
       {
         chat_history: history,
         query: message,
       },
-      {
-        version: 'v1',
-      },
+      emitter,
     );
 
-    this.handleStream(stream, emitter);
-
     return emitter;
   }
 }