feat(langgraph): replace toLangGraphEventStream with encoding argument (#1672)

.changeset/funny-houses-mate.md

@@ -0,0 +1,5 @@
+---
+"@langchain/langgraph": patch
+---
+
+Add `stream.encoding` option to emit LangGraph API events as Server-Sent Events. This allows for sending events through the wire by piping the stream to a `Response` object.

examples/stream-transport-vite/src/server.mts

@@ -1,6 +1,5 @@
 import type { BaseMessage } from "@langchain/core/messages";
 import { StateGraph, MessagesZodMeta, START } from "@langchain/langgraph";
-import { toLangGraphEventStreamResponse } from "@langchain/langgraph/ui";
 import { registry } from "@langchain/langgraph/zod";
 import { ChatOpenAI } from "@langchain/openai";
 import { z } from "zod/v4";
@@ -10,11 +9,11 @@ import { Hono } from "hono";
 
 const llm = new ChatOpenAI({ model: "gpt-4o-mini" });
 
-const graph = new StateGraph(
-  z.object({
-    messages: z.custom<BaseMessage[]>().register(registry, MessagesZodMeta),
-  })
-)
+const schema = z.object({
+  messages: z.custom<BaseMessage[]>().register(registry, MessagesZodMeta),
+});
+
+const graph = new StateGraph(schema)
   .addNode("agent", async ({ messages }) => ({
     messages: await llm.invoke(messages),
   }))
@@ -26,14 +25,15 @@ export type GraphType = typeof graph;
 const app = new Hono();
 
 app.post("/api/stream", async (c) => {
-  type InputType = GraphType["~InputType"];
-  const { input } = await c.req.json<{ input: InputType }>();
+  const { input } = z.object({ input: schema }).parse(await c.req.json());
 
-  return toLangGraphEventStreamResponse({
-    stream: graph.streamEvents(input, {
-      version: "v2",
-      streamMode: ["values", "messages"],
-    }),
+  const stream = await graph.stream(input, {
+    encoding: "text/event-stream",
+    streamMode: ["values", "messages", "updates"],
+  });
+
+  return new Response(stream, {
+    headers: { "Content-Type": "text/event-stream" },
   });
 });
 

libs/langgraph/package.json

@@ -178,17 +178,6 @@
       },
       "input": "./src/graph/zod/schema.ts"
     },
-    "./ui": {
-      "import": {
-        "types": "./dist/ui/index.d.ts",
-        "default": "./dist/ui/index.js"
-      },
-      "require": {
-        "types": "./dist/ui/index.d.cts",
-        "default": "./dist/ui/index.cjs"
-      },
-      "input": "./src/ui/index.ts"
-    },
     "./package.json": "./package.json"
   },
   "files": [

libs/langgraph/src/pregel/index.ts

@@ -86,6 +86,7 @@ import { PregelRunner } from "./runner.js";
 import {
   IterableReadableStreamWithAbortSignal,
   IterableReadableWritableStream,
+  toEventStream,
 } from "./stream.js";
 import type {
   Durability,
@@ -1827,11 +1828,19 @@ export class Pregel<
    */
   override async stream<
     TStreamMode extends StreamMode | StreamMode[] | undefined,
-    TSubgraphs extends boolean
+    TSubgraphs extends boolean,
+    TEncoding extends "text/event-stream" | undefined
   >(
     input: InputType | CommandType | null,
     options?: Partial<
-      PregelOptions<Nodes, Channels, ContextType, TStreamMode, TSubgraphs>
+      PregelOptions<
+        Nodes,
+        Channels,
+        ContextType,
+        TStreamMode,
+        TSubgraphs,
+        TEncoding
+      >
     >
   ): Promise<
     IterableReadableStream<
@@ -1842,7 +1851,8 @@ export class Pregel<
         StreamValuesType,
         keyof Nodes,
         NodeReturnType,
-        StreamCustom
+        StreamCustom,
+        TEncoding
       >
     >
   > {
@@ -1860,18 +1870,11 @@ export class Pregel<
         .signal,
     };
 
+    const stream = await super.stream(input, config);
     return new IterableReadableStreamWithAbortSignal(
-      (await super.stream(input, config)) as IterableReadableStream<
-        StreamOutputMap<
-          TStreamMode,
-          TSubgraphs,
-          StreamUpdatesType,
-          StreamValuesType,
-          keyof Nodes,
-          NodeReturnType,
-          StreamCustom
-        >
-      >,
+      options?.encoding === "text/event-stream"
+        ? toEventStream(stream)
+        : stream,
       abortController
     );
   }
@@ -1958,6 +1961,9 @@ export class Pregel<
     input: PregelInputType | Command,
     options?: Partial<PregelOptions<Nodes, Channels>>
   ): AsyncGenerator<PregelOutputType> {
+    // Skip LGP encoding option is `streamEvents` is used
+    const streamEncoding =
+      "version" in (options ?? {}) ? undefined : options?.encoding ?? undefined;
     const streamSubgraphs = options?.subgraphs;
     const inputConfig = ensureLangGraphConfig(this.config, options);
     if (
@@ -2144,6 +2150,14 @@ export class Pregel<
         }
         const [namespace, mode, payload] = chunk;
         if (streamMode.includes(mode)) {
+          if (streamEncoding === "text/event-stream") {
+            if (streamSubgraphs) {
+              yield [namespace, mode, payload];
+            } else {
+              yield [null, mode, payload];
+            }
+            continue;
+          }
           if (streamSubgraphs && !streamModeSingle) {
             yield [namespace, mode, payload];
           } else if (!streamModeSingle) {
@@ -2177,13 +2191,16 @@ export class Pregel<
    */
   override async invoke(
     input: InputType | CommandType | null,
-    options?: Partial<PregelOptions<Nodes, Channels, ContextType>>
+    options?: Partial<
+      Omit<PregelOptions<Nodes, Channels, ContextType>, "encoding">
+    >
   ): Promise<OutputType> {
     const streamMode = options?.streamMode ?? "values";
     const config = {
       ...options,
       outputKeys: options?.outputKeys ?? this.outputChannels,
       streamMode,
+      encoding: undefined,
     };
     const chunks = [];
     const stream = await this.stream(input, config);

libs/langgraph/src/pregel/stream.ts

@@ -1,9 +1,32 @@
 import { IterableReadableStream } from "@langchain/core/utils/stream";
-import { StreamMode } from "./types.js";
+import type { RunnableConfig } from "@langchain/core/runnables";
+import type { StreamMode, StreamOutputMap } from "./types.js";
 
 // [namespace, streamMode, payload]
 export type StreamChunk = [string[], StreamMode, unknown];
 
+type StreamCheckpointsOutput<StreamValues> = StreamOutputMap<
+  "checkpoints",
+  false,
+  StreamValues,
+  unknown,
+  string,
+  unknown,
+  unknown,
+  undefined
+>;
+
+type AnyStreamOutput = StreamOutputMap<
+  StreamMode[],
+  true,
+  unknown,
+  unknown,
+  string,
+  unknown,
+  unknown,
+  undefined
+>;
+
 /**
  * A wrapper around an IterableReadableStream that allows for aborting the stream when
  * {@link cancel} is called.
@@ -125,3 +148,145 @@ export class IterableReadableWritableStream extends IterableReadableStream<Strea
     this.controller.error(e);
   }
 }
+
+function _stringifyAsDict(obj: unknown) {
+  return JSON.stringify(obj, function (key: string | number, value: unknown) {
+    const rawValue = this[key];
+    if (
+      rawValue != null &&
+      typeof rawValue === "object" &&
+      "toDict" in rawValue &&
+      typeof rawValue.toDict === "function"
+    ) {
+      const { type, data } = rawValue.toDict();
+      return { ...data, type };
+    }
+
+    return value;
+  });
+}
+
+function _serializeError(error: unknown) {
+  // eslint-disable-next-line no-instanceof/no-instanceof
+  if (error instanceof Error) {
+    return { error: error.name, message: error.message };
+  }
+  return { error: "Error", message: JSON.stringify(error) };
+}
+
+function _isRunnableConfig(
+  config: unknown
+): config is RunnableConfig & { configurable: Record<string, unknown> } {
+  if (typeof config !== "object" || config == null) return false;
+  return (
+    "configurable" in config &&
+    typeof config.configurable === "object" &&
+    config.configurable != null
+  );
+}
+
+function _extractCheckpointFromConfig(
+  config: RunnableConfig | null | undefined
+) {
+  if (!_isRunnableConfig(config) || !config.configurable.thread_id) {
+    return null;
+  }
+
+  return {
+    thread_id: config.configurable.thread_id,
+    checkpoint_ns: config.configurable.checkpoint_ns || "",
+    checkpoint_id: config.configurable.checkpoint_id || null,
+    checkpoint_map: config.configurable.checkpoint_map || null,
+  };
+}
+
+function _serializeConfig(config: unknown) {
+  if (_isRunnableConfig(config)) {
+    const configurable = Object.fromEntries(
+      Object.entries(config.configurable).filter(
+        ([key]) => !key.startsWith("__")
+      )
+    );
+
+    const newConfig = { ...config, configurable };
+    delete newConfig.callbacks;
+    return newConfig;
+  }
+
+  return config;
+}
+
+function _serializeCheckpoint(payload: StreamCheckpointsOutput<unknown>) {
+  const result: Record<string, unknown> = {
+    ...payload,
+    checkpoint: _extractCheckpointFromConfig(payload.config),
+    parent_checkpoint: _extractCheckpointFromConfig(payload.parentConfig),
+
+    config: _serializeConfig(payload.config),
+    parent_config: _serializeConfig(payload.parentConfig),
+
+    tasks: payload.tasks.map((task) => {
+      if (_isRunnableConfig(task.state)) {
+        const checkpoint = _extractCheckpointFromConfig(task.state);
+        if (checkpoint != null) {
+          const cloneTask: Record<string, unknown> = { ...task, checkpoint };
+          delete cloneTask.state;
+          return cloneTask;
+        }
+      }
+
+      return task;
+    }),
+  };
+
+  delete result.parentConfig;
+  return result;
+}
+
+export function toEventStream(stream: AsyncGenerator) {
+  const encoder = new TextEncoder();
+  return new ReadableStream<Uint8Array>({
+    async start(controller) {
+      const enqueueChunk = (sse: {
+        id?: string;
+        event: string;
+        data: unknown;
+      }) => {
+        controller.enqueue(
+          encoder.encode(
+            `event: ${sse.event}\ndata: ${_stringifyAsDict(sse.data)}\n\n`
+          )
+        );
+      };
+
+      try {
+        for await (const payload of stream) {
+          const [ns, mode, chunk] = payload as AnyStreamOutput;
+
+          let data: unknown = chunk;
+          if (mode === "debug") {
+            const debugChunk = chunk;
+
+            if (debugChunk.type === "checkpoint") {
+              data = {
+                ...debugChunk,
+                payload: _serializeCheckpoint(debugChunk.payload),
+              };
+            }
+          }
+
+          if (mode === "checkpoints") {
+            data = _serializeCheckpoint(chunk);
+          }
+
+          const event = ns?.length ? `${mode}|${ns.join("|")}` : mode;
+          enqueueChunk({ event, data });
+        }
+      } catch (error) {
+        enqueueChunk({ event: "error", data: _serializeError(error) });
+      }
+
+      controller.close();
+    },
+  });
+}

libs/langgraph/src/pregel/types.ts

@@ -74,6 +74,12 @@ type StreamTasksOutput<StreamUpdates, StreamValues, Nodes = string> =
 
 type DefaultStreamMode = "updates";
 
+export type IsEventStream<T> = [T] extends ["text/event-stream"]
+  ? ["text/event-stream"] extends [T]
+    ? true
+    : false
+  : false;
+
 export type StreamOutputMap<
   TStreamMode extends StreamMode | StreamMode[] | undefined,
   TStreamSubgraphs extends boolean,
@@ -81,18 +87,21 @@ export type StreamOutputMap<
   StreamValues,
   Nodes,
   NodeReturnType,
-  StreamCustom
-> = (
-  undefined extends TStreamMode
-    ? []
-    : StreamMode | StreamMode[] extends TStreamMode
-    ? TStreamMode extends StreamMode[]
-      ? TStreamMode[number]
-      : TStreamMode
-    : TStreamMode extends StreamMode[]
-    ? TStreamMode[number]
-    : []
-) extends infer Multiple extends StreamMode
+  StreamCustom,
+  TEncoding extends "text/event-stream" | undefined
+> = IsEventStream<TEncoding> extends true
+  ? Uint8Array
+  : (
+      undefined extends TStreamMode
+        ? []
+        : StreamMode | StreamMode[] extends TStreamMode
+        ? TStreamMode extends StreamMode[]
+          ? TStreamMode[number]
+          : TStreamMode
+        : TStreamMode extends StreamMode[]
+        ? TStreamMode[number]
+        : []
+    ) extends infer Multiple extends StreamMode
   ? [TStreamSubgraphs] extends [true]
     ? {
         values: [string[], "values", StreamValues];
@@ -182,7 +191,10 @@ export interface PregelOptions<
     | StreamMode
     | StreamMode[]
     | undefined,
-  TSubgraphs extends boolean = boolean
+  TSubgraphs extends boolean = boolean,
+  TEncoding extends "text/event-stream" | undefined =
+    | "text/event-stream"
+    | undefined
 > extends RunnableConfig<ContextType> {
   /**
    * Controls what information is streamed during graph execution.
@@ -316,6 +328,14 @@ export interface PregelOptions<
    * Static context for the graph run, like `userId`, `dbConnection` etc.
    */
   context?: ContextType;
+
+  /**
+   * The encoding to use for the stream.
+   * - `undefined`: Use the default format.
+   * - `"text/event-stream"`: Use the Server-Sent Events format.
+   * @default undefined
+   */
+  encoding?: TEncoding;
 }
 
 /**

libs/langgraph/src/tests/pregel.test-d.ts

@@ -199,6 +199,16 @@ it("state graph annotation", async () => {
         ]
     )[]
   >();
+
+  expectTypeOf(
+    await gatherIterator(
+      graph.stream(input, {
+        encoding: "text/event-stream",
+        streamMode: "values",
+        subgraphs: true,
+      })
+    )
+  ).toExtend<Uint8Array[]>();
 });
 
 it("state graph configurable", async () => {

libs/langgraph/src/tests/ui.test-d.ts

@@ -1,68 +0,0 @@
-import { expectTypeOf, it } from "vitest";
-import { Annotation, StateGraph } from "../graph/index.js";
-import { MessagesAnnotation } from "../graph/messages_annotation.js";
-import { END, START } from "../constants.js";
-import { toLangGraphEventStream } from "../ui/stream.js";
-import type { SerializedMessage } from "../ui/types.message.js";
-
-it("toLangGraphEventStream", async () => {
-  const graph = new StateGraph(
-    Annotation.Root({
-      messages: MessagesAnnotation.spec.messages,
-      foo: Annotation<string>,
-    })
-  )
-    .addNode("one", () => ({
-      messages: { role: "ai", content: "one" as const },
-    }))
-    .addNode("two", () => ({
-      messages: { role: "ai", content: "two" as const },
-      foo: "foo:two" as const,
-    }))
-    .addEdge(START, "one")
-    .addEdge("one", "two")
-    .addEdge("two", END)
-    .compile();
-
-  type GraphType = typeof graph;
-  type ExtraType = { CustomType: string; InterruptType: number };
-
-  const stream = await toLangGraphEventStream<GraphType, ExtraType>(
-    graph.streamEvents({ messages: "input" }, { version: "v2" })
-  );
-
-  for await (const { event, data } of stream) {
-    if (event === "values") {
-      expectTypeOf(data).toExtend<{
-        messages: SerializedMessage.AnyMessage[];
-        foo: string;
-      }>();
-
-      expectTypeOf(data).toExtend<{ __interrupt__?: number }>();
-      expectTypeOf(data).not.toExtend<{ __interrupt__?: string }>();
-    }
-
-    if (event === "updates") {
-      expectTypeOf(data).toExtend<{
-        one?: { messages: { role: "ai"; content: "one" } };
-        two?: { messages: { role: "ai"; content: "two" }; foo: "foo:two" };
-      }>();
-
-      expectTypeOf(data).not.toExtend<{
-        one?: { messages?: { role: "human"; content: "two" } };
-      }>();
-      expectTypeOf(data).not.toExtend<{ two?: { foo: "invalid" } }>();
-
-      expectTypeOf(data).toExtend<{
-        [node: string]: {
-          messages?: SerializedMessage.AnyMessage[] | undefined;
-          foo?: string | undefined;
-        };
-      }>();
-    }
-
-    if (event === "custom") {
-      expectTypeOf(data).toExtend<string>();
-    }
-  }
-});

libs/langgraph/src/ui/consume.ts

@@ -1,206 +0,0 @@
-import { IterableReadableStream } from "@langchain/core/utils/stream";
-import type {
-  AnyPregelLike,
-  ExtraTypeBag,
-  InferLangGraphEventStream,
-} from "./types.infer.js";
-
-const CR = "\r".charCodeAt(0);
-const LF = "\n".charCodeAt(0);
-const NULL = "\0".charCodeAt(0);
-const COLON = ":".charCodeAt(0);
-const SPACE = " ".charCodeAt(0);
-
-const TRAILING_NEWLINE = [CR, LF];
-
-function BytesLineDecoder() {
-  let buffer: Uint8Array[] = [];
-  let trailingCr = false;
-
-  return new TransformStream<Uint8Array, Uint8Array>({
-    start() {
-      buffer = [];
-      trailingCr = false;
-    },
-
-    transform(chunk, controller) {
-      // See https://docs.python.org/3/glossary.html#term-universal-newlines
-      let text = chunk;
-
-      // Handle trailing CR from previous chunk
-      if (trailingCr) {
-        text = joinArrays([[CR], text]);
-        trailingCr = false;
-      }
-
-      // Check for trailing CR in current chunk
-      if (text.length > 0 && text.at(-1) === CR) {
-        trailingCr = true;
-        text = text.subarray(0, -1);
-      }
-
-      if (!text.length) return;
-      const trailingNewline = TRAILING_NEWLINE.includes(text.at(-1)!);
-
-      const lastIdx = text.length - 1;
-      const { lines } = text.reduce<{ lines: Uint8Array[]; from: number }>(
-        (acc, cur, idx) => {
-          if (acc.from > idx) return acc;
-
-          if (cur === CR || cur === LF) {
-            acc.lines.push(text.subarray(acc.from, idx));
-            if (cur === CR && text[idx + 1] === LF) {
-              acc.from = idx + 2;
-            } else {
-              acc.from = idx + 1;
-            }
-          }
-
-          if (idx === lastIdx && acc.from <= lastIdx) {
-            acc.lines.push(text.subarray(acc.from));
-          }
-
-          return acc;
-        },
-        { lines: [], from: 0 }
-      );
-
-      if (lines.length === 1 && !trailingNewline) {
-        buffer.push(lines[0]);
-        return;
-      }
-
-      if (buffer.length) {
-        // Include existing buffer in first line
-        buffer.push(lines[0]);
-        lines[0] = joinArrays(buffer);
-        buffer = [];
-      }
-
-      if (!trailingNewline) {
-        // If the last segment is not newline terminated,
-        // buffer it for the next chunk
-        if (lines.length) buffer = [lines.pop()!];
-      }
-
-      // Enqueue complete lines
-      for (const line of lines) {
-        controller.enqueue(line);
-      }
-    },
-
-    flush(controller) {
-      if (buffer.length) {
-        controller.enqueue(joinArrays(buffer));
-      }
-    },
-  });
-}
-
-function SSEDecoder<
-  Part extends {
-    id?: string | undefined;
-    event: string;
-    data: unknown;
-  }
->() {
-  let event = "";
-  let data: Uint8Array[] = [];
-  let lastEventId = "";
-  let retry: number | null = null;
-
-  const decoder = new TextDecoder();
-
-  return new TransformStream<Uint8Array, Part>({
-    transform(chunk, controller) {
-      // Handle empty line case
-      if (!chunk.length) {
-        if (!event && !data.length && !lastEventId && retry == null) return;
-
-        const sse = {
-          id: lastEventId || undefined,
-          event,
-          data: data.length ? decodeArraysToJson(decoder, data) : null,
-        };
-
-        // NOTE: as per the SSE spec, do not reset lastEventId
-        event = "";
-        data = [];
-        retry = null;
-
-        controller.enqueue(sse as Part);
-        return;
-      }
-
-      // Ignore comments
-      if (chunk[0] === COLON) return;
-
-      const sepIdx = chunk.indexOf(COLON);
-      if (sepIdx === -1) return;
-
-      const fieldName = decoder.decode(chunk.subarray(0, sepIdx));
-      let value = chunk.subarray(sepIdx + 1);
-      if (value[0] === SPACE) value = value.subarray(1);
-
-      if (fieldName === "event") {
-        event = decoder.decode(value);
-      } else if (fieldName === "data") {
-        data.push(value);
-      } else if (fieldName === "id") {
-        if (value.indexOf(NULL) === -1) lastEventId = decoder.decode(value);
-      } else if (fieldName === "retry") {
-        const retryNum = Number.parseInt(decoder.decode(value), 10);
-        if (!Number.isNaN(retryNum)) retry = retryNum;
-      }
-    },
-
-    flush(controller) {
-      if (event) {
-        controller.enqueue({
-          id: lastEventId || undefined,
-          event,
-          data: data.length ? decodeArraysToJson(decoder, data) : null,
-        } as Part);
-      }
-    },
-  });
-}
-
-function joinArrays(data: ArrayLike<number>[]) {
-  const totalLength = data.reduce((acc, curr) => acc + curr.length, 0);
-  const merged = new Uint8Array(totalLength);
-  let offset = 0;
-  for (const c of data) {
-    merged.set(c, offset);
-    offset += c.length;
-  }
-  return merged;
-}
-
-function decodeArraysToJson(decoder: TextDecoder, data: ArrayLike<number>[]) {
-  return JSON.parse(decoder.decode(joinArrays(data)));
-}
-
-export async function* stream<
-  TGraph extends AnyPregelLike,
-  TExtra extends ExtraTypeBag = ExtraTypeBag
->(options: {
-  url: string;
-  headers: HeadersInit;
-}): AsyncGenerator<InferLangGraphEventStream<TGraph, TExtra>> {
-  const res = await fetch(options.url, {
-    method: "POST",
-    headers: options.headers,
-    body: JSON.stringify({ type: "stream" }),
-  });
-
-  if (!res.ok) {
-    throw new Error(`Failed to stream: ${res.statusText}`);
-  }
-
-  yield* IterableReadableStream.fromReadableStream(
-    (res.body || new ReadableStream({ start: (ctrl) => ctrl.close() }))
-      .pipeThrough(BytesLineDecoder())
-      .pipeThrough(SSEDecoder<InferLangGraphEventStream<TGraph, TExtra>>())
-  );
-}

libs/langgraph/src/ui/index.ts

@@ -1,7 +0,0 @@
-export {
-  toLangGraphEventStream,
-  toLangGraphEventStreamResponse,
-} from "./stream.js";
-export type { SerializedMessage } from "./types.message.js";
-export type { LangGraphEventStream } from "./types.schema.js";
-export type { InferLangGraphEventStream, ExtraTypeBag } from "./types.infer.js";

libs/langgraph/src/ui/stream.ts

@@ -1,233 +0,0 @@
-import type { RunnableConfig } from "@langchain/core/runnables";
-import type { StreamEvent } from "@langchain/core/tracers/log_stream";
-import type { StreamMode, StreamOutputMap } from "../pregel/types.js";
-import type {
-  AnyPregelLike,
-  ExtraTypeBag,
-  InferLangGraphEventStream,
-} from "./types.infer.js";
-
-type StreamCheckpointsOutput<StreamValues> = StreamOutputMap<
-  "checkpoints",
-  false,
-  StreamValues,
-  unknown,
-  string,
-  unknown,
-  unknown
->;
-
-const serialiseAsDict = <T>(obj: T): T => {
-  return JSON.parse(
-    JSON.stringify(obj, function (key: string | number, value: unknown) {
-      const rawValue = this[key];
-      if (
-        rawValue != null &&
-        typeof rawValue === "object" &&
-        "toDict" in rawValue &&
-        typeof rawValue.toDict === "function"
-      ) {
-        const { type, data } = rawValue.toDict();
-        return { ...data, type };
-      }
-
-      return value;
-    })
-  );
-};
-
-const serializeError = (error: unknown) => {
-  // eslint-disable-next-line no-instanceof/no-instanceof
-  if (error instanceof Error) {
-    return { error: error.name, message: error.message };
-  }
-  return { error: "Error", message: JSON.stringify(error) };
-};
-
-const isRunnableConfig = (
-  config: unknown
-): config is RunnableConfig & { configurable: Record<string, unknown> } => {
-  if (typeof config !== "object" || config == null) return false;
-  return (
-    "configurable" in config &&
-    typeof config.configurable === "object" &&
-    config.configurable != null
-  );
-};
-
-const extractCheckpointFromConfig = (
-  config: RunnableConfig | null | undefined
-) => {
-  if (!isRunnableConfig(config) || !config.configurable.thread_id) {
-    return null;
-  }
-
-  return {
-    thread_id: config.configurable.thread_id,
-    checkpoint_ns: config.configurable.checkpoint_ns || "",
-    checkpoint_id: config.configurable.checkpoint_id || null,
-    checkpoint_map: config.configurable.checkpoint_map || null,
-  };
-};
-
-const serializeConfig = (config: unknown) => {
-  if (isRunnableConfig(config)) {
-    const configurable = Object.fromEntries(
-      Object.entries(config.configurable).filter(
-        ([key]) => !key.startsWith("__")
-      )
-    );
-
-    const newConfig = { ...config, configurable };
-    delete newConfig.callbacks;
-    return newConfig;
-  }
-
-  return config;
-};
-
-function serializeCheckpoint(payload: StreamCheckpointsOutput<unknown>) {
-  const result: Record<string, unknown> = {
-    ...payload,
-    checkpoint: extractCheckpointFromConfig(payload.config),
-    parent_checkpoint: extractCheckpointFromConfig(payload.parentConfig),
-
-    config: serializeConfig(payload.config),
-    parent_config: serializeConfig(payload.parentConfig),
-
-    tasks: payload.tasks.map((task) => {
-      if (isRunnableConfig(task.state)) {
-        const checkpoint = extractCheckpointFromConfig(task.state);
-        if (checkpoint != null) {
-          const cloneTask: Record<string, unknown> = { ...task, checkpoint };
-          delete cloneTask.state;
-          return cloneTask;
-        }
-      }
-
-      return task;
-    }),
-  };
-
-  delete result.parentConfig;
-  return result;
-}
-
-/**
- * Converts a `graph.streamEvents()` output into a LangGraph Platform compatible event stream.
- * @experimental Does not follow semver.
- *
- * @param events
- */
-export async function* toLangGraphEventStream<
-  TGraph extends AnyPregelLike,
-  TExtra extends ExtraTypeBag = ExtraTypeBag
->(
-  events: AsyncIterable<StreamEvent> | Promise<AsyncIterable<StreamEvent>>
-): AsyncGenerator<InferLangGraphEventStream<TGraph, TExtra>> {
-  let rootRunId: string | undefined;
-
-  try {
-    for await (const event of await events) {
-      if (event.event === "on_chain_start" && rootRunId == null) {
-        rootRunId = event.run_id;
-      }
-      if (event.tags?.includes("langsmith:hidden")) continue;
-      if (event.event === "on_chain_stream" && event.run_id === rootRunId) {
-        if (!Array.isArray(event.data.chunk)) {
-          continue;
-        }
-
-        type AnyStreamOutput = StreamOutputMap<
-          StreamMode[],
-          true,
-          unknown,
-          unknown,
-          string,
-          unknown,
-          unknown
-        >;
-
-        const [ns, mode, chunk] = (
-          event.data.chunk.length === 3
-            ? event.data.chunk
-            : [null, ...event.data.chunk]
-        ) as AnyStreamOutput;
-
-        // Listen for debug events and capture checkpoint
-        let data: unknown = chunk;
-        if (mode === "debug") {
-          const debugChunk = chunk;
-
-          if (debugChunk.type === "checkpoint") {
-            data = {
-              ...debugChunk,
-              payload: serializeCheckpoint(debugChunk.payload),
-            };
-          }
-        }
-
-        if (mode === "checkpoints") {
-          data = serializeCheckpoint(chunk);
-        }
-
-        // This needs to be done for LC.js V0 messages, since they
-        // by default serialize using the verbose Serializable protocol.
-        data = serialiseAsDict(data);
-
-        yield {
-          event: ns?.length ? `${mode}|${ns.join("|")}` : mode,
-          data,
-        } as InferLangGraphEventStream<TGraph, TExtra>;
-      }
-    }
-  } catch (error) {
-    yield { event: "error", data: serializeError(error) };
-  }
-}
-
-/**
- * Converts a `graph.streamEvents()` output into a LangGraph Platform compatible Web Response.
- * @experimental Does not follow semver.
- */
-export function toLangGraphEventStreamResponse(options: {
-  status?: number;
-  statusText?: string;
-  headers?: Headers | Record<string, string>;
-  stream: AsyncIterable<StreamEvent> | Promise<AsyncIterable<StreamEvent>>;
-}) {
-  const headers = new Headers({
-    "Content-Type": "text/event-stream",
-    "Cache-Control": "no-cache",
-    Connection: "keep-alive",
-  });
-
-  if (options.headers) {
-    if (
-      "forEach" in options.headers &&
-      typeof options.headers.forEach === "function"
-    ) {
-      options.headers.forEach((v, k) => headers.set(k, v));
-    } else {
-      Object.entries(options.headers).map(([k, v]) => headers.set(k, v));
-    }
-  }
-
-  return new Response(
-    new ReadableStream({
-      async start(controller) {
-        try {
-          for await (const { event, data } of toLangGraphEventStream(
-            options.stream
-          )) {
-            controller.enqueue(`event: ${event}\n`);
-            controller.enqueue(`data: ${JSON.stringify(data)}\n\n`);
-          }
-        } finally {
-          controller.close();
-        }
-      },
-    }),
-    { headers, status: options.status, statusText: options.statusText }
-  );
-}

libs/langgraph/src/ui/types.infer.ts

@@ -1,102 +0,0 @@
-/* eslint-disable @typescript-eslint/no-explicit-any */
-import type { BaseMessage } from "@langchain/core/messages";
-import type { SerializedMessage } from "./types.message.js";
-import type { LangGraphEventStream } from "./types.schema.js";
-
-type Equals<X, Y> = (<T>() => T extends X ? 1 : 2) extends <T>() => T extends Y
-  ? 1
-  : 2
-  ? true
-  : false;
-
-type MatchBaseMessage<T> = T extends BaseMessage ? BaseMessage : never;
-type MatchBaseMessageArray<T> = T extends Array<infer C>
-  ? Equals<MatchBaseMessage<C>, BaseMessage> extends true
-    ? BaseMessage[]
-    : never
-  : never;
-
-type ReplaceMessages<T, TDepth extends Array<0> = []> = TDepth extends [0, 0, 0]
-  ? any
-  : T extends unknown
-  ? {
-      [K in keyof T]: 0 extends 1 & T[K]
-        ? T[K]
-        : Equals<MatchBaseMessageArray<T[K]>, BaseMessage[]> extends true
-        ? SerializedMessage.AnyMessage[]
-        : Equals<MatchBaseMessage<T[K]>, BaseMessage> extends true
-        ? SerializedMessage.AnyMessage
-        : ReplaceMessages<T[K], [0, ...TDepth]>;
-    }
-  : never;
-
-/* eslint-disable @typescript-eslint/no-explicit-any */
-type Defactorify<T> = T extends (...args: any[]) => infer R
-  ? Awaited<R>
-  : Awaited<T>;
-
-type AnyPregel = {
-  lg_is_pregel: boolean;
-  stream: (...args: any[]) => any;
-  invoke: (...args: any[]) => any;
-};
-
-type AnyGraph = {
-  compiled: boolean;
-  compile: (...args: any[]) => any;
-};
-
-export type AnyPregelLike =
-  | AnyPregel
-  | AnyGraph
-  | ((...args: any[]) => AnyPregel | AnyGraph);
-
-type ReflectCompiledGraph<T> = T extends {
-  RunInput: infer State;
-  RunOutput: infer Update;
-  "~NodeReturnType"?: infer ReturnType;
-}
-  ? {
-      state: ReplaceMessages<State>;
-      update: ReplaceMessages<Update>;
-      returnType: ReturnType;
-    }
-  : T extends { "~InputType": infer InputType; "~OutputType": infer OutputType }
-  ? { state: ReplaceMessages<OutputType>; update: ReplaceMessages<InputType> }
-  : never;
-
-type InferGraph<T> = Defactorify<T> extends infer DT
-  ? DT extends {
-      compile(...args: any[]): infer Compiled;
-    }
-    ? ReflectCompiledGraph<Compiled>
-    : ReflectCompiledGraph<DT>
-  : never;
-
-export type ExtraTypeBag<TCustomType = unknown, TInterruptType = unknown> = {
-  CustomType?: TCustomType;
-  InterruptType?: TInterruptType;
-};
-
-type GetCustomType<Bag extends ExtraTypeBag> = Bag extends {
-  CustomType: unknown;
-}
-  ? Bag["CustomType"]
-  : unknown;
-
-type GetInterruptType<Bag extends ExtraTypeBag> = Bag extends {
-  InterruptType: unknown;
-}
-  ? Bag["InterruptType"]
-  : unknown;
-
-export type InferLangGraphEventStream<
-  TGraph extends AnyPregelLike,
-  TExtra extends ExtraTypeBag = ExtraTypeBag
-> = LangGraphEventStream<
-  InferGraph<TGraph>["state"],
-  InferGraph<TGraph>["update"],
-  GetCustomType<TExtra>,
-  GetInterruptType<TExtra>,
-  InferGraph<TGraph>["returnType"]
->;

libs/langgraph/src/ui/types.message.ts

@@ -1,96 +0,0 @@
-type ImageDetail = "auto" | "low" | "high";
-type MessageContentImageUrl = {
-  type: "image_url";
-  image_url: string | { url: string; detail?: ImageDetail | undefined };
-};
-type MessageContentText = { type: "text"; text: string };
-type MessageContentComplex = MessageContentText | MessageContentImageUrl;
-type MessageContent = string | MessageContentComplex[];
-
-/**
- * Model-specific additional kwargs, which is passed back to the underlying LLM.
- */
-type MessageAdditionalKwargs = Record<string, unknown>;
-
-// eslint-disable-next-line @typescript-eslint/no-namespace
-export namespace SerializedMessage {
-  type BaseMessage = {
-    additional_kwargs?: MessageAdditionalKwargs | undefined;
-    content: MessageContent;
-    id?: string | undefined;
-    name?: string | undefined;
-    response_metadata?: Record<string, unknown> | undefined;
-  };
-
-  export type HumanMessage = BaseMessage & {
-    type: "human";
-    example?: boolean | undefined;
-  };
-
-  export type AIMessage = BaseMessage & {
-    type: "ai";
-    example?: boolean | undefined;
-    tool_calls?:
-      | {
-          name: string;
-          // eslint-disable-next-line @typescript-eslint/no-explicit-any
-          args: { [x: string]: any };
-          id?: string | undefined;
-          type?: "tool_call" | undefined;
-        }[]
-      | undefined;
-    invalid_tool_calls?:
-      | {
-          name?: string | undefined;
-          args?: string | undefined;
-          id?: string | undefined;
-          error?: string | undefined;
-          type?: "invalid_tool_call" | undefined;
-        }[]
-      | undefined;
-    usage_metadata?:
-      | {
-          input_tokens: number;
-          output_tokens: number;
-          total_tokens: number;
-          input_token_details?:
-            | {
-                audio?: number | undefined;
-                cache_read?: number | undefined;
-                cache_creation?: number | undefined;
-              }
-            | undefined;
-          output_token_details?:
-            | { audio?: number | undefined; reasoning?: number | undefined }
-            | undefined;
-        }
-      | undefined;
-  };
-
-  export type ToolMessage = BaseMessage & {
-    type: "tool";
-    status?: "error" | "success" | undefined;
-    tool_call_id: string;
-    /**
-     * Artifact of the Tool execution which is not meant to be sent to the model.
-     *
-     * Should only be specified if it is different from the message content, e.g. if only
-     * a subset of the full tool output is being passed as message content but the full
-     * output is needed in other parts of the code.
-     */
-    // eslint-disable-next-line @typescript-eslint/no-explicit-any
-    artifact?: any;
-  };
-
-  export type SystemMessage = BaseMessage & { type: "system" };
-  export type FunctionMessage = BaseMessage & { type: "function" };
-  export type RemoveMessage = BaseMessage & { type: "remove" };
-
-  export type AnyMessage =
-    | HumanMessage
-    | AIMessage
-    | ToolMessage
-    | SystemMessage
-    | FunctionMessage
-    | RemoveMessage;
-}

libs/langgraph/src/ui/types.schema.ts

@@ -1,278 +0,0 @@
-import type { SerializedMessage } from "./types.message.js";
-
-type Optional<T> = T | null | undefined;
-
-type MessageTupleMetadata = {
-  tags: string[];
-  [key: string]: unknown;
-};
-
-type DefaultValues = Record<string, unknown>[] | Record<string, unknown>;
-
-interface ThreadState<ValuesType = DefaultValues> {
-  /** The state values */
-  values: ValuesType;
-
-  /** The next nodes to execute. If empty, the thread is done until new input is received */
-  next: string[];
-
-  /** Checkpoint of the thread state */
-  checkpoint: Checkpoint;
-
-  /** Metadata for this state */
-  metadata: CheckpointMetadata;
-
-  /** Time of state creation  */
-  created_at: Optional<string>;
-
-  /** The parent checkpoint. If missing, this is the root checkpoint */
-  parent_checkpoint: Optional<Checkpoint>;
-
-  /** Tasks to execute in this step. If already attempted, may contain an error */
-  tasks: Array<ThreadTask>;
-}
-
-interface ThreadTask {
-  id: string;
-  name: string;
-  result?: unknown;
-  error: Optional<string>;
-  interrupts: Array<Interrupt>;
-  checkpoint: Optional<Checkpoint>;
-  state: Optional<ThreadState>;
-}
-
-type Config = {
-  /**
-   * Tags for this call and any sub-calls (eg. a Chain calling an LLM).
-   * You can use these to filter calls.
-   */
-  tags?: string[];
-
-  /**
-   * Maximum number of times a call can recurse.
-   * If not provided, defaults to 25.
-   */
-  recursion_limit?: number;
-
-  /**
-   * Runtime values for attributes previously made configurable on this Runnable.
-   */
-  configurable?: {
-    /**
-     * ID of the thread
-     */
-    thread_id?: Optional<string>;
-
-    /**
-     * Timestamp of the state checkpoint
-     */
-    checkpoint_id?: Optional<string>;
-
-    [key: string]: unknown;
-  };
-};
-
-type CheckpointMetadata = Optional<{
-  source?: "input" | "loop" | "update" | (string & {}); // eslint-disable-line @typescript-eslint/ban-types
-  step?: number;
-  writes?: Record<string, unknown> | null;
-  parents?: Record<string, string>;
-  [key: string]: unknown;
-}>;
-
-interface Checkpoint {
-  thread_id: string;
-  checkpoint_ns: string;
-  checkpoint_id: Optional<string>;
-  checkpoint_map: Optional<Record<string, unknown>>;
-}
-
-/**
- * An interrupt thrown inside a thread.
- */
-interface Interrupt<TValue = unknown> {
-  /**
-   * The ID of the interrupt.
-   */
-  id?: string;
-
-  /**
-   * The value of the interrupt.
-   */
-  value?: TValue;
-}
-
-type AsSubgraph<TEvent extends { id?: string; event: string; data: unknown }> =
-  {
-    id?: TEvent["id"];
-    event: TEvent["event"] | `${TEvent["event"]}|${string}`;
-    data: TEvent["data"];
-  };
-
-/**
- * Stream event with values after completion of each step.
- */
-type ValuesStreamEvent<StateType, InterruptType> = AsSubgraph<{
-  id?: string;
-  event: "values";
-  data: unknown extends InterruptType
-    ? StateType
-    : StateType & { __interrupt__?: InterruptType };
-}>;
-
-/**
- * Stream event with message chunks coming from LLM invocations inside nodes.
- */
-type MessagesTupleStreamEvent = AsSubgraph<{
-  id?: string;
-  event: "messages";
-  // TODO: add types for message and config, which do not depend on LangChain
-  // while making sure it's easy to keep them in sync.
-  data: [message: SerializedMessage.AnyMessage, config: MessageTupleMetadata];
-}>;
-
-/**
- * Metadata stream event with information about the run and thread
- */
-type MetadataStreamEvent = {
-  id?: string;
-  event: "metadata";
-  data: { run_id: string; thread_id: string };
-};
-
-/**
- * Stream event with error information.
- */
-type SubgraphErrorStreamEvent = AsSubgraph<{
-  id?: string;
-  event: "error";
-  data: { error: string; message: string };
-}>;
-
-/**
- * Stream event with updates to the state after each step.
- * The streamed outputs include the name of the node that
- * produced the update as well as the update.
- */
-type UpdatesStreamEvent<UpdateType, NodeReturnType> = AsSubgraph<{
-  id?: string;
-  event: "updates";
-  data: NodeReturnType extends Record<string, unknown>
-    ? { [K in keyof NodeReturnType]?: NodeReturnType[K] } & {
-        [node: string]: UpdateType;
-      }
-    : Record<string, UpdateType>;
-}>;
-
-/**
- * Streaming custom data from inside the nodes.
- */
-type CustomStreamEvent<T> = AsSubgraph<{
-  id?: string;
-  event: "custom";
-  data: T;
-}>;
-
-type TasksStreamCreateEvent<StateType> = {
-  id?: string;
-  event: "tasks";
-  data: {
-    id: string;
-    name: string;
-    interrupts: Interrupt[];
-    input: StateType;
-    triggers: string[];
-  };
-};
-
-type TasksStreamResultEvent<UpdateType> = {
-  id?: string;
-  event: "tasks";
-  data: {
-    id: string;
-    name: string;
-    interrupts: Interrupt[];
-    result: [string, UpdateType][];
-  };
-};
-
-type TasksStreamErrorEvent = {
-  id?: string;
-  event: "tasks";
-  data: {
-    id: string;
-    name: string;
-    interrupts: Interrupt[];
-    error: string;
-  };
-};
-
-type TasksStreamEvent<StateType, UpdateType> =
-  | AsSubgraph<TasksStreamCreateEvent<StateType>>
-  | AsSubgraph<TasksStreamResultEvent<UpdateType>>
-  | AsSubgraph<TasksStreamErrorEvent>;
-
-type CheckpointsStreamEvent<StateType> = AsSubgraph<{
-  id?: string;
-  event: "checkpoints";
-  data: {
-    values: StateType;
-    next: string[];
-    config: Config;
-    metadata: CheckpointMetadata;
-    tasks: ThreadTask[];
-  };
-}>;
-
-/**
- * Stream event with detailed debug information.
- */
-type DebugStreamEvent = AsSubgraph<{
-  id?: string;
-  event: "debug";
-  data: unknown;
-}>;
-
-/**
- * Stream event with events occurring during execution.
- */
-type EventsStreamEvent = {
-  id?: string;
-  event: "events";
-  data: {
-    event:
-      | `on_${
-          | "chat_model"
-          | "llm"
-          | "chain"
-          | "tool"
-          | "retriever"
-          | "prompt"}_${"start" | "stream" | "end"}`
-      | (string & {}); // eslint-disable-line @typescript-eslint/ban-types
-    name: string;
-    tags: string[];
-    run_id: string;
-    metadata: Record<string, unknown>;
-    parent_ids: string[];
-    data: unknown;
-  };
-};
-
-export type LangGraphEventStream<
-  TStateType = unknown,
-  TUpdateType = TStateType,
-  TCustomType = unknown,
-  TInterruptType = unknown,
-  TNodeReturnType = unknown
-> =
-  | ValuesStreamEvent<TStateType, TInterruptType>
-  | UpdatesStreamEvent<TUpdateType, TNodeReturnType>
-  | CustomStreamEvent<TCustomType>
-  | DebugStreamEvent
-  | MessagesTupleStreamEvent
-  | EventsStreamEvent
-  | TasksStreamEvent<TStateType, TUpdateType>
-  | CheckpointsStreamEvent<TStateType>
-  | SubgraphErrorStreamEvent
-  | MetadataStreamEvent;