[feat] Introduce Loop tasks: Map, Reduce, and While

TODO.md

@@ -9,10 +9,10 @@ TODO.md
 - [ ] Datasets Package
   - [x] Documents dataset (mabye rename to DocumentDataset)
   - [ ] Chunks Package (or part of DocumentDataset?)
-  - [ ] Move Model repository to datasets package.
-- [ ] Chunks and nodes are not always the same.
-  - [ ] And we may need to save the chunk's node path. Or paths? or document range? Standard metadata?
-- [ ] Instead of passing doc_id around, pass a document key that is unknonwn (string or object)
+  - [x] Move Model repository to datasets package.
+- [x] Chunks and nodes are not always the same.
+  - [x] And we may need to save the chunk's node path. Or paths? or document range? Standard metadata?
+- [ ] Instead of passing doc_id around, pass a document key that is of type unknonwn (string or object)
 
 - [ ] Get a better model for question answering.
 - [ ] Get a better model for named entity recognition, the current one recognized everything as a token, not helpful.
@@ -25,16 +25,11 @@ TODO.md
 - [ ] rename the registration stuff to not look ugly: registerHuggingfaceTransformers() and registerHuggingfaceTransformersUsingWorkers() and registerHuggingfaceTransformersInsideWorker()
 - [ ] fix image transferables
 
-onnx-community/ModernBERT-finetuned-squad-ONNX - summarization
+- [ ] Consider different ways to connect tasks to queues. What is a task? What is a job?
+
+- [ ] Input and outputs are all scalar, arrays, or unions. But what about streams? Stream of items in an array, stream of content for a scalar like a string, etc.
 
-- [x] Auto-generated primary keys for TabularStorage
-  - [x] Schema annotation with `x-auto-generated: true`
-  - [x] Type system with `InsertEntity` for optional auto-generated keys
-  - [x] Support for autoincrement (integer) and UUID (string) strategies
-  - [x] Configurable client-provided keys: "never", "if-missing", "always"
-  - [x] Implementations: InMemory, SQLite, Postgres, Supabase, IndexedDB, FsFolder
-  - [x] Comprehensive test suite (342 tests pass)
-  - [x] Documentation updated
+onnx-community/ModernBERT-finetuned-squad-ONNX - summarization
 
 Rework the Document Dataset. Currently there is a Document storage of tabular storage type, and that should be registered as a "dataset:document:source" meaning the source material in node format. And there is already a "dataset:document-chunk" for the chunk/vector storage which should be registered as a "dataset:document:chunk" with a well defined metadata schema. The two combined should be registered as a "dataset:document" which is the complete document with its source and all its chunks and metadata. This is for convenience but not used by tasks or ai tasks.
 

examples/cli/src/TaskCLI.ts

@@ -4,14 +4,18 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-import { DownloadModelTask, getGlobalModelRepository, type ModelConfig } from "@workglow/ai";
+import {
+  DownloadModelTask,
+  getGlobalModelRepository,
+  registerAiTasks,
+  type ModelConfig,
+} from "@workglow/ai";
 import { HF_TRANSFORMERS_ONNX } from "@workglow/ai-provider";
-import { JsonTaskItem, TaskGraph, Workflow } from "@workglow/task-graph";
-import { DelayTask, JsonTask } from "@workglow/tasks";
+import { JsonTaskItem, registerBaseTasks, TaskGraph, Workflow } from "@workglow/task-graph";
+import { DelayTask, JsonTask, registerCommonTasks } from "@workglow/tasks";
 import type { Command } from "commander";
 import { readFile, writeFile } from "fs/promises";
 import { runTasks } from "./TaskGraphToUI";
-
 /**
  * Read image input from file or stdin
  */
@@ -539,8 +543,18 @@ export function AddBaseCommands(program: Command) {
   program
     .command("json")
     .description("run based on json input")
-    .argument("[json]", "json text to rewrite and vectorize")
-    .action(async (json) => {
+    .argument("[json]", "json text")
+    .option("--file <path>", "read JSON from file")
+    .action(async (jsonArg, options) => {
+      registerBaseTasks();
+      registerCommonTasks();
+      registerAiTasks();
+      let json = jsonArg;
+      if (!json && options.file) {
+        json = (await readFile(options.file, "utf-8")).trim();
+      } else if (!json && !process.stdin.isTTY) {
+        json = await readTextInput();
+      }
       if (!json) {
         const exampleJson: JsonTaskItem[] = [
           {

examples/cli/src/components/TaskUI.tsx

@@ -144,7 +144,9 @@ export const TaskUI: FC<{
       setError((prevErr) => (prevErr ? `${prevErr}\nAborted` : "Aborted"));
     };
     onRegenerate();
-    setDependantChildren(graph.getTargetTasks(task.config.id));
+    const targets = graph.getTargetTasks(task.config.id);
+    const unique = [...new Map(targets.map((t) => [t.config.id, t])).values()];
+    setDependantChildren(unique);
 
     task.on("start", onStart);
     task.on("progress", onProgress);

packages/ai/src/model/ModelRepository.ts

@@ -168,7 +168,7 @@ export class ModelRepository {
    * Enumerates all models in the repository
    * @returns Promise resolving to an array of model instances
    */
-  async enumerateAllModels() {
+  async enumerateAllModels(): Promise<ModelRecord[] | undefined> {
     const models = await this.modelTabularRepository.getAll();
     if (!models || models.length === 0) return undefined;
     return models;
@@ -179,7 +179,7 @@ export class ModelRepository {
    * @param modelId - The model_id of the model to find
    * @returns Promise resolving to the found model or undefined if not found
    */
-  async findByName(model_id: string) {
+  async findByName(model_id: string): Promise<ModelRecord | undefined> {
     if (typeof model_id != "string") return undefined;
     const model = await this.modelTabularRepository.get({ model_id });
     return model ?? undefined;