mirror of
https://github.com/langchain-ai/langgraphjs.git
synced 2026-07-22 17:15:25 -04:00
feat(langgraph): split streaming protocol from LGP (#1554)
This commit is contained in:
David Duong
@@ -0,0 +1,5 @@
|
||||
---
|
||||
"@langchain/langgraph": patch
|
||||
---
|
||||
|
||||
feat(langgraph): add `toLangGraphEventStream` method to stream events in LGP compatible format
|
||||
@@ -19,6 +19,7 @@ export const config = {
|
||||
remote: "remote",
|
||||
zod: "graph/zod/index",
|
||||
"zod/schema": "graph/zod/schema",
|
||||
"ui": "ui/index"
|
||||
},
|
||||
tsConfigPath: resolve("./tsconfig.json"),
|
||||
cjsSource: "./dist-cjs",
|
||||
|
||||
@@ -595,6 +595,8 @@ export class CompiledGraph<
|
||||
> {
|
||||
declare NodeType: N;
|
||||
|
||||
declare "~NodeReturnType": NodeReturnType;
|
||||
|
||||
declare RunInput: State;
|
||||
|
||||
declare RunOutput: Update;
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
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>();
|
||||
}
|
||||
}
|
||||
});
|
||||
@@ -0,0 +1,206 @@
|
||||
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>>())
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
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";
|
||||
@@ -0,0 +1,231 @@
|
||||
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
|
||||
>;
|
||||
|
||||
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
|
||||
>;
|
||||
|
||||
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 }
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,102 @@
|
||||
/* 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"]
|
||||
>;
|
||||
@@ -0,0 +1,96 @@
|
||||
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;
|
||||
}
|
||||
@@ -0,0 +1,278 @@
|
||||
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;
|
||||
Reference in New Issue
Block a user