keep createStreamFromWorkflowContext

Author: thucpn

templates/components/multiagent/typescript/express/chat.controller.ts

@@ -5,7 +5,10 @@ import {
   retrieveMessageContent,
 } from "./llamaindex/streaming/annotations";
 import { createWorkflow } from "./workflow/factory";
-import { createStreamFromWorkflowContext } from "./workflow/stream";
+import {
+  createStreamFromWorkflowContext,
+  streamToAsyncIterable,
+} from "./workflow/stream";
 
 export const chat = async (req: Request, res: Response) => {
   try {
@@ -27,10 +30,13 @@ export const chat = async (req: Request, res: Response) => {
 
     const { stream, dataStream } =
       await createStreamFromWorkflowContext(context);
-
-    const streamResponse = LlamaIndexAdapter.toDataStreamResponse(stream, {
-      data: dataStream,
-    });
+    const streamIterable = streamToAsyncIterable(stream);
+    const streamResponse = LlamaIndexAdapter.toDataStreamResponse(
+      streamIterable,
+      {
+        data: dataStream,
+      },
+    );
     if (streamResponse.body) {
       const reader = streamResponse.body.getReader();
       while (true) {

templates/components/multiagent/typescript/nextjs/route.ts

@@ -8,7 +8,10 @@ import {
   retrieveMessageContent,
 } from "./llamaindex/streaming/annotations";
 import { createWorkflow } from "./workflow/factory";
-import { createStreamFromWorkflowContext } from "./workflow/stream";
+import {
+  createStreamFromWorkflowContext,
+  streamToAsyncIterable,
+} from "./workflow/stream";
 
 initObservability();
 initSettings();
@@ -41,7 +44,10 @@ export async function POST(request: NextRequest) {
     });
     const { stream, dataStream } =
       await createStreamFromWorkflowContext(context);
-    return LlamaIndexAdapter.toDataStreamResponse(stream, { data: dataStream });
+    const streamIterable = streamToAsyncIterable(stream);
+    return LlamaIndexAdapter.toDataStreamResponse(streamIterable, {
+      data: dataStream,
+    });
   } catch (error) {
     console.error("[LlamaIndex]", error);
     return NextResponse.json(

templates/components/multiagent/typescript/workflow/stream.ts

@@ -13,7 +13,7 @@ import { AgentRunEvent } from "./type";
 
 export async function createStreamFromWorkflowContext<Input, Output, Context>(
   context: WorkflowContext<Input, Output, Context>,
-): Promise<{ stream: AsyncIterable<EngineResponse>; dataStream: StreamData }> {
+): Promise<{ stream: ReadableStream<string>; dataStream: StreamData }> {
   const dataStream = new StreamData();
   const encoder = new TextEncoder();
   let generator: AsyncGenerator<ChatResponseChunk> | undefined;
@@ -51,14 +51,10 @@ export async function createStreamFromWorkflowContext<Input, Output, Context>(
     },
   });
 
-  const stream = mainStream
-    .pipeThrough(createStreamDataTransformer())
-    .pipeThrough(new TextDecoderStream());
-
-  const streamIterable = streamToAsyncIterable(stream);
-
   return {
-    stream: streamIterable,
+    stream: mainStream
+      .pipeThrough(createStreamDataTransformer())
+      .pipeThrough(new TextDecoderStream()),
     dataStream,
   };
 }
@@ -80,7 +76,7 @@ function handleEvent(
   }
 }
 
-function streamToAsyncIterable(stream: ReadableStream<string>) {
+export function streamToAsyncIterable(stream: ReadableStream<string>) {
   const streamIterable: AsyncIterable<EngineResponse> = {
     [Symbol.asyncIterator]() {
       const reader = stream.getReader();