import { workflowEvent } from "@llamaindex/workflow-core"; const startEvent = workflowEvent<string>(); const stopEvent = workflowEvent<1 | -1>();

import { createWorkflow } from "@llamaindex/workflow-core"; const convertEvent = workflowEvent(); const workflow = createWorkflow(); workflow.handle([startEvent], (start) => { return convertEvent.with(Number.parseInt(start.data, 10)); }); workflow.handle([convertEvent], (convert) => { return stopEvent.with(convert.data > 0 ? 1 : -1); });

import { pipeline } from "node:stream/promises"; const { stream, sendEvent } = workflow.createContext(); sendEvent(startEvent.with()); const result = await pipeline(stream, async function (source) { for await (const event of source) { if (stopEvent.include(event)) { return "stop received!"; } } }); console.log(result); // stop received! // or const allEvents = await stream.until(stopEvent).toArray();

import { runWorkflow, runAndCollect, runWorkflowWithFilter, } from "@llamaindex/workflow-core/stream/run"; // Run workflow and get final result const result = await runWorkflow(workflow, startEvent.with("42"), stopEvent); // Run workflow and collect all events const allEvents = await runAndCollect( workflow, startEvent.with("42"), stopEvent, );

let condition = false; workflow.handle([startEvent], async (context, start) => { const { sendEvent, stream } = context; for (let i = 0; i < 10; i++) { sendEvent(convertEvent.with(i)); } // You define the condition to stop the workflow const results = await stream .until(() => condition) .filter(convertStopEvent) .toArray(); console.log(results.length); // 10 return stopEvent.with(); }); workflow.handle([convertEvent], (convert) => { if (convert.data === 9) { condition = true; } return convertStopEvent.with(/* ... */); });

import { from, pipe } from "rxjs"; const { stream, sendEvent } = workflow.createContext(); from(stream) .pipe(filter((ev) => eventSource(ev) === messageEvent)) .subscribe((ev) => { console.log(ev.data); }); sendEvent(fileParseWorkflow.startEvent(directory));

import { Hono } from "hono"; import { serve } from "@hono/node-server"; import { createHonoHandler } from "@llamaindex/workflow-core/interrupter/hono"; import { agentWorkflow, startEvent, stopEvent, } from "../workflows/tool-call-agent.js"; const app = new Hono(); app.post( "/workflow", createHonoHandler( agentWorkflow, async (ctx) => startEvent(await ctx.req.text()), stopEvent, ), ); serve(app, ({ port }) => { console.log(`Server started at http://localhost:${port}`); });

workflow.handle([convertEvent], (context) => { const { signal } = context; signal.onabort = () => { console.error("error in convert event:", abort.reason); }; });

workflow.handle([startEvent], async (context) => { const { sendEvent, stream, signal } = context; // Use context properties directly sendEvent(processEvent.with()); });

import { createStatefulMiddleware } from "@llamaindex/workflow-core/middleware/state"; const { withState } = createStatefulMiddleware(() => ({ pendingTasks: new Set<Promise<unknown>>(), })); const workflow = withState(createWorkflow()); workflow.handle([startEvent], (context) => { const { state } = context; state.pendingTasks.add( new Promise((resolve) => { setTimeout(() => { resolve(); }, 100); }), ); }); const { state } = workflow.createContext();

const { withState } = createStatefulMiddleware((input: { id: string }) => ({ id: input.id, })); const workflow = withState(createWorkflow()); const { state } = workflow.createContext({ id: "1" });

const { snapshot, resume } = workflow.createContext(); // create snapshot const snapshotData = await snapshot(); // resume workflow from snapshot const { stream, sendEvent } = workflow.resume(snapshotData); sendEvent(humanResponseEvent.with("hello"));

// before: workflow.handle([startEvent], (start) => {}); // after: workflow.handle([startEvent], (sendEvent, start) => {});

import { withValidation } from "@llamaindex/workflow-core/middleware/validation"; const startEvent = workflowEvent<void, "start">(); const disallowedEvent = workflowEvent<void, "disallowed">({ debugLabel: "disallowed", }); const parseEvent = workflowEvent<string, "parse">(); const stopEvent = workflowEvent<number, "stop">(); const workflow = withValidation(createWorkflow(), [ [[startEvent], [stopEvent]], [[startEvent], [parseEvent]], ]); workflow.strictHandle([startEvent], (sendEvent, start) => { sendEvent( disallowedEvent.with(), // <-- ❌ Type Check Failed, Runtime Error ); sendEvent(parseEvent.with("")); // <-- ✅ sendEvent(stopEvent.with(1)); // <-- ✅ });

import { withTraceEvents, runOnce, } from "@llamaindex/workflow-core/middleware/trace-events"; const workflow = withTraceEvents(createWorkflow()); workflow.handle( [messageEvent], runOnce(() => { console.log("This message handler will only run once"); }), ); workflow.handle([startEvent], (context) => { context.sendEvent(messageEvent.with()); context.sendEvent(messageEvent.with()); }); { const { sendEvent } = workflow.createContext(); sendEvent(startEvent.with()); sendEvent(messageEvent.with()); // This message handler will only run once! } { const { sendEvent } = workflow.createContext(); // For each new context, the decorator is isolated. sendEvent(startEvent.with()); sendEvent(messageEvent.with()); // This message handler will only run once! }

const ev = startEvent.with(); const { sendEvent, stream } = workflow.createContext(); sendEvent(ev); sendEvent(messageEvent.with()); // <- this will not be included in the substream const substream = workflow.substream(ev, stream);

import { createHandlerDecorator } from "@llamaindex/workflow-core/middleware/trace-events"; const noop: (...args: any[]) => void = function noop() {}; export const runOnce = createHandlerDecorator({ debugLabel: "onceHook", getInitialValue: () => false, onBeforeHandler: (handler, handlerContext, tracked) => tracked ? noop : handler, onAfterHandler: () => true, });

type BaseHandlerContext = { // ... some other properties are hidden handler: Handler<WorkflowEvent<any>[], any>; inputEvents: WorkflowEvent<any>[]; // events data that are accepted by the handler inputs: WorkflowEventData<any>[]; // events data that are emitted by the handler outputs: WorkflowEventData<any>[]; //#region linked list data structure prev: HandlerContext; next: Set<HandlerContext>; root: HandlerContext; //#endregion }; type SyncHandlerContext = BaseHandlerContext & { async: false; pending: null; }; type AsyncHandlerContext = BaseHandlerContext & { async: true; pending: Promise<WorkflowEventData<any> | void> | null; }; type HandlerContext = AsyncHandlerContext | SyncHandlerContext;

let once = false; workflow.handle([startEvent], (context) => { const { sendEvent } = context; if (once) { return; } once = true; sendEvent(messageEvent.with()); }); const { sendEvent } = workflow.createContext(); sendEvent(startEvent.with()); sendEvent(startEvent.with()); sendEvent(messageEvent.with());