DX feedback about the Workflow API

[himself65]

I made some example code using next.js + LITS + vercel ai sdk. I feel it not good to use

Screen.Recording.2024-09-25.at.17.32.04.mov

Some feedback:

1. some API type is vague and unclear.

1.1 the event.data type workflow.handleEvent is any
1.2 Workflow.prototype.run can run multiple startEvent, but streamEvents only accept run in single context. It's hard to maintain for example you put this workflow into a server handler
1.3 hard to add context data inside single context, I can only give input: string for a start event. What if there's more input data?
And context.get/set seems useful, but since I cannot access it in the streamEvents. It's useless, I have to build another context layer on top of the workflow api
1.4 StopEvent and StartEvent is kind of Redundant, I want customize my data type (maybe related to 1.1 or 1.3)
1.5 validate will throw error, and it accepts no parameter. I don't know what it's used for at first glance. I think it's better to remove this function.
1.6 return type of StepFunction accepct a new Event, in the meanwhile context allows writeEventToStream

I think Workflow.prototype.run should retuns Context and user call await context could get a StopEvent

2. unsafe implementation

WorkflowEvent.prototype.toString will leak data, imaging user have senstive input but you console.log(event) or result = ${event}

export class WorkflowEvent<T extends Record<string, any> = any> {
  data: T;

  constructor(data: T) {
    this.data = data;
  }

  toString() {
    return `${this.constructor.name}(${JSON.stringify(this.data)})`;
  }
}

3. other quetions

• How to interrupt to get current state for a context?
• How to intergate with other format of data, like in this example I have UI as return value instead of just plain text or json
• Is there clear task scheduler if there're multiple StartEvent, because it could have data racing issue if a event connect to a database

other than that, I think this API is a subset of XState, Workflow is kind of Finite State Machine. I personaly would to use xstate instead of Workflow API because it more customzable (even it's hard to learn)

[himself65]

Now I'm really thinking whether we can switch to use xstate as low level code, so we can have

1. xstate studio, show UI diagram from code
2. we just need design high level api and don't need consider the implemtation and validation
3. strong type, they have very strong ts check

[logan-markewich]

One thing to consider using another library vs. from scratch, will it support things that we know customers want

• human in the loop
• state management
• human on the loop (i.e. stepwise execution, editing state, undo/rewind)
• semantic caching
• retires
• etc.

[marcusschiesser]

@himself65 yes, i also had a couple of the issues you mentioned.

Workflow.prototype.run should retuns Context and user call await context could get a StopEvent
That sounds like a good idea; it's implemented as is to mimic the Python implementation
validate will throw error, and it accepts no parameter. I don't know what it's used for at first glance. I think it's better to remove this function.
mimics Python behavior at the moment. you need to define the output types of each step for it to work

[himself65]

For last days I read the source code and have some update:

I still don't understand why we allow await run to StopEvent but meanwhile streamEvents() will give totally different event queue. This gives user too much cognitive load. Because it means there're two queue parallel running, you cannot tell the order.

In the app-creator.ts, I find out it's used for display some message to user when there's a heavy task on going. In my idea, I suppose user to customizer their own even channel, it's more flexable, like in my case on top, I used vercel/ai/rsc, createStreamUI for each request, and I have my custom logic to handle workflow event loop.

Here're my solution for that examples code

solution 1: use AsyncLocalStorage

const uiAls = new AsyncLocalStorage();

workflow.addStep(SearchEvent, async () => {
  const ui = uiAls.getStore()
  ui.update('loading....')
  // heavy search
  llm.complete(...)
  return StopEvent('stop!')
})
const ui = createStreamUI()
const result = await uiAls.run(ui, () => workflow.run('search'))
ui.done(result)

solution 2: extend Context (need our support custom context)

class MyContext extends Context {
  // ...
}

[himself65]

Second is that we should support recover event from any point, imaging there's a step that need user input first (like search), for now we cannot pause a step to continue (have to wait until reach StopEvent). so I have to customize a StopEvent with my logic then add a another workflow.

So im imaging the low level API should be

// for the first run
const { input } = req.body()
const context = workflow.run('search ${input}')
for await (const event of context) {
  if (event.requireUserInput) {
    response.json({
      snapshot: context.snapshot()
    })
  }
}

// in the second run
const { snapshot, input } = req.body()
const context = Context.recover(snapshot)
for await (const event of context) {
  // ...
}

[himself65]

Now im writing up the new API

The benefit form new design:

1. strong type check in LITS
2. better workflow extend logic
3. independent context
   1. serialization support (when context data is follow structured clone algorithm)
   2. worker thread support (context is transferable because it's a Stream)
4. functional programming for debugging

const myWorkflow = Workflow
  .from(initialContext)
  .step(StartEvent, stepHandler) // unknown return event, weak validation
  .step([OtherEvent], [StopEvent], stepHandler) // known return event is StopEvent, strong validation

const context = myWorkflow.run('query')

const result = await context // you can await a context to get result

for await (const event of context) {
  // or you can for loop each event
}

// setup a new context from old context, will ensure pure function
// for same user input, giving different context data could give different result, for example Open AI vs Gemini AI
const context2 = context.with(newContextData)

context2 !== context

const result2 = await context2 

// extend old logic
const newWorkflow = myWorkflow
   // adding more step
  .step(StartEvent, newStepHandler)
  .remove(oldHandler)

[logan-markewich]

I think I like this design quite a bit

Can you clarify how you would create initialContext or newContextData ?

[himself65]

it could be anything, any JS value

const myWorkflow = Workflow
  .from({ 
    // empty
  })

so the initial context is just {}

put llm into context data

const myWorkflow = Workflow
  .from({
    llm: new OpenAI()
  })

same as newContextData.

note that this will follow JS object reference, which means

const initialContext = { a: 1 }
const myWorkflow = Workflow
  .from(initialContext)

const result = myWorkflow.run()
initialContext.a = 2 // update before run makes side effect
await result

const a = initialContext.a // <- maybe changed

from high level perspective, you can combine workflow with anything with your own data layer. like a react state manager, zustand, react context...

import { create } from 'zustand'
const useApp = create((set, get) => {
  workflow: Workflow.from({set, get}).addStep() // ... 
})
const Component = () => {
  const agent = useApp(app => workflow)
  const resultPromise = agent.run('hello')
  return <Display resultPromise={resultPromise} />
}

for a lower level API, we shouldn't handle what it is. the only one edge case is if user doesn't pass it, it's null

[himself65]

btw I will rename it to contextData not initialXXX

[logan-markewich]

That makes sense to me, thanks for the explanation 👍🏻

[himself65]

Most API is done in #1264, will close this discussion