LLM Streaming Response Becomes so Slow in Vercel World

[sijawara]

Hallo i am using DurableAgent in local world and vercel world, in local world almost there is no problem. But when i deploy on vercel streaming response (tokens per second) become so slow:

"outputTokens": {
          "total": 1964,
          "text": 1658,
          "reasoning": 306
        },
"attempts": [
  {
          "provider": "xai",
           "internalModelId": "xai:grok-code-fast-1",
           "providerApiModelId": "grok-code-fast-1",
           "credentialType": "system",
           "success": true,
            "startTime": 360785.635555,
            "endTime": 361101.433652, <--  more than 3 minutes
            "statusCode": 200
   }]

This is my code:

"use workflow";

export const createPrepareStepCallback = (context: any) => {
  return async (agentParams: any) => {
    // simulate your deep sanitize / serialization safety logic
    const sanitizeStep = (step: any) => {
      if (!step) return step;
      const { model, ...rest } = step;
      return {
        ...rest,
        modelId: typeof model === "string" ? model : model?.modelId
      };
    };

    const serializableAgentParams = {
      stepNumber: agentParams.stepNumber,
      steps: agentParams.steps?.map(sanitizeStep),
      messages: agentParams.messages,
      model:
        typeof agentParams.model === "string"
          ? agentParams.model
          : agentParams.model?.modelId
    };


    await new Promise((r) => setTimeout(r, 50));

    return {
      ...context,
      serializableAgentParams
    };
  };
};

// ------------------------------------------------------------------
// This is the "workflow" function
// ------------------------------------------------------------------
export async function chatWorkflow(params: any) {
  const {
    preFetchedData,
    streamId
  }: {
    preFetchedData: any;
    streamId: string;
  } = params;

  const systemInstruction = preFetchedData.systemInstruction;
  const processedMessages = preFetchedData.processedMessages;

  const modelName = preFetchedData.participantConfig?.model ?? "openai/gpt-4.1-mini";

  // writable used by agent.stream
  const writable = getWritable<UIMessageChunk>();

  const prepareStep = createPrepareStepCallback({
    streamId,
    modelName
  });

  const agent = new DurableAgent({
    model: modelName as any,
    system: systemInstruction,
    temperature: 0.2,
    toolChoice: "auto",
    tools: {}
  });

  const startAt = performance.now();
  console.log("[START] streamId=", streamId, "model=", modelName);

  await agent.stream({
    messages: processedMessages,
    writable,
    maxSteps: 3,
    prepareStep: prepareStep as any,

    onStepFinish: async (stepResult: any) => {
      console.log(
        "\n[onStepFinish] finishReason=",
        stepResult?.finishReason ?? "unknown"
      );
    },

    onFinish: async () => {
      console.log(
        "\n[FINISH] total stream duration:",
        ((performance.now() - startAt) / 1000).toFixed(2),
        "s"
      );
    },

    onError: async ({ error }: { error: any }) => {
      console.error("\n[ERROR]", error);
    }
  });
}

dependencies:

    "@ai-sdk/react": "^3.0.18",
    "@ai-sdk/gateway": "^3.0.9",
    "ai": "^6.0.18",
    "@workflow/ai": "4.0.1-beta.43",

Anyone facing the same issue or there is something wrong with my code? Thank you

[sijawara]

Found the root cause:
Inside node_modules/@workflow/world-vercel/dist/streamer.js, the writeToStream implementation makes one HTTP PUT request for each chunk.

async writeToStream(name, runId, chunk) {
    const resolvedRunId = await runId;
    const httpConfig = await getHttpConfig(config);
    // The bottleneck: Fetch (HTTP PUT) called when each chunk arrives
    await fetch(getStreamUrl(name, resolvedRunId, httpConfig), {
        method: 'PUT',
        body: chunk,
        headers: httpConfig.headers,
        duplex: 'half',
    });
}

Interim solution:

