Update docusaurus.config.js

Co-authored-by: Alex Yang <himself65@outlook.com>

.changeset/perfect-zebras-live.md

@@ -0,0 +1,6 @@
+---
+"@llamaindex/unit-test": patch
+"@llamaindex/workflow": patch
+---
+
+feat(workflow): allow send event with no output

.changeset/witty-worms-marry.md

@@ -0,0 +1,5 @@
+---
+"@llamaindex/core": patch
+---
+
+feat: add async support to BaseChatStore and BaseChatStoreMemory

README.md

@@ -1,12 +1,16 @@
-# LlamaIndex.TS
+<p align="center">
+  <img height="100" width="100" alt="LlamaIndex logo" src="https://ts.llamaindex.ai/square.svg" />
+</p>
+<h1 align="center">LlamaIndex.TS</h1>
+<h3 align="center">
+  Data framework for your LLM application.
+</h3>
 
 [![NPM Version](https://img.shields.io/npm/v/llamaindex)](https://www.npmjs.com/package/llamaindex)
 [![NPM License](https://img.shields.io/npm/l/llamaindex)](https://www.npmjs.com/package/llamaindex)
 [![NPM Downloads](https://img.shields.io/npm/dm/llamaindex)](https://www.npmjs.com/package/llamaindex)
 [![Discord](https://img.shields.io/discord/1059199217496772688)](https://discord.com/invite/eN6D2HQ4aX)
 
-LlamaIndex is a data framework for your LLM application.
-
 Use your own data with large language models (LLMs, OpenAI ChatGPT and others) in JS runtime environments with TypeScript support.
 
 Documentation: https://ts.llamaindex.ai/

apps/docs/CHANGELOG.md

@@ -1,5 +1,11 @@
 # docs
 
+## 0.0.117
+
+### Patch Changes
+
+- @llamaindex/examples@0.0.15
+
 ## 0.0.116
 
 ### Patch Changes

apps/docs/docusaurus.config.js

@@ -62,6 +62,12 @@ const config = {
     ({
       // Replace with your project's social card
       image: "img/favicon.png", // TODO change this
+      announcementBar: {
+        id: "migrate_to_next",
+        content:
+          'We are migrating to Next.js based documentation. Check it out <a href="https://ts.llamaindex.ai/docs/llamaindex">here</a>!',
+        isCloseable: false,
+      },
       navbar: {
         title: "LlamaIndex.TS",
         logo: {

apps/docs/package.json

@@ -1,6 +1,6 @@
 {
   "name": "docs",
-  "version": "0.0.116",
+  "version": "0.0.117",
   "private": true,
   "scripts": {
     "docusaurus": "docusaurus",

apps/next/CHANGELOG.md

@@ -1,5 +1,13 @@
 # @llamaindex/doc
 
+## 0.0.15
+
+### Patch Changes
+
+- Updated dependencies [ea92b69]
+- Updated dependencies [fadc8b8]
+  - @llamaindex/workflow@0.0.5
+
 ## 0.0.14
 
 ### Patch Changes

apps/next/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@llamaindex/doc",
-  "version": "0.0.14",
+  "version": "0.0.15",
   "private": true,
   "scripts": {
     "build": "pnpm run build:docs && next build",

examples/CHANGELOG.md

@@ -1,5 +1,13 @@
 # examples
 
+## 0.0.15
+
+### Patch Changes
+
+- Updated dependencies [ea92b69]
+- Updated dependencies [fadc8b8]
+  - @llamaindex/workflow@0.0.5
+
 ## 0.0.14
 
 ### Patch Changes

examples/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@llamaindex/examples",
   "private": true,
-  "version": "0.0.14",
+  "version": "0.0.15",
   "dependencies": {
     "@aws-crypto/sha256-js": "^5.2.0",
     "@azure/cosmos": "^4.1.1",
@@ -9,7 +9,7 @@
     "@datastax/astra-db-ts": "^1.4.1",
     "@llamaindex/core": "^0.4.7",
     "@llamaindex/readers": "^1.0.8",
-    "@llamaindex/workflow": "^0.0.4",
+    "@llamaindex/workflow": "^0.0.5",
     "@notionhq/client": "^2.2.15",
     "@pinecone-database/pinecone": "^3.0.2",
     "@vercel/postgres": "^0.10.0",

packages/core/src/memory/base.ts

@@ -71,15 +71,15 @@ export abstract class BaseChatStoreMemory<
     return this.chatStore.getMessages(this.chatStoreKey);
   }
 
-  put(messages: ChatMessage<AdditionalMessageOptions>) {
+  put(messages: ChatMessage<AdditionalMessageOptions>): void | Promise<void> {
     this.chatStore.addMessage(this.chatStoreKey, messages);
   }
 
-  set(messages: ChatMessage<AdditionalMessageOptions>[]) {
+  set(messages: ChatMessage<AdditionalMessageOptions>[]): void | Promise<void> {
     this.chatStore.setMessages(this.chatStoreKey, messages);
   }
 
-  reset() {
+  reset(): void | Promise<void> {
     this.chatStore.deleteMessages(this.chatStoreKey);
   }
 }

packages/core/src/storage/chat-store/base-chat-store.ts

@@ -19,5 +19,7 @@ export abstract class BaseChatStore<
   ): void;
   abstract deleteMessages(key: string): void;
   abstract deleteMessage(key: string, idx: number): void;
-  abstract getKeys(): IterableIterator<string>;
+  abstract getKeys():
+    | IterableIterator<string>
+    | Promise<IterableIterator<string>>;
 }

packages/workflow/CHANGELOG.md

@@ -1,5 +1,12 @@
 # @llamaindex/workflow
 
+## 0.0.5
+
+### Patch Changes
+
+- ea92b69: fix: output event check
+- fadc8b8: feat: recoverable context with error handling
+
 ## 0.0.4
 
 ### Patch Changes

packages/workflow/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@llamaindex/workflow",
   "description": "Workflow API",
-  "version": "0.0.4",
+  "version": "0.0.5",
   "type": "module",
   "types": "dist/index.d.ts",
   "module": "dist/index.js",

packages/workflow/src/workflow-context.ts

@@ -13,19 +13,18 @@ export type StepHandler<
     AnyWorkflowEventConstructor | StartEventConstructor,
     ...(AnyWorkflowEventConstructor | StopEventConstructor)[],
   ] = [AnyWorkflowEventConstructor | StartEventConstructor],
-  Out extends [
-    AnyWorkflowEventConstructor | StartEventConstructor,
-    ...(AnyWorkflowEventConstructor | StopEventConstructor)[],
-  ] = [AnyWorkflowEventConstructor | StopEventConstructor],
+  Out extends (AnyWorkflowEventConstructor | StopEventConstructor)[] = [],
 > = (
   context: HandlerContext<Data>,
   ...events: {
     [K in keyof Inputs]: InstanceType<Inputs[K]>;
   }
 ) => Promise<
-  {
-    [K in keyof Out]: InstanceType<Out[K]>;
-  }[number]
+  Out extends []
+    ? void
+    : {
+        [K in keyof Out]: InstanceType<Out[K]>;
+      }[number]
 >;
 
 export type ReadonlyStepMap<Data> = ReadonlyMap<
@@ -275,7 +274,7 @@ export class WorkflowContext<Start = string, Stop = string, Data = unknown>
    */
   #createStreamEvents(): AsyncIterableIterator<WorkflowEvent<unknown>> {
     const isPendingEvents = new WeakSet<WorkflowEvent<unknown>>();
-    const pendingTasks = new Set<Promise<WorkflowEvent<unknown>>>();
+    const pendingTasks = new Set<Promise<WorkflowEvent<unknown> | void>>();
     const enqueuedEvents = new Set<WorkflowEvent<unknown>>();
     const stream = new ReadableStream<WorkflowEvent<unknown>>({
       start: async (controller) => {
@@ -325,101 +324,104 @@ export class WorkflowContext<Start = string, Stop = string, Data = unknown>
                   }
                   const [steps, inputsMap, outputsMap] =
                     this.#getStepFunction(event);
-                  const nextEventPromises: Promise<WorkflowEvent<unknown>>[] = [
-                    ...steps,
-                  ]
-                    .map((step) => {
-                      const inputs = [...(inputsMap.get(step) ?? [])];
-                      const acceptableInputs: WorkflowEvent<unknown>[] =
-                        this.#pendingInputQueue.filter((event) =>
-                          inputs.some((input) => event instanceof input),
-                        );
-                      const events: WorkflowEvent<unknown>[] = flattenEvents(
-                        inputs,
-                        [event, ...acceptableInputs],
-                      );
-                      // remove the event from the queue, in case of infinite loop
-                      events.forEach((event) => {
-                        const protocolIdx = this.#queue.findIndex(
-                          (protocol) =>
-                            protocol.type === "event" &&
-                            protocol.event === event,
-                        );
-                        if (protocolIdx !== -1) {
-                          this.#queue.splice(protocolIdx, 1);
-                        }
-                      });
-                      if (events.length !== inputs.length) {
-                        if (this.#verbose) {
-                          console.log(
-                            `Not enough inputs for step ${step.name}, waiting for more events`,
+                  const nextEventPromises: Promise<WorkflowEvent<unknown> | void>[] =
+                    [...steps]
+                      .map((step) => {
+                        const inputs = [...(inputsMap.get(step) ?? [])];
+                        const acceptableInputs: WorkflowEvent<unknown>[] =
+                          this.#pendingInputQueue.filter((event) =>
+                            inputs.some((input) => event instanceof input),
                           );
-                        }
-                        // not enough to run the step, push back to the queue
-                        this.#sendEvent(event);
-                        isPendingEvents.add(event);
-                        return null;
-                      }
-                      if (isPendingEvents.has(event)) {
-                        isPendingEvents.delete(event);
-                      }
-                      if (this.#verbose) {
-                        console.log(
-                          `Running step ${step.name} with inputs ${events}`,
+                        const events: WorkflowEvent<unknown>[] = flattenEvents(
+                          inputs,
+                          [event, ...acceptableInputs],
                         );
-                      }
-                      const data = this.data;
-                      return (step as StepHandler<Data>)
-                        .call(
-                          null,
-                          {
-                            get data() {
-                              return data;
-                            },
-                            sendEvent: this.#sendEvent,
-                            requireEvent: this.#requireEvent,
-                          },
-                          // @ts-expect-error IDK why
-                          ...events.sort((a, b) => {
-                            const aIndex = inputs.indexOf(
-                              a.constructor as AnyWorkflowEventConstructor,
-                            );
-                            const bIndex = inputs.indexOf(
-                              b.constructor as AnyWorkflowEventConstructor,
-                            );
-                            return aIndex - bIndex;
-                          }),
-                        )
-                        .then((nextEvent) => {
+                        // remove the event from the queue, in case of infinite loop
+                        events.forEach((event) => {
+                          const protocolIdx = this.#queue.findIndex(
+                            (protocol) =>
+                              protocol.type === "event" &&
+                              protocol.event === event,
+                          );
+                          if (protocolIdx !== -1) {
+                            this.#queue.splice(protocolIdx, 1);
+                          }
+                        });
+                        if (events.length !== inputs.length) {
                           if (this.#verbose) {
                             console.log(
-                              `Step ${step.name} completed, next event is ${nextEvent}`,
+                              `Not enough inputs for step ${step.name}, waiting for more events`,
                             );
                           }
-                          const outputs = outputsMap.get(step) ?? [];
-                          const outputEvents = flattenEvents(outputs, [
-                            nextEvent,
-                          ]);
-                          if (outputEvents.length !== outputs.length) {
-                            if (this.#strict) {
-                              const error = Error(
-                                `Step ${step.name} returned an unexpected output event ${nextEvent}`,
+                          // not enough to run the step, push back to the queue
+                          this.#sendEvent(event);
+                          isPendingEvents.add(event);
+                          return null;
+                        }
+                        if (isPendingEvents.has(event)) {
+                          isPendingEvents.delete(event);
+                        }
+                        if (this.#verbose) {
+                          console.log(
+                            `Running step ${step.name} with inputs ${events}`,
+                          );
+                        }
+                        const data = this.data;
+                        return (step as StepHandler<Data>)
+                          .call(
+                            null,
+                            {
+                              get data() {
+                                return data;
+                              },
+                              sendEvent: this.#sendEvent,
+                              requireEvent: this.#requireEvent,
+                            },
+                            // @ts-expect-error IDK why
+                            ...events.sort((a, b) => {
+                              const aIndex = inputs.indexOf(
+                                a.constructor as AnyWorkflowEventConstructor,
                               );
-                              controller.error(error);
-                            } else {
-                              console.warn(
-                                `Step ${step.name} returned an unexpected output event ${nextEvent}`,
+                              const bIndex = inputs.indexOf(
+                                b.constructor as AnyWorkflowEventConstructor,
+                              );
+                              return aIndex - bIndex;
+                            }),
+                          )
+                          .then((nextEvent: void | WorkflowEvent<unknown>) => {
+                            if (nextEvent === undefined) {
+                              return;
+                            }
+                            if (this.#verbose) {
+                              console.log(
+                                `Step ${step.name} completed, next event is ${nextEvent}`,
                               );
                             }
-                          }
-                          if (!(nextEvent instanceof StopEvent)) {
-                            this.#pendingInputQueue.unshift(nextEvent);
-                            this.#sendEvent(nextEvent);
-                          }
-                          return nextEvent;
-                        });
-                    })
-                    .filter((promise) => promise !== null);
+                            const outputs = outputsMap.get(step) ?? [];
+                            if (
+                              !outputs.some(
+                                (output) => nextEvent.constructor === output,
+                              )
+                            ) {
+                              if (this.#strict) {
+                                const error = Error(
+                                  `Step ${step.name} returned an unexpected output event ${nextEvent}`,
+                                );
+                                controller.error(error);
+                              } else {
+                                console.warn(
+                                  `Step ${step.name} returned an unexpected output event ${nextEvent}`,
+                                );
+                              }
+                            }
+                            if (!(nextEvent instanceof StopEvent)) {
+                              this.#pendingInputQueue.unshift(nextEvent);
+                              this.#sendEvent(nextEvent);
+                            }
+                            return nextEvent;
+                          });
+                      })
+                      .filter((promise) => promise !== null);
                   nextEventPromises.forEach((promise) => {
                     pendingTasks.add(promise);
                     promise
@@ -432,6 +434,9 @@ export class WorkflowContext<Start = string, Stop = string, Data = unknown>
                   });
                   Promise.race(nextEventPromises)
                     .then((fastestNextEvent) => {
+                      if (fastestNextEvent === undefined) {
+                        return;
+                      }
                       if (!enqueuedEvents.has(fastestNextEvent)) {
                         controller.enqueue(fastestNextEvent);
                         enqueuedEvents.add(fastestNextEvent);
@@ -440,7 +445,10 @@ export class WorkflowContext<Start = string, Stop = string, Data = unknown>
                     })
                     .then(async (fastestNextEvent) =>
                       Promise.all(nextEventPromises).then((nextEvents) => {
-                        for (const nextEvent of nextEvents) {
+                        const events = nextEvents.filter(
+                          (event) => event !== undefined,
+                        );
+                        for (const nextEvent of events) {
                           // do not enqueue the same event twice
                           if (fastestNextEvent !== nextEvent) {
                             if (!enqueuedEvents.has(nextEvent)) {
@@ -452,6 +460,9 @@ export class WorkflowContext<Start = string, Stop = string, Data = unknown>
                       }),
                     )
                     .catch((err) => {
+                      // when the step raise an error, should go back to the previous step
+                      this.#sendEvent(event);
+                      isPendingEvents.add(event);
                       controller.error(err);
                     });
                 }

packages/workflow/src/workflow.ts

@@ -57,10 +57,7 @@ export class Workflow<ContextData, Start, Stop> {
       AnyWorkflowEventConstructor | StartEventConstructor,
       ...(AnyWorkflowEventConstructor | StopEventConstructor)[],
     ],
-    const Out extends [
-      AnyWorkflowEventConstructor | StopEventConstructor,
-      ...(AnyWorkflowEventConstructor | StopEventConstructor)[],
-    ],
+    const Out extends (AnyWorkflowEventConstructor | StopEventConstructor)[],
   >(
     parameters: StepParameters<In, Out>,
     stepFn: (
@@ -69,9 +66,11 @@ export class Workflow<ContextData, Start, Stop> {
         [K in keyof In]: InstanceType<In[K]>;
       }
     ) => Promise<
-      {
-        [K in keyof Out]: InstanceType<Out[K]>;
-      }[number]
+      Out extends []
+        ? void
+        : {
+            [K in keyof Out]: InstanceType<Out[K]>;
+          }[number]
     >,
   ): this {
     const { inputs, outputs } = parameters;

pnpm-lock.yaml

@@ -590,7 +590,7 @@ importers:
         specifier: ^1.0.8
         version: link:../packages/readers
       '@llamaindex/workflow':
-        specifier: ^0.0.4
+        specifier: ^0.0.5
         version: link:../packages/workflow
       '@notionhq/client':
         specifier: ^2.2.15

unit/CHANGELOG.md

@@ -1,5 +1,13 @@
 # @llamaindex/unit-test
 
+## 0.0.22
+
+### Patch Changes
+
+- Updated dependencies [ea92b69]
+- Updated dependencies [fadc8b8]
+  - @llamaindex/workflow@0.0.5
+
 ## 0.0.21
 
 ### Patch Changes

unit/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@llamaindex/unit-test",
   "private": true,
-  "version": "0.0.21",
+  "version": "0.0.22",
   "type": "module",
   "scripts": {
     "test": "vitest run"

unit/workflow/workflow.test.ts

@@ -531,6 +531,21 @@ describe("workflow basic", () => {
     const result = await myWorkflow.run("start");
     expect(result.data).toBe("query result");
   });
+
+  test("allow output with send event", async () => {
+    const myFlow = new Workflow<unknown, string, string>({ verbose: true });
+    myFlow.addStep(
+      {
+        inputs: [StartEvent<string>],
+        outputs: [],
+      },
+      async (context, ev) => {
+        context.sendEvent(new StopEvent(`Hello ${ev.data}!`));
+      },
+    );
+    const result = myFlow.run("world");
+    expect((await result).data).toBe("Hello world!");
+  });
 });
 
 describe("workflow event loop", () => {
@@ -794,6 +809,21 @@ describe("workflow event loop", () => {
       }
     `);
   });
+
+  test("workflow multiple output", async () => {
+    const myFlow = new Workflow<unknown, string, string>({ verbose: true });
+    myFlow.addStep(
+      {
+        inputs: [StartEvent<string>],
+        outputs: [StopEvent<string>, StopEvent<string>],
+      },
+      async (_context, ev) => {
+        return new StopEvent(`Hello ${ev.data}!`);
+      },
+    );
+    const result = await myFlow.run("world").strict();
+    expect(result.data).toBe("Hello world!");
+  });
 });
 
 describe("snapshot", async () => {
@@ -869,3 +899,83 @@ describe("snapshot", async () => {
     expect(fn).toHaveBeenCalledTimes(1);
   });
 });
+
+describe("error", () => {
+  test("error in handler", async () => {
+    const myFlow = new Workflow<boolean, string, string>({ verbose: true });
+    myFlow.addStep(
+      {
+        inputs: [StartEvent<string>],
+        outputs: [StopEvent<string>],
+      },
+      async ({ data }) => {
+        if (!data) {
+          throw new Error("Something went wrong");
+        } else {
+          return new StopEvent(`Hello ${data}!`);
+        }
+      },
+    );
+    await expect(myFlow.run("world")).rejects.toThrow("Something went wrong");
+    {
+      const context = myFlow.run("world");
+      try {
+        for await (const _ of context) {
+          // do nothing
+        }
+      } catch (error) {
+        expect(error).toBeInstanceOf(Error);
+        expect((error as Error).message).toBe("Something went wrong");
+        const snapshot = context.snapshot();
+        const newContext = myFlow.recover(snapshot).with(true);
+        expect((await newContext).data).toBe("Hello true!");
+      }
+    }
+  });
+
+  test("recover in the middle of workflow", async () => {
+    const myFlow = new Workflow<string | undefined, string, string>({
+      verbose: true,
+    });
+
+    class AEvent extends WorkflowEvent<string> {}
+
+    myFlow.addStep(
+      {
+        inputs: [StartEvent<string>],
+        outputs: [AEvent],
+      },
+      async ({ data }) => {
+        if (data !== undefined) {
+          throw new Error("Something went wrong");
+        }
+        return new AEvent("world");
+      },
+    );
+    myFlow.addStep(
+      {
+        inputs: [AEvent],
+        outputs: [StopEvent],
+      },
+      async ({ data }, ev) => {
+        if (data === undefined) {
+          throw new Error("Something went wrong");
+        }
+        return new StopEvent(`Hello, ${data}!`);
+      },
+    );
+    // no context, so will throw error
+    const context = myFlow.run("world");
+    try {
+      for await (const _ of context) {
+        // do nothing
+      }
+    } catch (error) {
+      expect(error).toBeInstanceOf(Error);
+      expect((error as Error).message).toBe("Something went wrong");
+      const snapshot = context.snapshot();
+      const newContext = myFlow.recover(snapshot).with("Recovered Data");
+      expect((await newContext).data).toBe("Hello, Recovered Data!");
+    }
+  });
+});