How to make c.stream() works with Ollama (...or a simple json stream)

[charnould]

Hi,
I'm currently trying to reproduce c.stream() example (from 3.7.0) using not OpenAI but Ollama.
My objective: a simple chat where a user asks a question, and AI responds.

Locally, when I cURL this:

curl http://localhost:11434/api/generate -d '{
"model": "llama2",   
"prompt":"Hello",
"stream":true
}'

The answer is:

{"model":"llama2","created_at":"2023-12-04T17:24:30.442092Z","response":"Hello","done":false}
{"model":"llama2","created_at":"2023-12-04T17:24:30.474753Z","response":" there","done":false}
{"model":"llama2","created_at":"2023-12-04T17:24:30.508147Z","response":"!","done":false}
{"model":"llama2","created_at":"2023-12-04T17:24:30.542424Z","response":" How","done":false}
{"model":"llama2","created_at":"2023-12-04T17:24:30.578529Z","response":" are","done":false}
{"model":"llama2","created_at":"2023-12-04T17:24:30.616623Z","response":" you","done":false}
{"model":"llama2","created_at":"2023-12-04T17:24:30.655301Z","response":" today","done":false}
{"model":"llama2","created_at":"2023-12-04T17:24:30.695035Z","response":"?","done":false}
{"model":"llama2","created_at":"2023-12-04T17:24:30.736839Z","response":"","done":true,"context":[518,25580,29962,3532,14816,29903,29958,5299,829,14816,29903,6778,13,13,10994,518,29914,25580,29962,13,10994,727,29991,1128,526,366,9826,29973],"total_duration":8638807917,"load_duration":8152700875,"prompt_eval_count":21,"prompt_eval_duration":216709000,"eval_count":8,"eval_duration":261011000}

So far, so good.

Then, I wrote a simple Hono app with this route heavily inspired from Hono itself:

app.get("/ai", async (c) => {

  const chatStream = await fetch("http://localhost:11434/api/generate", {
    body: JSON.stringify({
      model: "llama2",
      prompt: "hello",
      stream: true,
    }),
    headers: { "Content-Type": "application/json" },
    method: "POST",
  });

  console.log("BODY: ", chatStream.body);

  return c.streamText(async (stream) => {
    for await (const message of chatStream) {
      console.log("MESSAGE: ", message);
      await stream.write(message);
    }
  });
});

However, the code gives an error I can't understand and debug (I'm a newbie)

BODY:  ReadableStream {
  [Symbol(state)]: 4,
  [Symbol(reader)]: undefined,
  [Symbol(storedError)]: undefined,
  [Symbol(disturbed)]: false,
  [Symbol(readableStreamController)]: null,
  [Symbol(bunNativeType)]: 4,
  [Symbol(asyncContext)]: undefined,
  [Symbol(highWaterMark)]: undefined,
  [Symbol(underlyingSource)]: undefined,
  [Symbol(start)]: [Function],
  locked: [Getter],
  cancel: [Function: cancel],
  getReader: [Function: getReader],
  pipeTo: [Function: pipeTo],
  pipeThrough: [Function: pipeThrough],
  tee: [Function: tee],
  values: [Function: values],
  [Symbol(Symbol.asyncIterator)]: [Function: lazyAsyncIterator],
}
57 |   });
58 | 
59 |   console.log("BODY: ", chatStream.body);
60 | 
61 |   return c.streamText(async (stream) => {
62 |     for await (const message of chatStream) {
                          ^
TypeError: undefined is not a function (near '...message of chatStream...')

Do you have any clue to make this streamed response works?
Thanks a lot.

[charnould]

I've made some progress. Now I can print the AI answer in my webpage, but it's not streamed.
I think things are not in the right order. Any clue? Thanks.

app.get("/ai", async (c) => {
  const arr = [];

  await fetch("http://localhost:11434/api/generate", {
    body: JSON.stringify({
      prompt: "Why the sun is hot in one short sentence?",
      model: "llama2",
      stream: true,
    }),
    headers: { "Content-Type": "application/json" },
    method: "POST",
  })
    .then((response) => response.text())
    .then((data) => {
      const splited = data.split("\n");
      for (const s of splited) if (s !== "") arr.push(JSON.parse(s));     
    });

  return c.streamText(async (stream) => {
    for await (const item of arr) {
    //  console.log(item);
      await stream.write(item.response);
    }
  });
});

[charnould]

I' think I made it (it works with a client side lib), even if it doesn't seem elegant.

export const controller = async (c: Context) => {
  const prompt = c.req.query("prompt");

  const output = await fetch("http://127.0.0.1:11434/api/generate", {
    body: JSON.stringify({
      system: `Hello world`,
      model: "mistral",
      stream: true
    }),
    headers: { "Content-Type": "application/json" },
    method: "POST",
  });

  const reader = output.body?.getReader();

  return streamSSE(c, async (stream) => {
    while (true) {
      const { value, done } = await reader.read();
      if (done) break;
      const chunk = new TextDecoder().decode(value);
      const jsonchunk = JSON.parse(chunk);
      await stream.writeSSE({ data: jsonchunk.response });
    }
  });
};

[yusukebe]

@sor4chi Can you answer this?

[sor4chi]

Hi, @charnould
This code should works correctly.

Note that you have to get the reader of the readable stream and that there is a bug that causes ollama's json response to be split.

import { Hono } from "hono";

const app = new Hono();

app.get("/ai", async (c) => {
  const { body } = await fetch("http://localhost:11434/api/generate", {
    body: JSON.stringify({
      model: "llama2",
      prompt: "hello",
      stream: true,
    }),
    headers: { "Content-Type": "application/json" },
    method: "POST",
  });

  if (body == null) {
    return c.text("No body", 500);
  }

  const reader = body.getReader();

  return c.streamText(async (stream) => {
    let received = "";
    while (true) {
      const { done, value: message } = await reader.read();
      if (done) break;
      const decoded = new TextDecoder().decode(message);
      try {
        received += decoded;
        const json = JSON.parse(received);
        received = "";
        await stream.write(json.response);
      } catch (e) {
        // console.error(e);
      }
    }
  });
});

app.onError((err, c) => {
  console.error(err);
  return c.text("Internal Server Error", 500);
});

export default app;

2023-12-07.18.44.42.mov