diff --git a/dist/streamer.js b/dist/streamer.js
index c5403faff58c9e24ba523830fb5be98818ddce3a..55ecdf70e90e19b142887f59e439e769ab82b292 100644
--- a/dist/streamer.js
+++ b/dist/streamer.js
@@ -6,19 +6,87 @@ function getStreamUrl(name, runId, httpConfig) {
     return new URL(`${httpConfig.baseUrl}/v1/stream/${encodeURIComponent(name)}`);
 }
 export function createStreamer(config) {
-    return {
-        async writeToStream(name, runId, chunk) {
-            // Await runId if it's a promise to ensure proper flushing
+    const streamStates = new Map();
+    async function flush(name, runId, state) {
+        if (state.flushing || state.buffer.length === 0)
+            return;
+        state.flushing = true;
+        const chunks = state.buffer;
+        state.buffer = [];
+        let merged;
+        if (chunks.length === 1) {
+            merged = chunks[0];
+        }
+        else {
+            const totalLength = chunks.reduce((acc, c) => acc + c.length, 0);
+            merged = new Uint8Array(totalLength);
+            let offset = 0;
+            for (const c of chunks) {
+                merged.set(c, offset);
+                offset += c.length;
+            }
+        }
+        try {
             const resolvedRunId = await runId;
             const httpConfig = await getHttpConfig(config);
             await fetch(getStreamUrl(name, resolvedRunId, httpConfig), {
                 method: 'PUT',
-                body: chunk,
+                body: merged,
                 headers: httpConfig.headers,
                 duplex: 'half',
             });
+        }
+        finally {
+            state.flushing = false;
+            // If more data arrived while flushing, flush again
+            if (state.buffer.length > 0) {
+                flush(name, runId, state);
+            }
+            if (state.resolveQueue && state.resolveQueue.length > 0) {
+                const queue = state.resolveQueue;
+                state.resolveQueue = [];
+                for (const resolve of queue)
+                    resolve();
+            }
+        }
+    }
+    return {
+        async writeToStream(name, runId, chunk) {
+            let state = streamStates.get(name);
+            if (!state) {
+                state = {
+                    buffer: [],
+                    flushing: false,
+                    timer: null,
+                    resolveQueue: []
+                };
+                streamStates.set(name, state);
+            }
+            state.buffer.push(chunk);
+            if (!state.timer && !state.flushing) {
+                state.timer = setTimeout(() => {
+                    state.timer = null;
+                    flush(name, runId, state);
+                }, 10);
+            }
+            return Promise.resolve();
         },
         async closeStream(name, runId) {
+            const state = streamStates.get(name);
+            if (state) {
+                if (state.timer) {
+                    clearTimeout(state.timer);
+                    state.timer = null;
+                }
+                while (state.flushing || state.buffer.length > 0) {
+                    if (!state.flushing && state.buffer.length > 0) {
+                        await flush(name, runId, state);
+                    }
+                    else {
+                        await new Promise(r => setTimeout(r, 10));
+                    }
+                }
+            }
             // Await runId if it's a promise to ensure proper flushing
             const resolvedRunId = await runId;
             const httpConfig = await getHttpConfig(config);
@@ -27,6 +95,7 @@ export function createStreamer(config) {
                 method: 'PUT',
                 headers: httpConfig.headers,
             });
+            streamStates.delete(name);
         },
         async readFromStream(name, startIndex) {
             const httpConfig = await getHttpConfig(config);

Internal Buffering:

• Grouping Chunks: Each data chunk (Uint8Array) is now held in a temporary in-memory buffer for 10ms, or until multiple chunks have accumulated.
• Batch Upload: The accumulated data is then merged and sent in a single HTTP PUT request.
• Background Flushing: Uploading is performed asynchronously so it doesn’t block the text-generation loop from the AI SDK.
• Reliability: I also added logic in closeStream to ensure any remaining buffered data is fully flushed (sent) before the connection is closed.

And it works and significantly faster.

[TooTallNate]

Hi there, thank you for pointing this out. I have a different approach implemented in #867.

Could you please give it a try in your application and let me know if it improves performance to a satisfactory level for you? You can install by running npm i https://workflow-docs-a8ah8od0g.vercel.sh/workflow.tgz (or equivalent with you package manager of choice).

[gdaybrice]

Thanks @TooTallNate - any chance to publish this as a -canary version to npm instead please? Bun has trouble resolving interdependent packages from tarball URLs.

[VaguelySerious]

@gdaybrice We have shipped these changes to npm a while ago, any updates on your side for this?

[gdaybrice]

@VaguelySerious we had to move away from workflows for streaming because of single region support, streaming from syd to iad was to slow (the real issue here).

[VaguelySerious]

@gdaybrice Makes sense! We're planning to go multi-region eventually. Will close this issue in the meantime

[spesnova]

@VaguelySerious

I had the same issue before.

I upgraded vercel/workflow to 4.1.0-beta.63 today, and the streaming latency dropped to about 1/10 of what it was before. Thanks!

I'm using Vercel Workflow from Japan, so I'm also looking forward to a Japan region.