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:

``shell
npm install mux-web-streams # or pnpm or yarn
`

Usage

To use mux-web-streams, import the desired functions from the library:

`typescript
import { demuxer, muxer } from 'mux-web-streams';
`

$3

The muxer function is used to multiplex an array of ReadableStreams into a single stream.

`typescript
import { muxer } from 'mux-web-streams';

// Multiplex readable streams into a single stream
const multiplexedReadableStream: ReadableStream = muxer([
readableStream0,
readableStream1,
readableStream2,
readableStream3,
readableStream4,
]);
`

$3

The demuxer function is used to demultiplex a multiplexed stream back into the original array of ReadableStreams.

`typescript
import { demuxer } from 'mux-web-streams';

// Demultiplex the stream and listen for emitted streams
const [
readableStream0,
readableStream1,
readableStream2,
readableStream3,
readableStream4,
] = demuxer(multiplexedReadableStream, 5);

// Use your streams!
readableStream0.pipeTo(/ ... /);
`

API

$3

Multiplexes an array of ReadableStreams into a single stream.

- streams: An array of ReadableStreams to be multiplexed.

$3

Demultiplexes a single multiplexed ReadableStream into an array of ReadableStreams.

- stream: The multiplexed stream from muxer().
- numberOfStreams: The number of streams passed into muxer().

$3

The ReadableStreams 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

$3

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

`typescript
// app/api/chat/route.ts
import { muxer } from 'mux-web-streams';

export async function POST(request: Request): Promise {
// Get several ReadableStreams
const chatResponseStream: ReadableStream = createReadableStream();
const chainValuesStream: ReadableStream = createReadableStream();

// Multiplex the streams into a single stream
const multiplexedStream = muxer([chatResponseStream, chainValuesStream]);

// Stream the multiplexed data
return new Response(multiplexedStream, {
headers: {
'Content-Type': 'text/event-stream',
Connection: 'keep-alive',
'Cache-Control': 'no-cache, no-transform',
},
});
}
`

#### Client-side demuxer

`tsx
// components/chat.tsx

import type { ChainValues } from 'langchain/schema';
import { demuxer } from 'mux-web-streams';

export const Chat = () => {
const [chatResponse, setChatResponse] = useState('');
const [chainValues, setChainValues] = useState>({});

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');
}

// Demultiplex the multiplexed stream
const [chatResponseStream, chainValuesStream] = demuxer(res.body, 2);

// Write generation to the UI as it comes
chatResponseStream.pipeThrough(new TextDecoderStream()).pipeTo(
new WritableStream({
write(chunk) {
// Update text of the most recently added element (the AI message)
setChatResponse(chatResponse + chunk);
},
}),
);

// Render the chain values when they come
chainValuesStream.pipeTo(
new WritableStream({
write(chunk) {
setChainValues(JSON.parse(chunk));
},
}),
);
};

return (



);
};
``

License

MIT