[feat] Enhance streaming capabilities to be per-port not per-task.

- Added support for streaming output in various AI tasks, including TextGenerationTask, TextQuestionAnswerTask, and TextRewriterTask, allowing real-time data processing.
- Updated task registration and execution methods to accommodate new streaming functionalities, improving responsiveness.
- Refactored package dependencies in bun.lock and package.json to use catalog references for better version management.
- Introduced new streaming-related UI components and styles in the web examples, enhancing user experience during task execution.
- Improved task graph handling for streaming events, ensuring accurate updates and visual feedback in the UI.

Author: sroussey

bun.lock

@@ -296,6 +296,7 @@ export function AddBaseCommands(program: Command) {
     .option("--model-path <path>", "model path or URI")
     .option("--model-dtype <dtype>", "model dtype (default: auto)")
     .option("--model-pipeline <pipeline>", "override inferred pipeline type")
+    .option("--stream", "enable streaming output")
     .action(async (text, options) => {
       let model: string | ModelConfig;
 
@@ -337,6 +338,7 @@ export function AddBaseCommands(program: Command) {
     .option("--model-path <path>", "model path or URI")
     .option("--model-dtype <dtype>", "model dtype (default: auto)")
     .option("--model-pipeline <pipeline>", "override inferred pipeline type")
+    .option("--stream", "enable streaming output")
     .action(async (text, options) => {
       let model: string | ModelConfig;
 
@@ -511,6 +513,7 @@ export function AddBaseCommands(program: Command) {
     .option("--model-path <path>", "model path or URI")
     .option("--model-dtype <dtype>", "model dtype (default: auto)")
     .option("--model-pipeline <pipeline>", "override inferred pipeline type")
+    .option("--stream", "enable streaming output")
     .action(async (question, context, options) => {
       let model: string | ModelConfig;
 

examples/cli/src/TaskCLI.ts

@@ -5,10 +5,10 @@
  */
 
 import { DownloadModelTask } from "@workglow/ai";
-import { ITask, ITaskGraph, TaskStatus } from "@workglow/task-graph";
+import { ITask, ITaskGraph, TaskStatus, type StreamEvent } from "@workglow/task-graph";
 import { ArrayTask } from "@workglow/tasks";
 import type { FC } from "react";
-import { memo, useEffect, useState } from "react";
+import { memo, useEffect, useRef, useState } from "react";
 import { Box, Text } from "retuink";
 import { createBar, Spinner, symbols } from "./Elements";
 
@@ -22,6 +22,10 @@ const StatusIcon = memo(
       sym = <Spinner color="yellow" />;
     }
 
+    if (status === TaskStatus.STREAMING) {
+      sym = <Spinner color="cyan" />;
+    }
+
     if (status === TaskStatus.ABORTING) {
       sym = <Text color="yellow">{symbols.warning}</Text>;
     }
@@ -70,6 +74,9 @@ export const TaskUI: FC<{
   const [arrayProgress, setArrayProgress] = useState<{ completed: number; total: number } | null>(
     null
   );
+  const [streamingText, setStreamingText] = useState<string>("");
+  const [isStreaming, setIsStreaming] = useState<boolean>(false);
+  const streamingTextRef = useRef<string>("");
 
   useEffect(() => {
     const onStart = () => {
@@ -143,6 +150,29 @@ export const TaskUI: FC<{
       setStatus(TaskStatus.ABORTING);
       setError((prevErr) => (prevErr ? `${prevErr}\nAborted` : "Aborted"));
     };
+
+    const onStreamStart = () => {
+      setStatus(TaskStatus.STREAMING);
+      setIsStreaming(true);
+      setStreamingText("");
+      streamingTextRef.current = "";
+    };
+
+    const onStreamChunk = (event: StreamEvent) => {
+      if (event.type === "text-delta") {
+        streamingTextRef.current += event.textDelta;
+        setStreamingText(streamingTextRef.current);
+      } else if (event.type === "snapshot") {
+        const text = typeof event.data === "string" ? event.data : JSON.stringify(event.data);
+        streamingTextRef.current = text;
+        setStreamingText(text);
+      }
+    };
+
+    const onStreamEnd = () => {
+      setIsStreaming(false);
+    };
+
     onRegenerate();
     const targets = graph.getTargetTasks(task.config.id);
     const unique = [...new Map(targets.map((t) => [t.config.id, t])).values()];
@@ -154,6 +184,9 @@ export const TaskUI: FC<{
     task.on("error", onError);
     task.on("regenerate", onRegenerate);
     task.on("abort", onAbort);
+    task.on("stream_start", onStreamStart);
+    task.on("stream_chunk", onStreamChunk);
+    task.on("stream_end", onStreamEnd);
 
     return () => {
       task.off("start", onStart);
@@ -162,6 +195,9 @@ export const TaskUI: FC<{
       task.off("error", onError);
       task.off("regenerate", onRegenerate);
       task.off("abort", onAbort);
+      task.off("stream_start", onStreamStart);
+      task.off("stream_chunk", onStreamChunk);
+      task.off("stream_end", onStreamEnd);
     };
   }, [task, graph]);
 
@@ -181,6 +217,12 @@ export const TaskUI: FC<{
           </Box>
         )}
 
+        {status === TaskStatus.STREAMING && (
+          <Box marginLeft={2} flexShrink={1}>
+            <Text color="cyan">[streaming]</Text>
+          </Box>
+        )}
+
         {status === TaskStatus.PROCESSING && progress > 0 && (
           <Box marginLeft={2} flexShrink={1}>
             <Text dimColor>[{status}]</Text>
@@ -216,6 +258,11 @@ export const TaskUI: FC<{
             <Text color="gray">{`${symbols.arrowDashedRight} ${createBar(progress / 100, 10)} ${progressGenerationText ?? ""}`}</Text>
           </Box>
         )}
+      {(status === TaskStatus.STREAMING || isStreaming) && streamingText && (
+        <Box marginLeft={2}>
+          <Text color="cyan" wrap="truncate">{`${symbols.arrowDashedRight} ${streamingText}`}</Text>
+        </Box>
+      )}
       {arrayProgress && (
         <Box marginLeft={2}>
           <Text color="gray">{`${symbols.arrowDashedRight} Processing array tasks: ${arrayProgress.completed}/${arrayProgress.total} completed ${createBar(arrayProgress.completed / arrayProgress.total, 10)}`}</Text>

examples/cli/src/components/TaskUI.tsx

@@ -13,20 +13,29 @@ export const ProgressBar: React.FC<{
   progress: number;
   status: TaskStatus;
   showText: boolean;
-}> = ({ progress, status, showText }) => (
-  <>
-    <div className="w-full bg-[rgba(28,35,50,0.6)] rounded-full overflow-hidden h-2 my-2">
-      <div
-        className={`h-full rounded-full transition-[width] duration-300 ease-in-out ${
-          status === TaskStatus.PROCESSING
-            ? "bg-gradient-to-r from-[#2a8af6] via-[#a853ba] to-[#2a8af6] bg-[length:200%_100%] animate-progress"
-            : getStatusColorBg(status)
-        }`}
-        style={{
-          width: `${Math.round(progress)}%`,
-        }}
-      />
-    </div>
-    {showText && <div className="text-xs text-gray-500">Progress: {Math.round(progress)}%</div>}
-  </>
-);
+}> = ({ progress, status, showText }) => {
+  const isStreaming = status === TaskStatus.STREAMING;
+
+  return (
+    <>
+      <div className="w-full bg-[rgba(28,35,50,0.6)] rounded-full overflow-hidden h-2 my-2">
+        <div
+          className={`h-full rounded-full transition-[width] duration-300 ease-in-out ${
+            isStreaming
+              ? "bg-blue-500 animate-streaming-pulse"
+              : status === TaskStatus.PROCESSING
+                ? "bg-gradient-to-r from-[#2a8af6] via-[#a853ba] to-[#2a8af6] bg-[length:200%_100%] animate-progress"
+                : getStatusColorBg(status)
+          }`}
+          style={{
+            width: isStreaming ? "100%" : `${Math.round(progress)}%`,
+          }}
+        />
+      </div>
+      {showText && !isStreaming && (
+        <div className="text-xs text-gray-500">Progress: {Math.round(progress)}%</div>
+      )}
+      {showText && isStreaming && <div className="text-xs text-blue-400">Streaming...</div>}
+    </>
+  );
+};

examples/web/src/components/ProgressBar.tsx

@@ -332,6 +332,36 @@
   overflow-y: auto;
 }
 
+/* Streaming progress bar */
+.progress-bar.streaming {
+  background: #3b82f6;
+  width: 100%;
+  animation: streaming-pulse 1.5s ease-in-out infinite;
+}
+
+@keyframes streaming-pulse {
+  0%,
+  100% {
+    opacity: 1;
+  }
+  50% {
+    opacity: 0.5;
+  }
+}
+
+.react-flow__node .wrapper.gradient.streaming:before {
+  content: "";
+  background: conic-gradient(
+    from -160deg at 50% 50%,
+    #3b82f6 0deg,
+    #60a5fa 120deg,
+    #3b82f6 240deg,
+    rgba(59, 130, 246, 0) 360deg
+  );
+  animation: spinner 3s linear infinite;
+  transform: translate(-50%, -50%) rotate(0deg);
+}
+
 /* Transitions */
 .fade-in {
   animation: fade-in 0.3s ease-in-out;

examples/web/src/graph/RunGraphFlow.css

@@ -189,6 +189,10 @@ export const RunGraphFlow: React.FC<{
           unsubscribes.push(unsub);
         });
 
+        // Streaming status: update node when a task starts streaming
+        const streamStartUnsub = task.subscribe("stream_start", () => updateNode(setNodes, task));
+        unsubscribes.push(streamStartUnsub);
+
         // Progress events (just node update)
         const progressUnsub = task.subscribe("progress", () => updateNode(setNodes, task));
         unsubscribes.push(progressUnsub);
@@ -201,8 +205,9 @@ export const RunGraphFlow: React.FC<{
       });
 
       const dataflows = graph.getDataflows();
+      const dataflowEvents = [...statusEvents, "streaming"] as const;
       dataflows.forEach((dataflow) => {
-        statusEvents.forEach((eventName) => {
+        dataflowEvents.forEach((eventName) => {
           const unsub = dataflow.subscribe(eventName, () => updateEdgeStatus(dataflow));
           unsubscribes.push(unsub);
         });