Queueing User Messages

Inject messages during an agent's turn, before tool calls complete or while the model is reasoning.

When using multi-turn workflows, messages typically arrive between agent turns. The workflow waits at a hook, receives a message, then starts a new turn. But sometimes you need to inject messages during an agent's turn, before tool calls complete or while the model is reasoning.

DurableAgent's prepareStep callback enables this by running before each step in the agent loop, giving you a chance to inject queued messages into the conversation. prepareStep also allows you to modify the model choice and existing messages mid-turn, see AI SDK's prepareStep callback for more details.

Message queueing is useful when:

Users send follow-up messages while the agent is still searching for flights or processing bookings
External systems need to inject context mid-turn (e.g., a flight status webhook fires during processing)
You want messages to influence the agent's next step rather than waiting for the current turn to complete

If you just need basic multi-turn conversations where messages arrive between turns, see Chat Session Modeling. This guide covers the more advanced case of injecting messages during turns.

The prepareStep callback runs before each step in the agent loop. It receives the current state and can modify the messages sent to the model:

import type { ModelMessage, LanguageModel } from "ai";

interface PrepareStepInfo {
  model: string | (() => Promise<LanguageModel>);    // Current model
  stepNumber: number;                                // 0-indexed step count
  steps: StepResult[];                               // Previous step results
  messages: ModelMessage[];                          // Messages to be sent
}

interface PrepareStepResult {
  model?: string | (() => Promise<LanguageModel>);   // Override model
  messages?: ModelMessage[];                         // Override messages
}

Once you have a multi-turn workflow, you can combine a message queue with prepareStep to inject messages that arrive during processing:

import { DurableAgent } from "@workflow/ai/agent";
import { getWritable, getWorkflowMetadata } from "workflow";
import { chatMessageHook } from "./hooks/chat-message";
import { flightBookingTools, FLIGHT_ASSISTANT_PROMPT } from "./steps/tools";
import type { UIMessageChunk, ModelMessage } from "ai";

export async function chat(initialMessages: ModelMessage[]) {
  "use workflow";

  const { workflowRunId: runId } = getWorkflowMetadata();
  const writable = getWritable<UIMessageChunk>();
  const messageQueue: Array<{ role: "user"; content: string }> = []; 

  const agent = new DurableAgent({
    model: "bedrock/claude-haiku-4-5-20251001-v1",
    instructions: FLIGHT_ASSISTANT_PROMPT,
    tools: flightBookingTools,
  });

  // Listen for messages in background (non-blocking)
  const hook = chatMessageHook.create({ token: runId }); 
  hook.then(({ message }) => { 
    messageQueue.push({ role: "user", content: message }); 
  }); 

  await agent.stream({
    messages: initialMessages,
    writable,
    prepareStep: ({ messages: currentMessages }) => { 
      // Inject any queued messages before the next LLM call
      if (messageQueue.length > 0) { 
        const newMessages = messageQueue.splice(0); // Drain queue
        return { 
          messages: [ 
            ...currentMessages, 
            ...newMessages.map((m) => ({ 
              role: m.role, 
              content: [{ type: "text" as const, text: m.content }], 
            })), 
          ], 
        }; 
      } 
      return {}; 
    }, 
  });
}

Messages sent via chatMessageHook.resume() accumulate in the queue and get injected before the next step, whether that's a tool call or another LLM request.

The prepareStep callback receives messages in ModelMessage[] format (with content arrays), which is the internal format used by the AI SDK.

You can also combine message queueing with the standard multi-turn pattern:

import { DurableAgent } from "@workflow/ai/agent";
import { getWritable, getWorkflowMetadata } from "workflow";
import { chatMessageHook } from "./hooks/chat-message";
import type { UIMessageChunk, ModelMessage } from "ai";

export async function chat(initialMessages: ModelMessage[]) {
  "use workflow";

  const { workflowRunId: runId } = getWorkflowMetadata();
  const writable = getWritable<UIMessageChunk>();
  const messages: ModelMessage[] = [...initialMessages];
  const messageQueue: Array<{ role: "user"; content: string }> = [];

  const agent = new DurableAgent({ /* ... */ });
  const hook = chatMessageHook.create({ token: runId });

  while (true) {
    // Set up non-blocking listener for mid-turn messages
    let pendingMessage: string | null = null; 
    hook.then(({ message }) => { 
      if (message === "/done") return; 
      messageQueue.push({ role: "user", content: message }); 
      pendingMessage = message; 
    }); 

    const result = await agent.stream({
      messages,
      writable,
      preventClose: true,
      prepareStep: ({ messages: currentMessages }) => {
        // Inject queued messages during turn
        if (messageQueue.length > 0) {
          const newMessages = messageQueue.splice(0);
          return {
            messages: [
              ...currentMessages,
              ...newMessages.map((m) => ({
                role: m.role,
                content: [{ type: "text" as const, text: m.content }],
              })),
            ],
          };
        }
        return {};
      },
    });

    messages.push(...result.messages.slice(messages.length));

    // Wait for next message (either queued during turn or new)
    const { message: followUp } = pendingMessage ? { message: pendingMessage } : await hook; 
    if (followUp === "/done") break;

    messages.push({ role: "user", content: followUp });
  }
}

Chat Session Modeling - Single-turn vs multi-turn patterns
Building Durable AI Agents - Complete guide to creating durable agents
DurableAgent API Reference - Full API documentation
defineHook() API Reference - Hook configuration options

推荐订阅源

Workflow SDK Documentation