refactor: move streamToAsyncIterable to private

Author: thucpn

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

@@ -1,6 +1,5 @@
 import { initObservability } from "@/app/observability";
-import { LlamaIndexAdapter, parseDataStreamPart, type Message } from "ai";
-import { EngineResponse } from "llamaindex";
+import { LlamaIndexAdapter, type Message } from "ai";
 import { NextRequest, NextResponse } from "next/server";
 import { initSettings } from "./engine/settings";
 import {
@@ -42,8 +41,7 @@ export async function POST(request: NextRequest) {
     });
     const { stream, dataStream: data } =
       await createStreamFromWorkflowContext(context);
-    const streamIterable = streamToAsyncIterable(stream);
-    return LlamaIndexAdapter.toDataStreamResponse(streamIterable, { data });
+    return LlamaIndexAdapter.toDataStreamResponse(stream, { data });
   } catch (error) {
     console.error("[LlamaIndex]", error);
     return NextResponse.json(
@@ -56,25 +54,3 @@ export async function POST(request: NextRequest) {
     );
   }
 }
-
-function streamToAsyncIterable(stream: ReadableStream<string>) {
-  const streamIterable: AsyncIterable<EngineResponse> = {
-    [Symbol.asyncIterator]() {
-      const reader = stream.getReader();
-      return {
-        async next() {
-          const { done, value } = await reader.read();
-          if (done) {
-            return { done: true, value: undefined };
-          }
-          const delta = parseDataStreamPart(value)?.value.toString() || "";
-          return {
-            done: false,
-            value: { delta } as unknown as EngineResponse,
-          };
-        },
-      };
-    },
-  };
-  return streamIterable;
-}

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

@@ -3,13 +3,17 @@ import {
   WorkflowContext,
   WorkflowEvent,
 } from "@llamaindex/workflow";
-import { StreamData, createStreamDataTransformer } from "ai";
-import { ChatResponseChunk } from "llamaindex";
+import {
+  StreamData,
+  createStreamDataTransformer,
+  parseDataStreamPart,
+} from "ai";
+import { ChatResponseChunk, EngineResponse } from "llamaindex";
 import { AgentRunEvent } from "./type";
 
 export async function createStreamFromWorkflowContext<Input, Output, Context>(
   context: WorkflowContext<Input, Output, Context>,
-): Promise<{ stream: ReadableStream<string>; dataStream: StreamData }> {
+): Promise<{ stream: AsyncIterable<EngineResponse>; dataStream: StreamData }> {
   const dataStream = new StreamData();
   const encoder = new TextEncoder();
   let generator: AsyncGenerator<ChatResponseChunk> | undefined;
@@ -47,10 +51,14 @@ export async function createStreamFromWorkflowContext<Input, Output, Context>(
     },
   });
 
+  const stream = mainStream
+    .pipeThrough(createStreamDataTransformer())
+    .pipeThrough(new TextDecoderStream());
+
+  const streamIterable = streamToAsyncIterable(stream);
+
   return {
-    stream: mainStream
-      .pipeThrough(createStreamDataTransformer())
-      .pipeThrough(new TextDecoderStream()),
+    stream: streamIterable,
     dataStream,
   };
 }
@@ -71,3 +79,25 @@ function handleEvent(
     });
   }
 }
+
+function streamToAsyncIterable(stream: ReadableStream<string>) {
+  const streamIterable: AsyncIterable<EngineResponse> = {
+    [Symbol.asyncIterator]() {
+      const reader = stream.getReader();
+      return {
+        async next() {
+          const { done, value } = await reader.read();
+          if (done) {
+            return { done: true, value: undefined };
+          }
+          const delta = parseDataStreamPart(value)?.value.toString() || "";
+          return {
+            done: false,
+            value: { delta } as unknown as EngineResponse,
+          };
+        },
+      };
+    },
+  };
+  return streamIterable;
+}