diff --git a/docs/workflows/common_patterns/branching.mdx b/docs/workflows/common_patterns/branching.mdx index 207576b..e1e20a3 100644 --- a/docs/workflows/common_patterns/branching.mdx +++ b/docs/workflows/common_patterns/branching.mdx @@ -20,27 +20,31 @@ const processNumberEvent = workflowEvent(); const successEvent = workflowEvent(); workflow.handle([inputEvent], async (context, event) => { - const { sendEvent } = context; if (typeof event.data === "string") { - sendEvent(processStringEvent.with({ data: event.data })); + return processStringEvent.with(event.data); } else { - sendEvent(processNumberEvent.with({ data: event.data })); + return processNumberEvent.with(event.data); } }); workflow.handle([processStringEvent], async (context, event) => { - const { sendEvent } = context; - sendEvent(successEvent.with({ data: `Processed string ${event.data}` })); + return successEvent.with(`Processed string ${event.data}`); }); workflow.handle([processNumberEvent], async (context, event) => { - const { sendEvent } = context; - sendEvent(successEvent.with({ data: `Processed number ${event.data}` })); + return successEvent.with(`Processed number ${event.data}`); }); -const { stream } = workflow.createContext(); -sendEvent(inputEvent.with("I am some data")); -const result = await stream.until(successEvent).take(1).toArray(); -console.log(result); +let context1 = workflow.createContext(); +context1.sendEvent(inputEvent.with("I am some data")); + +const result = await context1.stream.until(successEvent).toArray(); +console.log(result.at(-1)!.data); + +let context2 = workflow.createContext(); +context2.sendEvent(inputEvent.with(1)); + +const result2 = await context2.stream.until(successEvent).toArray(); +console.log(result2.at(-1)!.data); ``` diff --git a/docs/workflows/common_patterns/fan_in_fan_out.mdx b/docs/workflows/common_patterns/fan_in_fan_out.mdx index 838f3da..1ad3eda 100644 --- a/docs/workflows/common_patterns/fan_in_fan_out.mdx +++ b/docs/workflows/common_patterns/fan_in_fan_out.mdx @@ -24,6 +24,7 @@ import { createWorkflow, workflowEvent, } from "@llamaindex/workflow-core"; +import { createStatefulMiddleware } from "@llamaindex/workflow-core/middleware/state"; // Define the events we'll use const startEvent = workflowEvent(); // Triggers the fan-out process @@ -32,15 +33,15 @@ const resultEvent = workflowEvent(); // Results from processed items const completeEvent = workflowEvent(); // Final aggregated results ``` -Next, we create our workflow and set up tracking variables: +Next, we create our workflow and set up tracking variables in the state: ```javascript -const workflow = createWorkflow(); - -// Shared variables to track progress -let itemsToProcess = 10; // Total number of items to process -let itemsProcessed = 0; // Counter for completed items -let processResults: string[] = [] +const { withState } = createStatefulMiddleware(() => ({ + itemsToProcess: 10, + itemsProcessed: 0, + processResults: [] as string[], +})); +const workflow = withState(createWorkflow()); ``` **2. Create the fan-out handler** @@ -49,11 +50,11 @@ This handler receives the start event and fans out multiple processing tasks: ```ts workflow.handle([startEvent], async (context, start) => { - const { sendEvent, stream } = context; - itemsProcessed = 0; // Reset counter for this execution + const { sendEvent, state} = context; + state.itemsProcessed = 0; // Reset counter for this execution // Fan out: emit multiple events to be processed in parallel - for (let i = 0; i < itemsToProcess; i++) { + for (let i = 0; i < state.itemsToProcess; i++) { sendEvent(processItemEvent.with(i)); } }); @@ -67,7 +68,7 @@ This handler processes each individual item: ```ts workflow.handle([processItemEvent], async (context, event) => { - const { sendEvent } = context; + const { sendEvent, state } = context; // Simulate some async work (like API calls, database operations, etc.) await new Promise((resolve) => setTimeout(resolve, Math.random() * 100)); @@ -76,7 +77,7 @@ workflow.handle([processItemEvent], async (context, event) => { const processedValue = `Processed: ${event.data}`; // Update the shared counter after processing completes - itemsProcessed++; + state.itemsProcessed++; // Return the result event sendEvent(resultEvent.with(processedValue)); @@ -91,14 +92,14 @@ In the last step of our workflow, we need to collect the results from the proces ```ts workflow.handle([resultEvent], async (context, event) => { - const { sendEvent } = context; + const { sendEvent, state } = context; // store the processed message - processResults.push(event.data) + state.processResults.push(event.data) // return completeEvent if the processing is completed - if (itemsProcessed === itemsToProcess) { - sendEvent(completeEvent.with(processResults)) + if (state.itemsProcessed === state.itemsToProcess) { + sendEvent(completeEvent.with(state.processResults)) } }); ``` @@ -130,5 +131,5 @@ async function runFanOut() { } } -runFanOut(); +runFanOut().catch(console.error); ``` \ No newline at end of file diff --git a/docs/workflows/common_patterns/human_in_the_loop.mdx b/docs/workflows/common_patterns/human_in_the_loop.mdx index 053583a..fb7c9ac 100644 --- a/docs/workflows/common_patterns/human_in_the_loop.mdx +++ b/docs/workflows/common_patterns/human_in_the_loop.mdx @@ -16,6 +16,7 @@ Here's how to implement a workflow that pauses for human input: ```ts import { createWorkflow, workflowEvent } from "@llamaindex/workflow-core"; +import { createStatefulMiddleware } from "@llamaindex/workflow-core/middleware/state"; // Define events const startEvent = workflowEvent(); @@ -23,15 +24,16 @@ const humanRequestEvent = workflowEvent(); const humanResponseEvent = workflowEvent(); const stopEvent = workflowEvent(); -const workflow = createWorkflow(); +const { withState } = createStatefulMiddleware(() => ({})); +const workflow = withState(createWorkflow()); // Workflow that needs human input workflow.handle([startEvent], () => { return humanRequestEvent.with(); }); -workflow.handle([humanResponseEvent], (context, { data }) => { - return stopEvent.with(`Human said: ${data}`); +workflow.handle([humanResponseEvent], (context, event) => { + return stopEvent.with(`Human said: ${event.data}`); }); // Usage with snapshot/resume diff --git a/docs/workflows/common_patterns/loops.mdx b/docs/workflows/common_patterns/loops.mdx index 421e038..9134bfd 100644 --- a/docs/workflows/common_patterns/loops.mdx +++ b/docs/workflows/common_patterns/loops.mdx @@ -14,9 +14,12 @@ Since workflows are event-driven, and every step is triggered by an event, you c Here is how you can do it with code: ```ts +import { createWorkflow, workflowEvent } from "@llamaindex/workflow-core"; +import { createStatefulMiddleware } from "@llamaindex/workflow-core/middleware/state"; + type AgentWorkflowState = { - counter: number = 0 - max_counter: number = 5 + counter: number, + max_counter: number }; const { withState } = createStatefulMiddleware( @@ -31,7 +34,7 @@ const stopEvent = workflowEvent(); workflow.handle([startEvent], async (context, { data }) => { const { sendEvent, state } = context; if (state.counter < state.max_counter) { - sendEvent(increaseCounterEvent.with({})) + sendEvent(increaseCounterEvent.with()) } else { sendEvent(stopEvent.with(state.counter)) } @@ -40,7 +43,7 @@ workflow.handle([startEvent], async (context, { data }) => { workflow.handle([increaseCounterEvent], async (context, { data }) => { const { sendEvent, state } = context; state.counter += 1 - sendEvent(startEvent.with({})) + sendEvent(startEvent.with()) }) const { stream, sendEvent } = workflow.createContext({ @@ -48,11 +51,11 @@ const { stream, sendEvent } = workflow.createContext({ max_counter: 5, }); -sendEvent( - startEvent.with({}), -); +sendEvent(startEvent.with(),); const result = await stream.until(stopEvent).toArray(); + +// should print 5 since the workflow is looping console.log(result[result.length - 1].data) ``` diff --git a/docs/workflows/common_patterns/state.mdx b/docs/workflows/common_patterns/state.mdx index cdf89ec..44dc9e6 100644 --- a/docs/workflows/common_patterns/state.mdx +++ b/docs/workflows/common_patterns/state.mdx @@ -14,20 +14,19 @@ The workflow state is _typed_, meaning that it can be defined as a `type` and ca Here is an example: ```ts -type AgentWorkflowState = { - user_message: string, - retrieved_information: string[], - num_iterations: number, +type MyWorkflowState = { + previous_message: string, }; ``` You can then add a state to your workflow in this way: ```ts +import { createWorkflow } from "@llamaindex/workflow-core"; import { createStatefulMiddleware } from "@llamaindex/workflow-core/middleware/state"; const { withState } = createStatefulMiddleware( - (state: AgentWorkflowState) => state, + (state: MyWorkflowState) => state, ); const workflow = withState(createWorkflow()); ``` @@ -35,27 +34,19 @@ const workflow = withState(createWorkflow()); And then, in your step handling logic, you can access and modify state properties: ```ts -workflow.handle([userInputEvent], async (context, { data }) => { +import { workflowEvent } from "@llamaindex/workflow-core"; + +const startEvent = workflowEvent<{ userInput: string }>(); +const stopEvent = workflowEvent<{ result: string }>(); + +workflow.handle([startEvent], async (context, { data }) => { const { sendEvent, state } = context; const { userInput } = data; + + const previous_message = state.previous_message; + state.previous_message = userInput; - state.user_message = userInput; - - retrievedInfo = await retriever.retrieve({query: userInput, top_k: 5}) - - if (retrievedInfo && retrievedInfo.length > 0) { - state.retrieved_information.push(...retrievedInfo) - response = await llm.complete(`Please reply to this message: ${userInput} based on this information: ${retrievedInfo.join('\n\n---\n\n')}.`) - sendEvent(finalResponseEvent.with({result: response})) - } else { - if (state.num_iterations < 5) { - state.num_iterations += 1 - newQuery = await llm.complete(`Please rewrite this message: ${userInput} so that it can retrieve more relevant contextual information.`) - sendEvent(userInputEvent.with({userInput: newQuery})) - } else { - sendEvent(finalResponseEvent.with({result: "Unable to find a response to your query based on the information in our database"})); - } - } + return stopEvent.with({ result: "Processed message: " + userInput + " previous message: " + previous_message }); }); ``` @@ -63,8 +54,11 @@ Before running the workflow, remember to initialize the context object with the ```typescript const { stream, sendEvent } = workflow.createContext({ - user_message: "What is the number I need to call for customer support?", - retrieved_information: [], - num_iterations: 0, + previous_message: "my initial previous message", }); + +sendEvent(startEvent.with({ userInput: "Hello, how are you?" })); + +const result = await stream.until(stopEvent).toArray(); +console.log(result[result.length - 1].data); ``` \ No newline at end of file