How to stream responses with Cloudflare AI?

[osseonews]

I am trying to stream responses with the Cloudflare AI, using their starter example here: https://developers.cloudflare.com/workers-ai/tutorials/build-a-retrieval-augmented-generation-ai/

My basic code is below. The problem is that I can stream the message, but the browser shows the entire response like this:
data: {"response":" ","p":"abcdefghijklmnopqrstuvwxyz01234567"} data: {"response":" Based","p":"abcdefghijklmnopqrstuvwxyz0123456789abcde"}

Obviously, the user just needs to see the response. But, how do you stream that? Here is my code:

const ai = new Ai(c.env.AI);
const streamMessage = await ai.run(
    '@cf/meta/llama-2-7b-chat-int8',
    {
      messages: [
        ...(notes.length ? [{ role: 'system', content: contextMessage }] : []),
        { role: 'system', content: systemPrompt },
        { role: 'user', content: question}
      ],
      stream: true
    }
  )
return streamText(c, async (stream) => {
    // Write a text with a new line ('\n').
    for await (const message of streamMessage) {
     
      await stream.write(message ?? '') //this works, but returns everything, how do you just stream the response part??
     
    }
  })

[Levelleor]

Hi, have you figured it out by any chance? It looks like the response is actually in the SSE format, intentionally prepended with the 'data: '. I assume I can strip it from the string but I don't understand why I have to do that instead of actually getting a valid object directly from the model.

[abasu0713]

Did anyone find a way to handle this?

[abasu0713]

@Levelleor @osseonews A bit late, but I figured out a way to hack it..

Just respond with:

const responseStream = await ai.run('@cf/google/gemma-2b-it-lora', {
    stream: true,
    messages: messages }
});
return new Response(responseStream as ReadableStream, {
    headers: {
      "content-type": "text/event-stream",
    },
});

Now you can consume it on the client side (for me it's a Nuxt App) like this:

const url = "/api/test";
const source = new EventSource(url, null);
source.onmessage = (event) => {
      if (event.data === "[DONE]") {
        source.close();
        return;
      }
      const data = JSON.parse(event.data)
      // console.log(data.response);
      quote.value += data.response;
}

Hope this helps.

[oyal]

This doesn't seem to be a streaming transmission, but rather a one-time return of all data.

server

app.post('/chat', async (c) => {
  const { messages } = await c.req.json<{ messages: Message[] }>()
  const responseStream = await c.env.AI.run('@cf/meta/llama-2-7b-chat-int8', {
    messages,
    stream: true
  })

  return new Response(responseStream as ReadableStream, {
    headers: {
      "content-type": "text/event-stream"
    }
  })
})

client

<div id="chat"></div>
<input type="text" id="message" placeholder="Type your message here...">
<button id="send">Send</button>

<script>
  const chat = document.getElementById('chat');
  const messageInput = document.getElementById('message');
  const sendButton = document.getElementById('send');

  sendButton.addEventListener('click', async () => {
    const message = messageInput.value;
    if (!message) return;

    // Display the user's message
    const userMessage = document.createElement('div');
    userMessage.textContent = `User: ${message}`;
    chat.appendChild(userMessage);

    // Send the message to the server
    const response = await fetch('http://localhost:8787/chat', {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json'
      },
      body: JSON.stringify({ messages: [{ role: 'user', content: message }] })
    });

    // Read the response stream
    const reader = response.body.getReader();
    const decoder = new TextDecoder('utf-8');
    let result = '';
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      result += decoder.decode(value, { stream: true });
    }

    // Display the assistant's response
    const assistantMessage = document.createElement('div');
    assistantMessage.textContent = `Assistant: ${result}`;
    chat.appendChild(assistantMessage);

    // Clear the input
    messageInput.value = '';
  });
</script>

EventStream is empty.

[abasu0713]

You're right. It only works in development mode. Try this:

Return as sendStream:

export default defineEventHandler(async (event) => {
  const ai = hubAI(); // access AI bindings 
  const responseStream = await ai.run('@cf/google/gemma-2b-it-lora', {
    stream: true,
    messages: messages });
  return sendStream(event, responseStream as ReadableStream);
})

Then consume it like this.

try {
    const url = "/api/hardWorkAi";
    const response = await $fetch<ReadableStream>(url, {
      method: 'POST',
      body: {},
      responseType: 'stream',
    })
    const reader = (response as ReadableStream).pipeThrough(new TextDecoderStream()).getReader()
    let buffer = '';
    while (true) {
      const { done, value } = await reader.read()
      // console.log(value);
      if (done) {
        if (buffer.trim()) {
          console.warn('Stream ended with unparsed data:', buffer);
        }
        break
      }
      buffer += value;
      const lines = buffer.split('\n');
      buffer = lines.pop() || '';
      for (const line of lines) {
        isLoading.value = false;
        if(line.startsWith('data: ')) {
          const data = line.slice('data: '.length).trim();
          if (data === '[DONE]') break;

          try {
            const jsonData = JSON.parse(data);
            if (jsonData.response) {
              quote.value += jsonData.response;
            }
          } catch (parseError) {
            console.warn('Error parsing JSON:', parseError);
          }
        }
      }
    }
  } catch (err) {
    console.error(`Error: ${err}`);
    error.value = `An error occurred.`;
  } finally {
    isLoading.value = false;
  }

Tested it in prod after deployments. Streaming is working..