How would I use Cloudflare D1-Backed Chat Memory AND stream the response back to the user?

[StavrosHacker]

Checked other resources

• I added a very descriptive title to this question.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.

Commit to Help

• I commit to help with one of those options 👆

Example Code

const parser = new HttpResponseOutputParser({ contentType: "text/event-stream" });

    try {
      const stream = await chain.pipe(parser).stream(chainInput);

      await memory.saveContext(chainInput, {
        output: stream,
      });

      return new Response(stream, {
        headers: { "Content-Type": "text/event-stream" },
      });
    } catch (err) {
      console.error("Error during chain execution:", err);
    }

Description

I'm trying to use Cloudflare D1-Backed Chat Memory so that I can have persistent storage for each chat session, however I also want to stream the response back to the user using an event stream so that the user can receive the response faster. The problem is i receive back a ReadableStream which is not text so the memory.saveContext function does not accept it. If i read the ReadableStream so that I can get the text and save it, I'm not able to return the stream back to the user as the ReadableStream can only be read once. Is there a better way of approaching this?

System Info

langchain@0.3.6
@langchain/cloudflare@0.1.0
@langchain/core@0.3.19
windows 11
Node.js v22.11.0

[dosubot[bot]]

Hey @StavrosHacker! 👋 I'm here to help you with your code and answer any questions you have. Let's dive into this together!

To handle both storing the response in Cloudflare D1-Backed Chat Memory and streaming it back to the user, you can use a technique to duplicate the stream. Since a ReadableStream can only be read once, you need to create a mechanism to read the stream, store its content, and then re-stream it to the user. Here's a possible approach:

1. Convert the Stream to Text: Read the stream into a string or buffer so you can store it in memory.

2. Clone the Stream: Use the stored text to create a new stream that can be sent back to the user.

Here's an example of how you might implement this:

async function streamToString(stream: ReadableStream): Promise<string> {
  const reader = stream.getReader();
  const decoder = new TextDecoder();
  let result = '';
  let done = false;

  while (!done) {
    const { value, done: doneReading } = await reader.read();
    done = doneReading;
    if (value) {
      result += decoder.decode(value, { stream: true });
    }
  }

  return result;
}

async function handleRequest(chainInput) {
  const parser = new HttpResponseOutputParser({ contentType: "text/event-stream" });

  try {
    const stream = await chain.pipe(parser).stream(chainInput);

    // Convert the stream to a string
    const responseText = await streamToString(stream);

    // Save the context
    await memory.saveContext(chainInput, { output: responseText });

    // Create a new stream from the response text
    const newStream = new ReadableStream({
      start(controller) {
        controller.enqueue(new TextEncoder().encode(responseText));
        controller.close();
      }
    });

    return new Response(newStream, {
      headers: { "Content-Type": "text/event-stream" },
    });
  } catch (err) {
    console.error("Error during chain execution:", err);
  }
}

This code reads the original stream into a string, saves it to memory, and then creates a new stream from the stored string to send back to the user. This way, you can both store the response and stream it back to the user efficiently.

To continue talking to Dosu, mention @dosu.

[StavrosHacker]

I ended up using a TransformStream to get each chunk and concatenate it together while the stream is being returned to the user, and once the stream is completed, it the saves the full string to D1. Fairly simple, for more info look at this: https://developers.cloudflare.com/workers/runtime-apis/streams/