Merge pull request #2 from datastax/streaming

Implements easy handling of streaming

Author: philnash

README.md

@@ -14,6 +14,7 @@ This package provides an easy way to use the [Langflow API](https://docs.datasta
   - [Initialization](#initialization)
   - [Calling a flow](#calling-a-flow)
     - [Flow reponses](#flow-reponses)
+  - [Streaming](#streaming)
   - [File upload](#file-upload)
   - [Aborting requests](#aborting-requests)
 - [Contributing](#contributing)
@@ -142,6 +143,48 @@ const response = client.flow(flowId).run(input);
 console.log(response.chatOutputText());
 ```
 
+### Streaming
+
+The Langflow API supports streaming responses. Instead of calling `run` on a `Flow` object, you can call `stream` with the same arguments and the response will be a [`ReadableStream`](https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream) of objects.
+
+```js
+const response = await client.flow(flowId).stream(input);
+
+for await (const event of response) {
+  console.log(event);
+}
+```
+
+There are three different events: `add_message`, `token`, and `end`. The events mean:
+
+- `add_message`: a message being added to the chat and can refer to a human input message or a response from an AI
+- `token`: a token that is emitted as part of a message being generated by the flow
+- `end`: all tokens have been returned, this message will also contain a full `FlowResponse`
+
+Event objects have the format:
+
+```typescript
+{
+  event: "add_message" | "token" | "end",
+  data: object
+}
+```
+
+The `event.data` is different per event type. The `token` event type is the simplest and looks like this:
+
+```typescript
+{
+  "event": "token",
+  "data": {
+    "chunk": "hello ",
+    "id": "6686ff20-0c95-40bb-8879-fd90ed3d634e",
+    "timestamp": "2025-02-12 22:18:04 UTC"
+  }
+}
+```
+
+There's more [documentation and examples of a streaming response in the Langflow docs](https://docs.langflow.org/api-reference-api-examples#run-flow).
+
 ### File upload
 
 Chat input components support files as input as well as text. You need to upload your file first, using the file upload function, then provide the file path to the flow as a tweak.

src/errors.ts

@@ -21,21 +21,11 @@ export class LangflowError extends Error {
     super(message, { cause: response });
     this.cause = response;
   }
-
-  response(): Response {
-    return this.cause;
-  }
 }
 
 export class LangflowRequestError extends Error {
-  cause: Error;
-
   constructor(message: string, error: Error) {
     super(message, { cause: error });
     this.cause = error;
   }
-
-  error(): Error {
-    return this.cause;
-  }
 }

src/flow.ts

@@ -78,6 +78,37 @@ export class Flow {
     return new FlowResponse(result);
   }
 
+  async stream(
+    input_value: string,
+    options: Partial<Omit<FlowRequestOptions, "input_value">> = {}
+  ) {
+    const {
+      input_type = "chat",
+      output_type = "chat",
+      session_id,
+      signal,
+    } = options;
+    const tweaks = { ...this.tweaks, ...options.tweaks };
+    const body = JSON.stringify({
+      input_type,
+      output_type,
+      input_value,
+      tweaks,
+      session_id,
+    });
+    const headers = new Headers();
+    headers.set("Content-Type", "application/json");
+    headers.set("Accept", "application/json");
+    const streamingResult = await this.client.stream({
+      path: `/run/${this.id}`,
+      method: "POST",
+      body,
+      headers,
+      signal,
+    });
+    return streamingResult;
+  }
+
   async uploadFile(path: string, options: { signal?: AbortSignal } = {}) {
     const data = await readFile(path);
     const { signal } = options;

src/index.ts

@@ -12,15 +12,21 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-import { fetch, FormData } from "undici";
+import { fetch } from "undici";
 
 import pkg from "../package.json" with { type: "json" };
 import { LangflowError, LangflowRequestError } from "./errors.js";
 import { Flow } from "./flow.js";
-import type { LangflowClientOptions, Tweaks } from "./types.js";
+import type {
+  LangflowClientOptions,
+  RequestOptions,
+  StreamEvent,
+  Tweaks,
+} from "./types.js";
 import { DATASTAX_LANGFLOW_BASE_URL } from "./consts.js";
 
 import { platform, arch } from "os";
+import { NDJSONStream } from "./ndjson.js";
 
 export class LangflowClient {
   baseUrl: string;
@@ -75,33 +81,25 @@ export class LangflowClient {
     }
   }
 
-  flow(flowId: string, tweaks?: Tweaks): Flow {
-    return new Flow(this, flowId, tweaks);
-  }
-
-  async request(options: {
-    path: string;
-    method: string;
-    body: string | FormData;
-    headers: Headers;
-    query?: Record<string, string>;
-    signal?: AbortSignal;
-  }): Promise<unknown> {
-    const { path, method, body, headers, query, signal } = options;
-    const url = new URL(`${this.baseUrl}${this.basePath}${path}`);
+  #setHeaders(headers: Headers) {
     for (const [header, value] of this.defaultHeaders.entries()) {
       if (!headers.has(header)) {
         headers.set(header, value);
       }
     }
-    if (query) {
-      Object.entries(query).forEach(([key, value]) => {
-        url.searchParams.set(key, value);
-      });
-    }
     if (this.apiKey) {
       this.#setApiKey(this.apiKey, headers);
     }
+  }
+
+  flow(flowId: string, tweaks?: Tweaks): Flow {
+    return new Flow(this, flowId, tweaks);
+  }
+
+  async request(options: RequestOptions): Promise<unknown> {
+    const { path, method, body, headers, signal } = options;
+    const url = new URL(`${this.baseUrl}${this.basePath}${path}`);
+    this.#setHeaders(headers);
     try {
       signal?.throwIfAborted();
       const response = await this.fetch(url, { method, body, headers, signal });
@@ -128,4 +126,46 @@ export class LangflowClient {
       throw error;
     }
   }
+
+  async stream(options: RequestOptions): Promise<ReadableStream<StreamEvent>> {
+    const { path, method, body, headers, signal } = options;
+    const url = new URL(`${this.baseUrl}${this.basePath}${path}`);
+    url.searchParams.set("stream", "true");
+    this.#setHeaders(headers);
+    try {
+      signal?.throwIfAborted();
+      const response = await this.fetch(url, {
+        method,
+        body,
+        headers,
+        signal,
+      });
+      if (!response.ok) {
+        throw new LangflowError(
+          `${response.status} - ${response.statusText}`,
+          response
+        );
+      }
+      const ndjsonStream = NDJSONStream<StreamEvent>();
+      if (response.body) {
+        return response.body
+          .pipeThrough(new TextDecoderStream(), { signal })
+          .pipeThrough(ndjsonStream, { signal });
+      } else {
+        throw new LangflowError("No body in the response", response);
+      }
+    } catch (error) {
+      if (
+        error instanceof LangflowError ||
+        (error instanceof DOMException &&
+          (error.name === "AbortError" || error.name === "TimeoutError"))
+      ) {
+        throw error;
+      }
+      if (error instanceof Error) {
+        throw new LangflowRequestError(error.message, error);
+      }
+      throw error;
+    }
+  }
 }

src/ndjson.ts

@@ -0,0 +1,42 @@
+// Copyright DataStax, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+import { QueuingStrategy } from "node:stream/web";
+
+export function NDJSONStream<T>(
+  writableStrategy?: QueuingStrategy<string>,
+  readableStrategy?: QueuingStrategy<T>
+) {
+  let buffer = "";
+  return new TransformStream<string, T>(
+    {
+      transform(chunk, controller) {
+        const text = chunk.toString();
+        const lines = text.split("\n");
+        for (const line of lines) {
+          buffer += line;
+          try {
+            const result = JSON.parse(buffer);
+            controller.enqueue(result);
+            buffer = "";
+          } catch {
+            // Not a valid JSON object
+          }
+        }
+      },
+    },
+    writableStrategy,
+    readableStrategy
+  );
+}