From a1dad4360298f3a54c3159c47e86717aed26d7a9 Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Sat, 8 Feb 2025 20:05:31 -0800 Subject: [PATCH 1/7] feat(sdk-js): strongly-typed messages, state/update-type, stream mode --- libs/sdk-js/src/client.ts | 106 +++++++++++++------- libs/sdk-js/src/index.ts | 21 ++++ libs/sdk-js/src/types.messages.ts | 108 +++++++++++++++++++++ libs/sdk-js/src/types.stream.ts | 155 ++++++++++++++++++++++++++++++ libs/sdk-js/src/types.ts | 29 ++---- 5 files changed, 361 insertions(+), 58 deletions(-) create mode 100644 libs/sdk-js/src/types.messages.ts create mode 100644 libs/sdk-js/src/types.stream.ts diff --git a/libs/sdk-js/src/client.ts b/libs/sdk-js/src/client.ts index da88c84df..4c7562e28 100644 --- a/libs/sdk-js/src/client.ts +++ b/libs/sdk-js/src/client.ts @@ -27,7 +27,7 @@ import { createParser, } from "./utils/eventsource-parser/index.js"; import { IterableReadableStream } from "./utils/stream.js"; -import { +import type { RunsCreatePayload, RunsStreamPayload, RunsWaitPayload, @@ -38,6 +38,7 @@ import { import { mergeSignals } from "./utils/signals.js"; import { getEnvironmentVariable } from "./utils/env.js"; import { _getFetchImplementation } from "./singletons/fetch.js"; +import type { TypedAsyncGenerator, StreamMode } from "./types.stream.js"; /** * Get the API key from the environment. * Precedence: @@ -457,15 +458,17 @@ export class AssistantsClient extends BaseClient { } } -export class ThreadsClient extends BaseClient { +export class ThreadsClient extends BaseClient { /** * Get a thread by ID. * * @param threadId ID of the thread. * @returns The thread. */ - async get(threadId: string): Promise { - return this.fetch(`/threads/${threadId}`); + async get( + threadId: string, + ): Promise> { + return this.fetch>(`/threads/${threadId}`); } /** @@ -481,8 +484,8 @@ export class ThreadsClient extends BaseClient { metadata?: Metadata; threadId?: string; ifExists?: OnConflictBehavior; - }): Promise { - return this.fetch(`/threads`, { + }): Promise> { + return this.fetch>(`/threads`, { method: "POST", json: { metadata: payload?.metadata, @@ -497,8 +500,8 @@ export class ThreadsClient extends BaseClient { * @param threadId ID of the thread to be copied * @returns Newly copied thread */ - async copy(threadId: string): Promise { - return this.fetch(`/threads/${threadId}/copy`, { + async copy(threadId: string): Promise> { + return this.fetch>(`/threads/${threadId}/copy`, { method: "POST", }); } @@ -542,7 +545,7 @@ export class ThreadsClient extends BaseClient { * @param query Query options * @returns List of threads */ - async search(query?: { + async search(query?: { /** * Metadata to filter threads by. */ @@ -561,8 +564,8 @@ export class ThreadsClient extends BaseClient { * Must be one of 'idle', 'busy', 'interrupted' or 'error'. */ status?: ThreadStatus; - }): Promise { - return this.fetch("/threads/search", { + }): Promise[]> { + return this.fetch[]>("/threads/search", { method: "POST", json: { metadata: query?.metadata ?? undefined, @@ -579,7 +582,7 @@ export class ThreadsClient extends BaseClient { * @param threadId ID of the thread. * @returns Thread state. */ - async getState( + async getState( threadId: string, checkpoint?: Checkpoint | string, options?: { subgraphs?: boolean }, @@ -613,7 +616,7 @@ export class ThreadsClient extends BaseClient { * @param threadId The ID of the thread. * @returns */ - async updateState( + async updateState( threadId: string, options: { values: ValuesType; @@ -672,7 +675,7 @@ export class ThreadsClient extends BaseClient { * @param options Additional options. * @returns List of thread states. */ - async getHistory( + async getHistory( threadId: string, options?: { limit?: number; @@ -696,24 +699,43 @@ export class ThreadsClient extends BaseClient { } } -export class RunsClient extends BaseClient { - stream( +export class RunsClient< + TStateType = DefaultValues, + TUpdateType = TStateType, + TCustomEventType = unknown, +> extends BaseClient { + stream< + TStreamMode extends StreamMode | StreamMode[] = [], + TSubgraphs extends boolean = false, + >( threadId: null, assistantId: string, - payload?: Omit, - ): AsyncGenerator<{ - event: StreamEvent; - data: any; - }>; + payload?: Omit< + RunsStreamPayload, + "multitaskStrategy" | "onCompletion" + >, + ): TypedAsyncGenerator< + TStreamMode, + TSubgraphs, + TStateType, + TUpdateType, + TCustomEventType + >; - stream( + stream< + TStreamMode extends StreamMode | StreamMode[] = [], + TSubgraphs extends boolean = false, + >( threadId: string, assistantId: string, - payload?: RunsStreamPayload, - ): AsyncGenerator<{ - event: StreamEvent; - data: any; - }>; + payload?: RunsStreamPayload, + ): TypedAsyncGenerator< + TStreamMode, + TSubgraphs, + TStateType, + TUpdateType, + TCustomEventType + >; /** * Create a run and stream the results. @@ -722,14 +744,20 @@ export class RunsClient extends BaseClient { * @param assistantId Assistant ID to use for this run. * @param payload Payload for creating a run. */ - async *stream( + async *stream< + TStreamMode extends StreamMode | StreamMode[] = [], + TSubgraphs extends boolean = false, + >( threadId: string | null, assistantId: string, - payload?: RunsStreamPayload, - ): AsyncGenerator<{ - event: StreamEvent; - data: any; - }> { + payload?: RunsStreamPayload, + ): TypedAsyncGenerator< + TStreamMode, + TSubgraphs, + TStateType, + TUpdateType, + TCustomEventType + > { const json: Record = { input: payload?.input, command: payload?.command, @@ -767,7 +795,7 @@ export class RunsClient extends BaseClient { let onEndEvent: () => void; const textDecoder = new TextDecoder(); - const stream: ReadableStream<{ event: string; data: any }> = ( + const stream: ReadableStream<{ event: any; data: any }> = ( response.body || new ReadableStream({ start: (ctrl) => ctrl.close() }) ).pipeThrough( new TransformStream({ @@ -1284,7 +1312,11 @@ export class StoreClient extends BaseClient { } } -export class Client { +export class Client< + TStateType = DefaultValues, + TUpdateType = TStateType, + TCustomEventType = unknown, +> { /** * The client for interacting with assistants. */ @@ -1293,12 +1325,12 @@ export class Client { /** * The client for interacting with threads. */ - public threads: ThreadsClient; + public threads: ThreadsClient; /** * The client for interacting with runs. */ - public runs: RunsClient; + public runs: RunsClient; /** * The client for interacting with cron runs. diff --git a/libs/sdk-js/src/index.ts b/libs/sdk-js/src/index.ts index 6474f8f69..94e27539a 100644 --- a/libs/sdk-js/src/index.ts +++ b/libs/sdk-js/src/index.ts @@ -26,3 +26,24 @@ export type { export { overrideFetchImplementation } from "./singletons/fetch.js"; export type { OnConflictBehavior, Command } from "./types.js"; + +export type { StreamMode } from "./types.stream.js"; +export type { + ValuesPayload, + MessagesTuplePayload, + MetadataPayload, + UpdatesPayload, + CustomPayload, + MessagesPayload, + DebugPayload, + EventsPayload, +} from "./types.stream.js"; +export type { + Message, + HumanMessage, + AIMessage, + ToolMessage, + SystemMessage, + FunctionMessage, + RemoveMessage, +} from "./types.messages.js"; diff --git a/libs/sdk-js/src/types.messages.ts b/libs/sdk-js/src/types.messages.ts new file mode 100644 index 000000000..074ed7151 --- /dev/null +++ b/libs/sdk-js/src/types.messages.ts @@ -0,0 +1,108 @@ +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 MessageAdditionalKwargs = { + [x: string]: unknown; + + function_call?: { arguments: string; name: string } | undefined; + tool_calls?: + | { + id: string; + function: { arguments: string; name: string }; + type: "function"; + index?: number | undefined; + }[] + | undefined; +}; + +export type HumanMessage = { + type: "human"; + id?: string | undefined; + content: string | MessageContentComplex[]; +}; + +export type AIMessage = { + type: "ai"; + id?: string | undefined; + content: string | MessageContentComplex[]; + tool_calls?: + | { + name: string; + args: { [x: string]: { [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; + additional_kwargs?: MessageAdditionalKwargs | undefined; + response_metadata?: Record | undefined; +}; + +export type ToolMessage = { + type: "tool"; + name?: string | undefined; + id?: string | undefined; + content: string | MessageContentComplex[]; + status?: "error" | "success" | undefined; + lc_direct_tool_output: boolean; + tool_call_id: string; + additional_kwargs?: MessageAdditionalKwargs | undefined; + response_metadata?: Record | undefined; +}; + +export type SystemMessage = { + type: "system"; + id?: string | undefined; + content: string | MessageContentComplex[]; +}; + +export type FunctionMessage = { + type: "function"; + id?: string | undefined; + content: string | MessageContentComplex[]; +}; + +export type RemoveMessage = { + type: "remove"; + id: string; + content: string | MessageContentComplex[]; +}; + +export type Message = + | HumanMessage + | AIMessage + | ToolMessage + | SystemMessage + | FunctionMessage + | RemoveMessage; diff --git a/libs/sdk-js/src/types.stream.ts b/libs/sdk-js/src/types.stream.ts new file mode 100644 index 000000000..9e1ff75ba --- /dev/null +++ b/libs/sdk-js/src/types.stream.ts @@ -0,0 +1,155 @@ +import type { Message } from "./types.messages.js"; + +/** + * Stream modes + * - "values": Stream only the state values. + * - "messages": Stream complete messages. + * - "messages-tuple": Stream (message chunk, metadata) tuples. + * - "updates": Stream updates to the state. + * - "events": Stream events occurring during execution. + * - "debug": Stream detailed debug information. + * - "custom": Stream custom events. + */ +export type StreamMode = + | "values" + | "messages" + | "updates" + | "events" + | "debug" + | "custom" + | "messages-tuple"; + +type MessageTupleMetadata = { + tags: string[]; + [key: string]: unknown; +}; + +type AsSubgraph = { + event: TEvent["event"] | `${TEvent["event"]}|${string}`; + data: TEvent["data"]; +}; + +export type ValuesPayload = { event: "values"; data: StateType }; + +/** @internal */ +export type ValuesPayloadSubgraphs = AsSubgraph< + ValuesPayload +>; + +export type MessagesTuplePayload = { + 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: Message, config: MessageTupleMetadata]; +}; + +/** @internal */ +export type MessagesTuplePayloadSubgraphs = AsSubgraph; + +export type MetadataPayload = { + event: "metadata"; + data: { run_id: string; thread_id: string }; +}; + +/** @internal */ +export type MetadataPayloadSubgraphs = AsSubgraph; + +export type UpdatesPayload = { + event: "updates"; + data: { [node: string]: UpdateType }; +}; + +/** @internal */ +export type UpdatesPayloadSubgraphs = AsSubgraph< + UpdatesPayload +>; + +export type CustomPayload = { event: "custom"; data: T }; + +/** @internal */ +export type CustomPayloadSubgraphs = AsSubgraph>; + +type MessagesMetadataPayload = { + event: "messages/metadata"; + data: { [messageId: string]: { metadata: unknown } }; +}; +type MessagesCompletePayload = { + event: "messages/complete"; + data: Message[]; +}; +type MessagesPartialPayload = { + event: "messages/partial"; + data: Message[]; +}; + +export type MessagesPayload = + | MessagesMetadataPayload + | MessagesCompletePayload + | MessagesPartialPayload; + +/** @internal */ +export type MessagesPayloadSubgraphs = + | AsSubgraph + | AsSubgraph + | AsSubgraph; + +export type DebugPayload = { event: "debug"; data: unknown }; + +/** @internal */ +export type DebugPayloadSubgraphs = AsSubgraph; + +export type EventsPayload = { event: "events"; data: unknown }; + +/** @internal */ +export type EventsPayloadSubgraphs = AsSubgraph; + +type GetStreamModeMap< + TStreamMode extends StreamMode | StreamMode[], + TStateType = unknown, + TUpdateType = TStateType, + TCustomType = unknown, +> = + | { + values: ValuesPayload; + updates: UpdatesPayload; + custom: CustomPayload; + debug: DebugPayload; + messages: MessagesPayload; + "messages-tuple": MessagesTuplePayload; + events: EventsPayload; + }[TStreamMode extends StreamMode[] ? TStreamMode[number] : TStreamMode] + | MetadataPayload; + +type GetSubgraphsStreamModeMap< + TStreamMode extends StreamMode | StreamMode[], + TStateType = unknown, + TUpdateType = TStateType, + TCustomType = unknown, +> = + | { + values: ValuesPayloadSubgraphs; + updates: UpdatesPayloadSubgraphs; + custom: CustomPayloadSubgraphs; + debug: DebugPayloadSubgraphs; + messages: MessagesPayloadSubgraphs; + "messages-tuple": MessagesTuplePayloadSubgraphs; + events: EventsPayloadSubgraphs; + }[TStreamMode extends StreamMode[] ? TStreamMode[number] : TStreamMode] + | MetadataPayloadSubgraphs; + +export type TypedAsyncGenerator< + TStreamMode extends StreamMode | StreamMode[] = [], + TSubgraphs extends boolean = false, + TStateType = unknown, + TUpdateType = TStateType, + TCustomType = unknown, +> = AsyncGenerator< + TSubgraphs extends true + ? GetSubgraphsStreamModeMap< + TStreamMode, + TStateType, + TUpdateType, + TCustomType + > + : GetStreamModeMap +>; diff --git a/libs/sdk-js/src/types.ts b/libs/sdk-js/src/types.ts index 27e5c1857..643128e83 100644 --- a/libs/sdk-js/src/types.ts +++ b/libs/sdk-js/src/types.ts @@ -1,23 +1,6 @@ import { Checkpoint, Config, Metadata } from "./schema.js"; +import { StreamMode } from "./types.stream.js"; -/** - * Stream modes - * - "values": Stream only the state values. - * - "messages": Stream complete messages. - * - "messages-tuple": Stream (message chunk, metadata) tuples. - * - "updates": Stream updates to the state. - * - "events": Stream events occurring during execution. - * - "debug": Stream detailed debug information. - * - "custom": Stream custom events. - */ -export type StreamMode = - | "values" - | "messages" - | "updates" - | "events" - | "debug" - | "custom" - | "messages-tuple"; export type MultitaskStrategy = "reject" | "interrupt" | "rollback" | "enqueue"; export type OnConflictBehavior = "raise" | "do_nothing"; export type OnCompletionBehavior = "complete" | "continue"; @@ -31,6 +14,7 @@ export type StreamEvent = | "messages/partial" | "messages/metadata" | "messages/complete" + | "messages" | (string & {}); export interface Send { @@ -148,16 +132,19 @@ interface RunsInvokePayload { command?: Command; } -export interface RunsStreamPayload extends RunsInvokePayload { +export interface RunsStreamPayload< + TStreamMode extends StreamMode | StreamMode[] = [], + TSubgraphs extends boolean = false, +> extends RunsInvokePayload { /** * One of `"values"`, `"messages"`, `"messages-tuple"`, `"updates"`, `"events"`, `"debug"`, `"custom"`. */ - streamMode?: StreamMode | Array; + streamMode?: TStreamMode; /** * Stream output from subgraphs. By default, streams only the top graph. */ - streamSubgraphs?: boolean; + streamSubgraphs?: TSubgraphs; /** * Pass one or more feedbackKeys if you want to request short-lived signed URLs From 066525b3358b1fa9cecea0a409d29b3a8e344666 Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Mon, 10 Feb 2025 12:21:34 -0800 Subject: [PATCH 2/7] Update content --- libs/sdk-js/src/types.messages.ts | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/libs/sdk-js/src/types.messages.ts b/libs/sdk-js/src/types.messages.ts index 074ed7151..79ed78473 100644 --- a/libs/sdk-js/src/types.messages.ts +++ b/libs/sdk-js/src/types.messages.ts @@ -6,6 +6,7 @@ type MessageContentImageUrl = { type MessageContentText = { type: "text"; text: string }; type MessageContentComplex = MessageContentText | MessageContentImageUrl; +type MessageContent = string | MessageContentComplex[]; type MessageAdditionalKwargs = { [x: string]: unknown; @@ -24,13 +25,13 @@ type MessageAdditionalKwargs = { export type HumanMessage = { type: "human"; id?: string | undefined; - content: string | MessageContentComplex[]; + content: MessageContent; }; export type AIMessage = { type: "ai"; id?: string | undefined; - content: string | MessageContentComplex[]; + content: MessageContent; tool_calls?: | { name: string; @@ -73,9 +74,8 @@ export type ToolMessage = { type: "tool"; name?: string | undefined; id?: string | undefined; - content: string | MessageContentComplex[]; + content: MessageContent; status?: "error" | "success" | undefined; - lc_direct_tool_output: boolean; tool_call_id: string; additional_kwargs?: MessageAdditionalKwargs | undefined; response_metadata?: Record | undefined; @@ -84,19 +84,19 @@ export type ToolMessage = { export type SystemMessage = { type: "system"; id?: string | undefined; - content: string | MessageContentComplex[]; + content: MessageContent; }; export type FunctionMessage = { type: "function"; id?: string | undefined; - content: string | MessageContentComplex[]; + content: MessageContent; }; export type RemoveMessage = { type: "remove"; id: string; - content: string | MessageContentComplex[]; + content: MessageContent; }; export type Message = From cbca07e3db339c2ae62955704d8d1aaf42e2ca37 Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Mon, 10 Feb 2025 12:58:11 -0800 Subject: [PATCH 3/7] Rename to more sane event type --- libs/sdk-js/src/index.ts | 16 ++--- libs/sdk-js/src/types.stream.ts | 114 ++++++++++++++++++++------------ 2 files changed, 79 insertions(+), 51 deletions(-) diff --git a/libs/sdk-js/src/index.ts b/libs/sdk-js/src/index.ts index 94e27539a..ca30e5d54 100644 --- a/libs/sdk-js/src/index.ts +++ b/libs/sdk-js/src/index.ts @@ -29,14 +29,14 @@ export type { OnConflictBehavior, Command } from "./types.js"; export type { StreamMode } from "./types.stream.js"; export type { - ValuesPayload, - MessagesTuplePayload, - MetadataPayload, - UpdatesPayload, - CustomPayload, - MessagesPayload, - DebugPayload, - EventsPayload, + ValuesStreamEvent, + MessagesTupleStreamEvent, + MetadataStreamEvent, + UpdatesStreamEvent, + CustomStreamEvent, + MessagesStreamEvent, + DebugStreamEvent, + EventsStreamEvent, } from "./types.stream.js"; export type { Message, diff --git a/libs/sdk-js/src/types.stream.ts b/libs/sdk-js/src/types.stream.ts index 9e1ff75ba..4a36652d7 100644 --- a/libs/sdk-js/src/types.stream.ts +++ b/libs/sdk-js/src/types.stream.ts @@ -29,14 +29,20 @@ type AsSubgraph = { data: TEvent["data"]; }; -export type ValuesPayload = { event: "values"; data: StateType }; +/** + * Stream event with values after completion of each step. + */ +export type ValuesStreamEvent = { event: "values"; data: StateType }; /** @internal */ -export type ValuesPayloadSubgraphs = AsSubgraph< - ValuesPayload +export type SubgraphValuesStreamEvent = AsSubgraph< + ValuesStreamEvent >; -export type MessagesTuplePayload = { +/** + * Stream event with message chunks coming from LLM invocations inside nodes. + */ +export type MessagesTupleStreamEvent = { 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. @@ -44,64 +50,86 @@ export type MessagesTuplePayload = { }; /** @internal */ -export type MessagesTuplePayloadSubgraphs = AsSubgraph; +export type SubgraphMessagesTupleStreamEvent = + AsSubgraph; -export type MetadataPayload = { +/** + * Metadata stream event with information about the run and thread + */ +export type MetadataStreamEvent = { event: "metadata"; data: { run_id: string; thread_id: string }; }; /** @internal */ -export type MetadataPayloadSubgraphs = AsSubgraph; +export type SubgraphMetadataStreamEvent = AsSubgraph; -export type UpdatesPayload = { +/** + * 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. + */ +export type UpdatesStreamEvent = { event: "updates"; data: { [node: string]: UpdateType }; }; /** @internal */ -export type UpdatesPayloadSubgraphs = AsSubgraph< - UpdatesPayload +export type SubgraphUpdatesStreamEvent = AsSubgraph< + UpdatesStreamEvent >; -export type CustomPayload = { event: "custom"; data: T }; +/** + * Streaming custom data from inside the nodes. + */ +export type CustomStreamEvent = { event: "custom"; data: T }; /** @internal */ -export type CustomPayloadSubgraphs = AsSubgraph>; +export type SubgraphCustomStreamEvent = AsSubgraph>; -type MessagesMetadataPayload = { +type MessagesMetadataStreamEvent = { event: "messages/metadata"; data: { [messageId: string]: { metadata: unknown } }; }; -type MessagesCompletePayload = { +type MessagesCompleteStreamEvent = { event: "messages/complete"; data: Message[]; }; -type MessagesPartialPayload = { +type MessagesPartialStreamEvent = { event: "messages/partial"; data: Message[]; }; -export type MessagesPayload = - | MessagesMetadataPayload - | MessagesCompletePayload - | MessagesPartialPayload; +/** + * Message stream event specific to LangGraph Server. + * @deprecated Use `streamMode: "messages-tuple"` instead. + */ +export type MessagesStreamEvent = + | MessagesMetadataStreamEvent + | MessagesCompleteStreamEvent + | MessagesPartialStreamEvent; /** @internal */ -export type MessagesPayloadSubgraphs = - | AsSubgraph - | AsSubgraph - | AsSubgraph; +export type SubgraphMessagesStreamEvent = + | AsSubgraph + | AsSubgraph + | AsSubgraph; -export type DebugPayload = { event: "debug"; data: unknown }; +/** + * Stream event with detailed debug information. + */ +export type DebugStreamEvent = { event: "debug"; data: unknown }; /** @internal */ -export type DebugPayloadSubgraphs = AsSubgraph; +export type SubgraphDebugStreamEvent = AsSubgraph; -export type EventsPayload = { event: "events"; data: unknown }; +/** + * Stream event with events occurring during execution. + */ +export type EventsStreamEvent = { event: "events"; data: unknown }; /** @internal */ -export type EventsPayloadSubgraphs = AsSubgraph; +export type SubgraphEventsStreamEvent = AsSubgraph; type GetStreamModeMap< TStreamMode extends StreamMode | StreamMode[], @@ -110,15 +138,15 @@ type GetStreamModeMap< TCustomType = unknown, > = | { - values: ValuesPayload; - updates: UpdatesPayload; - custom: CustomPayload; - debug: DebugPayload; - messages: MessagesPayload; - "messages-tuple": MessagesTuplePayload; - events: EventsPayload; + values: ValuesStreamEvent; + updates: UpdatesStreamEvent; + custom: CustomStreamEvent; + debug: DebugStreamEvent; + messages: MessagesStreamEvent; + "messages-tuple": MessagesTupleStreamEvent; + events: EventsStreamEvent; }[TStreamMode extends StreamMode[] ? TStreamMode[number] : TStreamMode] - | MetadataPayload; + | MetadataStreamEvent; type GetSubgraphsStreamModeMap< TStreamMode extends StreamMode | StreamMode[], @@ -127,15 +155,15 @@ type GetSubgraphsStreamModeMap< TCustomType = unknown, > = | { - values: ValuesPayloadSubgraphs; - updates: UpdatesPayloadSubgraphs; - custom: CustomPayloadSubgraphs; - debug: DebugPayloadSubgraphs; - messages: MessagesPayloadSubgraphs; - "messages-tuple": MessagesTuplePayloadSubgraphs; - events: EventsPayloadSubgraphs; + values: SubgraphValuesStreamEvent; + updates: SubgraphUpdatesStreamEvent; + custom: SubgraphCustomStreamEvent; + debug: SubgraphDebugStreamEvent; + messages: SubgraphMessagesStreamEvent; + "messages-tuple": SubgraphMessagesTupleStreamEvent; + events: SubgraphEventsStreamEvent; }[TStreamMode extends StreamMode[] ? TStreamMode[number] : TStreamMode] - | MetadataPayloadSubgraphs; + | SubgraphMetadataStreamEvent; export type TypedAsyncGenerator< TStreamMode extends StreamMode | StreamMode[] = [], From 7fd693120041c582873e8ba87d4d377c6dc99ba0 Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Mon, 10 Feb 2025 13:23:23 -0800 Subject: [PATCH 4/7] Cleanup types --- libs/sdk-js/src/client.ts | 15 +++++++++------ libs/sdk-js/src/index.ts | 1 - 2 files changed, 9 insertions(+), 7 deletions(-) diff --git a/libs/sdk-js/src/client.ts b/libs/sdk-js/src/client.ts index 4c7562e28..54ce7733b 100644 --- a/libs/sdk-js/src/client.ts +++ b/libs/sdk-js/src/client.ts @@ -458,7 +458,10 @@ export class AssistantsClient extends BaseClient { } } -export class ThreadsClient extends BaseClient { +export class ThreadsClient< + TStateType = DefaultValues, + TUpdateType = TStateType, +> extends BaseClient { /** * Get a thread by ID. * @@ -616,7 +619,7 @@ export class ThreadsClient extends BaseClient { * @param threadId The ID of the thread. * @returns */ - async updateState( + async updateState( threadId: string, options: { values: ValuesType; @@ -705,7 +708,7 @@ export class RunsClient< TCustomEventType = unknown, > extends BaseClient { stream< - TStreamMode extends StreamMode | StreamMode[] = [], + TStreamMode extends StreamMode | StreamMode[] = StreamMode, TSubgraphs extends boolean = false, >( threadId: null, @@ -723,7 +726,7 @@ export class RunsClient< >; stream< - TStreamMode extends StreamMode | StreamMode[] = [], + TStreamMode extends StreamMode | StreamMode[] = StreamMode, TSubgraphs extends boolean = false, >( threadId: string, @@ -745,7 +748,7 @@ export class RunsClient< * @param payload Payload for creating a run. */ async *stream< - TStreamMode extends StreamMode | StreamMode[] = [], + TStreamMode extends StreamMode | StreamMode[] = StreamMode, TSubgraphs extends boolean = false, >( threadId: string | null, @@ -1325,7 +1328,7 @@ export class Client< /** * The client for interacting with threads. */ - public threads: ThreadsClient; + public threads: ThreadsClient; /** * The client for interacting with runs. diff --git a/libs/sdk-js/src/index.ts b/libs/sdk-js/src/index.ts index ca30e5d54..956f5f64b 100644 --- a/libs/sdk-js/src/index.ts +++ b/libs/sdk-js/src/index.ts @@ -26,7 +26,6 @@ export type { export { overrideFetchImplementation } from "./singletons/fetch.js"; export type { OnConflictBehavior, Command } from "./types.js"; - export type { StreamMode } from "./types.stream.js"; export type { ValuesStreamEvent, From 7a2eb614dcb52218e57b33a1e502e7c4f8cdbcf5 Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Mon, 10 Feb 2025 17:54:18 -0800 Subject: [PATCH 5/7] Add ErrorStreamEvent --- libs/sdk-js/src/index.ts | 1 + libs/sdk-js/src/types.stream.ts | 17 +++++++++++++++-- 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/libs/sdk-js/src/index.ts b/libs/sdk-js/src/index.ts index 956f5f64b..eb399a9b2 100644 --- a/libs/sdk-js/src/index.ts +++ b/libs/sdk-js/src/index.ts @@ -36,6 +36,7 @@ export type { MessagesStreamEvent, DebugStreamEvent, EventsStreamEvent, + ErrorStreamEvent, } from "./types.stream.js"; export type { Message, diff --git a/libs/sdk-js/src/types.stream.ts b/libs/sdk-js/src/types.stream.ts index 4a36652d7..7ca7742d1 100644 --- a/libs/sdk-js/src/types.stream.ts +++ b/libs/sdk-js/src/types.stream.ts @@ -64,6 +64,17 @@ export type MetadataStreamEvent = { /** @internal */ export type SubgraphMetadataStreamEvent = AsSubgraph; +/** + * Stream event with error information. + */ +export type ErrorStreamEvent = { + event: "error"; + data: { error: string; message: string }; +}; + +/** @internal */ +export type SubgraphErrorStreamEvent = AsSubgraph; + /** * Stream event with updates to the state after each step. * The streamed outputs include the name of the node that @@ -146,7 +157,8 @@ type GetStreamModeMap< "messages-tuple": MessagesTupleStreamEvent; events: EventsStreamEvent; }[TStreamMode extends StreamMode[] ? TStreamMode[number] : TStreamMode] - | MetadataStreamEvent; + | MetadataStreamEvent + | ErrorStreamEvent; type GetSubgraphsStreamModeMap< TStreamMode extends StreamMode | StreamMode[], @@ -163,7 +175,8 @@ type GetSubgraphsStreamModeMap< "messages-tuple": SubgraphMessagesTupleStreamEvent; events: SubgraphEventsStreamEvent; }[TStreamMode extends StreamMode[] ? TStreamMode[number] : TStreamMode] - | SubgraphMetadataStreamEvent; + | SubgraphMetadataStreamEvent + | SubgraphErrorStreamEvent; export type TypedAsyncGenerator< TStreamMode extends StreamMode | StreamMode[] = [], From bb3193c83e8e3b70e0975c51f5f3ad1f4ef303ac Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Mon, 10 Feb 2025 17:54:58 -0800 Subject: [PATCH 6/7] fix(sdk-js): non-ok response is not throwing anymore --- libs/sdk-js/src/client.ts | 9 +++------ libs/sdk-js/src/utils/async_caller.ts | 4 +++- 2 files changed, 6 insertions(+), 7 deletions(-) diff --git a/libs/sdk-js/src/client.ts b/libs/sdk-js/src/client.ts index 54ce7733b..c4ffcb7ed 100644 --- a/libs/sdk-js/src/client.ts +++ b/libs/sdk-js/src/client.ts @@ -165,8 +165,7 @@ class BaseClient { signal?: AbortSignal; }, ): Promise { - const response = await this.asyncCaller.call( - _getFetchImplementation(), + const response = await this.asyncCaller.fetch( ...this.prepareFetchOptions(path, options), ); if (response.status === 202 || response.status === 204) { @@ -784,8 +783,7 @@ export class RunsClient< const endpoint = threadId == null ? `/runs/stream` : `/threads/${threadId}/runs/stream`; - const response = await this.asyncCaller.call( - _getFetchImplementation(), + const response = await this.asyncCaller.fetch( ...this.prepareFetchOptions(endpoint, { method: "POST", json, @@ -1077,8 +1075,7 @@ export class RunsClient< ? { signal: options } : options; - const response = await this.asyncCaller.call( - _getFetchImplementation(), + const response = await this.asyncCaller.fetch( ...this.prepareFetchOptions(`/threads/${threadId}/runs/${runId}/stream`, { method: "GET", timeoutMs: null, diff --git a/libs/sdk-js/src/utils/async_caller.ts b/libs/sdk-js/src/utils/async_caller.ts index d5de9131a..72beefb54 100644 --- a/libs/sdk-js/src/utils/async_caller.ts +++ b/libs/sdk-js/src/utils/async_caller.ts @@ -1,5 +1,6 @@ import pRetry from "p-retry"; import PQueueMod from "p-queue"; +import { _getFetchImplementation } from "../singletons/fetch.js"; const STATUS_NO_RETRY = [ 400, // Bad Request @@ -210,7 +211,8 @@ export class AsyncCaller { } fetch(...args: Parameters): ReturnType { - const fetchFn = this.customFetch ?? fetch; + const fetchFn = + this.customFetch ?? (_getFetchImplementation() as typeof fetch); return this.call(() => fetchFn(...args).then((res) => (res.ok ? res : Promise.reject(res))), ); From 69311a41354a2ab64f2370b7744f4ca1f9eeaa01 Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Tue, 11 Feb 2025 08:41:26 -0800 Subject: [PATCH 7/7] Add feedback stream event --- libs/sdk-js/src/index.ts | 1 + libs/sdk-js/src/types.stream.ts | 20 ++++++++++++++------ 2 files changed, 15 insertions(+), 6 deletions(-) diff --git a/libs/sdk-js/src/index.ts b/libs/sdk-js/src/index.ts index eb399a9b2..0626144cd 100644 --- a/libs/sdk-js/src/index.ts +++ b/libs/sdk-js/src/index.ts @@ -37,6 +37,7 @@ export type { DebugStreamEvent, EventsStreamEvent, ErrorStreamEvent, + FeedbackStreamEvent, } from "./types.stream.js"; export type { Message, diff --git a/libs/sdk-js/src/types.stream.ts b/libs/sdk-js/src/types.stream.ts index 7ca7742d1..441745aec 100644 --- a/libs/sdk-js/src/types.stream.ts +++ b/libs/sdk-js/src/types.stream.ts @@ -61,9 +61,6 @@ export type MetadataStreamEvent = { data: { run_id: string; thread_id: string }; }; -/** @internal */ -export type SubgraphMetadataStreamEvent = AsSubgraph; - /** * Stream event with error information. */ @@ -142,6 +139,15 @@ export type EventsStreamEvent = { event: "events"; data: unknown }; /** @internal */ export type SubgraphEventsStreamEvent = AsSubgraph; +/** + * Stream event with a feedback key to signed URL map. Set `feedbackKeys` in + * the `RunsStreamPayload` to receive this event. + */ +export type FeedbackStreamEvent = { + event: "feedback"; + data: { [feedbackKey: string]: string }; +}; + type GetStreamModeMap< TStreamMode extends StreamMode | StreamMode[], TStateType = unknown, @@ -157,8 +163,9 @@ type GetStreamModeMap< "messages-tuple": MessagesTupleStreamEvent; events: EventsStreamEvent; }[TStreamMode extends StreamMode[] ? TStreamMode[number] : TStreamMode] + | ErrorStreamEvent | MetadataStreamEvent - | ErrorStreamEvent; + | FeedbackStreamEvent; type GetSubgraphsStreamModeMap< TStreamMode extends StreamMode | StreamMode[], @@ -175,8 +182,9 @@ type GetSubgraphsStreamModeMap< "messages-tuple": SubgraphMessagesTupleStreamEvent; events: SubgraphEventsStreamEvent; }[TStreamMode extends StreamMode[] ? TStreamMode[number] : TStreamMode] - | SubgraphMetadataStreamEvent - | SubgraphErrorStreamEvent; + | SubgraphErrorStreamEvent + | MetadataStreamEvent + | FeedbackStreamEvent; export type TypedAsyncGenerator< TStreamMode extends StreamMode | StreamMode[] = [],