Mux (and Demux) Web Streams
mux-web-streams
enables you to multiplex and demultiplex (AKA "mux" and "demux") streams. Stream multiplexing combines multiple streams into a single stream, so that they can be sent over one communication channel, such as in a single HTTP response. Stream demultiplexing is the opposite operation – it takes a single stream and splits it into multiple streams.
mux-web-streams
uses WHATWG-standard Web Streams, which work across Browsers, Node, Bun, Deno.
At Transcend, we use mux-web-streams
to stream LLM responses when using langchain on Lamdba functions with Vercel. This allows us to stream responses as they're generated, while also passing other metadata to the client, such as ChainValues.
Installation
You can install mux-web-streams
using npm:
npm install mux-web-streams # or `pnpm` or `yarn`
Usage
To use mux-web-streams
, import the desired functions from the library:
import { demuxer, muxer } from 'mux-web-streams';
Multiplexing streams
The muxer
function is used to multiplex an array of ReadableStream
s into a single stream.
import { muxer } from 'mux-web-streams';
const multiplexedReadableStream: ReadableStream<Uint8Array> = muxer([
readableStream0,
readableStream1,
readableStream2,
readableStream3,
readableStream4,
]);
Demultiplexing streams
The demuxer
function is used to demultiplex a multiplexed stream back into the original array of ReadableStream
s.
import { demuxer } from 'mux-web-streams';
const [
readableStream0,
readableStream1,
readableStream2,
readableStream3,
readableStream4,
] = demuxer(multiplexedReadableStream, 5);
readableStream0.pipeTo();
API
muxer(streams: ReadableStream<SerializableData>[]): ReadableStream<Uint8Array>
Multiplexes an array of ReadableStream
s into a single stream.
streams
: An array of ReadableStream
s to be multiplexed.
demuxer(stream: ReadableStream, numberOfStreams: number): ReadableStream<SerializableData>[]
Demultiplexes a single multiplexed ReadableStream
into an array of ReadableStream
s.
stream
: The multiplexed stream from muxer()
.numberOfStreams
: The number of streams passed into muxer()
.
Type SerializableData
The ReadableStream
s passed into muxer()
must emit SerializableData
. This can be a Uint8Array
, or anything that's deserializable from JSON, such as string
, number
, boolean
, null
, or objects and arrays composed of those primitive types.
Examples
Streaming from a Vercel function
Here's an example using Langchain on Vercel functions. We want to render the AI chat completion as it comes, while also passing the chain values to the client, allowing the end-user to review the source documents behind the chatbot's answer. The chain values return the source documents that were provided to the LLM chat model using a RAG architecture. A more elaborate example might include error messages and other data.
Server-side muxer
import { muxer } from 'mux-web-streams';
export async function POST(request: Request): Promise<Response> {
const chatResponseStream: ReadableStream<string> = createReadableStream();
const chainValuesStream: ReadableStream<ChainValues> = createReadableStream();
const multiplexedStream = muxer([chatResponseStream, chainValuesStream]);
return new Response(multiplexedStream, {
headers: {
'Content-Type': 'text/event-stream',
Connection: 'keep-alive',
'Cache-Control': 'no-cache, no-transform',
},
});
}
Client-side demuxer
import type { ChainValues } from 'langchain/schema';
import { demuxer } from 'mux-web-streams';
export const Chat = () => {
const [chatResponse, setChatResponse] = useState<string>('');
const [chainValues, setChainValues] = useState<Record<string, any>>({});
const onClick = async () => {
const res = await fetch('/api/chat', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({ myInput: 0 }),
});
if (!res.body) {
throw new Error('No body');
}
const [chatResponseStream, chainValuesStream] = demuxer(res.body, 2);
chatResponseStream.pipeThrough(new TextDecoderStream()).pipeTo(
new WritableStream({
write(chunk) {
setChatResponse(chatResponse + chunk);
},
}),
);
chainValuesStream.pipeTo(
new WritableStream({
write(chunk) {
setChainValues(JSON.parse(chunk));
},
}),
);
};
return (
<div>
<button onClick={onClick}>Get multiplexed stream</button>
<p>Demuxed results:</p>
<ul>
<li>Chat response: {chatResponse}</li>
<li>
Chain values:{' '}
<pre>
<code>{JSON.stringify(chainValues, null, 2)}</code>
</pre>
</li>
</ul>
</div>
);
};
License
MIT