Function usePipe

packages/core/src/react/use-pipe.ts:29–308

({
	apiRoute = '/langbase/pipes/run-stream',
	onResponse,
	onFinish,
	onConnect,
	onError,
	threadId: initialThreadId,
	initialMessages = [],
	stream = true,
}: UsePipeOptions = {})

const externalThreadIdSchema = uuidSchema.optional();

export function usePipe({
	apiRoute = '/langbase/pipes/run-stream',
	onResponse,
	onFinish,
	onConnect,
	onError,
	threadId: initialThreadId,
	initialMessages = [],
	stream = true,
}: UsePipeOptions = {}) {
	const [messages, setMessages] = useState<Message[]>(initialMessages);
	const [input, setInput] = useState('');
	const [isLoading, setIsLoading] = useState(false);
	const [error, setError] = useState<Error | null>(null);

	const abortControllerRef = useRef<AbortController | null>(null);
	const threadIdRef = useRef<string | undefined>(
		initialThreadId || undefined,
	);
	const messagesRef = useRef<Message[]>(initialMessages);
	const isFirstRequestRef = useRef<boolean>(true);

	const updateMessages = useCallback((newMessages: Message[]) => {
		messagesRef.current = newMessages;
		setMessages(newMessages);
	}, []);

	const processStreamResponse = useCallback(
		async (runner: Runner) => {
			let assistantMessage: Message = {role: 'assistant', content: ''};
			updateMessages([...messagesRef.current, assistantMessage]);

			for await (const chunk of runner) {
				if (abortControllerRef.current?.signal.aborted) break;

				const content = chunk.choices[0]?.delta?.content || '';
				assistantMessage.content += content;

				updateMessages([
					...messagesRef.current.slice(0, -1),
					{...assistantMessage},
				]);
				onResponse?.({...assistantMessage});
			}

			onFinish?.(messagesRef.current);
		},
		[updateMessages, onResponse, onFinish],
	);

	const processNonStreamResponse = useCallback(
		(result: RunResponse) => {
			const assistantMessage: Message = {
				role: 'assistant',
				content: result.completion,
			};
			const newMessages = [...messagesRef.current, assistantMessage];
			updateMessages(newMessages);