Merge branch 'dani/rag' of https://github.com/Clarifai/clarifai-nodejs …

…into dani/logging

package.json

@@ -39,7 +39,6 @@
     "@parcel/packager-ts": "^2.11.0",
     "@parcel/transformer-typescript-types": "^2.11.0",
     "@types/async": "^3.2.24",
-    "@types/cli-progress": "^3.11.5",
     "@types/google-protobuf": "^3.15.12",
     "@types/js-yaml": "^4.0.9",
     "@types/lodash": "^4.17.0",
@@ -69,7 +68,6 @@
     "async": "^3.2.5",
     "chalk": "^5.3.0",
     "clarifai-nodejs-grpc": "^10.3.2",
-    "cli-progress": "^3.12.0",
     "csv-parse": "^5.5.5",
     "from-protobuf-object": "^1.0.2",
     "google-protobuf": "^3.21.2",

src/client/dataset.ts

@@ -7,7 +7,7 @@ import { APIError, UserError } from "../errors";
 import { ClarifaiUrl, ClarifaiUrlHelper } from "../urls/helper";
 import { AuthConfig } from "../utils/types";
 import { Lister } from "./lister";
-import { Input } from "./input";
+import { Input, InputBulkUpload } from "./input";
 import {
   DeleteDatasetVersionsRequest,
   ListDatasetVersionsRequest,
@@ -144,11 +144,13 @@ export class Dataset extends Lister {
     inputType,
     labels = false,
     batchSize = this.batchSize,
+    uploadProgressEmitter,
   }: {
     folderPath: string;
     inputType: "image" | "text";
     labels: boolean;
     batchSize?: number;
+    uploadProgressEmitter?: InputBulkUpload;
   }): Promise<void> {
     if (["image", "text"].indexOf(inputType) === -1) {
       throw new UserError("Invalid input type");
@@ -171,6 +173,7 @@ export class Dataset extends Lister {
     await this.input.bulkUpload({
       inputs: inputProtos,
       batchSize: batchSize,
+      uploadProgressEmitter,
     });
   }
 
@@ -180,12 +183,14 @@ export class Dataset extends Lister {
     csvType,
     labels = true,
     batchSize = 128,
+    uploadProgressEmitter,
   }: {
     csvPath: string;
     inputType?: "image" | "text" | "video" | "audio";
     csvType: "raw" | "url" | "file";
     labels?: boolean;
     batchSize?: number;
+    uploadProgressEmitter?: InputBulkUpload;
   }): Promise<void> {
     if (!["image", "text", "video", "audio"].includes(inputType)) {
       throw new UserError(
@@ -214,6 +219,7 @@ export class Dataset extends Lister {
     await this.input.bulkUpload({
       inputs: inputProtos,
       batchSize: batchSize,
+      uploadProgressEmitter,
     });
   }
 }

src/client/input.ts

@@ -15,7 +15,7 @@ import {
   Text,
   Video,
 } from "clarifai-nodejs-grpc/proto/clarifai/api/resources_pb";
-import { AuthConfig } from "../utils/types";
+import { AuthConfig, Polygon as PolygonType } from "../utils/types";
 import { Lister } from "./lister";
 import { Buffer } from "buffer";
 import fs from "fs";
@@ -42,11 +42,11 @@ import { StatusCode } from "clarifai-nodejs-grpc/proto/clarifai/api/status/statu
 import os from "os";
 import chunk from "lodash/chunk";
 import { Status } from "clarifai-nodejs-grpc/proto/clarifai/api/status/status_pb";
-import cliProgress from "cli-progress";
 import async from "async";
 import { MAX_RETRIES } from "../constants/dataset";
 import { APIError, UserError } from "../errors";
 import { logger } from "../utils/logging";
+import { EventEmitter } from "events";
 
 interface CSVRecord {
   inputid: string;
@@ -56,6 +56,29 @@ interface CSVRecord {
   geopoints: string;
 }
 
+interface UploadEvents {
+  start: ProgressEvent;
+  progress: ProgressEvent;
+  error: ErrorEvent;
+  end: ProgressEvent;
+}
+
+interface ProgressEvent {
+  current: number;
+  total: number;
+}
+
+interface ErrorEvent {
+  error: Error;
+}
+
+type BulkUploadEventEmitter<T> = EventEmitter & {
+  emit<K extends keyof T>(event: K, payload: T[K]): boolean;
+  on<K extends keyof T>(event: K, listener: (payload: T[K]) => void): void;
+};
+
+export type InputBulkUpload = BulkUploadEventEmitter<UploadEvents>;
+
 /**
  * Inputs is a class that provides access to Clarifai API endpoints related to Input information.
  * @noInheritDoc
@@ -743,9 +766,9 @@ export class Input extends Lister {
   }: {
     inputId: string;
     label: string;
-    polygons: number[][][];
+    polygons: PolygonType[];
   }): Annotation {
-    const polygonsSchema = z.array(z.array(z.array(z.number())));
+    const polygonsSchema = z.array(z.array(z.tuple([z.number(), z.number()])));
     try {
       polygonsSchema.parse(polygons);
     } catch {
@@ -1013,19 +1036,18 @@ export class Input extends Lister {
   bulkUpload({
     inputs,
     batchSize: providedBatchSize = 128,
+    uploadProgressEmitter,
   }: {
     inputs: GrpcInput[];
     batchSize?: number;
+    uploadProgressEmitter?: InputBulkUpload;
   }): Promise<void> {
     const batchSize = Math.min(128, providedBatchSize);
     const chunkedInputs = chunk(inputs, batchSize);
 
-    const progressBar = new cliProgress.SingleBar(
-      {},
-      cliProgress.Presets.shades_classic,
-    );
-
-    progressBar.start(chunkedInputs.length, 0);
+    let currentProgress = 0;
+    const total = chunkedInputs.length;
+    uploadProgressEmitter?.emit("start", { current: currentProgress, total });
 
     return new Promise<void>((resolve, reject) => {
       async.mapLimit(
@@ -1037,7 +1059,11 @@ export class Input extends Lister {
               this.retryUploads({
                 failedInputs,
               }).finally(() => {
-                progressBar.increment();
+                currentProgress++;
+                uploadProgressEmitter?.emit("progress", {
+                  current: currentProgress,
+                  total,
+                });
                 callback(null, failedInputs);
               });
             })
@@ -1048,9 +1074,10 @@ export class Input extends Lister {
         (err) => {
           if (err) {
             logger.error(`Error processing batches ${err.message}`);
+            uploadProgressEmitter?.emit("error");
             reject(err);
           }
-          progressBar.stop();
+          uploadProgressEmitter?.emit("end", { current: total, total });
           logger.info("All inputs processed");
           resolve();
         },

src/client/rag.ts

@@ -54,8 +54,6 @@ type workflowSchema = ReturnType<typeof validateWorkflow>;
 export class RAG {
   private authConfig: AuthConfig;
 
-  private chatStateId: string = "";
-
   public promptWorkflow: Workflow;
 
   public app: App;
@@ -402,40 +400,17 @@ export class RAG {
       );
     }
 
-    if (clientManageState) {
-      const singlePrompt = convertMessagesToStr(messages);
-      const inputProto = Input.getTextInput({
-        inputId: uuidv4(),
-        rawText: singlePrompt,
-      });
-      const response = await this.promptWorkflow.predict({
-        inputs: [inputProto],
-      });
-      const outputsList = response.resultsList?.[0]?.outputsList;
-      const output = outputsList[outputsList.length - 1];
-      messages.push(formatAssistantMessage(output?.data?.text?.raw ?? ""));
-      return messages;
-    }
-
-    // Server side chat state management
-    const message = messages[messages.length - 1].content;
-    if (!message.length) {
-      throw new UserError("Empty message supplied.");
-    }
-
-    const chatStateId = this.chatStateId !== null ? this.chatStateId : "init";
+    const singlePrompt = convertMessagesToStr(messages);
     const inputProto = Input.getTextInput({
       inputId: uuidv4(),
-      rawText: message,
+      rawText: singlePrompt,
     });
     const response = await this.promptWorkflow.predict({
       inputs: [inputProto],
-      workflowStateId: chatStateId,
     });
-
-    this.chatStateId = response.workflowState?.id ?? "";
     const outputsList = response.resultsList?.[0]?.outputsList;
     const output = outputsList[outputsList.length - 1];
-    return [formatAssistantMessage(output?.data?.text?.raw ?? "")];
+    messages.push(formatAssistantMessage(output?.data?.text?.raw ?? ""));
+    return messages;
   }
 }

src/datasets/upload/features.ts

@@ -1,4 +1,5 @@
 import { JavaScriptValue } from "google-protobuf/google/protobuf/struct_pb";
+import { Polygon } from "../../utils/types";
 
 export interface TextFeatures {
   imagePath?: undefined;
@@ -33,7 +34,7 @@ export interface VisualDetectionFeatures {
 export interface VisualSegmentationFeatures {
   imagePath: string;
   labels: Array<string | number>;
-  polygons: Array<Array<Array<number>>>;
+  polygons: Polygon[];
   geoInfo?: [number, number];
   id?: number;
   metadata?: Record<string, JavaScriptValue>;

src/utils/misc.ts

@@ -71,7 +71,7 @@ export function mergeObjects(obj1: AuthConfig, obj2: AuthConfig): AuthConfig {
 export class BackoffIterator {
   private count: number;
 
-  constructor({ count = 0 }: { count?: number } = { count: 0 }) {
+  constructor({ count } = { count: 0 }) {
     this.count = count;
   }
 

src/utils/types.ts

@@ -21,3 +21,6 @@ export type GrpcWithCallback<TRequest, TResponse> = (
 export type PaginationRequestParams<T extends Record<string, unknown>> =
   | Omit<Partial<T>, "userAppId" | "pageNo" | "perPage">
   | Record<string, never>;
+
+export type Point = [number, number];
+export type Polygon = Point[];

tests/client/rag.integration.test.ts

@@ -64,7 +64,7 @@ describe("Rag", async () => {
     const messages = [{ role: "human", content: "What is 1 + 1?" }];
     const newMessages = await rag.chat({ messages, clientManageState: true });
     expect(newMessages.length).toBe(2);
-  });
+  }, 10000);
 
   // TODO: Server side state management is not supported yet - work in progress
   it.skip("should predict & manage state on the server", async () => {

tests/client/search.integration.test.ts

@@ -1,9 +1,10 @@
 import path from "path";
 import { getSchema } from "../../src/schema/search";
 import { z } from "zod";
-import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
 import { App, Dataset, Input, Search, User } from "../../src/index";
 import { Hit } from "clarifai-nodejs-grpc/proto/clarifai/api/resources_pb";
+import EventEmitter from "events";
 
 const NOW = Date.now().toString() + "-search";
 const CREATE_APP_USER_ID = import.meta.env.VITE_CLARIFAI_USER_ID;
@@ -197,11 +198,48 @@ describe("Search", () => {
       },
       datasetId: datasetObj.id,
     });
+    const eventEmitter = new EventEmitter();
+    const eventHandler = {
+      start: (...args: unknown[]) => console.log("start", args),
+      progress: (...args: unknown[]) => console.log("progress", args),
+      end: (...args: unknown[]) => console.log("end", args),
+      error: (...args: unknown[]) => console.log("error", args),
+    };
+    const startSpy = vi.spyOn(eventHandler, "start");
+    const progressSpy = vi.spyOn(eventHandler, "progress");
+    const endSpy = vi.spyOn(eventHandler, "end");
+    const errorSpy = vi.spyOn(eventHandler, "error");
+    eventEmitter.on("start", (start) => {
+      eventHandler.start(start);
+    });
+    eventEmitter.on("progress", (progress) => {
+      eventHandler.progress(progress);
+    });
+    eventEmitter.on("end", (progress) => {
+      eventHandler.end(progress);
+    });
+    eventEmitter.on("error", (error) => {
+      eventHandler.error(error);
+    });
     await dataset.uploadFromFolder({
       folderPath: DATASET_IMAGES_DIR,
       inputType: "image",
       labels: false,
+      uploadProgressEmitter: eventEmitter,
     });
+    expect(startSpy).toHaveBeenNthCalledWith(
+      1,
+      expect.objectContaining({ current: 0, total: 1 }),
+    );
+    expect(progressSpy).toHaveBeenNthCalledWith(
+      1,
+      expect.objectContaining({ current: 1, total: 1 }),
+    );
+    expect(endSpy).toHaveBeenNthCalledWith(
+      1,
+      expect.objectContaining({ current: 1, total: 1 }),
+    );
+    expect(errorSpy).not.toHaveBeenCalled();
   }, 50000);
 
   it("should get expected hits for filters", async () => {

vitest.config.mjs

@@ -15,6 +15,7 @@ export default defineConfig({
   test: {
     coverage: {
       reporter: ["text", "json", "html", "clover", "json-summary"],
+      include: ["src/**/*"],
     },
   },
 });