chore (ai): data stream updates (#6158)

## Background

`DataStream` and `streamText` are fairly tightly coupled because of
their evolution. Going forward, data streams (or something similar)
should be available more broadly to allow for flexible usage.

## Summary
* remove `StreamTextResult.mergeIntoDataStream` method
* rename `DataStreamOptions.getErrorMessage` to `onError`
* add `pipeTextStreamToResponse` function
* add `createTextStreamResponse` function
* change `createDataStreamResponse` function to accept a `DataStream`
and not a `DataStreamWriter`
* change `pipeDataStreamToResponse` function to accept a `DataStream`
and not a `DataStreamWriter`
* change `pipeDataStreamToResponse` function to have a single parameter

Author: lgrammel

.changeset/tender-comics-rescue.md

@@ -0,0 +1,11 @@
+---
+'ai': major
+---
+
+remove StreamTextResult.mergeIntoDataStream method
+rename DataStreamOptions.getErrorMessage to onError
+add pipeTextStreamToResponse function
+add createTextStreamResponse function
+change createDataStreamResponse function to accept a DataStream and not a DataStreamWriter
+change pipeDataStreamToResponse function to accept a DataStream and not a DataStreamWriter
+change pipeDataStreamToResponse function to have a single parameter

examples/express/src/server.ts

@@ -1,5 +1,5 @@
 import { openai } from '@ai-sdk/openai';
-import { pipeDataStreamToResponse, streamText } from 'ai';
+import { createDataStream, pipeDataStreamToResponse, streamText } from 'ai';
 import 'dotenv/config';
 import express, { Request, Response } from 'express';
 
@@ -16,23 +16,26 @@ app.post('/', async (req: Request, res: Response) => {
 
 app.post('/stream-data', async (req: Request, res: Response) => {
   // immediately start streaming the response
-  pipeDataStreamToResponse(res, {
-    execute: async dataStreamWriter => {
-      dataStreamWriter.writeData('initialized call');
+
+  const dataStream = createDataStream({
+    execute: async writer => {
+      writer.writeData('initialized call');
 
       const result = streamText({
         model: openai('gpt-4o'),
         prompt: 'Invent a new holiday and describe its traditions.',
       });
 
-      result.mergeIntoDataStream(dataStreamWriter);
+      writer.merge(result.toDataStream());
     },
     onError: error => {
       // Error messages are masked by default for security reasons.
       // If you want to expose the error message to the client, you can do so here:
       return error instanceof Error ? error.message : String(error);
     },
   });
+
+  pipeDataStreamToResponse({ response: res, dataStream });
 });
 
 app.listen(8080, () => {

examples/fastify/src/server.ts

@@ -21,15 +21,15 @@ fastify.post('/', async function (request, reply) {
 fastify.post('/stream-data', async function (request, reply) {
   // immediately start streaming the response
   const dataStream = createDataStream({
-    execute: async dataStreamWriter => {
-      dataStreamWriter.writeData('initialized call');
+    execute: async writer => {
+      writer.writeData('initialized call');
 
       const result = streamText({
         model: openai('gpt-4o'),
         prompt: 'Invent a new holiday and describe its traditions.',
       });
 
-      result.mergeIntoDataStream(dataStreamWriter);
+      writer.merge(result.toDataStream());
     },
     onError: error => {
       // Error messages are masked by default for security reasons.

examples/hono/src/server.ts

@@ -23,15 +23,15 @@ app.post('/', async c => {
 app.post('/stream-data', async c => {
   // immediately start streaming the response
   const dataStream = createDataStream({
-    execute: async dataStreamWriter => {
-      dataStreamWriter.writeData('initialized call');
+    execute: async writer => {
+      writer.writeData('initialized call');
 
       const result = streamText({
         model: openai('gpt-4o'),
         prompt: 'Invent a new holiday and describe its traditions.',
       });
 
-      result.mergeIntoDataStream(dataStreamWriter);
+      writer.merge(result.toDataStream());
     },
     onError: error => {
       // Error messages are masked by default for security reasons.

examples/nest/src/app.controller.ts

@@ -1,6 +1,6 @@
 import { openai } from '@ai-sdk/openai';
 import { Controller, Post, Res } from '@nestjs/common';
-import { pipeDataStreamToResponse, streamText } from 'ai';
+import { createDataStream, pipeDataStreamToResponse, streamText } from 'ai';
 import { Response } from 'express';
 
 @Controller()
@@ -17,22 +17,24 @@ export class AppController {
 
   @Post('/stream-data')
   async streamData(@Res() res: Response) {
-    pipeDataStreamToResponse(res, {
-      execute: async (dataStreamWriter) => {
-        dataStreamWriter.writeData('initialized call');
+    const dataStream = createDataStream({
+      execute: async (writer) => {
+        writer.writeData('initialized call');
 
         const result = streamText({
           model: openai('gpt-4o'),
           prompt: 'Invent a new holiday and describe its traditions.',
         });
 
-        result.mergeIntoDataStream(dataStreamWriter);
+        writer.merge(result.toDataStream());
       },
       onError: (error) => {
         // Error messages are masked by default for security reasons.
         // If you want to expose the error message to the client, you can do so here:
         return error instanceof Error ? error.message : String(error);
       },
     });
+
+    pipeDataStreamToResponse({ response: res, dataStream });
   }
 }

examples/next-openai/app/api/use-chat-custom-sources/route.ts

@@ -1,13 +1,13 @@
 import { openai } from '@ai-sdk/openai';
-import { createDataStreamResponse, streamText } from 'ai';
+import { createDataStream, createDataStreamResponse, streamText } from 'ai';
 
 export async function POST(req: Request) {
   const { messages } = await req.json();
 
-  return createDataStreamResponse({
-    execute: dataStream => {
+  const dataStream = createDataStream({
+    execute: writer => {
       // write a custom url source to the stream:
-      dataStream.writeSource({
+      writer.writeSource({
         type: 'source',
         sourceType: 'url',
         id: 'source-1',
@@ -20,7 +20,9 @@ export async function POST(req: Request) {
         messages,
       });
 
-      result.mergeIntoDataStream(dataStream);
+      writer.merge(result.toDataStream());
     },
   });
+
+  return createDataStreamResponse({ dataStream });
 }

examples/next-openai/app/api/use-chat-human-in-the-loop/route.ts

@@ -1,5 +1,10 @@
 import { openai } from '@ai-sdk/openai';
-import { createDataStreamResponse, UIMessage, streamText } from 'ai';
+import {
+  createDataStreamResponse,
+  UIMessage,
+  streamText,
+  createDataStream,
+} from 'ai';
 import { processToolCalls } from './utils';
 import { tools } from './tools';
 
@@ -9,14 +14,14 @@ export const maxDuration = 30;
 export async function POST(req: Request) {
   const { messages }: { messages: UIMessage[] } = await req.json();
 
-  return createDataStreamResponse({
-    execute: async dataStream => {
+  const dataStream = createDataStream({
+    execute: async writer => {
       // Utility function to handle tools that require human confirmation
       // Checks for confirmation in last message and then runs associated tool
       const processedMessages = await processToolCalls(
         {
           messages,
-          dataStream,
+          dataStream: writer,
           tools,
         },
         {
@@ -36,7 +41,9 @@ export async function POST(req: Request) {
         tools,
       });
 
-      result.mergeIntoDataStream(dataStream);
+      writer.merge(result.toDataStream());
     },
   });
+
+  return createDataStreamResponse({ dataStream });
 }

examples/next-openai/app/api/use-chat-persistence-single-message-image-output/route.ts

@@ -3,6 +3,7 @@ import { loadChat, saveChat } from '@util/chat-store';
 import {
   appendClientMessage,
   appendResponseMessages,
+  createDataStream,
   createDataStreamResponse,
   createIdGenerator,
   streamText,
@@ -25,8 +26,8 @@ export async function POST(req: Request) {
   });
 
   // immediately start streaming (solves RAG issues with status, etc.)
-  return createDataStreamResponse({
-    execute: dataStream => {
+  const dataStream = createDataStream({
+    execute: writer => {
       const result = streamText({
         model: google('gemini-2.0-flash-exp'),
         providerOptions: {
@@ -49,12 +50,14 @@ export async function POST(req: Request) {
         },
       });
 
-      result.mergeIntoDataStream(dataStream);
+      writer.merge(result.toDataStream());
     },
     onError: error => {
       // Error messages are masked by default for security reasons.
       // If you want to expose the error message to the client, you can do so here:
       return error instanceof Error ? error.message : String(error);
     },
   });
+
+  return createDataStreamResponse({ dataStream });
 }

examples/next-openai/app/api/use-chat-persistence-single-message-tools/route.ts

@@ -3,6 +3,7 @@ import { loadChat, saveChat } from '@util/chat-store';
 import {
   appendClientMessage,
   appendResponseMessages,
+  createDataStream,
   createDataStreamResponse,
   createIdGenerator,
   streamText,
@@ -29,9 +30,9 @@ export async function POST(req: Request) {
   });
 
   // immediately start streaming (solves RAG issues with status, etc.)
-  return createDataStreamResponse({
-    execute: dataStream => {
-      dataStream.writeMessageAnnotation({
+  const dataStream = createDataStream({
+    execute: writer => {
+      writer.writeMessageAnnotation({
         start: 'start',
         count: count++,
       });
@@ -63,7 +64,7 @@ export async function POST(req: Request) {
                   Math.floor(Math.random() * weatherOptions.length)
                 ];
 
-              dataStream.writeMessageAnnotation({
+              writer.writeMessageAnnotation({
                 city,
                 weather,
               });
@@ -103,12 +104,14 @@ export async function POST(req: Request) {
         },
       });
 
-      result.mergeIntoDataStream(dataStream);
+      writer.merge(result.toDataStream());
     },
     onError: error => {
       // Error messages are masked by default for security reasons.
       // If you want to expose the error message to the client, you can do so here:
       return error instanceof Error ? error.message : String(error);
     },
   });
+
+  return createDataStreamResponse({ dataStream });
 }

examples/next-openai/app/api/use-chat-resume/route.ts

@@ -41,7 +41,7 @@ export async function POST(req: Request) {
   await appendStreamId({ chatId: id, streamId });
 
   const stream = createDataStream({
-    execute: dataStream => {
+    execute: writer => {
       const result = streamText({
         model: openai('gpt-4o'),
         messages,
@@ -56,7 +56,7 @@ export async function POST(req: Request) {
         },
       });
 
-      result.mergeIntoDataStream(dataStream);
+      writer.merge(result.toDataStream());
     },
   });