Agent SDK

[ericallam]

Depending on your definition of an Agent, Trigger.dev is already a pretty great platform for building and executing them, with the ability to wait, trigger subtasks dynamically, observe everything that is going on, and not have to worry about timeouts. Combined with Build Extensions allowing you to add all sorts of great packages to your builds, like ffmpeg and chrome, we're uniquely positioned for the agent-sphere.

One thing we're currently not very good at is the "agent-loop" where each iteration continues from some external event. With Run Engine v2 and our new waitpoints (and wait.forToken), it will start to be possible to pause execution while waiting for an external event.

We should provide an easy way to run an "agent" loop inside a task that uses our Realtime system + Run engine v2.

A manual example

Here's an example of doing this completely manually:

const PayloadSchema = z.object({
  model: z.string().default("gpt-3.5-turbo"),
  prompt: z.string().default("Hello, how are you?"),
});

export const agentTask = schemaTask({
  id: "agent-task",
  schema: PayloadSchema,
  run: async (payload) => {
    const model = openai(payload.model);

    let messages: CoreMessage[] = [
      {
        role: "user",
        content: payload.prompt,
      }
    ];

    let iteration = 0;

    while (true) {
      iteration++;

      const result = streamText({
        model,
        messages,
      });

      await metadata.stream(`openai.${iteration}`, result.fullStream);
      const systemResponse = await result.text;

      messages.push({
        role: "system",
        content: systemResponse,
      });

      // This token will expire in 1 day
      const token = await wait.createToken({ timeout: "1d" });

      metadata.set("resumeToken", token); // So we can access the token through Realtime

      // Now wait for the next message
      const nextMessage = await wait.forToken<CoreUserMessage>(token);

      if (nextMessage.ok) {
        messages.push(nextMessage.data);
      } else {
        break;
      }
    }
  },
});

Using a built-in helper function

We could provide a helper function like so to make this standardized:

// Define our context type for this specific task
interface ChatContext {
  messages: CoreMessage[];
}

export const agentTask = schemaTask({
  id: "agent-task",
  schema: PayloadSchema,
  run: async (payload) => {
    const model = openai(payload.model);

    // Initial context with starting messages
    const initialContext: ChatContext = {
      messages: [
        {
          role: "user",
          content: payload.prompt,
        },
      ],
    };

    // Use the reusable conversation loop with explicit generic types and anonymous functions
    const finalContext = await agentLoop<ChatContext, CoreUserMessage>(
      initialContext,
      async (context, iteration) => {
        const result = streamText({
          model,
          messages: context.messages,
        });

        await metadata.stream(`openai.${iteration}`, result.fullStream);
        const systemResponse = await result.text;

        return {
          ...context,
          messages: [
            ...context.messages,
            {
              role: "system",
              content: systemResponse,
            },
          ],
        };
      },
      (context, userMessage) => ({
        ...context,
        messages: [...context.messages, userMessage],
      })
    );

    // Optionally do something with the final context
    return finalContext;
  },
});

[brennanmceachran]

Agree on the notion of the idea. I've extracted a set of three tasks to act as a generic Evaluator Optimizer and really like how it feels

Unsure i this basic loop should sit natively within trigger, be upstreamed to ai-sdk, or just be cookbook examples.

But if there's help on llm streaming and metadata sharing that might be worth it.

(edit: also eagerly awaiting v2)