diff --git a/docs/docs/cloud/how-tos/use_stream_react.md b/docs/docs/cloud/how-tos/use_stream_react.md index 9329d5c79..9d9ec2413 100644 --- a/docs/docs/cloud/how-tos/use_stream_react.md +++ b/docs/docs/cloud/how-tos/use_stream_react.md @@ -1,8 +1,8 @@ -# How to integrate LangGraph into your React application +How to integrate LangGraph into your React application# How to integrate LangGraph into your React application -!!! info "Prerequisites" +!!! info "Prerequisites" - - [LangGraph Platform](../../concepts/langgraph_platform.md) + - [LangGraph Platform](../../concepts/langgraph_platform.md) - [LangGraph Server](../../concepts/langgraph_server.md) The `useStream()` React hook provides a seamless way to integrate LangGraph into your React applications. It handles all the complexities of streaming, state management, and branching logic, letting you focus on building great chat experiences. @@ -113,6 +113,115 @@ export default function App() { } ``` +### Resume a stream after page refresh + +The `useStream()` hook can automatically resume an already ongoing run on mount by passing `reconnectOnMount: true`. This is useful for resuming an ongoing stream after a full page refresh without losing any messages that were generated in the meantime. + +```tsx +const thread = useStream<{ messages: Message[] }>({ + apiUrl: "http://localhost:2024", + assistantId: "agent", + reconnectOnMount: true, +}); +``` + +By default the ID of the created run is stored in `window.sessionStorage` under `lg:stream:${threadId}` key. This can be swapped by passing a custom storage in `reconnectOnMount`. + +```tsx +const thread = useStream<{ messages: Message[] }>({ + apiUrl: "http://localhost:2024", + assistantId: "agent", + reconnectOnMount: () => window.localStorage, +}); +``` + +Finally, you can manually handle the lifecycle of stream resuming by using the run callbacks and `joinStream` method. Make sure to pass `streamResumable: true` and `onDisconnect: "continue"` when creating the run. + +````tsx +import type { Message } from "@langchain/langgraph-sdk"; +import { useStream } from "@langchain/langgraph-sdk/react"; +import { useCallback, useState, useEffect, useRef } from "react"; + +export default function App() { + const [threadId, onThreadId] = useSearchParam("threadId"); + + const thread = useStream<{ messages: Message[] }>({ + apiUrl: "http://localhost:2024", + assistantId: "agent", + + threadId, + onThreadId, + + onCreated: (run) => { + window.sessionStorage.setItem(`resume:${run.thread_id}`, run.run_id); + }, + onFinish: (_, run) => { + window.sessionStorage.removeItem(`resume:${run?.thread_id}`); + }, + }); + + // Ensure that we only join the stream once per thread. + const joinedThreadId = useRef(null); + useEffect(() => { + if (!threadId) return; + + const resume = window.sessionStorage.getItem(`resume:${threadId}`); + if (resume && joinedThreadId.current !== threadId) { + thread.joinStream(resume); + joinedThreadId.current = threadId; + } + }, [threadId]); + + return ( +
{ + e.preventDefault(); + const form = e.target as HTMLFormElement; + const message = new FormData(form).get("message") as string; + thread.submit( + { messages: [{ type: "human", content: message }] }, + { streamResumable: true } + ); + }} + > +
+ {thread.messages.map((message) => ( +
{JSON.stringify(message.content)}
+ ))} +
+ + +
+ ); +} + +// Utility method to retrieve and persist data in URL as search param +function useSearchParam(key: string) { + const [value, setValue] = useState(() => { + const params = new URLSearchParams(window.location.search); + return params.get(key) ?? null; + }); + + const update = useCallback( + (value: string | null) => { + setValue(value); + + const url = new URL(window.location.href); + if (value == null) { + url.searchParams.delete(key); + } else { + url.searchParams.set(key, value); + } + + window.history.pushState({}, "", url.toString()); + }, + [key] + ); + + return [value, update] as const; +} +``` + ### Thread Management Keep track of conversations with built-in thread management. You can access the current thread ID and get notified when new threads are created: @@ -127,7 +236,7 @@ const thread = useStream<{ messages: Message[] }>({ threadId: threadId, onThreadId: setThreadId, }); -``` +```` We recommend storing the `threadId` in your URL's query parameters to let users resume conversations after page refreshes. diff --git a/libs/sdk-js/package.json b/libs/sdk-js/package.json index 795131974..90324e17d 100644 --- a/libs/sdk-js/package.json +++ b/libs/sdk-js/package.json @@ -1,6 +1,6 @@ { "name": "@langchain/langgraph-sdk", - "version": "0.0.78", + "version": "0.0.79", "description": "Client library for interacting with the LangGraph API", "type": "module", "packageManager": "yarn@1.22.19", diff --git a/libs/sdk-js/src/react/stream.tsx b/libs/sdk-js/src/react/stream.tsx index ba65c50c9..6db58a587 100644 --- a/libs/sdk-js/src/react/stream.tsx +++ b/libs/sdk-js/src/react/stream.tsx @@ -405,6 +405,11 @@ type GetCustomEventType = Bag extends { ? Bag["CustomEventType"] : unknown; +interface RunCallbackMeta { + run_id: string; + thread_id: string; +} + export interface UseStreamOptions< StateType extends Record = Record, Bag extends BagTemplate = BagTemplate, @@ -450,17 +455,20 @@ export interface UseStreamOptions< /** * Callback that is called when an error occurs. */ - onError?: (error: unknown) => void; + onError?: (error: unknown, run: RunCallbackMeta | undefined) => void; /** * Callback that is called when the stream is finished. */ - onFinish?: (state: ThreadState) => void; + onFinish?: ( + state: ThreadState, + run: RunCallbackMeta | undefined, + ) => void; /** * Callback that is called when a new stream is created. */ - onCreated?: (run: { run_id: string; thread_id: string }) => void; + onCreated?: (run: RunCallbackMeta) => void; /** * Callback that is called when an update event is received. @@ -846,8 +854,12 @@ export function useStream< action: (signal: AbortSignal) => Promise<{ onSuccess: () => Promise[]>; stream: AsyncGenerator; + getCallbackMeta: () => { thread_id: string; run_id: string } | undefined; }>, ) { + let getCallbackMeta: + | (() => { thread_id: string; run_id: string } | undefined) + | undefined; try { setIsLoading(true); setStreamError(undefined); @@ -856,6 +868,7 @@ export function useStream< abortRef.current = new AbortController(); const run = await action(abortRef.current.signal); + getCallbackMeta = run.getCallbackMeta; let streamError: StreamError | undefined; for await (const { event, data } of run.stream) { @@ -916,7 +929,7 @@ export function useStream< if (streamError != null) throw streamError; const lastHead = result.at(0); - if (lastHead) onFinish?.(lastHead); + if (lastHead) onFinish?.(lastHead, getCallbackMeta?.()); } catch (error) { if ( !( @@ -926,7 +939,7 @@ export function useStream< ) { console.error(error); setStreamError(error); - onError?.(error); + onError?.(error, getCallbackMeta?.()); } } finally { setIsLoading(false); @@ -953,6 +966,7 @@ export function useStream< return history.mutate(threadId); }, stream, + getCallbackMeta: () => ({ thread_id: threadId, run_id: runId }), }; }); }; @@ -1004,6 +1018,9 @@ export function useStream< // @ts-expect-error if (checkpoint != null) delete checkpoint.thread_id; let rejoinKey: `lg:stream:${string}` | undefined; + let callbackMeta: RunCallbackMeta | undefined; + const streamResumable = + submitOptions?.streamResumable ?? !!runMetadataStorage; const stream = client.runs.stream(usableThreadId, assistantId, { input: values as Record, @@ -1017,29 +1034,31 @@ export function useStream< onCompletion: submitOptions?.onCompletion, onDisconnect: submitOptions?.onDisconnect ?? - (runMetadataStorage ? "continue" : "cancel"), + (streamResumable ? "continue" : "cancel"), signal, checkpoint, streamMode, streamSubgraphs: submitOptions?.streamSubgraphs, - streamResumable: submitOptions?.streamResumable ?? !!runMetadataStorage, + streamResumable, onRunCreated(params) { - const runParams = { + callbackMeta = { run_id: params.run_id, thread_id: params.thread_id ?? usableThreadId, }; + if (runMetadataStorage) { - rejoinKey = `lg:stream:${runParams.thread_id}`; - runMetadataStorage.setItem(rejoinKey, runParams.run_id); + rejoinKey = `lg:stream:${callbackMeta.thread_id}`; + runMetadataStorage.setItem(rejoinKey, callbackMeta.run_id); } - onCreated?.(runParams); + onCreated?.(callbackMeta); }, }) as AsyncGenerator; return { stream, + getCallbackMeta: () => callbackMeta, onSuccess: () => { if (rejoinKey) runMetadataStorage?.removeItem(rejoinKey); return history.mutate(usableThreadId);