From 25f88740c18855b6fff7ba65f8daffe2e248f1ec Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Sat, 8 Feb 2025 20:36:42 -0800 Subject: [PATCH 01/16] feat(sdk-js): add experimental useStream hook --- libs/sdk-js/.gitignore | 4 + libs/sdk-js/langchain.config.js | 4 +- libs/sdk-js/package.json | 25 +- libs/sdk-js/src/react/index.ts | 1 + libs/sdk-js/src/react/stream.tsx | 559 +++++++++++++++++++++++++++++++ libs/sdk-js/tsconfig.json | 1 + libs/sdk-js/yarn.lock | 133 +++++++- 7 files changed, 715 insertions(+), 12 deletions(-) create mode 100644 libs/sdk-js/src/react/index.ts create mode 100644 libs/sdk-js/src/react/stream.tsx diff --git a/libs/sdk-js/.gitignore b/libs/sdk-js/.gitignore index fd0d01027..4d7124e5f 100644 --- a/libs/sdk-js/.gitignore +++ b/libs/sdk-js/.gitignore @@ -6,6 +6,10 @@ client.cjs client.js client.d.ts client.d.cts +react.cjs +react.js +react.d.ts +react.d.cts node_modules dist .yarn diff --git a/libs/sdk-js/langchain.config.js b/libs/sdk-js/langchain.config.js index eeb79ab47..b644f4b3c 100644 --- a/libs/sdk-js/langchain.config.js +++ b/libs/sdk-js/langchain.config.js @@ -10,8 +10,8 @@ function abs(relativePath) { } export const config = { - internals: [], - entrypoints: { index: "index", client: "client" }, + internals: [/react/], + entrypoints: { index: "index", client: "client", react: "react/index" }, tsConfigPath: resolve("./tsconfig.json"), cjsSource: "./dist-cjs", cjsDestination: "./dist", diff --git a/libs/sdk-js/package.json b/libs/sdk-js/package.json index aae3d0219..1181feaf8 100644 --- a/libs/sdk-js/package.json +++ b/libs/sdk-js/package.json @@ -33,7 +33,15 @@ "ts-jest": "^29.1.2", "typedoc": "^0.26.1", "typedoc-plugin-markdown": "^4.1.0", - "typescript": "^5.4.5" + "typescript": "^5.4.5", + "react": "^18.3.1", + "@langchain/core": "^0.3.31", + "@types/react": "18.3.2" + }, + "peerDependencies": { + "react": "*", + "@types/react": "*", + "@langchain/core": "*" }, "exports": { ".": { @@ -54,6 +62,15 @@ "import": "./client.js", "require": "./client.cjs" }, + "./react": { + "types": { + "import": "./react.d.ts", + "require": "./react.d.cts", + "default": "./react.d.ts" + }, + "import": "./react.js", + "require": "./react.cjs" + }, "./package.json": "./package.json" }, "files": [ @@ -65,6 +82,10 @@ "client.cjs", "client.js", "client.d.ts", - "client.d.cts" + "client.d.cts", + "react.cjs", + "react.js", + "react.d.ts", + "react.d.cts" ] } diff --git a/libs/sdk-js/src/react/index.ts b/libs/sdk-js/src/react/index.ts new file mode 100644 index 000000000..7fc550ada --- /dev/null +++ b/libs/sdk-js/src/react/index.ts @@ -0,0 +1 @@ +export { useStream, LangGraphConfig } from "./stream.js"; diff --git a/libs/sdk-js/src/react/stream.tsx b/libs/sdk-js/src/react/stream.tsx new file mode 100644 index 000000000..8b5b84562 --- /dev/null +++ b/libs/sdk-js/src/react/stream.tsx @@ -0,0 +1,559 @@ +/* __LC_ALLOW_ENTRYPOINT_SIDE_EFFECTS__ */ +"use client"; + +import { Client } from "../client.js"; +import type { Command } from "../types.js"; +import type { Message } from "../types.messages.js"; +import type { Config, ThreadState } from "../schema.js"; +import type { + CustomPayload, + DebugPayload, + EventsPayload, + MessagesPayload, + MessagesTuplePayload, + UpdatesPayload, + ValuesPayload, +} from "../types.stream.js"; + +import { + type MutableRefObject, + type ReactNode, + createContext, + useCallback, + useContext, + useEffect, + useMemo, + useRef, + useState, +} from "react"; +import { + type BaseMessageChunk, + type BaseMessage, + coerceMessageLikeToMessage, + convertToChunk, +} from "@langchain/core/messages"; + +class MessageTupleManager { + chunks: Record = {}; + + constructor() { + this.chunks = {}; + } + + add(serialized: Message): string | null { + const chunk = convertToChunk(coerceMessageLikeToMessage(serialized)); + + const id = chunk.id; + if (!id) return null; + + this.chunks[id] ??= {}; + this.chunks[id].chunk = this.chunks[id]?.chunk?.concat(chunk) ?? chunk; + + return id; + } + + clear() { + this.chunks = {}; + } + + get(id: string, defaultIndex: number) { + if (this.chunks[id] == null) return null; + this.chunks[id].index ??= defaultIndex; + + return this.chunks[id]; + } +} + +const toMessageDict = (chunk: BaseMessage): Message => { + const { type, data } = chunk.toDict(); + return { ...data, type } as Message; +}; + +function unique(array: T[]) { + return [...new Set(array)] as T[]; +} + +function findLastIndex(array: T[], predicate: (item: T) => boolean) { + for (let i = array.length - 1; i >= 0; i--) { + if (predicate(array[i])) return i; + } + return -1; +} + +interface Node { + type: "node"; + value: ThreadState; + path: string[]; +} + +interface Fork { + type: "fork"; + items: Array>; +} + +interface Sequence { + type: "sequence"; + items: Array | Fork>; +} + +interface ValidFork { + type: "fork"; + items: Array>; +} + +interface ValidSequence { + type: "sequence"; + items: [Node, ...(Node | ValidFork)[]]; +} + +// forks +export type CheckpointBranchPath = string[]; + +export type MessageBranch = { + current: CheckpointBranchPath; + options: CheckpointBranchPath[]; +}; + +const mergeConfig = (...configs: (Config | undefined)[]) => { + const result: Config = { configurable: {} }; + + for (const config of configs) { + if (config == null) continue; + // TODO: Just assigning the latest configurable is iffy. + result.configurable = config.configurable; + } + + return result; +}; + +function fetchHistory>( + client: Client, + threadId: string, +) { + return client.threads.getHistory(threadId, { limit: 1000 }); +} + +function useThreadHistory>( + threadId: string | undefined | null, + client: Client, + submittingRef: MutableRefObject, +) { + const [history, setHistory] = useState[]>([]); + + const fetcher = useCallback( + (threadId: string | undefined | null): Promise => { + if (threadId != null) { + return fetchHistory(client, threadId).then((history) => + setHistory(history), + ); + } + + setHistory([]); + return Promise.resolve(); + }, + [], + ); + + useEffect(() => { + if (submittingRef.current) return; + fetcher(threadId); + }, [fetcher, submittingRef, threadId]); + + return { + data: history, + mutate: (mutateId?: string) => fetcher(mutateId ?? threadId), + }; +} + +interface LangGraphConfig { + withMessages?: string; + onError?: (error: unknown) => void; + client?: Client; +} + +const ConfigProvider = createContext(null!); +export const LangGraphConfig = (props: { + config: LangGraphConfig; + children?: ReactNode; +}) => { + return ( + + {props.children} + + ); +}; + +export function useStream< + StateType extends Record = Record, + UpdateType extends Record = Partial, + CustomType = unknown, +>(options?: { + client?: Client; + + withMessages?: string; + onError?: (error: unknown) => void; + + // TODO: can we make threadId uncontrollable / controllable? + threadId?: string | null; + onThreadId?: (threadId: string) => void; +}) { + type EventPayload = + | ValuesPayload + | UpdatesPayload + | CustomPayload + | DebugPayload + | MessagesPayload + | MessagesTuplePayload + | EventsPayload; + + const contextConfig = useContext(ConfigProvider); + const { withMessages, onError, threadId, client } = Object.assign( + {}, + contextConfig, + options, + ); + + if (client == null) { + throw new Error( + "LangGraph SDK not provided. Either pass a client to `useStream` or wrap your app in a `LangGraphConfig` provider and pass the client there.", + ); + } + + const [branchPath, setBranchPath] = useState([]); + + const [error, setError] = useState(undefined); + const [events, setEvents] = useState([]); + + const [streamValues, setStreamValues] = useState(null); + + const [streamMode, setStreamMode] = useState< + Array<"values" | "updates" | "events" | "custom" | "messages-tuple"> + >(["values", "messages-tuple"]); + + const manager = useRef(new MessageTupleManager()); + const submittingRef = useRef(false); + + // TODO: is this a responsibility of SWR / React Query? + // Maybe we should use that instead to allow rehydration? + const history = useThreadHistory(threadId, client, submittingRef); + + const getMessages = useMemo(() => { + if (withMessages == null) return undefined; + return (value: StateType) => + Array.isArray(value[withMessages]) + ? (value[withMessages] as Message[]) + : []; + }, [withMessages]); + + const [sequence, pathMap] = (() => { + const childrenMap: Record[]> = {}; + + // First pass - collect nodes for each checkpoint + history.data.forEach((state) => { + const checkpointId = state.parent_checkpoint?.checkpoint_id ?? "$"; + childrenMap[checkpointId] ??= []; + childrenMap[checkpointId].push(state); + }); + + // Second pass - create a tree of sequences + type Task = { id: string; sequence: Sequence; path: string[] }; + const rootSequence: Sequence = { type: "sequence", items: [] }; + const queue: Task[] = [{ id: "$", sequence: rootSequence, path: [] }]; + + const paths: string[][] = []; + + const visited = new Set(); + while (queue.length > 0) { + const task = queue.shift()!; + if (visited.has(task.id)) continue; + visited.add(task.id); + + const children = childrenMap[task.id]; + if (children == null || children.length === 0) continue; + + // If we've encountered a fork (2+ children), push the fork + // to the sequence and add a new sequence for each child + let fork: Fork | undefined; + if (children.length > 1) { + fork = { type: "fork", items: [] }; + task.sequence.items.push(fork); + } + + for (const value of children) { + const id = value.checkpoint.checkpoint_id!; + + let sequence = task.sequence; + let path = task.path; + if (fork != null) { + sequence = { type: "sequence", items: [] }; + fork.items.unshift(sequence); + + path = path.slice(); + path.push(id); + paths.push(path); + } + + sequence.items.push({ type: "node", value, path }); + queue.push({ id, sequence, path }); + } + } + + // Third pass, create a map for available forks + const pathMap: Record = {}; + for (const path of paths) { + const parent = path.at(-2) ?? "$"; + pathMap[parent] ??= []; + pathMap[parent].unshift(path); + } + + return [rootSequence as ValidSequence, pathMap]; + })(); + + const [flatValues, checkpointPathMap] = (() => { + const result: ThreadState[] = []; + + // TODO: this is kinda ugly + const checkpointPathMap: Record = {}; + + const forkStack = branchPath.slice(); + const queue: (Node | Fork)[] = [...sequence.items]; + + while (queue.length > 0) { + const item = queue.shift()!; + + if (item.type === "node") { + result.push(item.value); + checkpointPathMap[item.value.checkpoint.checkpoint_id!] = { + current: item.path, + options: + item.path.length > 0 ? pathMap[item.path.at(-2) ?? "$"] ?? [] : [], + }; + } + if (item.type === "fork") { + const forkId = forkStack.shift(); + const index = + forkId != null + ? item.items.findIndex((value) => { + const firstItem = value.items.at(0); + if (!firstItem || firstItem.type !== "node") return false; + return firstItem.value.checkpoint.checkpoint_id === forkId; + }) + : -1; + + const nextItems = item.items.at(index)?.items ?? []; + queue.push(...nextItems); + } + } + + return [result, checkpointPathMap]; + })(); + + const lastSeenValue = flatValues.at(-1); + const historyValues = lastSeenValue?.values ?? ({} as StateType); + + const messageMeta = (() => { + if (getMessages == null) return undefined; + + const alreadyShown = new Set(); + return getMessages(historyValues).map((message, idx) => { + const messageId = message.id ?? idx; + const firstSeenIdx = findLastIndex(history.data, (state) => + getMessages(state.values) + .map((m, idx) => m.id ?? idx) + .includes(messageId), + ); + + const firstState = history.data[firstSeenIdx] as + | ThreadState + | undefined; + + let branch = firstState + ? checkpointPathMap[firstState.checkpoint.checkpoint_id!] + : undefined; + if (!branch?.current.length) branch = undefined; + + const optionsShown = branch?.options?.flat(2).join(","); + if (optionsShown) { + if (alreadyShown.has(optionsShown)) branch = undefined; + alreadyShown.add(optionsShown); + } + + return { messageId, firstState, branch }; + }); + })(); + + const handleSubmit = async ( + values: UpdateType | undefined, + submitOptions?: { + config?: Config; + command?: Command; + optimisticValues?: + | Partial + | ((prev: StateType) => Partial); + }, + ) => { + try { + // TODO: have loading state as well + submittingRef.current = true; + + // This is used to reset the path to make sure we always fetch the + // latest generation / edit. + // TODO: make sure it's actually aware of the config passed to handleSubmit + setBranchPath((path) => path.slice(0, -1)); + + let usableThreadId = threadId; + if (!usableThreadId) { + const thread = await client.threads.create(); + options?.onThreadId?.(thread.thread_id); + usableThreadId = thread.thread_id; + } + + // TODO: why non-existent assistant ID does not throw an error here? + const run = (await client.runs.stream(usableThreadId, "agent", { + input: values as Record, + config: mergeConfig( + { configurable: lastSeenValue?.checkpoint }, + submitOptions?.config, + ), + streamMode, + })) as AsyncGenerator; + + // Assumption: we're setting the initial value + // Used for instant feedback + if (submitOptions?.optimisticValues != null) { + setStreamValues((streamValues) => { + const values = { ...historyValues, ...streamValues }; + return { + ...values, + ...(typeof submitOptions.optimisticValues === "function" + ? submitOptions.optimisticValues(values) + : submitOptions.optimisticValues), + }; + }); + } + + for await (const { event, data } of run) { + setEvents((events) => [...events, { event, data } as EventPayload]); + + if (event === "values") { + setStreamValues(data); + } else if (event === "messages") { + if (!getMessages) continue; + + const [serialized] = data; + + const messageId = manager.current.add(serialized); + if (!messageId) { + console.warn( + "Failed to add message to manager, no message ID found", + ); + continue; + } + + setStreamValues((streamValues) => { + const values = { ...historyValues, ...streamValues }; + + // Assumption: we're concating the message + const messages = getMessages(values).slice(); + const { chunk, index } = + manager.current.get(messageId, messages.length) ?? {}; + + if (!chunk || index == null) return values; + messages[index] = toMessageDict(chunk); + + return { ...values, [withMessages!]: messages }; + }); + } + } + + // TODO: add a "checkpoint" stream mode to get the branches directly + await history.mutate(usableThreadId); + setStreamValues(null); + } catch (error) { + setError(error); + onError?.(error); + } finally { + // Assumption: messages are already handled, we can clear the manager + manager.current.clear(); + submittingRef.current = false; + } + }; + + const values = streamValues ?? historyValues; + const stream = { + get custom() { + if (!streamMode.includes("custom")) { + setStreamMode((mode) => unique([...mode, "custom"])); + } + + return events + .filter((item) => item.event === "custom") + .map(({ data }) => data as CustomType); + }, + + get events() { + if (!streamMode.includes("events")) { + setStreamMode((mode) => unique([...mode, "events"])); + } + + return events; + }, + get updates() { + if (!streamMode.includes("updates")) { + setStreamMode((mode) => unique([...mode, "updates"])); + } + + return events + .filter( + (item): item is UpdatesPayload => + item.event === "updates", + ) + .map(({ data }) => data); + }, + }; + + return { + error, + handleSubmit, + setBranchPath, + + sequence, + stream, + + get messages() { + if (!streamMode.includes("messages-tuple")) { + setStreamMode((mode) => unique([...mode, "messages-tuple"])); + } + + if (getMessages == null) { + throw new Error( + "No messages key provided. Make sure that `useStream` contains the `messagesKey` property.", + ); + } + + return getMessages(values); + }, + + getMessagesMeta(message: Message, index?: number) { + if (!streamMode.includes("messages-tuple")) { + setStreamMode((mode) => unique([...mode, "messages-tuple"])); + } + + if (getMessages == null) { + throw new Error( + "No messages key provided. Make sure that `useStream` contains the `messagesKey` property.", + ); + } + + return messageMeta?.find((m) => m.messageId === (message.id ?? index)); + }, + + get values() { + if (!streamMode.includes("values")) { + setStreamMode((mode) => unique([...mode, "values"])); + } + + return values; + }, + }; +} diff --git a/libs/sdk-js/tsconfig.json b/libs/sdk-js/tsconfig.json index f72c108ed..9c6561a09 100644 --- a/libs/sdk-js/tsconfig.json +++ b/libs/sdk-js/tsconfig.json @@ -19,6 +19,7 @@ "strictPropertyInitialization": false, "allowJs": true, "strict": true, + "jsx": "react-jsx", "outDir": "dist" }, "include": [ diff --git a/libs/sdk-js/yarn.lock b/libs/sdk-js/yarn.lock index fbde16b49..4d4e4ccf8 100644 --- a/libs/sdk-js/yarn.lock +++ b/libs/sdk-js/yarn.lock @@ -296,6 +296,11 @@ resolved "https://registry.yarnpkg.com/@bcoe/v8-coverage/-/v8-coverage-0.2.3.tgz#75a2e8b51cb758a7553d6804a5932d7aace75c39" integrity sha512-0hYQ8SB4Db5zvZB4axdMHGwEaQjkZzFjQiN9LVYvIFB2nSUHW9tYpxWriPrWDASIxiaXax83REcLxuSdnGPZtw== +"@cfworker/json-schema@^4.0.2": + version "4.1.1" + resolved "https://registry.yarnpkg.com/@cfworker/json-schema/-/json-schema-4.1.1.tgz#4a2a3947ee9fa7b7c24be981422831b8674c3be6" + integrity sha512-gAmrUZSGtKc3AiBL71iNWxDsyUC5uMaKKGdvzYsBoTW/xi42JQHl7eKV2OYzCUqvc+D2RCcf7EXY2iCyFIk6og== + "@isaacs/cliui@^8.0.2": version "8.0.2" resolved "https://registry.yarnpkg.com/@isaacs/cliui/-/cliui-8.0.2.tgz#b37667b7bc181c168782259bab42474fbf52b550" @@ -548,6 +553,24 @@ "@jridgewell/resolve-uri" "^3.1.0" "@jridgewell/sourcemap-codec" "^1.4.14" +"@langchain/core@^0.3.31": + version "0.3.39" + resolved "https://registry.yarnpkg.com/@langchain/core/-/core-0.3.39.tgz#81002e0cacacb7c1011264d6612f923816a3965f" + integrity sha512-muXs4asy1A7qDtcdznxqyBfxf4N6qxofY/S0c95vbsWa0r9YAE2PttHIjcuxSy1q2jUiTkpCcgFEjNJRQRVhEw== + dependencies: + "@cfworker/json-schema" "^4.0.2" + ansi-styles "^5.0.0" + camelcase "6" + decamelize "1.2.0" + js-tiktoken "^1.0.12" + langsmith ">=0.2.8 <0.4.0" + mustache "^4.2.0" + p-queue "^6.6.2" + p-retry "4" + uuid "^10.0.0" + zod "^3.22.4" + zod-to-json-schema "^3.22.3" + "@langchain/scripts@^0.1.4": version "0.1.4" resolved "https://registry.yarnpkg.com/@langchain/scripts/-/scripts-0.1.4.tgz#8c5d03d627686f20b9522213c12e2b329e73e381" @@ -950,6 +973,19 @@ resolved "https://registry.yarnpkg.com/@types/normalize-package-data/-/normalize-package-data-2.4.4.tgz#56e2cc26c397c038fab0e3a917a12d5c5909e901" integrity sha512-37i+OaWTh9qeK4LSHPsyRC7NahnGotNuZvjLSgcPzblpHB3rrCJxAOgI5gCdKm7coonsaX1Of0ILiTcnZjbfxA== +"@types/prop-types@*": + version "15.7.14" + resolved "https://registry.yarnpkg.com/@types/prop-types/-/prop-types-15.7.14.tgz#1433419d73b2a7ebfc6918dcefd2ec0d5cd698f2" + integrity sha512-gNMvNH49DJ7OJYv+KAKn0Xp45p8PLl6zo2YnvDIbTd4J6MER2BmWN49TG7n9LvkyihINxeKW8+3bfS2yDC9dzQ== + +"@types/react@18.3.2": + version "18.3.2" + resolved "https://registry.yarnpkg.com/@types/react/-/react-18.3.2.tgz#462ae4904973bc212fa910424d901e3d137dbfcd" + integrity sha512-Btgg89dAnqD4vV7R3hlwOxgqobUQKgx3MmrQRi0yYbs/P0ym8XozIAlkqVilPqHQwXs4e9Tf63rrCgl58BcO4w== + dependencies: + "@types/prop-types" "*" + csstype "^3.0.2" + "@types/retry@0.12.0": version "0.12.0" resolved "https://registry.yarnpkg.com/@types/retry/-/retry-0.12.0.tgz#2b35eccfcee7d38cd72ad99232fbd58bffb3c84d" @@ -965,6 +1001,11 @@ resolved "https://registry.yarnpkg.com/@types/unist/-/unist-2.0.10.tgz#04ffa7f406ab628f7f7e97ca23e290cd8ab15efc" integrity sha512-IfYcSBWE3hLpBg8+X2SEa8LVkJdJEkT2Ese2aaLs3ptGdVtABxndrMaxuFlQ1qdFf9Q5rDvDpxI3WwgvKFAsQA== +"@types/uuid@^10.0.0": + version "10.0.0" + resolved "https://registry.yarnpkg.com/@types/uuid/-/uuid-10.0.0.tgz#e9c07fe50da0f53dc24970cca94d619ff03f6f6d" + integrity sha512-7gqG38EyHgyP1S+7+xomFtL+ZNHcKv6DwNaCZmJmo1vgMugyF3TCnXVg4t1uk89mLNwnLtnY3TpOpCOyp1/xHQ== + "@types/uuid@^9.0.1": version "9.0.8" resolved "https://registry.yarnpkg.com/@types/uuid/-/uuid-9.0.8.tgz#7545ba4fc3c003d6c756f651f3bf163d8f0f29ba" @@ -1181,6 +1222,11 @@ balanced-match@^1.0.0: resolved "https://registry.yarnpkg.com/balanced-match/-/balanced-match-1.0.2.tgz#e83e3a7e3f300b34cb9d87f615fa0cbf357690ee" integrity sha512-3oSeUO0TMV67hN1AmbXsK4yaqU7tjiHlbxRDZOpH0KW9+CeX4bRAaX0Anxt0tx2MrpRpWwQaPwIlISEJhYU5Pw== +base64-js@^1.5.1: + version "1.5.1" + resolved "https://registry.yarnpkg.com/base64-js/-/base64-js-1.5.1.tgz#1b1b440160a5bf7ad40b650f095963481903930a" + integrity sha512-AKpaYlHn8t4SVbOHCy+b5+KKgvR4vrsD8vbvrbiQJps7fKDTkjkDry6ji0rUJjC0kzbNePLwzxq8iypo41qeWA== + before-after-hook@^3.0.2: version "3.0.2" resolved "https://registry.yarnpkg.com/before-after-hook/-/before-after-hook-3.0.2.tgz#d5665a5fa8b62294a5aa0a499f933f4a1016195d" @@ -1262,16 +1308,16 @@ camelcase-keys@^6.2.2: map-obj "^4.0.0" quick-lru "^4.0.1" +camelcase@6, camelcase@^6.2.0: + version "6.3.0" + resolved "https://registry.yarnpkg.com/camelcase/-/camelcase-6.3.0.tgz#5685b95eb209ac9c0c177467778c9c84df58ba9a" + integrity sha512-Gmy6FhYlCY7uOElZUSbxo2UCDH8owEk996gkbrpsgGtrJLM3J7jGxl9Ic7Qwwj4ivOE5AWZWRMecDdF7hqGjFA== + camelcase@^5.3.1: version "5.3.1" resolved "https://registry.yarnpkg.com/camelcase/-/camelcase-5.3.1.tgz#e3c9b31569e106811df242f715725a1f4c494320" integrity sha512-L28STB170nwWS63UjtlEOE3dldQApaJXZkOI1uMFfzf3rRuPegHaHesyee+YxQ+W6SvRDQV6UrdOdRiR153wJg== -camelcase@^6.2.0: - version "6.3.0" - resolved "https://registry.yarnpkg.com/camelcase/-/camelcase-6.3.0.tgz#5685b95eb209ac9c0c177467778c9c84df58ba9a" - integrity sha512-Gmy6FhYlCY7uOElZUSbxo2UCDH8owEk996gkbrpsgGtrJLM3J7jGxl9Ic7Qwwj4ivOE5AWZWRMecDdF7hqGjFA== - caniuse-lite@^1.0.30001688: version "1.0.30001692" resolved "https://registry.yarnpkg.com/caniuse-lite/-/caniuse-lite-1.0.30001692.tgz#4585729d95e6b95be5b439da6ab55250cd125bf9" @@ -1291,7 +1337,7 @@ chalk@^2.4.2: escape-string-regexp "^1.0.5" supports-color "^5.3.0" -chalk@^4.0.0, chalk@^4.0.2: +chalk@^4.0.0, chalk@^4.0.2, chalk@^4.1.2: version "4.1.2" resolved "https://registry.yarnpkg.com/chalk/-/chalk-4.1.2.tgz#aac4e2b7734a740867aeb16bf02aad556a1e7a01" integrity sha512-oKnbhFyRIXpUuez8iBMmyEa4nbj4IOQyuhc/wy9kY7/WVPcwIO9VA668Pu8RkO7+0G76SLROeyw9CpQ061i4mA== @@ -1406,6 +1452,13 @@ concat-md@^0.5.1: meow "^9.0.0" transform-markdown-links "^2.0.0" +console-table-printer@^2.12.1: + version "2.12.1" + resolved "https://registry.yarnpkg.com/console-table-printer/-/console-table-printer-2.12.1.tgz#4a9646537a246a6d8de57075d4fae1e08abae267" + integrity sha512-wKGOQRRvdnd89pCeH96e2Fn4wkbenSP6LMHfjfyNLMbGuHEFbMqQNuxXqd0oXG9caIOQ1FTvc5Uijp9/4jujnQ== + dependencies: + simple-wcswidth "^1.0.1" + convert-source-map@^2.0.0: version "2.0.0" resolved "https://registry.yarnpkg.com/convert-source-map/-/convert-source-map-2.0.0.tgz#4b560f649fc4e918dd0ab75cf4961e8bc882d82a" @@ -1442,6 +1495,11 @@ cross-spawn@^7.0.3: shebang-command "^2.0.0" which "^2.0.1" +csstype@^3.0.2: + version "3.1.3" + resolved "https://registry.yarnpkg.com/csstype/-/csstype-3.1.3.tgz#d80ff294d114fb0e6ac500fbf85b60137d7eff81" + integrity sha512-M1uQkMl8rQK/szD0LNhtqxIPLpimGm8sOBwU7lLnCpSbTyY3yeU1Vc7l4KT5zT4s/yOxHH5O7tIuuLOCnLADRw== + data-view-buffer@^1.0.1: version "1.0.1" resolved "https://registry.yarnpkg.com/data-view-buffer/-/data-view-buffer-1.0.1.tgz#8ea6326efec17a2e42620696e671d7d5a8bc66b2" @@ -1491,7 +1549,7 @@ decamelize-keys@^1.1.0: decamelize "^1.1.0" map-obj "^1.0.0" -decamelize@^1.1.0, decamelize@^1.2.0: +decamelize@1.2.0, decamelize@^1.1.0, decamelize@^1.2.0: version "1.2.0" resolved "https://registry.yarnpkg.com/decamelize/-/decamelize-1.2.0.tgz#f6534d15148269b20352e7bee26f501f9a191290" integrity sha512-z2S+W9X73hAUUki+N+9Za2lBlun89zigOyGrsax+KUQ6wKW4ZoWpEYBkGhQjwAjjDCkWxhY0VKEhk8wzY7F5cA== @@ -2794,7 +2852,14 @@ jest@^29.7.0: import-local "^3.0.2" jest-cli "^29.7.0" -js-tokens@^4.0.0: +js-tiktoken@^1.0.12: + version "1.0.18" + resolved "https://registry.yarnpkg.com/js-tiktoken/-/js-tiktoken-1.0.18.tgz#aaf68dda155bc693e6f0a572b9a359d569cb53df" + integrity sha512-hFYx4xYf6URgcttcGvGuOBJhTxPYZ2R5eIesqCaNRJmYH8sNmsfTeWg4yu//7u1VD/qIUkgKJTpGom9oHXmB4g== + dependencies: + base64-js "^1.5.1" + +"js-tokens@^3.0.0 || ^4.0.0", js-tokens@^4.0.0: version "4.0.0" resolved "https://registry.yarnpkg.com/js-tokens/-/js-tokens-4.0.0.tgz#19203fb59991df98e3a287050d4647cdeaf32499" integrity sha512-RdJUflcE3cUzKiMqQgsCu06FPu9UdIJO0beYbPhHN4k6apgJtifcoCtT9bcxOpYBtpD2kCM6Sbzg4CausW/PKQ== @@ -2832,6 +2897,19 @@ kleur@^3.0.3: resolved "https://registry.yarnpkg.com/kleur/-/kleur-3.0.3.tgz#a79c9ecc86ee1ce3fa6206d1216c501f147fc07e" integrity sha512-eTIzlVOSUR+JxdDFepEYcBMtZ9Qqdef+rnzWdRZuMbOywu5tO2w2N7rqjoANZ5k9vywhL6Br1VRjUIgTQx4E8w== +"langsmith@>=0.2.8 <0.4.0": + version "0.3.7" + resolved "https://registry.yarnpkg.com/langsmith/-/langsmith-0.3.7.tgz#c29362f78ea2872252a60a680d6adb8b67e18b74" + integrity sha512-wakN1hxGkm1JR2PpAV7fiT7oC99LKcgxiuUrYGZWPbuj7Y8EPF19F7VNr4B+hA219bfaeWTa4Lxy2YrtPSKnQA== + dependencies: + "@types/uuid" "^10.0.0" + chalk "^4.1.2" + console-table-printer "^2.12.1" + p-queue "^6.6.2" + p-retry "4" + semver "^7.6.3" + uuid "^10.0.0" + leven@^3.1.0: version "3.1.0" resolved "https://registry.yarnpkg.com/leven/-/leven-3.1.0.tgz#77891de834064cccba82ae7842bb6b14a13ed7f2" @@ -2876,6 +2954,13 @@ longest-streak@^2.0.0: resolved "https://registry.yarnpkg.com/longest-streak/-/longest-streak-2.0.4.tgz#b8599957da5b5dab64dee3fe316fa774597d90e4" integrity sha512-vM6rUVCVUJJt33bnmHiZEvr7wPT78ztX7rojL+LW51bHtLh6HTjx84LA5W4+oa6aKEJA7jJu5LR6vQRBpA5DVg== +loose-envify@^1.1.0: + version "1.4.0" + resolved "https://registry.yarnpkg.com/loose-envify/-/loose-envify-1.4.0.tgz#71ee51fa7be4caec1a63839f7e682d8132d30caf" + integrity sha512-lyuxPGr/Wfhrlem2CL/UcnUc1zcqKAImBDzukY7Y5F/yQiNdko6+fRLevlw1HgMySw7f611UIY408EtxRSoK3Q== + dependencies: + js-tokens "^3.0.0 || ^4.0.0" + lru-cache@^10.2.0: version "10.4.3" resolved "https://registry.yarnpkg.com/lru-cache/-/lru-cache-10.4.3.tgz#410fc8a17b70e598013df257c2446b7f3383f119" @@ -3234,6 +3319,11 @@ ms@^2.1.3: resolved "https://registry.yarnpkg.com/ms/-/ms-2.1.3.tgz#574c8138ce1d2b5861f0b44579dbadd60c6615b2" integrity sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA== +mustache@^4.2.0: + version "4.2.0" + resolved "https://registry.yarnpkg.com/mustache/-/mustache-4.2.0.tgz#e5892324d60a12ec9c2a73359edca52972bf6f64" + integrity sha512-71ippSywq5Yb7/tVYyGbkBggbU8H3u5Rz56fH60jGFgr8uHwxs+aSKeqmluIVzM0m0kB7xQjKS6qPfd0b2ZoqQ== + natural-compare@^1.4.0: version "1.4.0" resolved "https://registry.yarnpkg.com/natural-compare/-/natural-compare-1.4.0.tgz#4abebfeed7541f2c27acfb29bdbbd15c8d5ba4f7" @@ -3518,6 +3608,13 @@ react-is@^18.0.0: resolved "https://registry.yarnpkg.com/react-is/-/react-is-18.3.1.tgz#e83557dc12eae63a99e003a46388b1dcbb44db7e" integrity sha512-/LLMVyas0ljjAtoYiPqYiL8VWXzUUdThrmU5+n20DZv+a+ClRoevUzw5JxU+Ieh5/c87ytoTBV9G1FiKfNJdmg== +react@^18.3.1: + version "18.3.1" + resolved "https://registry.yarnpkg.com/react/-/react-18.3.1.tgz#49ab892009c53933625bd16b2533fc754cab2891" + integrity sha512-wS+hAgJShR0KhEvPJArfuPVN1+Hz1t0Y6n5jLrGQbkb4urgPE/0Rve+1kMB1v/oWgHgm4WIcV+i7F2pTVj+2iQ== + dependencies: + loose-envify "^1.1.0" + read-pkg-up@^7.0.1: version "7.0.1" resolved "https://registry.yarnpkg.com/read-pkg-up/-/read-pkg-up-7.0.1.tgz#f3a6135758459733ae2b95638056e1854e7ef507" @@ -3787,6 +3884,11 @@ signal-exit@^4.0.1: resolved "https://registry.yarnpkg.com/signal-exit/-/signal-exit-4.1.0.tgz#952188c1cbd546070e2dd20d0f41c0ae0530cb04" integrity sha512-bzyZ1e88w9O1iNJbKnOlvYTrWPDl46O1bG0D3XInv+9tkPrxrN8jUUTiFlDkkmKWgn1M6CfIA13SuGqOa9Korw== +simple-wcswidth@^1.0.1: + version "1.0.1" + resolved "https://registry.yarnpkg.com/simple-wcswidth/-/simple-wcswidth-1.0.1.tgz#8ab18ac0ae342f9d9b629604e54d2aa1ecb018b2" + integrity sha512-xMO/8eNREtaROt7tJvWJqHBDTMFN4eiQ5I4JRMuilwfnFcV5W9u7RUkueNkdw0jPqGMX36iCywelS5yilTuOxg== + sisteransi@^1.0.5: version "1.0.5" resolved "https://registry.yarnpkg.com/sisteransi/-/sisteransi-1.0.5.tgz#134d681297756437cc05ca01370d3a7a571075ed" @@ -4230,6 +4332,11 @@ update-section@^0.3.3: resolved "https://registry.yarnpkg.com/update-section/-/update-section-0.3.3.tgz#458f17820d37820dc60e20b86d94391b00123158" integrity sha512-BpRZMZpgXLuTiKeiu7kK0nIPwGdyrqrs6EDSaXtjD/aQ2T+qVo9a5hRC3HN3iJjCMxNT/VxoLGQ7E/OzE5ucnw== +uuid@^10.0.0: + version "10.0.0" + resolved "https://registry.yarnpkg.com/uuid/-/uuid-10.0.0.tgz#5a95aa454e6e002725c79055fd42aaba30ca6294" + integrity sha512-8XkAphELsDnEGrDxUOHB3RGvXz6TeuYSGEZBOjtTtPm2lwhGBjLgOzLHB63IUWfBpNucQjND6d3AOudO+H3RWQ== + uuid@^9.0.0: version "9.0.1" resolved "https://registry.yarnpkg.com/uuid/-/uuid-9.0.1.tgz#e188d4c8853cc722220392c424cd637f32293f30" @@ -4394,6 +4501,16 @@ yocto-queue@^0.1.0: resolved "https://registry.yarnpkg.com/yocto-queue/-/yocto-queue-0.1.0.tgz#0294eb3dee05028d31ee1a5fa2c556a6aaf10a1b" integrity sha512-rVksvsnNCdJ/ohGc6xgPwyN8eheCxsiLM8mxuE/t/mOVqJewPuO1miLpTHQiRgTKCLexL4MeAFVagts7HmNZ2Q== +zod-to-json-schema@^3.22.3: + version "3.24.1" + resolved "https://registry.yarnpkg.com/zod-to-json-schema/-/zod-to-json-schema-3.24.1.tgz#f08c6725091aadabffa820ba8d50c7ab527f227a" + integrity sha512-3h08nf3Vw3Wl3PK+q3ow/lIil81IT2Oa7YpQyUUDsEWbXveMesdfK1xBd2RhCkynwZndAxixji/7SYJJowr62w== + +zod@^3.22.4: + version "3.24.1" + resolved "https://registry.yarnpkg.com/zod/-/zod-3.24.1.tgz#27445c912738c8ad1e9de1bea0359fa44d9d35ee" + integrity sha512-muH7gBL9sI1nciMZV67X5fTKKBLtwpZ5VBp1vsOQzj1MhrBZ4wlVCm3gedKZWLp0Oyel8sIGfeiz54Su+OVT+A== + zwitch@^1.0.0: version "1.0.5" resolved "https://registry.yarnpkg.com/zwitch/-/zwitch-1.0.5.tgz#d11d7381ffed16b742f6af7b3f223d5cd9fe9920" From bd2268404c2dc417668d3ef5a3c6d6298e78eade Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Mon, 10 Feb 2025 13:28:24 -0800 Subject: [PATCH 02/16] Update types --- libs/sdk-js/src/react/stream.tsx | 61 ++++++++++++++------------------ 1 file changed, 26 insertions(+), 35 deletions(-) diff --git a/libs/sdk-js/src/react/stream.tsx b/libs/sdk-js/src/react/stream.tsx index 8b5b84562..d3941ce1e 100644 --- a/libs/sdk-js/src/react/stream.tsx +++ b/libs/sdk-js/src/react/stream.tsx @@ -6,13 +6,13 @@ import type { Command } from "../types.js"; import type { Message } from "../types.messages.js"; import type { Config, ThreadState } from "../schema.js"; import type { - CustomPayload, - DebugPayload, - EventsPayload, - MessagesPayload, - MessagesTuplePayload, - UpdatesPayload, - ValuesPayload, + CustomStreamEvent, + DebugStreamEvent, + EventsStreamEvent, + MessagesStreamEvent, + MessagesTupleStreamEvent, + UpdatesStreamEvent, + ValuesStreamEvent, } from "../types.stream.js"; import { @@ -114,18 +114,6 @@ export type MessageBranch = { options: CheckpointBranchPath[]; }; -const mergeConfig = (...configs: (Config | undefined)[]) => { - const result: Config = { configurable: {} }; - - for (const config of configs) { - if (config == null) continue; - // TODO: Just assigning the latest configurable is iffy. - result.configurable = config.configurable; - } - - return result; -}; - function fetchHistory>( client: Client, threadId: string, @@ -197,14 +185,14 @@ export function useStream< threadId?: string | null; onThreadId?: (threadId: string) => void; }) { - type EventPayload = - | ValuesPayload - | UpdatesPayload - | CustomPayload - | DebugPayload - | MessagesPayload - | MessagesTuplePayload - | EventsPayload; + type EventStreamEvent = + | ValuesStreamEvent + | UpdatesStreamEvent + | CustomStreamEvent + | DebugStreamEvent + | MessagesStreamEvent + | MessagesTupleStreamEvent + | EventsStreamEvent; const contextConfig = useContext(ConfigProvider); const { withMessages, onError, threadId, client } = Object.assign( @@ -222,7 +210,7 @@ export function useStream< const [branchPath, setBranchPath] = useState([]); const [error, setError] = useState(undefined); - const [events, setEvents] = useState([]); + const [events, setEvents] = useState([]); const [streamValues, setStreamValues] = useState(null); @@ -411,12 +399,15 @@ export function useStream< // TODO: why non-existent assistant ID does not throw an error here? const run = (await client.runs.stream(usableThreadId, "agent", { input: values as Record, - config: mergeConfig( - { configurable: lastSeenValue?.checkpoint }, - submitOptions?.config, - ), + config: { + ...submitOptions?.config, + configurable: { + ...lastSeenValue?.checkpoint, + ...submitOptions?.config?.configurable, + }, + }, streamMode, - })) as AsyncGenerator; + })) as AsyncGenerator; // Assumption: we're setting the initial value // Used for instant feedback @@ -433,7 +424,7 @@ export function useStream< } for await (const { event, data } of run) { - setEvents((events) => [...events, { event, data } as EventPayload]); + setEvents((events) => [...events, { event, data } as EventStreamEvent]); if (event === "values") { setStreamValues(data); @@ -505,7 +496,7 @@ export function useStream< return events .filter( - (item): item is UpdatesPayload => + (item): item is UpdatesStreamEvent => item.event === "updates", ) .map(({ data }) => data); From cb9405bcee5fca6ec15484a684c2ecacff8f13b2 Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Mon, 10 Feb 2025 17:52:51 -0800 Subject: [PATCH 03/16] Allow update: null --- libs/sdk-js/src/types.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libs/sdk-js/src/types.ts b/libs/sdk-js/src/types.ts index 643128e83..6945e9eba 100644 --- a/libs/sdk-js/src/types.ts +++ b/libs/sdk-js/src/types.ts @@ -26,7 +26,7 @@ export interface Command { /** * An object to update the thread state with. */ - update?: Record | [string, unknown][]; + update?: Record | [string, unknown][] | null; /** * The value to return from an `interrupt` function call. From a33437964c292586ac1564463369e68bb08deaed Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Mon, 10 Feb 2025 17:53:09 -0800 Subject: [PATCH 04/16] Add debugger, that is not exported --- libs/sdk-js/src/react/debug.tsx | 102 ++++++++++++++++++++++++++++++++ 1 file changed, 102 insertions(+) create mode 100644 libs/sdk-js/src/react/debug.tsx diff --git a/libs/sdk-js/src/react/debug.tsx b/libs/sdk-js/src/react/debug.tsx new file mode 100644 index 000000000..b5fc462fc --- /dev/null +++ b/libs/sdk-js/src/react/debug.tsx @@ -0,0 +1,102 @@ +import { ThreadState } from "../schema.js"; + +interface Node { + type: "node"; + value: ThreadState; + path: string[]; +} + +interface Fork { + type: "fork"; + items: Array>; +} + +interface Sequence { + type: "sequence"; + items: Array | Fork>; +} + +interface ValidFork { + type: "fork"; + items: Array>; +} + +interface ValidSequence { + type: "sequence"; + items: [Node, ...(Node | ValidFork)[]]; +} + +// forks +export type CheckpointBranchPath = string[]; + +export type MessageBranch = { + current: CheckpointBranchPath; + options: CheckpointBranchPath[]; +}; + +export function DebugSegmentsView(props: { + sequence: ValidSequence; +}) { + const concatContent = (value: ThreadState) => { + let content; + try { + content = value.values?.messages?.at(-1)?.content ?? ""; + } catch { + content = JSON.stringify(value.values); + } + + content = content.replace(/(\n|\r\n)/g, ""); + if (content.length <= 23) return content; + return `${content.slice(0, 10)}...${content.slice(-10)}`; + }; + + return ( +
+ {props.sequence.items.map((item, index) => { + if (item.type === "fork") { + return ( +
+ {item.items.map((fork, idx) => { + const [first] = fork.items; + return ( +
+ + Fork{" "} + + ...{first.path.at(-1)?.slice(-4)} + + +
+ +
+
+ ); + })} +
+ ); + } + + if (item.type === "node") { + return ( +
+ 
+                ({item.value.metadata?.step}) ...
+                {item.value.checkpoint.checkpoint_id?.slice(-4)} (
+                {item.value.metadata?.source}): {concatContent(item.value)}
+
+ +
+ ); + } + + return null; + })} +
+ ); +} From 735a76a16c0573209a66f1ecc64f1f317998689e Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Mon, 10 Feb 2025 17:53:22 -0800 Subject: [PATCH 05/16] Clean up the API --- libs/sdk-js/src/react/index.ts | 2 +- libs/sdk-js/src/react/stream.tsx | 291 +++++++++++++++++-------------- 2 files changed, 159 insertions(+), 134 deletions(-) diff --git a/libs/sdk-js/src/react/index.ts b/libs/sdk-js/src/react/index.ts index 7fc550ada..97a58f3d1 100644 --- a/libs/sdk-js/src/react/index.ts +++ b/libs/sdk-js/src/react/index.ts @@ -1 +1 @@ -export { useStream, LangGraphConfig } from "./stream.js"; +export { useStream, type MessageMetadata } from "./stream.js"; diff --git a/libs/sdk-js/src/react/stream.tsx b/libs/sdk-js/src/react/stream.tsx index d3941ce1e..aca6bf098 100644 --- a/libs/sdk-js/src/react/stream.tsx +++ b/libs/sdk-js/src/react/stream.tsx @@ -4,23 +4,23 @@ import { Client } from "../client.js"; import type { Command } from "../types.js"; import type { Message } from "../types.messages.js"; -import type { Config, ThreadState } from "../schema.js"; +import type { Checkpoint, Config, ThreadState } from "../schema.js"; import type { CustomStreamEvent, DebugStreamEvent, + ErrorStreamEvent, EventsStreamEvent, MessagesStreamEvent, MessagesTupleStreamEvent, + MetadataStreamEvent, + StreamMode, UpdatesStreamEvent, ValuesStreamEvent, } from "../types.stream.js"; import { type MutableRefObject, - type ReactNode, - createContext, useCallback, - useContext, useEffect, useMemo, useRef, @@ -33,6 +33,12 @@ import { convertToChunk, } from "@langchain/core/messages"; +class StreamError extends Error { + constructor(data: { error: string; message: string }) { + super([data.error, data.message].filter(Boolean).join(": ")); + } +} + class MessageTupleManager { chunks: Record = {}; @@ -106,12 +112,12 @@ interface ValidSequence { items: [Node, ...(Node | ValidFork)[]]; } -// forks -export type CheckpointBranchPath = string[]; +export type MessageMetadata> = { + messageId: string; + firstSeenState: ThreadState | undefined; -export type MessageBranch = { - current: CheckpointBranchPath; - options: CheckpointBranchPath[]; + branch: string | undefined; + branchOptions: string[] | undefined; }; function fetchHistory>( @@ -129,15 +135,18 @@ function useThreadHistory>( const [history, setHistory] = useState[]>([]); const fetcher = useCallback( - (threadId: string | undefined | null): Promise => { + ( + threadId: string | undefined | null, + ): Promise[]> => { if (threadId != null) { - return fetchHistory(client, threadId).then((history) => - setHistory(history), - ); + return fetchHistory(client, threadId).then((history) => { + setHistory(history); + return history; + }); } setHistory([]); - return Promise.resolve(); + return Promise.resolve([]); }, [], ); @@ -153,33 +162,18 @@ function useThreadHistory>( }; } -interface LangGraphConfig { - withMessages?: string; - onError?: (error: unknown) => void; - client?: Client; -} - -const ConfigProvider = createContext(null!); -export const LangGraphConfig = (props: { - config: LangGraphConfig; - children?: ReactNode; -}) => { - return ( - - {props.children} - - ); -}; - export function useStream< StateType extends Record = Record, UpdateType extends Record = Partial, CustomType = unknown, ->(options?: { - client?: Client; +>(options: { + assistantId: string; + client: Client; withMessages?: string; + onError?: (error: unknown) => void; + onFinish?: (state: ThreadState) => void; // TODO: can we make threadId uncontrollable / controllable? threadId?: string | null; @@ -192,14 +186,12 @@ export function useStream< | DebugStreamEvent | MessagesStreamEvent | MessagesTupleStreamEvent - | EventsStreamEvent; + | EventsStreamEvent + | MetadataStreamEvent + | ErrorStreamEvent; - const contextConfig = useContext(ConfigProvider); - const { withMessages, onError, threadId, client } = Object.assign( - {}, - contextConfig, - options, - ); + const { assistantId, threadId, client, withMessages, onError, onFinish } = + options; if (client == null) { throw new Error( @@ -207,22 +199,29 @@ export function useStream< ); } - const [branchPath, setBranchPath] = useState([]); - + const [branchPath, setBranchPath] = useState([]); + const [isLoading, setIsLoading] = useState(false); const [error, setError] = useState(undefined); const [events, setEvents] = useState([]); const [streamValues, setStreamValues] = useState(null); - const [streamMode, setStreamMode] = useState< + const messageManagerRef = useRef(new MessageTupleManager()); + const submittingRef = useRef(false); + const trackStreamModeRef = useRef< Array<"values" | "updates" | "events" | "custom" | "messages-tuple"> >(["values", "messages-tuple"]); - const manager = useRef(new MessageTupleManager()); - const submittingRef = useRef(false); + const trackStreamMode = useCallback( + (mode: Exclude) => { + if (!trackStreamModeRef.current.includes(mode)) + trackStreamModeRef.current.push(mode); + }, + [], + ); - // TODO: is this a responsibility of SWR / React Query? - // Maybe we should use that instead to allow rehydration? + // TODO: this should be done on the server to avoid pagination + // TODO: should we permit adapter? SWR / React Query? const history = useThreadHistory(threadId, client, submittingRef); const getMessages = useMemo(() => { @@ -297,11 +296,14 @@ export function useStream< return [rootSequence as ValidSequence, pathMap]; })(); - const [flatValues, checkpointPathMap] = (() => { + const [flatValues, flatPaths] = (() => { const result: ThreadState[] = []; // TODO: this is kinda ugly - const checkpointPathMap: Record = {}; + const flatPaths: Record< + string, + { current: string[] | undefined; branches: string[][] | undefined } + > = {}; const forkStack = branchPath.slice(); const queue: (Node | Fork)[] = [...sequence.items]; @@ -311,9 +313,9 @@ export function useStream< if (item.type === "node") { result.push(item.value); - checkpointPathMap[item.value.checkpoint.checkpoint_id!] = { + flatPaths[item.value.checkpoint.checkpoint_id!] = { current: item.path, - options: + branches: item.path.length > 0 ? pathMap[item.path.at(-2) ?? "$"] ?? [] : [], }; } @@ -333,62 +335,68 @@ export function useStream< } } - return [result, checkpointPathMap]; + return [result, flatPaths]; })(); - const lastSeenValue = flatValues.at(-1); - const historyValues = lastSeenValue?.values ?? ({} as StateType); + const threadHead: ThreadState | undefined = flatValues.at(-1); + const historyValues = threadHead?.values ?? ({} as StateType); - const messageMeta = (() => { + const messageMetadata = (() => { if (getMessages == null) return undefined; const alreadyShown = new Set(); - return getMessages(historyValues).map((message, idx) => { - const messageId = message.id ?? idx; - const firstSeenIdx = findLastIndex(history.data, (state) => - getMessages(state.values) - .map((m, idx) => m.id ?? idx) - .includes(messageId), - ); + return getMessages(historyValues).map( + (message, idx): MessageMetadata => { + const messageId = message.id ?? idx; + const firstSeenIdx = findLastIndex(history.data, (state) => + getMessages(state.values) + .map((m, idx) => m.id ?? idx) + .includes(messageId), + ); - const firstState = history.data[firstSeenIdx] as - | ThreadState - | undefined; + const firstSeen = history.data[firstSeenIdx] as + | ThreadState + | undefined; - let branch = firstState - ? checkpointPathMap[firstState.checkpoint.checkpoint_id!] - : undefined; - if (!branch?.current.length) branch = undefined; + let branch = firstSeen + ? flatPaths[firstSeen.checkpoint.checkpoint_id!] + : undefined; - const optionsShown = branch?.options?.flat(2).join(","); - if (optionsShown) { - if (alreadyShown.has(optionsShown)) branch = undefined; - alreadyShown.add(optionsShown); - } + if (!branch?.current?.length) branch = undefined; - return { messageId, firstState, branch }; - }); + // serialize branches + const optionsShown = branch?.branches?.flat(2).join(","); + if (optionsShown) { + if (alreadyShown.has(optionsShown)) branch = undefined; + alreadyShown.add(optionsShown); + } + + return { + messageId: messageId.toString(), + firstSeenState: firstSeen, + branch: branch?.current?.join(">"), + branchOptions: branch?.branches?.map((b) => b.join(">")), + }; + }, + ); })(); const handleSubmit = async ( values: UpdateType | undefined, submitOptions?: { config?: Config; + checkpoint?: Omit | null; command?: Command; + streamMode?: Array; optimisticValues?: | Partial | ((prev: StateType) => Partial); }, ) => { try { - // TODO: have loading state as well + setIsLoading(true); submittingRef.current = true; - // This is used to reset the path to make sure we always fetch the - // latest generation / edit. - // TODO: make sure it's actually aware of the config passed to handleSubmit - setBranchPath((path) => path.slice(0, -1)); - let usableThreadId = threadId; if (!usableThreadId) { const thread = await client.threads.create(); @@ -396,36 +404,51 @@ export function useStream< usableThreadId = thread.thread_id; } + const streamMode = unique([ + ...(submitOptions?.streamMode ?? []), + ...trackStreamModeRef.current, + ]); + + const checkpoint = + submitOptions?.checkpoint ?? threadHead?.checkpoint ?? undefined; + // @ts-expect-error + if (checkpoint != null) delete checkpoint.thread_id; + // TODO: why non-existent assistant ID does not throw an error here? - const run = (await client.runs.stream(usableThreadId, "agent", { + const run = (await client.runs.stream(usableThreadId, assistantId, { input: values as Record, - config: { - ...submitOptions?.config, - configurable: { - ...lastSeenValue?.checkpoint, - ...submitOptions?.config?.configurable, - }, - }, + config: submitOptions?.config, + checkpoint, streamMode, })) as AsyncGenerator; // Assumption: we're setting the initial value // Used for instant feedback - if (submitOptions?.optimisticValues != null) { - setStreamValues((streamValues) => { - const values = { ...historyValues, ...streamValues }; + setStreamValues(() => { + const values = { ...historyValues }; + + if (submitOptions?.optimisticValues != null) { return { ...values, ...(typeof submitOptions.optimisticValues === "function" ? submitOptions.optimisticValues(values) : submitOptions.optimisticValues), }; - }); - } + } + + return values; + }); for await (const { event, data } of run) { setEvents((events) => [...events, { event, data } as EventStreamEvent]); + if (event === "error") { + const error = new StreamError(data); + setError(error); + onError?.(error); + break; + } + if (event === "values") { setStreamValues(data); } else if (event === "messages") { @@ -433,7 +456,7 @@ export function useStream< const [serialized] = data; - const messageId = manager.current.add(serialized); + const messageId = messageManagerRef.current.add(serialized); if (!messageId) { console.warn( "Failed to add message to manager, no message ID found", @@ -447,7 +470,7 @@ export function useStream< // Assumption: we're concating the message const messages = getMessages(values).slice(); const { chunk, index } = - manager.current.get(messageId, messages.length) ?? {}; + messageManagerRef.current.get(messageId, messages.length) ?? {}; if (!chunk || index == null) return values; messages[index] = toMessageDict(chunk); @@ -457,15 +480,20 @@ export function useStream< } } - // TODO: add a "checkpoint" stream mode to get the branches directly - await history.mutate(usableThreadId); + // TODO: stream created checkpoints to avoid an unnecessary network request + const result = await history.mutate(usableThreadId); setStreamValues(null); + + const lastHead = result.at(0); + if (lastHead) onFinish?.(lastHead); } catch (error) { setError(error); onError?.(error); } finally { + setIsLoading(false); + // Assumption: messages are already handled, we can clear the manager - manager.current.clear(); + messageManagerRef.current.clear(); submittingRef.current = false; } }; @@ -473,9 +501,7 @@ export function useStream< const values = streamValues ?? historyValues; const stream = { get custom() { - if (!streamMode.includes("custom")) { - setStreamMode((mode) => unique([...mode, "custom"])); - } + trackStreamMode("custom"); return events .filter((item) => item.event === "custom") @@ -483,17 +509,12 @@ export function useStream< }, get events() { - if (!streamMode.includes("events")) { - setStreamMode((mode) => unique([...mode, "events"])); - } - + trackStreamMode("events"); return events; }, - get updates() { - if (!streamMode.includes("updates")) { - setStreamMode((mode) => unique([...mode, "updates"])); - } + get updates() { + trackStreamMode("updates"); return events .filter( (item): item is UpdatesStreamEvent => @@ -503,18 +524,27 @@ export function useStream< }, }; - return { - error, - handleSubmit, - setBranchPath, + const setBranch = useCallback( + (path: string) => setBranchPath(path.split(">")), + [setBranchPath], + ); + + return { + get values() { + trackStreamMode("values"); + return values; + }, + + error, + isLoading, + + handleSubmit, + setBranch, - sequence, stream, get messages() { - if (!streamMode.includes("messages-tuple")) { - setStreamMode((mode) => unique([...mode, "messages-tuple"])); - } + trackStreamMode("messages-tuple"); if (getMessages == null) { throw new Error( @@ -525,10 +555,11 @@ export function useStream< return getMessages(values); }, - getMessagesMeta(message: Message, index?: number) { - if (!streamMode.includes("messages-tuple")) { - setStreamMode((mode) => unique([...mode, "messages-tuple"])); - } + getMessagesMetadata( + message: Message, + index?: number, + ): MessageMetadata | undefined { + trackStreamMode("messages-tuple"); if (getMessages == null) { throw new Error( @@ -536,15 +567,9 @@ export function useStream< ); } - return messageMeta?.find((m) => m.messageId === (message.id ?? index)); - }, - - get values() { - if (!streamMode.includes("values")) { - setStreamMode((mode) => unique([...mode, "values"])); - } - - return values; + return messageMetadata?.find( + (m) => m.messageId === (message.id ?? index), + ); }, }; } From ea8025b7193ff8ffa4339c9ac41d07a3dedfd8ae Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Mon, 10 Feb 2025 23:02:47 -0800 Subject: [PATCH 06/16] Add other stream parameters --- libs/sdk-js/src/react/stream.tsx | 68 ++++++++++++++++++++++++++------ 1 file changed, 55 insertions(+), 13 deletions(-) diff --git a/libs/sdk-js/src/react/stream.tsx b/libs/sdk-js/src/react/stream.tsx index aca6bf098..40520f819 100644 --- a/libs/sdk-js/src/react/stream.tsx +++ b/libs/sdk-js/src/react/stream.tsx @@ -2,9 +2,14 @@ "use client"; import { Client } from "../client.js"; -import type { Command } from "../types.js"; +import type { + Command, + DisconnectMode, + MultitaskStrategy, + OnCompletionBehavior, +} from "../types.js"; import type { Message } from "../types.messages.js"; -import type { Checkpoint, Config, ThreadState } from "../schema.js"; +import type { Checkpoint, Config, Metadata, ThreadState } from "../schema.js"; import type { CustomStreamEvent, DebugStreamEvent, @@ -34,8 +39,16 @@ import { } from "@langchain/core/messages"; class StreamError extends Error { - constructor(data: { error: string; message: string }) { - super([data.error, data.message].filter(Boolean).join(": ")); + constructor(data: { error?: string; name?: string; message: string }) { + super([data.error ?? data.name, data.message].filter(Boolean).join(": ")); + } + + static isStructuredError(error: unknown): error is { + error?: string; + name?: string; + message: string; + } { + return typeof error === "object" && error != null && "message" in error; } } @@ -201,9 +214,9 @@ export function useStream< const [branchPath, setBranchPath] = useState([]); const [isLoading, setIsLoading] = useState(false); - const [error, setError] = useState(undefined); const [events, setEvents] = useState([]); + const [streamError, setStreamError] = useState(undefined); const [streamValues, setStreamValues] = useState(null); const messageManagerRef = useRef(new MessageTupleManager()); @@ -340,6 +353,21 @@ export function useStream< const threadHead: ThreadState | undefined = flatValues.at(-1); const historyValues = threadHead?.values ?? ({} as StateType); + const historyError = (() => { + const error = threadHead?.tasks?.at(-1)?.error; + if (error == null) return undefined; + try { + const parsed = JSON.parse(error) as unknown; + if (StreamError.isStructuredError(parsed)) { + return new StreamError(parsed); + } + + return parsed; + } catch { + // do nothing + } + return error; + })(); const messageMetadata = (() => { if (getMessages == null) return undefined; @@ -387,6 +415,13 @@ export function useStream< config?: Config; checkpoint?: Omit | null; command?: Command; + interruptBefore?: "*" | string[]; + interruptAfter?: "*" | string[]; + metadata?: Metadata; + multitaskStrategy?: MultitaskStrategy; + onCompletion?: OnCompletionBehavior; + onDisconnect?: DisconnectMode; + feedbackKeys?: string[]; streamMode?: Array; optimisticValues?: | Partial @@ -395,6 +430,8 @@ export function useStream< ) => { try { setIsLoading(true); + setStreamError(undefined); + submittingRef.current = true; let usableThreadId = threadId; @@ -418,6 +455,16 @@ export function useStream< const run = (await client.runs.stream(usableThreadId, assistantId, { input: values as Record, config: submitOptions?.config, + command: submitOptions?.command, + + interruptBefore: submitOptions?.interruptBefore, + interruptAfter: submitOptions?.interruptAfter, + metadata: submitOptions?.metadata, + multitaskStrategy: submitOptions?.multitaskStrategy, + onCompletion: submitOptions?.onCompletion, + onDisconnect: submitOptions?.onDisconnect, + feedbackKeys: submitOptions?.feedbackKeys, + checkpoint, streamMode, })) as AsyncGenerator; @@ -442,13 +489,7 @@ export function useStream< for await (const { event, data } of run) { setEvents((events) => [...events, { event, data } as EventStreamEvent]); - if (event === "error") { - const error = new StreamError(data); - setError(error); - onError?.(error); - break; - } - + if (event === "error") throw new StreamError(data); if (event === "values") { setStreamValues(data); } else if (event === "messages") { @@ -487,7 +528,7 @@ export function useStream< const lastHead = result.at(0); if (lastHead) onFinish?.(lastHead); } catch (error) { - setError(error); + setStreamError(error); onError?.(error); } finally { setIsLoading(false); @@ -498,6 +539,7 @@ export function useStream< } }; + const error = isLoading ? streamError : historyError; const values = streamValues ?? historyValues; const stream = { get custom() { From 98976e016abc507c012800b113ace9537a63119b Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Mon, 10 Feb 2025 23:05:01 -0800 Subject: [PATCH 07/16] Prevent double Error: Error --- libs/sdk-js/src/react/stream.tsx | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/libs/sdk-js/src/react/stream.tsx b/libs/sdk-js/src/react/stream.tsx index 40520f819..86f85cc86 100644 --- a/libs/sdk-js/src/react/stream.tsx +++ b/libs/sdk-js/src/react/stream.tsx @@ -40,7 +40,8 @@ import { class StreamError extends Error { constructor(data: { error?: string; name?: string; message: string }) { - super([data.error ?? data.name, data.message].filter(Boolean).join(": ")); + super(data.message); + this.name = data.name ?? data.error ?? "StreamError"; } static isStructuredError(error: unknown): error is { From 2c5ddceeb1cc77a42d5665cc9be733d561ce1cae Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Mon, 10 Feb 2025 23:42:43 -0800 Subject: [PATCH 08/16] Add submit / stop --- libs/sdk-js/src/react/stream.tsx | 49 ++++++++++++++++++++++++++------ 1 file changed, 41 insertions(+), 8 deletions(-) diff --git a/libs/sdk-js/src/react/stream.tsx b/libs/sdk-js/src/react/stream.tsx index 86f85cc86..1726b5920 100644 --- a/libs/sdk-js/src/react/stream.tsx +++ b/libs/sdk-js/src/react/stream.tsx @@ -15,6 +15,7 @@ import type { DebugStreamEvent, ErrorStreamEvent, EventsStreamEvent, + FeedbackStreamEvent, MessagesStreamEvent, MessagesTupleStreamEvent, MetadataStreamEvent, @@ -202,7 +203,8 @@ export function useStream< | MessagesTupleStreamEvent | EventsStreamEvent | MetadataStreamEvent - | ErrorStreamEvent; + | ErrorStreamEvent + | FeedbackStreamEvent; const { assistantId, threadId, client, withMessages, onError, onFinish } = options; @@ -222,6 +224,8 @@ export function useStream< const messageManagerRef = useRef(new MessageTupleManager()); const submittingRef = useRef(false); + const abortRef = useRef(null); + const trackStreamModeRef = useRef< Array<"values" | "updates" | "events" | "custom" | "messages-tuple"> >(["values", "messages-tuple"]); @@ -410,7 +414,12 @@ export function useStream< ); })(); - const handleSubmit = async ( + const stop = useCallback(() => { + if (abortRef.current != null) abortRef.current.abort(); + abortRef.current = null; + }, []); + + const submit = async ( values: UpdateType | undefined, submitOptions?: { config?: Config; @@ -434,6 +443,7 @@ export function useStream< setStreamError(undefined); submittingRef.current = true; + abortRef.current = new AbortController(); let usableThreadId = threadId; if (!usableThreadId) { @@ -463,8 +473,11 @@ export function useStream< metadata: submitOptions?.metadata, multitaskStrategy: submitOptions?.multitaskStrategy, onCompletion: submitOptions?.onCompletion, - onDisconnect: submitOptions?.onDisconnect, + onDisconnect: submitOptions?.onDisconnect ?? "cancel", + + // TODO: check if integration on FE would work nice feedbackKeys: submitOptions?.feedbackKeys, + signal: abortRef.current.signal, checkpoint, streamMode, @@ -487,13 +500,20 @@ export function useStream< return values; }); + let streamError: StreamError | undefined; for await (const { event, data } of run) { setEvents((events) => [...events, { event, data } as EventStreamEvent]); - if (event === "error") throw new StreamError(data); + if (event === "error") { + streamError = new StreamError(data); + break; + } + if (event === "values") { setStreamValues(data); - } else if (event === "messages") { + } + + if (event === "messages") { if (!getMessages) continue; const [serialized] = data; @@ -524,19 +544,31 @@ export function useStream< // TODO: stream created checkpoints to avoid an unnecessary network request const result = await history.mutate(usableThreadId); + + // TODO: write tests verifying that stream values are properly handled lifecycle-wise setStreamValues(null); + if (streamError != null) throw streamError; + const lastHead = result.at(0); if (lastHead) onFinish?.(lastHead); } catch (error) { - setStreamError(error); - onError?.(error); + if ( + !( + error instanceof Error && + (error.name === "AbortError" || error.name === "TimeoutError") + ) + ) { + setStreamError(error); + onError?.(error); + } } finally { setIsLoading(false); // Assumption: messages are already handled, we can clear the manager messageManagerRef.current.clear(); submittingRef.current = false; + abortRef.current = null; } }; @@ -581,7 +613,8 @@ export function useStream< error, isLoading, - handleSubmit, + stop, + submit, setBranch, stream, From f3fe30380c6176eda93888d7998dafdeb5094fca Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Tue, 11 Feb 2025 09:01:39 -0800 Subject: [PATCH 09/16] Make sure we clear when switching threads --- libs/sdk-js/src/react/stream.tsx | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/libs/sdk-js/src/react/stream.tsx b/libs/sdk-js/src/react/stream.tsx index 1726b5920..047eff812 100644 --- a/libs/sdk-js/src/react/stream.tsx +++ b/libs/sdk-js/src/react/stream.tsx @@ -145,6 +145,7 @@ function fetchHistory>( function useThreadHistory>( threadId: string | undefined | null, client: Client, + clearCallbackRef: MutableRefObject<(() => void) | undefined>, submittingRef: MutableRefObject, ) { const [history, setHistory] = useState[]>([]); @@ -161,6 +162,7 @@ function useThreadHistory>( } setHistory([]); + clearCallbackRef.current?.(); return Promise.resolve([]); }, [], @@ -238,9 +240,20 @@ export function useStream< [], ); + const clearCallbackRef = useRef<() => void>(null!); + clearCallbackRef.current = () => { + setStreamError(undefined); + setStreamValues(null); + }; + // TODO: this should be done on the server to avoid pagination // TODO: should we permit adapter? SWR / React Query? - const history = useThreadHistory(threadId, client, submittingRef); + const history = useThreadHistory( + threadId, + client, + clearCallbackRef, + submittingRef, + ); const getMessages = useMemo(() => { if (withMessages == null) return undefined; From ac05955222734e2389c5a1bbb66aea27c1fac7e2 Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Tue, 11 Feb 2025 09:09:08 -0800 Subject: [PATCH 10/16] Update peer dependencies --- libs/sdk-js/package.json | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/libs/sdk-js/package.json b/libs/sdk-js/package.json index 1181feaf8..dfa35d1ae 100644 --- a/libs/sdk-js/package.json +++ b/libs/sdk-js/package.json @@ -39,9 +39,8 @@ "@types/react": "18.3.2" }, "peerDependencies": { - "react": "*", - "@types/react": "*", - "@langchain/core": "*" + "react": "^18 || ^19", + "@langchain/core": ">=0.2.31 <0.4.0" }, "exports": { ".": { From 3163a60466be6737d8587044acb531d63fb60db4 Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Tue, 11 Feb 2025 09:58:19 -0800 Subject: [PATCH 11/16] Fix typo --- libs/sdk-js/src/react/stream.tsx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libs/sdk-js/src/react/stream.tsx b/libs/sdk-js/src/react/stream.tsx index 047eff812..5eeafb6ed 100644 --- a/libs/sdk-js/src/react/stream.tsx +++ b/libs/sdk-js/src/react/stream.tsx @@ -542,7 +542,7 @@ export function useStream< setStreamValues((streamValues) => { const values = { ...historyValues, ...streamValues }; - // Assumption: we're concating the message + // Assumption: we're concatenating the message const messages = getMessages(values).slice(); const { chunk, index } = messageManagerRef.current.get(messageId, messages.length) ?? {}; From d9ed1ef52ec3db945c6ae94a3f18ae8e9c65deab Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Tue, 11 Feb 2025 10:10:54 -0800 Subject: [PATCH 12/16] Branching --- libs/sdk-js/src/react/stream.tsx | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/libs/sdk-js/src/react/stream.tsx b/libs/sdk-js/src/react/stream.tsx index 5eeafb6ed..a8df9e626 100644 --- a/libs/sdk-js/src/react/stream.tsx +++ b/libs/sdk-js/src/react/stream.tsx @@ -329,8 +329,6 @@ export function useStream< const [flatValues, flatPaths] = (() => { const result: ThreadState[] = []; - - // TODO: this is kinda ugly const flatPaths: Record< string, { current: string[] | undefined; branches: string[][] | undefined } @@ -475,7 +473,6 @@ export function useStream< // @ts-expect-error if (checkpoint != null) delete checkpoint.thread_id; - // TODO: why non-existent assistant ID does not throw an error here? const run = (await client.runs.stream(usableThreadId, assistantId, { input: values as Record, config: submitOptions?.config, @@ -496,6 +493,12 @@ export function useStream< streamMode, })) as AsyncGenerator; + // Unbranch things + const newPath = submitOptions?.checkpoint?.checkpoint_id + ? flatPaths[submitOptions?.checkpoint?.checkpoint_id]?.current + : undefined; + if (newPath != null) setBranchPath(newPath ?? []); + // Assumption: we're setting the initial value // Used for instant feedback setStreamValues(() => { From 77a3cffa7fbb4c167551d11bf9c3f2c7999ffc26 Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Wed, 12 Feb 2025 08:50:01 -0800 Subject: [PATCH 13/16] Allow omitting client --- libs/sdk-js/src/react/stream.tsx | 22 ++++++++++++++++++---- 1 file changed, 18 insertions(+), 4 deletions(-) diff --git a/libs/sdk-js/src/react/stream.tsx b/libs/sdk-js/src/react/stream.tsx index a8df9e626..01a7811be 100644 --- a/libs/sdk-js/src/react/stream.tsx +++ b/libs/sdk-js/src/react/stream.tsx @@ -1,7 +1,7 @@ /* __LC_ALLOW_ENTRYPOINT_SIDE_EFFECTS__ */ "use client"; -import { Client } from "../client.js"; +import { Client, ClientConfig } from "../client.js"; import type { Command, DisconnectMode, @@ -185,7 +185,12 @@ export function useStream< CustomType = unknown, >(options: { assistantId: string; - client: Client; + + apiUrl: ClientConfig["apiUrl"]; + apiKey?: ClientConfig["apiKey"]; + callerOptions?: ClientConfig["callerOptions"]; + timeoutMs?: ClientConfig["timeoutMs"]; + defaultHeaders?: ClientConfig["defaultHeaders"]; withMessages?: string; @@ -208,8 +213,17 @@ export function useStream< | ErrorStreamEvent | FeedbackStreamEvent; - const { assistantId, threadId, client, withMessages, onError, onFinish } = - options; + const { assistantId, threadId, withMessages, onError, onFinish } = options; + const [client] = useState( + () => + new Client({ + apiUrl: options.apiUrl, + apiKey: options.apiKey, + callerOptions: options.callerOptions, + timeoutMs: options.timeoutMs, + defaultHeaders: options.defaultHeaders, + }), + ); if (client == null) { throw new Error( From b6aff521e5c6e49a49200bded6750e680eef6215 Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Wed, 12 Feb 2025 09:11:44 -0800 Subject: [PATCH 14/16] Make event listeners a callback --- libs/sdk-js/src/client.ts | 2 +- libs/sdk-js/src/react/stream.tsx | 80 +++++++++++++------------------- 2 files changed, 33 insertions(+), 49 deletions(-) diff --git a/libs/sdk-js/src/client.ts b/libs/sdk-js/src/client.ts index c4ffcb7ed..0406ae4a1 100644 --- a/libs/sdk-js/src/client.ts +++ b/libs/sdk-js/src/client.ts @@ -68,7 +68,7 @@ export function getApiKey(apiKey?: string): string | undefined { return undefined; } -interface ClientConfig { +export interface ClientConfig { apiUrl?: string; apiKey?: string; callerOptions?: AsyncCallerParams; diff --git a/libs/sdk-js/src/react/stream.tsx b/libs/sdk-js/src/react/stream.tsx index 01a7811be..0f229da40 100644 --- a/libs/sdk-js/src/react/stream.tsx +++ b/libs/sdk-js/src/react/stream.tsx @@ -1,7 +1,7 @@ /* __LC_ALLOW_ENTRYPOINT_SIDE_EFFECTS__ */ "use client"; -import { Client, ClientConfig } from "../client.js"; +import { Client, type ClientConfig } from "../client.js"; import type { Command, DisconnectMode, @@ -188,15 +188,16 @@ export function useStream< apiUrl: ClientConfig["apiUrl"]; apiKey?: ClientConfig["apiKey"]; - callerOptions?: ClientConfig["callerOptions"]; - timeoutMs?: ClientConfig["timeoutMs"]; - defaultHeaders?: ClientConfig["defaultHeaders"]; withMessages?: string; onError?: (error: unknown) => void; onFinish?: (state: ThreadState) => void; + onUpdateEvent?: (data: UpdatesStreamEvent["data"]) => void; + onCustomEvent?: (data: CustomStreamEvent["data"]) => void; + onMetadataEvent?: (data: MetadataStreamEvent["data"]) => void; + // TODO: can we make threadId uncontrollable / controllable? threadId?: string | null; onThreadId?: (threadId: string) => void; @@ -214,26 +215,14 @@ export function useStream< | FeedbackStreamEvent; const { assistantId, threadId, withMessages, onError, onFinish } = options; - const [client] = useState( - () => - new Client({ - apiUrl: options.apiUrl, - apiKey: options.apiKey, - callerOptions: options.callerOptions, - timeoutMs: options.timeoutMs, - defaultHeaders: options.defaultHeaders, - }), + const client = useMemo( + () => new Client({ apiUrl: options.apiUrl, apiKey: options.apiKey }), + [options.apiKey, options.apiUrl], ); - if (client == null) { - throw new Error( - "LangGraph SDK not provided. Either pass a client to `useStream` or wrap your app in a `LangGraphConfig` provider and pass the client there.", - ); - } - const [branchPath, setBranchPath] = useState([]); const [isLoading, setIsLoading] = useState(false); - const [events, setEvents] = useState([]); + const [_, setEvents] = useState([]); const [streamError, setStreamError] = useState(undefined); const [streamValues, setStreamValues] = useState(null); @@ -254,6 +243,16 @@ export function useStream< [], ); + const hasUpdateListener = options.onUpdateEvent != null; + const hasCustomListener = options.onCustomEvent != null; + + const callbackStreamMode = useMemo(() => { + const modes: Exclude[] = []; + if (hasUpdateListener) modes.push("updates"); + if (hasCustomListener) modes.push("custom"); + return modes; + }, [hasUpdateListener, hasCustomListener]); + const clearCallbackRef = useRef<() => void>(null!); clearCallbackRef.current = () => { setStreamError(undefined); @@ -480,6 +479,7 @@ export function useStream< const streamMode = unique([ ...(submitOptions?.streamMode ?? []), ...trackStreamModeRef.current, + ...callbackStreamMode, ]); const checkpoint = @@ -499,8 +499,6 @@ export function useStream< onCompletion: submitOptions?.onCompletion, onDisconnect: submitOptions?.onDisconnect ?? "cancel", - // TODO: check if integration on FE would work nice - feedbackKeys: submitOptions?.feedbackKeys, signal: abortRef.current.signal, checkpoint, @@ -539,6 +537,18 @@ export function useStream< break; } + if (event === "updates") { + options.onUpdateEvent?.(data); + } + + if (event === "custom") { + options.onCustomEvent?.(data); + } + + if (event === "metadata") { + options.onMetadataEvent?.(data); + } + if (event === "values") { setStreamValues(data); } @@ -604,30 +614,6 @@ export function useStream< const error = isLoading ? streamError : historyError; const values = streamValues ?? historyValues; - const stream = { - get custom() { - trackStreamMode("custom"); - - return events - .filter((item) => item.event === "custom") - .map(({ data }) => data as CustomType); - }, - - get events() { - trackStreamMode("events"); - return events; - }, - - get updates() { - trackStreamMode("updates"); - return events - .filter( - (item): item is UpdatesStreamEvent => - item.event === "updates", - ) - .map(({ data }) => data); - }, - }; const setBranch = useCallback( (path: string) => setBranchPath(path.split(">")), @@ -647,8 +633,6 @@ export function useStream< submit, setBranch, - stream, - get messages() { trackStreamMode("messages-tuple"); From 15fe44ddf8e779a0e6071b2919e3eca967542af2 Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Wed, 12 Feb 2025 09:13:10 -0800 Subject: [PATCH 15/16] Update deps --- libs/sdk-js/package.json | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/libs/sdk-js/package.json b/libs/sdk-js/package.json index dfa35d1ae..2555acbd3 100644 --- a/libs/sdk-js/package.json +++ b/libs/sdk-js/package.json @@ -22,11 +22,13 @@ }, "devDependencies": { "@jest/globals": "^29.7.0", + "@langchain/core": "^0.3.31", "@langchain/scripts": "^0.1.4", "@tsconfig/recommended": "^1.0.2", "@types/jest": "^29.5.12", "@types/node": "^20.12.12", "@types/uuid": "^9.0.1", + "@types/react": "18.3.2", "concat-md": "^0.5.1", "jest": "^29.7.0", "prettier": "^3.2.5", @@ -34,9 +36,7 @@ "typedoc": "^0.26.1", "typedoc-plugin-markdown": "^4.1.0", "typescript": "^5.4.5", - "react": "^18.3.1", - "@langchain/core": "^0.3.31", - "@types/react": "18.3.2" + "react": "^18.3.1" }, "peerDependencies": { "react": "^18 || ^19", From d333e52fce410c61a8c5502b95d030e8e2827fbf Mon Sep 17 00:00:00 2001 From: Tat Dat Duong Date: Wed, 12 Feb 2025 09:16:07 -0800 Subject: [PATCH 16/16] Simplify types for additional kwargs --- libs/sdk-js/src/types.messages.ts | 17 ++++------------- 1 file changed, 4 insertions(+), 13 deletions(-) diff --git a/libs/sdk-js/src/types.messages.ts b/libs/sdk-js/src/types.messages.ts index 79ed78473..a2eb34a9a 100644 --- a/libs/sdk-js/src/types.messages.ts +++ b/libs/sdk-js/src/types.messages.ts @@ -8,19 +8,10 @@ type MessageContentText = { type: "text"; text: string }; type MessageContentComplex = MessageContentText | MessageContentImageUrl; type MessageContent = string | MessageContentComplex[]; -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; -}; +/** + * Model-specific additional kwargs, which is passed back to the underlying LLM. + */ +type MessageAdditionalKwargs = Record; export type HumanMessage = { type: "human";