feat (ai/core): add langchain stream event v2 support to LangChainAdapter (#2340)

Author: lgrammel

Diff for: .changeset/tiny-plants-press.md

@@ -0,0 +1,5 @@
+---
+'ai': patch
+---
+
+feat (ai/core): add langchain stream event v2 support to LangChainAdapter

Diff for: content/docs/07-reference/stream-helpers/16-langchain-adapter.mdx

@@ -8,6 +8,12 @@ description: API Reference for LangChainAdapter.
 The `LangChainAdapter` module provides a way to transform LangChain output streams into AI streams.
 See the [LangChain Adapter documentation](/providers/adapters/langchain) for more information.
 
+It supports:
+
+- LangChain StringOutputParser streams
+- LangChain AIMessageChunk streams
+- LangChain StreamEvents v2 streams
+
 ## Import
 
 <Snippet text={`import { LangChainAdapter } from "ai"`} prompt={false} />

Diff for: packages/core/streams/langchain-adapter.test.ts

@@ -29,4 +29,18 @@ describe('toAIStream', () => {
       ['0:"Hello"\n', '0:"World"\n'],
     );
   });
+
+  it('should convert ReadableStream<LangChainStreamEvent>', async () => {
+    const inputStream = convertArrayToReadableStream([
+      { event: 'on_chat_model_stream', data: { chunk: { content: 'Hello' } } },
+      { event: 'on_chat_model_stream', data: { chunk: { content: 'World' } } },
+    ]);
+
+    assert.deepStrictEqual(
+      await convertReadableStreamToArray(
+        toAIStream(inputStream).pipeThrough(new TextDecoderStream()),
+      ),
+      ['0:"Hello"\n', '0:"World"\n'],
+    );
+  });
 });

Diff for: packages/core/streams/langchain-adapter.ts

@@ -37,6 +37,12 @@ type LangChainAIMessageChunk = {
   content: LangChainMessageContent;
 };
 
+// LC stream event v2
+type LangChainStreamEvent = {
+  event: string;
+  data: any;
+};
+
 /**
 Converts LangChain output streams to AIStream. 
 
@@ -45,28 +51,57 @@ The following streams are supported:
 - `string` streams (LangChain `StringOutputParser` output)
  */
 export function toAIStream(
-  stream: ReadableStream<LangChainAIMessageChunk> | ReadableStream<string>,
+  stream:
+    | ReadableStream<LangChainStreamEvent>
+    | ReadableStream<LangChainAIMessageChunk>
+    | ReadableStream<string>,
   callbacks?: AIStreamCallbacksAndOptions,
 ) {
   return stream
     .pipeThrough(
-      new TransformStream<LangChainAIMessageChunk | string>({
-        transform: async (chunk, controller) => {
-          if (typeof chunk === 'string') {
-            controller.enqueue(chunk);
-          } else if (typeof chunk.content === 'string') {
-            controller.enqueue(chunk.content);
-          } else {
-            const content: LangChainMessageContentComplex[] = chunk.content;
-            for (const item of content) {
-              if (item.type === 'text') {
-                controller.enqueue(item.text);
-              }
+      new TransformStream<
+        LangChainStreamEvent | LangChainAIMessageChunk | string
+      >({
+        transform: async (value, controller) => {
+          // text stream:
+          if (typeof value === 'string') {
+            controller.enqueue(value);
+            return;
+          }
+
+          // LC stream events v2:
+          if ('event' in value) {
+            // chunk is AIMessage Chunk for on_chat_model_stream event:
+            if (value.event === 'on_chat_model_stream') {
+              forwardAIMessageChunk(
+                value.data?.chunk as LangChainAIMessageChunk,
+                controller,
+              );
             }
+            return;
           }
+
+          // AI Message chunk stream:
+          forwardAIMessageChunk(value, controller);
         },
       }),
     )
     .pipeThrough(createCallbacksTransformer(callbacks))
     .pipeThrough(createStreamDataTransformer());
 }
+
+function forwardAIMessageChunk(
+  chunk: LangChainAIMessageChunk,
+  controller: TransformStreamDefaultController<any>,
+) {
+  if (typeof chunk.content === 'string') {
+    controller.enqueue(chunk.content);
+  } else {
+    const content: LangChainMessageContentComplex[] = chunk.content;
+    for (const item of content) {
+      if (item.type === 'text') {
+        controller.enqueue(item.text);
+      }
+    }
+  }
+}