[feat] event triggered agents

[pkarw]

Related to #8. I can imagine having a node like this:

<on eventName="gmailMessage">
 <sequence>
</sequence>
</on>

In this case, we must provide a way to emit an event inside the flow. I can imagine having an HTTP server - waiting for a webhook call (for example, from the GMail/zapier integration), and just as the webhook arrives, it emits the event within the deserialized flow (serialized as a way of saving the state for #8)

[pkarw]

I'm thinking about how the handling of asynchronous resumptions (a.k.a. events) should work—so that we can implement mechanisms like we had in Fabrica. You know, that asynchronous flow which waits for a response from a tool's execution.

Here, I believe suspending execution to wait on tools isn't optimal—especially since they're invoked by the Vercel SDK, which we don't want to modify.

What do you think about having a type of node that simply waits for an event, along with a mechanism for emitting events within the flow?

Consider the following scenario:

export const githubProjectHealthAnalysisFlow = 
event({
  eventName: 'github-project-name',
  input: sequence([
    {
      agent: 'githubAgent',
      name: 'getIssues',
      input: 'Go to Github and get the top 3 most popular issues and the number of open ones'
      // ...
    },
  ])
});

Here’s the event element:

/**
 * Use this agent to implement a workflow where each step is executed in parallel.
 */
const eventAgent: Agent<EventFlowDefinition> = async ({ eventName, input }, context) => {
  return context.on(eventName, input, context); // `input` should be a single step, for example <sequence>
};

So, if it's run as a long-running CLI process, context.on() halts execution and waits on some await for the event. If it's running in serverless mode, context.on() interrupts the application—meaning it returns control to the system.

In that case, we need to serialize context—which must store which event it was last waiting on, along with the serialized input and context.

Upon the next invocation, once we deserialize context, we can simply continue running the program.

Hence, the eventAgent effectively freeze-frames the execution (or waits for the event if it's a long-running process).

Moreover, context should have a method like emit(eventName, payload) that simply resumes the execution of whatever is stored in context.input.

By the way, in this setup, I'd probably change the name from input to ops (operations) or children. In the case of agents that receive parameters from input, we'd then have both children and input, which would be more logical.

[pkarw]

By the way, I noticed that in some places context is typed as a string, which seems like a bug (because it's generally passed via JSON.stringify to the LLM). Since it's passed to the LLM, it's probably not really the “context” I'm referring to in the proposed event-based architecture. We’ll likely need to add event handling elsewhere, maybe in flow? Then it could be something like flow.instance or flow.execution, which would handle events and the subsequent steps: flow.execution.on(), flow.execution.input, flow.execution.stacktrace, and flow.execution.context.

That’s how I envision it. @grabbou - let me know what you think. It’s necessary to implement any asynchronous integration—currently it’s impossible to do so.

Ultimately, we could use such an execution for telemetry and display it in a GUI like [Inngest](https://www.inngest.com/ai?ref=nav) does, with profiling, etc.

[pkarw]

@grabbou looking for your comments on this one?

[grabbou]

Related to #8. I can imagine having a node like this:

In this case, we must provide a way to emit an event inside the flow. I can imagine having an `HTTP` server - waiting for a webhook call (for example, from the `GMail`/`zapier` integration), and just as the webhook arrives, it `emits` the event within the deserialized flow (serialized as a way of saving the state for [#8](https://github.com//issues/8))

Why would you want to do this, instead of simply doing an event emitter (or whatever your app requires), such as:

ws.on('message', () => {
   execute(newMessageProcessingFlow)
})

[grabbou]

I'm thinking about how the handling of asynchronous resumptions (a.k.a. events) should work—so that we can implement mechanisms like we had in Fabrica. You know, that asynchronous flow which waits for a response from a tool's execution. Here, I believe suspending execution to wait on tools isn't optimal—especially since they're invoked by the Vercel SDK, which we don't want to modify.

Yes, I definitely agree on this one. We don't want that mechanism - I think that approach of suspending on tool calls was "generally" fine, but did not age well and required too many ticks to run.

What do you think about having a type of node that simply waits for an event, along with a mechanism for emitting events within the flow?

I think the beauty of this design is that you can easily implement it anyway, and experiment. The flow can be suspended, so if you decide to return a Promise that keeps hanging until an event arrives, that is fine.

I guess I would do it more like this:

const makeEventAgent = (eventEmitter) = ({ input, on }, context) => {
   return new Promise((resolve) => {
       eventEmitter.on(on, async (payload) => {
           const res =  await input.agent(payload, context)
           resolve(res)
       })
   })
}

Then, in the user land:

const eventAgent = makeEventAgent(new WebSocket("localhost:3000"))

Obviously, what you pass to makeEventAgent is totally up to you. In this case, I just required event emitter, but you don't have to, depends on your application.

Flow will be paused until that given event arrives.

[grabbou]

So, if it's run as a long-running CLI process, context.on() halts execution and waits on some await for the event. If it's running in serverless mode, context.on() interrupts the application—meaning it returns control to the system.

I don't think we should extend context to be an event emitter and include interruptions there or any type of logic. We should instead think of how this can be done (potentially) in the user land to satisfy this and other async scenarios and support user in making this possible.

To make this happen, we have to first figure out how to pause execution at any time (by returning special signal, for example) and maintaining execution state so that it can be resumed.

I am not sure what is best way to do it at the moment, so I will have to take some time to think about it. I am definitely not happy with the way we have done this before, so I am taking extra time.

[grabbou]

By the way, I noticed that in some places context is typed as a string, which seems like a bug (because it's generally passed via JSON.stringify to the LLM).

That has been changed to be an array, at this point, context is just messages from previous (parent) flows that may be useful to current agent. We might change it to something else in the future, but again, taking baby steps.

We’ll likely need to add event handling elsewhere, maybe in flow? Then it could be something like flow.instance or flow.execution, which would handle events and the subsequent steps: flow.execution.on(), flow.execution.input, flow.execution.stacktrace, and flow.execution.context.

As you can see, the abstraction has challenge of growing really quickly, which is something we should think about carefully. I think I am more a fan of having dedicated serverless execute equivalent, but unfortunately, it's not as trivial.