I want to stream an LLM (ollama) response using fastapi and react. I can successfully get an answer from the LLM without streaming, but when I try to stream it, I get an error in react. The LLM answer streams successfully when I print each chunk in fastapi. I’m using @microsoft/fetch-event-source to stream the response.

I’ve create a github repo to help you help me.

Here’s the fastapi and react code in question:

react

Something is happening to cause console.log("another error onopen") to trigger.

import { useState } from "react";
import {
  fetchEventSource,
  EventStreamContentType,
} from "@microsoft/fetch-event-source";

class FatalError extends Error {}

const StreamResponse = ({ input }: { input: string }) => {
  const [answer, setAnswer] = useState("");

  const handleClick = () => {
    fetchEventSource("http://localhost:8000/stream", {
      body: JSON.stringify({ question: input }),
      method: "POST",
      headers: {
        "Content-type": "application/json",
        Accept: "text/event-stream",
      },

      async onopen(response) {
        console.log("onopen");
        console.log(response);

        if (
          response.ok &&
          response.headers.get("content-type") === EventStreamContentType
        ) {
          // setAnswer(response.answer)
          return; // everything's good
        } else if (
          response.status >= 400 &&
          response.status < 500 &&
          response.status !== 429
        ) {
          console.log("fatal error onopen");
          // client-side errors are usually non-retriable:
          // throw new FatalError();
        } else {
          console.log("another error onopen");
          // throw new RetriableError();
        }
      },
      onmessage(msg) {
        // if the server emits an error message, throw an exception
        // so it gets handled by the onerror callback below:
        console.log("onmessage");
        console.log(msg);
        if (msg.event === "FatalError") {
          throw new FatalError(msg.data);
        }
      },
      onclose() {
        console.log("onclose");
        // if the server closes the connection unexpectedly, retry:
        // throw new RetriableError();
      },
      onerror(err) {
        console.log("onerror");
        console.log(err);
        if (err instanceof FatalError) {
          throw err; // rethrow to stop the operation
        } else {
          // do nothing to automatically retry. You can also
          // return a specific retry interval here.
        }
      },
    });
  };

  return (
    <div style={{ width: "100%" }}>
      <div style={{ display: "flex", gap: 10, alignItems: "center" }}>
        <button onClick={handleClick} style={{ height: 24 }}>
          Submit question with streaming
        </button>
      </div>
      <p>Response</p>
      <div style={{ border: 1, borderStyle: "solid", height: 100 }}>
        {answer}
      </div>
    </div>
  );
};

export default StreamResponse;

fastapi

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

from langchain_community.llms import Ollama
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser


app = FastAPI()

origins = ['*']

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=False,
    allow_methods=['*'],
    allow_headers=['*']
)

ollama = Ollama(
    base_url="http://localhost:11434",
    model="llama3"
)

system_prompt = (
    "You are an assistant for question-answering tasks. "
    "Use the following pieces of retrieved context to answer "
    "the question. If you don't know the answer, say that you "
    "don't know. Use three sentences maximum and keep the "
    "answer concise."
)

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system_prompt),
        ("human", "{input}"),
    ]
)

chain = (prompt | ollama | StrOutputParser())


class Question(BaseModel):
    question: str


@app.get('/test')
def read_test():
    return {'hello': 'world'}


@app.post('/nostream')
def no_stream_llm(question: Question):
    answer = chain.invoke({'input': question.question})
    print(answer)
    return {'answer': answer}


def stream_answer(question):
    for chunk in chain.stream(question):
        print(chunk, end='', flush=True)
        yield chunk


@app.post('/stream')
def stream_response_from_llm(question: Question):
    return StreamingResponse(stream_answer(question=question.question), media_type="text/event-stream")